Claude (Opus 4.6)
f2a277f7b5
Co-authored-by: Claude (Opus 4.6) <claude@hermes.local> Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
461 lines
17 KiB
Python
461 lines
17 KiB
Python
"""Unit tests for timmy.kimi_delegation — Kimi research delegation via Gitea labels."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from timmy.kimi_delegation import (
|
|
KIMI_LABEL_COLOR,
|
|
KIMI_READY_LABEL,
|
|
_build_research_template,
|
|
_extract_action_items,
|
|
_slugify,
|
|
delegate_research_to_kimi,
|
|
exceeds_local_capacity,
|
|
)
|
|
|
|
# ── Constants ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_kimi_ready_label():
|
|
assert KIMI_READY_LABEL == "kimi-ready"
|
|
|
|
|
|
def test_kimi_label_color_is_hex():
|
|
assert KIMI_LABEL_COLOR.startswith("#")
|
|
assert len(KIMI_LABEL_COLOR) == 7
|
|
|
|
|
|
# ── exceeds_local_capacity ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestExceedsLocalCapacity:
|
|
def test_keyword_comprehensive(self):
|
|
assert exceeds_local_capacity("Do a comprehensive review of X") is True
|
|
|
|
def test_keyword_deep_research(self):
|
|
assert exceeds_local_capacity("deep research into neural networks") is True
|
|
|
|
def test_keyword_benchmark(self):
|
|
assert exceeds_local_capacity("benchmark these five models") is True
|
|
|
|
def test_keyword_exhaustive(self):
|
|
assert exceeds_local_capacity("exhaustive list of options") is True
|
|
|
|
def test_keyword_case_insensitive(self):
|
|
assert exceeds_local_capacity("COMPREHENSIVE analysis") is True
|
|
|
|
def test_keyword_survey(self):
|
|
assert exceeds_local_capacity("survey all available tools") is True
|
|
|
|
def test_keyword_extensive(self):
|
|
assert exceeds_local_capacity("extensive documentation needed") is True
|
|
|
|
def test_short_simple_task(self):
|
|
assert exceeds_local_capacity("fix the login bug") is False
|
|
|
|
def test_long_task_exceeds_word_threshold(self):
|
|
long_task = " ".join(["word"] * 55)
|
|
assert exceeds_local_capacity(long_task) is True
|
|
|
|
def test_exactly_at_threshold(self):
|
|
at_threshold = " ".join(["word"] * 50)
|
|
assert exceeds_local_capacity(at_threshold) is True
|
|
|
|
def test_just_below_threshold(self):
|
|
short = " ".join(["word"] * 49)
|
|
assert exceeds_local_capacity(short) is False
|
|
|
|
def test_empty_string(self):
|
|
assert exceeds_local_capacity("") is False
|
|
|
|
|
|
# ── _slugify ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSlugify:
|
|
def test_simple_text(self):
|
|
assert _slugify("Hello World") == "hello-world"
|
|
|
|
def test_special_characters_removed(self):
|
|
assert _slugify("Hello, World!") == "hello-world"
|
|
|
|
def test_underscores_become_dashes(self):
|
|
assert _slugify("hello_world") == "hello-world"
|
|
|
|
def test_multiple_spaces(self):
|
|
assert _slugify("hello world") == "hello-world"
|
|
|
|
def test_truncates_to_60(self):
|
|
long = "a" * 80
|
|
result = _slugify(long)
|
|
assert len(result) <= 60
|
|
|
|
def test_no_leading_trailing_dashes(self):
|
|
result = _slugify(" hello ")
|
|
assert not result.startswith("-")
|
|
assert not result.endswith("-")
|
|
|
|
def test_empty_string(self):
|
|
assert _slugify("") == ""
|
|
|
|
|
|
# ── _build_research_template ──────────────────────────────────────────────────
|
|
|
|
|
|
class TestBuildResearchTemplate:
|
|
def test_contains_task(self):
|
|
body = _build_research_template("My Task", "some context", "What is X?")
|
|
assert "My Task" in body
|
|
|
|
def test_contains_question(self):
|
|
body = _build_research_template("Task", "ctx", "What is the answer?")
|
|
assert "What is the answer?" in body
|
|
|
|
def test_contains_context(self):
|
|
body = _build_research_template("Task", "project background", "Q?")
|
|
assert "project background" in body
|
|
|
|
def test_contains_kimi_ready_label(self):
|
|
body = _build_research_template("Task", "ctx", "Q?")
|
|
assert KIMI_READY_LABEL in body
|
|
|
|
def test_default_priority_normal(self):
|
|
body = _build_research_template("Task", "ctx", "Q?")
|
|
assert "normal" in body
|
|
|
|
def test_custom_priority_high(self):
|
|
body = _build_research_template("Task", "ctx", "Q?", priority="high")
|
|
assert "high" in body
|
|
|
|
def test_contains_deliverables_section(self):
|
|
body = _build_research_template("Task", "ctx", "Q?")
|
|
assert "Deliverables" in body
|
|
|
|
def test_slug_in_artifact_path(self):
|
|
body = _build_research_template("My Research Task", "ctx", "Q?")
|
|
assert "my-research-task" in body
|
|
|
|
def test_contains_research_request_header(self):
|
|
body = _build_research_template("Task", "ctx", "Q?")
|
|
assert "## Research Request" in body
|
|
|
|
|
|
# ── _extract_action_items ─────────────────────────────────────────────────────
|
|
|
|
|
|
class TestExtractActionItems:
|
|
def test_checkbox_items(self):
|
|
text = "- [ ] Do thing A\n- [ ] Do thing B"
|
|
items = _extract_action_items(text)
|
|
assert "Do thing A" in items
|
|
assert "Do thing B" in items
|
|
|
|
def test_numbered_list(self):
|
|
text = "1. First step\n2. Second step\n3. Third step"
|
|
items = _extract_action_items(text)
|
|
assert "First step" in items
|
|
assert "Second step" in items
|
|
assert "Third step" in items
|
|
|
|
def test_action_prefix(self):
|
|
text = "Action: Implement caching layer"
|
|
items = _extract_action_items(text)
|
|
assert "Implement caching layer" in items
|
|
|
|
def test_todo_prefix(self):
|
|
text = "TODO: Write tests"
|
|
items = _extract_action_items(text)
|
|
assert "Write tests" in items
|
|
|
|
def test_next_step_prefix(self):
|
|
text = "Next step: Deploy to staging"
|
|
items = _extract_action_items(text)
|
|
assert "Deploy to staging" in items
|
|
|
|
def test_case_insensitive_prefixes(self):
|
|
text = "TODO: Upper\ntodo: lower\nTodo: Mixed"
|
|
items = _extract_action_items(text)
|
|
assert len(items) == 3
|
|
|
|
def test_deduplication(self):
|
|
text = "1. Do the thing\n2. Do the thing"
|
|
items = _extract_action_items(text)
|
|
assert items.count("Do the thing") == 1
|
|
|
|
def test_empty_text(self):
|
|
assert _extract_action_items("") == []
|
|
|
|
def test_no_action_items(self):
|
|
text = "This is just a paragraph with no action items."
|
|
assert _extract_action_items(text) == []
|
|
|
|
def test_returns_list(self):
|
|
assert isinstance(_extract_action_items("1. Item"), list)
|
|
|
|
|
|
# ── delegate_research_to_kimi ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestDelegateResearchToKimi:
|
|
@pytest.mark.asyncio
|
|
async def test_empty_task_returns_error(self):
|
|
result = await delegate_research_to_kimi("", "context", "question?")
|
|
assert result["success"] is False
|
|
assert "task" in result["error"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_whitespace_task_returns_error(self):
|
|
result = await delegate_research_to_kimi(" ", "context", "question?")
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_question_returns_error(self):
|
|
result = await delegate_research_to_kimi("Task title", "context", "")
|
|
assert result["success"] is False
|
|
assert "question" in result["error"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_whitespace_question_returns_error(self):
|
|
result = await delegate_research_to_kimi("Task", "ctx", " ")
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delegates_to_create_issue(self):
|
|
with patch(
|
|
"timmy.kimi_delegation.create_kimi_research_issue",
|
|
new_callable=AsyncMock,
|
|
return_value={
|
|
"success": True,
|
|
"issue_number": 42,
|
|
"issue_url": "http://x/42",
|
|
"error": None,
|
|
},
|
|
) as mock_create:
|
|
result = await delegate_research_to_kimi("Task", "ctx", "What is X?", "high")
|
|
mock_create.assert_awaited_once_with("Task", "ctx", "What is X?", "high")
|
|
assert result["success"] is True
|
|
assert result["issue_number"] == 42
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_passes_default_priority(self):
|
|
with patch(
|
|
"timmy.kimi_delegation.create_kimi_research_issue",
|
|
new_callable=AsyncMock,
|
|
return_value={"success": True, "issue_number": 1, "issue_url": "", "error": None},
|
|
) as mock_create:
|
|
await delegate_research_to_kimi("Task", "ctx", "Q?")
|
|
_, _, _, priority = mock_create.call_args.args
|
|
assert priority == "normal"
|
|
|
|
|
|
# ── create_kimi_research_issue ────────────────────────────────────────────────
|
|
|
|
|
|
class TestCreateKimiResearchIssue:
|
|
@pytest.mark.asyncio
|
|
async def test_no_gitea_token_returns_error(self):
|
|
from timmy.kimi_delegation import create_kimi_research_issue
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = ""
|
|
|
|
with patch("config.settings", mock_settings):
|
|
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_gitea_disabled_returns_error(self):
|
|
from timmy.kimi_delegation import create_kimi_research_issue
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.gitea_enabled = False
|
|
mock_settings.gitea_token = "tok"
|
|
|
|
with patch("config.settings", mock_settings):
|
|
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 = "fake-token"
|
|
mock_settings.gitea_url = "http://gitea.local"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
|
|
label_resp = MagicMock()
|
|
label_resp.status_code = 200
|
|
label_resp.json.return_value = [{"name": "kimi-ready", "id": 7}]
|
|
|
|
issue_resp = MagicMock()
|
|
issue_resp.status_code = 201
|
|
issue_resp.json.return_value = {
|
|
"number": 101,
|
|
"html_url": "http://gitea.local/issues/101",
|
|
}
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = label_resp
|
|
mock_client.post.return_value = issue_resp
|
|
|
|
async_ctx = AsyncMock()
|
|
async_ctx.__aenter__.return_value = mock_client
|
|
async_ctx.__aexit__.return_value = False
|
|
|
|
with (
|
|
patch("config.settings", mock_settings),
|
|
patch("httpx.AsyncClient", return_value=async_ctx),
|
|
):
|
|
result = await create_kimi_research_issue("Task", "ctx", "Q?")
|
|
|
|
assert result["success"] is True
|
|
assert result["issue_number"] == 101
|
|
assert result["error"] is None
|
|
|
|
@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://gitea.local"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
|
|
label_resp = MagicMock()
|
|
label_resp.status_code = 200
|
|
label_resp.json.return_value = [{"name": "kimi-ready", "id": 7}]
|
|
|
|
issue_resp = MagicMock()
|
|
issue_resp.status_code = 500
|
|
issue_resp.text = "Internal Server Error"
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.get.return_value = label_resp
|
|
mock_client.post.return_value = issue_resp
|
|
|
|
async_ctx = AsyncMock()
|
|
async_ctx.__aenter__.return_value = mock_client
|
|
async_ctx.__aexit__.return_value = False
|
|
|
|
with (
|
|
patch("config.settings", mock_settings),
|
|
patch("httpx.AsyncClient", return_value=async_ctx),
|
|
):
|
|
result = await create_kimi_research_issue("Task", "ctx", "Q?")
|
|
|
|
assert result["success"] is False
|
|
assert "500" in result["error"]
|
|
|
|
|
|
# ── index_kimi_artifact ───────────────────────────────────────────────────────
|
|
|
|
|
|
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(42, "Title", "")
|
|
assert result["success"] is False
|
|
assert "Empty" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_whitespace_only_artifact_returns_error(self):
|
|
from timmy.kimi_delegation import index_kimi_artifact
|
|
|
|
result = await index_kimi_artifact(42, "Title", " \n ")
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_indexing(self):
|
|
from timmy.kimi_delegation import index_kimi_artifact
|
|
|
|
mock_entry = MagicMock()
|
|
mock_entry.id = "mem-abc-123"
|
|
|
|
with patch("timmy.memory_system.store_memory", return_value=mock_entry) as mock_store:
|
|
result = await index_kimi_artifact(55, "Research Title", "Artifact content here.")
|
|
|
|
assert result["success"] is True
|
|
assert result["memory_id"] == "mem-abc-123"
|
|
mock_store.assert_called_once()
|
|
call_kwargs = mock_store.call_args.kwargs
|
|
assert call_kwargs["source"] == "kimi"
|
|
assert call_kwargs["context_type"] == "document"
|
|
assert call_kwargs["task_id"] == "55"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_memory_exception_returns_error(self):
|
|
from timmy.kimi_delegation import index_kimi_artifact
|
|
|
|
with patch(
|
|
"timmy.memory_system.store_memory",
|
|
side_effect=RuntimeError("DB error"),
|
|
):
|
|
result = await index_kimi_artifact(1, "T", "Some content")
|
|
assert result["success"] is False
|
|
assert "DB error" in result["error"]
|
|
|
|
|
|
# ── extract_and_create_followups ──────────────────────────────────────────────
|
|
|
|
|
|
class TestExtractAndCreateFollowups:
|
|
@pytest.mark.asyncio
|
|
async def test_no_action_items_returns_empty_list(self):
|
|
from timmy.kimi_delegation import extract_and_create_followups
|
|
|
|
result = await extract_and_create_followups("No action items here.", 10)
|
|
assert result["success"] is True
|
|
assert result["created"] == []
|
|
assert result["error"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitea_not_configured(self):
|
|
from timmy.kimi_delegation import extract_and_create_followups
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.gitea_enabled = False
|
|
mock_settings.gitea_token = ""
|
|
|
|
with patch("config.settings", mock_settings):
|
|
result = await extract_and_create_followups("1. Do the thing", 10)
|
|
assert result["success"] is False
|
|
assert result["created"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_followup_issues(self):
|
|
from timmy.kimi_delegation import extract_and_create_followups
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = "tok"
|
|
mock_settings.gitea_url = "http://gitea.local"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
|
|
issue_resp = MagicMock()
|
|
issue_resp.status_code = 201
|
|
issue_resp.json.return_value = {"number": 200}
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.post.return_value = issue_resp
|
|
|
|
async_ctx = AsyncMock()
|
|
async_ctx.__aenter__.return_value = mock_client
|
|
async_ctx.__aexit__.return_value = False
|
|
|
|
with (
|
|
patch("config.settings", mock_settings),
|
|
patch("httpx.AsyncClient", return_value=async_ctx),
|
|
):
|
|
result = await extract_and_create_followups("1. Do the thing\n2. Do another thing", 10)
|
|
|
|
assert result["success"] is True
|
|
assert 200 in result["created"]
|