From bde7232ece49ab06df896e2de6d5de07f000f92d Mon Sep 17 00:00:00 2001
From: "Claude (Opus 4.6)" <claude@hermes.local>
Date: Tue, 24 Mar 2026 01:54:44 +0000
Subject: [PATCH] [claude] Add unit tests for kimi_delegation.py (#1295)
 (#1303)

---
 tests/timmy/test_kimi_delegation.py | 643 ++++++++++++++++++++++++++++
 1 file changed, 643 insertions(+)
 create mode 100644 tests/timmy/test_kimi_delegation.py

diff --git a/tests/timmy/test_kimi_delegation.py b/tests/timmy/test_kimi_delegation.py
new file mode 100644
index 00000000..998dfd3a
--- /dev/null
+++ b/tests/timmy/test_kimi_delegation.py
@@ -0,0 +1,643 @@
+"""Unit tests for timmy.kimi_delegation — Kimi research delegation pipeline."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# exceeds_local_capacity
+# ---------------------------------------------------------------------------
+
+
+class TestExceedsLocalCapacity:
+    def test_heavy_keyword_triggers_delegation(self):
+        from timmy.kimi_delegation import exceeds_local_capacity
+
+        assert exceeds_local_capacity("Do a comprehensive review of the codebase") is True
+
+    def test_all_heavy_keywords_detected(self):
+        from timmy.kimi_delegation import _HEAVY_RESEARCH_KEYWORDS, exceeds_local_capacity
+
+        for kw in _HEAVY_RESEARCH_KEYWORDS:
+            assert exceeds_local_capacity(f"Please {kw} the topic") is True, f"Missed keyword: {kw}"
+
+    def test_long_task_triggers_delegation(self):
+        from timmy.kimi_delegation import _HEAVY_WORD_THRESHOLD, exceeds_local_capacity
+
+        long_task = " ".join(["word"] * (_HEAVY_WORD_THRESHOLD + 1))
+        assert exceeds_local_capacity(long_task) is True
+
+    def test_short_simple_task_returns_false(self):
+        from timmy.kimi_delegation import exceeds_local_capacity
+
+        assert exceeds_local_capacity("Fix the typo in README") is False
+
+    def test_exactly_at_word_threshold_triggers(self):
+        from timmy.kimi_delegation import _HEAVY_WORD_THRESHOLD, exceeds_local_capacity
+
+        task = " ".join(["word"] * _HEAVY_WORD_THRESHOLD)
+        assert exceeds_local_capacity(task) is True
+
+    def test_keyword_case_insensitive(self):
+        from timmy.kimi_delegation import exceeds_local_capacity
+
+        assert exceeds_local_capacity("Run a COMPREHENSIVE analysis") is True
+
+    def test_empty_string_returns_false(self):
+        from timmy.kimi_delegation import exceeds_local_capacity
+
+        assert exceeds_local_capacity("") is False
+
+
+# ---------------------------------------------------------------------------
+# _slugify
+# ---------------------------------------------------------------------------
+
+
+class TestSlugify:
+    def test_basic_text(self):
+        from timmy.kimi_delegation import _slugify
+
+        assert _slugify("Hello World") == "hello-world"
+
+    def test_special_characters_removed(self):
+        from timmy.kimi_delegation import _slugify
+
+        assert _slugify("Research: AI & ML!") == "research-ai--ml"
+
+    def test_underscores_become_dashes(self):
+        from timmy.kimi_delegation import _slugify
+
+        assert _slugify("some_snake_case") == "some-snake-case"
+
+    def test_long_text_truncated_to_60(self):
+        from timmy.kimi_delegation import _slugify
+
+        long_text = "a" * 100
+        result = _slugify(long_text)
+        assert len(result) <= 60
+
+    def test_leading_trailing_dashes_stripped(self):
+        from timmy.kimi_delegation import _slugify
+
+        result = _slugify("  hello  ")
+        assert not result.startswith("-")
+        assert not result.endswith("-")
+
+    def test_multiple_spaces_become_single_dash(self):
+        from timmy.kimi_delegation import _slugify
+
+        assert _slugify("one   two") == "one-two"
+
+
+# ---------------------------------------------------------------------------
+# _build_research_template
+# ---------------------------------------------------------------------------
+
+
+class TestBuildResearchTemplate:
+    def test_contains_task_title(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("My Task", "background", "the question?")
+        assert "My Task" in body
+
+    def test_contains_question(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("task", "context", "What is X?")
+        assert "What is X?" in body
+
+    def test_contains_context(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("task", "some context here", "q?")
+        assert "some context here" in body
+
+    def test_default_priority_normal(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("task", "ctx", "q?")
+        assert "normal" in body
+
+    def test_custom_priority_included(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("task", "ctx", "q?", priority="high")
+        assert "high" in body
+
+    def test_kimi_label_mentioned(self):
+        from timmy.kimi_delegation import KIMI_READY_LABEL, _build_research_template
+
+        body = _build_research_template("task", "ctx", "q?")
+        assert KIMI_READY_LABEL in body
+
+    def test_slugified_task_in_artifact_path(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("My Research Task", "ctx", "q?")
+        assert "my-research-task" in body
+
+    def test_sections_present(self):
+        from timmy.kimi_delegation import _build_research_template
+
+        body = _build_research_template("task", "ctx", "q?")
+        assert "## Research Request" in body
+        assert "### Research Question" in body
+        assert "### Background / Context" in body
+        assert "### Deliverables" in body
+
+
+# ---------------------------------------------------------------------------
+# _extract_action_items
+# ---------------------------------------------------------------------------
+
+
+class TestExtractActionItems:
+    def test_checkbox_items_extracted(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "- [ ] Fix the bug\n- [ ] Write tests\n"
+        items = _extract_action_items(text)
+        assert "Fix the bug" in items
+        assert "Write tests" in items
+
+    def test_numbered_list_extracted(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "1. Deploy to staging\n2. Run smoke tests\n"
+        items = _extract_action_items(text)
+        assert "Deploy to staging" in items
+        assert "Run smoke tests" in items
+
+    def test_action_prefix_extracted(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "Action: Update the config file\n"
+        items = _extract_action_items(text)
+        assert "Update the config file" in items
+
+    def test_todo_prefix_extracted(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "TODO: Add error handling\n"
+        items = _extract_action_items(text)
+        assert "Add error handling" in items
+
+    def test_next_step_prefix_extracted(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "Next step: Validate results\n"
+        items = _extract_action_items(text)
+        assert "Validate results" in items
+
+    def test_case_insensitive_prefixes(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "todo: lowercase todo\nACTION: uppercase action\n"
+        items = _extract_action_items(text)
+        assert "lowercase todo" in items
+        assert "uppercase action" in items
+
+    def test_deduplication(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "1. Do the thing\n2. Do the thing\n"
+        items = _extract_action_items(text)
+        assert items.count("Do the thing") == 1
+
+    def test_empty_text_returns_empty_list(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        assert _extract_action_items("") == []
+
+    def test_no_action_items_returns_empty_list(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "This is just plain prose with no action items here."
+        assert _extract_action_items(text) == []
+
+    def test_mixed_sources_combined(self):
+        from timmy.kimi_delegation import _extract_action_items
+
+        text = "- [ ] checkbox item\n1. numbered item\nAction: action item\n"
+        items = _extract_action_items(text)
+        assert len(items) == 3
+
+
+# ---------------------------------------------------------------------------
+# _get_or_create_label (async)
+# ---------------------------------------------------------------------------
+
+
+class TestGetOrCreateLabel:
+    @pytest.mark.asyncio
+    async def test_returns_existing_label_id(self):
+        from timmy.kimi_delegation import KIMI_READY_LABEL, _get_or_create_label
+
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = [{"name": KIMI_READY_LABEL, "id": 42}]
+
+        client = MagicMock()
+        client.get = AsyncMock(return_value=mock_resp)
+
+        result = await _get_or_create_label(client, "http://git", {"Authorization": "token x"}, "owner/repo")
+        assert result == 42
+
+    @pytest.mark.asyncio
+    async def test_creates_label_when_missing(self):
+        from timmy.kimi_delegation import _get_or_create_label
+
+        list_resp = MagicMock()
+        list_resp.status_code = 200
+        list_resp.json.return_value = []  # no existing labels
+
+        create_resp = MagicMock()
+        create_resp.status_code = 201
+        create_resp.json.return_value = {"id": 99}
+
+        client = MagicMock()
+        client.get = AsyncMock(return_value=list_resp)
+        client.post = AsyncMock(return_value=create_resp)
+
+        result = await _get_or_create_label(client, "http://git", {"Authorization": "token x"}, "owner/repo")
+        assert result == 99
+
+    @pytest.mark.asyncio
+    async def test_returns_none_on_list_exception(self):
+        from timmy.kimi_delegation import _get_or_create_label
+
+        client = MagicMock()
+        client.get = AsyncMock(side_effect=Exception("network error"))
+
+        result = await _get_or_create_label(client, "http://git", {}, "owner/repo")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_on_create_exception(self):
+        from timmy.kimi_delegation import _get_or_create_label
+
+        list_resp = MagicMock()
+        list_resp.status_code = 200
+        list_resp.json.return_value = []
+
+        client = MagicMock()
+        client.get = AsyncMock(return_value=list_resp)
+        client.post = AsyncMock(side_effect=Exception("create failed"))
+
+        result = await _get_or_create_label(client, "http://git", {}, "owner/repo")
+        assert result is None
+
+
+# ---------------------------------------------------------------------------
+# create_kimi_research_issue (async)
+# ---------------------------------------------------------------------------
+
+
+class TestCreateKimiResearchIssue:
+    @pytest.mark.asyncio
+    async def test_returns_error_when_gitea_disabled(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        with patch("timmy.kimi_delegation.settings") as mock_settings:
+            mock_settings.gitea_enabled = False
+            mock_settings.gitea_token = ""
+            result = await create_kimi_research_issue("task", "ctx", "q?")
+
+        assert result["success"] is False
+        assert "not configured" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_returns_error_when_no_token(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        with patch("timmy.kimi_delegation.settings") as mock_settings:
+            mock_settings.gitea_enabled = True
+            mock_settings.gitea_token = ""
+            result = await create_kimi_research_issue("task", "ctx", "q?")
+
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_successful_issue_creation(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://git"
+        mock_settings.gitea_repo = "owner/repo"
+
+        label_resp = MagicMock()
+        label_resp.status_code = 200
+        label_resp.json.return_value = [{"name": "kimi-ready", "id": 5}]
+
+        issue_resp = MagicMock()
+        issue_resp.status_code = 201
+        issue_resp.json.return_value = {"number": 42, "html_url": "http://git/issues/42"}
+
+        async_client = AsyncMock()
+        async_client.get = AsyncMock(return_value=label_resp)
+        async_client.post = AsyncMock(return_value=issue_resp)
+        async_client.__aenter__ = AsyncMock(return_value=async_client)
+        async_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("timmy.kimi_delegation.settings", mock_settings),
+            patch("timmy.kimi_delegation.httpx") as mock_httpx,
+        ):
+            mock_httpx.AsyncClient.return_value = async_client
+            result = await create_kimi_research_issue("task", "ctx", "q?")
+
+        assert result["success"] is True
+        assert result["issue_number"] == 42
+        assert "http://git/issues/42" in result["issue_url"]
+
+    @pytest.mark.asyncio
+    async def test_api_error_returns_failure(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://git"
+        mock_settings.gitea_repo = "owner/repo"
+
+        label_resp = MagicMock()
+        label_resp.status_code = 200
+        label_resp.json.return_value = []
+
+        create_label_resp = MagicMock()
+        create_label_resp.status_code = 201
+        create_label_resp.json.return_value = {"id": 1}
+
+        issue_resp = MagicMock()
+        issue_resp.status_code = 500
+        issue_resp.text = "Internal Server Error"
+
+        async_client = AsyncMock()
+        async_client.get = AsyncMock(return_value=label_resp)
+        async_client.post = AsyncMock(side_effect=[create_label_resp, issue_resp])
+        async_client.__aenter__ = AsyncMock(return_value=async_client)
+        async_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("timmy.kimi_delegation.settings", mock_settings),
+            patch("timmy.kimi_delegation.httpx") as mock_httpx,
+        ):
+            mock_httpx.AsyncClient.return_value = async_client
+            result = await create_kimi_research_issue("task", "ctx", "q?")
+
+        assert result["success"] is False
+        assert "500" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_exception_returns_failure(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://git"
+        mock_settings.gitea_repo = "owner/repo"
+
+        async_client = AsyncMock()
+        async_client.__aenter__ = AsyncMock(side_effect=Exception("connection refused"))
+        async_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("timmy.kimi_delegation.settings", mock_settings),
+            patch("timmy.kimi_delegation.httpx") as mock_httpx,
+        ):
+            mock_httpx.AsyncClient.return_value = async_client
+            result = await create_kimi_research_issue("task", "ctx", "q?")
+
+        assert result["success"] is False
+        assert result["error"] != ""
+
+
+# ---------------------------------------------------------------------------
+# poll_kimi_issue (async)
+# ---------------------------------------------------------------------------
+
+
+class TestPollKimiIssue:
+    @pytest.mark.asyncio
+    async def test_returns_error_when_gitea_not_configured(self):
+        from timmy.kimi_delegation import poll_kimi_issue
+
+        with patch("timmy.kimi_delegation.settings") as mock_settings:
+            mock_settings.gitea_enabled = False
+            mock_settings.gitea_token = ""
+            result = await poll_kimi_issue(123)
+
+        assert result["completed"] is False
+        assert "not configured" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_returns_completed_when_issue_closed(self):
+        from timmy.kimi_delegation import poll_kimi_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://git"
+        mock_settings.gitea_repo = "owner/repo"
+
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = {"state": "closed", "body": "Done!"}
+
+        async_client = AsyncMock()
+        async_client.get = AsyncMock(return_value=resp)
+        async_client.__aenter__ = AsyncMock(return_value=async_client)
+        async_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("timmy.kimi_delegation.settings", mock_settings),
+            patch("timmy.kimi_delegation.httpx") as mock_httpx,
+        ):
+            mock_httpx.AsyncClient.return_value = async_client
+            result = await poll_kimi_issue(42, poll_interval=0, max_wait=1)
+
+        assert result["completed"] is True
+        assert result["state"] == "closed"
+        assert result["body"] == "Done!"
+
+    @pytest.mark.asyncio
+    async def test_times_out_when_issue_stays_open(self):
+        from timmy.kimi_delegation import poll_kimi_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://git"
+        mock_settings.gitea_repo = "owner/repo"
+
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = {"state": "open", "body": ""}
+
+        async_client = AsyncMock()
+        async_client.get = AsyncMock(return_value=resp)
+        async_client.__aenter__ = AsyncMock(return_value=async_client)
+        async_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("timmy.kimi_delegation.settings", mock_settings),
+            patch("timmy.kimi_delegation.httpx") as mock_httpx,
+            patch("timmy.kimi_delegation.asyncio.sleep", new_callable=AsyncMock),
+        ):
+            mock_httpx.AsyncClient.return_value = async_client
+            # poll_interval > max_wait so it exits immediately after first sleep
+            result = await poll_kimi_issue(42, poll_interval=10, max_wait=5)
+
+        assert result["completed"] is False
+        assert result["state"] == "timeout"
+
+
+# ---------------------------------------------------------------------------
+# index_kimi_artifact (async)
+# ---------------------------------------------------------------------------
+
+
+class TestIndexKimiArtifact:
+    @pytest.mark.asyncio
+    async def test_empty_artifact_returns_error(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        result = await index_kimi_artifact(1, "title", "   ")
+        assert result["success"] is False
+        assert "Empty artifact" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_successful_indexing(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        mock_entry = MagicMock()
+        mock_entry.id = "mem-123"
+
+        with patch("timmy.kimi_delegation.asyncio.to_thread", new_callable=AsyncMock) as mock_thread:
+            mock_thread.return_value = mock_entry
+            result = await index_kimi_artifact(42, "My Research", "Some research content here")
+
+        assert result["success"] is True
+        assert result["memory_id"] == "mem-123"
+
+    @pytest.mark.asyncio
+    async def test_exception_returns_failure(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        with patch("timmy.kimi_delegation.asyncio.to_thread", new_callable=AsyncMock) as mock_thread:
+            mock_thread.side_effect = Exception("DB error")
+            result = await index_kimi_artifact(42, "title", "some content")
+
+        assert result["success"] is False
+        assert result["error"] != ""
+
+
+# ---------------------------------------------------------------------------
+# extract_and_create_followups (async)
+# ---------------------------------------------------------------------------
+
+
+class TestExtractAndCreateFollowups:
+    @pytest.mark.asyncio
+    async def test_no_action_items_returns_empty_created(self):
+        from timmy.kimi_delegation import extract_and_create_followups
+
+        result = await extract_and_create_followups("Plain prose, nothing to do.", 1)
+        assert result["success"] is True
+        assert result["created"] == []
+
+    @pytest.mark.asyncio
+    async def test_gitea_not_configured_returns_error(self):
+        from timmy.kimi_delegation import extract_and_create_followups
+
+        text = "1. Do something important\n"
+
+        with patch("timmy.kimi_delegation.settings") as mock_settings:
+            mock_settings.gitea_enabled = False
+            mock_settings.gitea_token = ""
+            result = await extract_and_create_followups(text, 5)
+
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_creates_followup_issues(self):
+        from timmy.kimi_delegation import extract_and_create_followups
+
+        text = "1. Deploy the service\n2. Run integration tests\n"
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://git"
+        mock_settings.gitea_repo = "owner/repo"
+
+        issue_resp = MagicMock()
+        issue_resp.status_code = 201
+        issue_resp.json.return_value = {"number": 10}
+
+        async_client = AsyncMock()
+        async_client.post = AsyncMock(return_value=issue_resp)
+        async_client.__aenter__ = AsyncMock(return_value=async_client)
+        async_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("timmy.kimi_delegation.settings", mock_settings),
+            patch("timmy.kimi_delegation.httpx") as mock_httpx,
+        ):
+            mock_httpx.AsyncClient.return_value = async_client
+            result = await extract_and_create_followups(text, 5)
+
+        assert result["success"] is True
+        assert len(result["created"]) == 2
+
+
+# ---------------------------------------------------------------------------
+# delegate_research_to_kimi (async)
+# ---------------------------------------------------------------------------
+
+
+class TestDelegateResearchToKimi:
+    @pytest.mark.asyncio
+    async def test_empty_task_returns_error(self):
+        from timmy.kimi_delegation import delegate_research_to_kimi
+
+        result = await delegate_research_to_kimi("", "ctx", "q?")
+        assert result["success"] is False
+        assert "required" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_whitespace_task_returns_error(self):
+        from timmy.kimi_delegation import delegate_research_to_kimi
+
+        result = await delegate_research_to_kimi("   ", "ctx", "q?")
+        assert result["success"] is False
+        assert "required" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_empty_question_returns_error(self):
+        from timmy.kimi_delegation import delegate_research_to_kimi
+
+        result = await delegate_research_to_kimi("valid task", "ctx", "")
+        assert result["success"] is False
+        assert "required" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_delegates_to_create_issue(self):
+        from timmy.kimi_delegation import delegate_research_to_kimi
+
+        with patch(
+            "timmy.kimi_delegation.create_kimi_research_issue",
+            new_callable=AsyncMock,
+        ) as mock_create:
+            mock_create.return_value = {"success": True, "issue_number": 7, "issue_url": "http://x", "error": None}
+            result = await delegate_research_to_kimi("Research X", "ctx", "What is X?", priority="high")
+
+        assert result["success"] is True
+        assert result["issue_number"] == 7
+        mock_create.assert_awaited_once_with("Research X", "ctx", "What is X?", "high")