644 lines
23 KiB
Python
644 lines
23 KiB
Python
"""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")
|