forked from Rockachopa/Timmy-time-dashboard
362 lines
11 KiB
Python
362 lines
11 KiB
Python
"""Unit tests for timmy.vassal.dispatch — routing and label helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from timmy.vassal.backlog import AgentTarget, TriagedIssue
|
|
from timmy.vassal.dispatch import (
|
|
DispatchRecord,
|
|
_apply_label_to_issue,
|
|
_get_or_create_label,
|
|
_post_dispatch_comment,
|
|
clear_dispatch_registry,
|
|
get_dispatch_registry,
|
|
)
|
|
|
|
|
|
def _make_triaged(
|
|
number: int,
|
|
title: str,
|
|
agent: AgentTarget,
|
|
priority: int = 50,
|
|
) -> TriagedIssue:
|
|
return TriagedIssue(
|
|
number=number,
|
|
title=title,
|
|
body="",
|
|
agent_target=agent,
|
|
priority_score=priority,
|
|
rationale="test rationale",
|
|
url=f"http://gitea/issues/{number}",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_registry_starts_empty():
|
|
clear_dispatch_registry()
|
|
assert get_dispatch_registry() == {}
|
|
|
|
|
|
def test_registry_returns_copy():
|
|
clear_dispatch_registry()
|
|
reg = get_dispatch_registry()
|
|
reg[999] = None # type: ignore[assignment]
|
|
assert 999 not in get_dispatch_registry()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# dispatch_issue — Timmy self-dispatch (no Gitea required)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_timmy_self_no_gitea():
|
|
"""Timmy self-dispatch records without hitting Gitea."""
|
|
clear_dispatch_registry()
|
|
|
|
issue = _make_triaged(1, "Fix docs typo", AgentTarget.TIMMY)
|
|
from timmy.vassal.dispatch import dispatch_issue
|
|
|
|
record = await dispatch_issue(issue)
|
|
|
|
assert isinstance(record, DispatchRecord)
|
|
assert record.issue_number == 1
|
|
assert record.agent == AgentTarget.TIMMY
|
|
assert 1 in get_dispatch_registry()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_claude_no_gitea_token():
|
|
"""Claude dispatch gracefully degrades when Gitea token is absent."""
|
|
clear_dispatch_registry()
|
|
|
|
issue = _make_triaged(2, "Refactor auth", AgentTarget.CLAUDE)
|
|
from timmy.vassal.dispatch import dispatch_issue
|
|
|
|
record = await dispatch_issue(issue)
|
|
|
|
assert record.issue_number == 2
|
|
assert record.agent == AgentTarget.CLAUDE
|
|
# label/comment not applied — no token
|
|
assert record.label_applied is False
|
|
assert 2 in get_dispatch_registry()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_kimi_no_gitea_token():
|
|
clear_dispatch_registry()
|
|
|
|
issue = _make_triaged(3, "Research embeddings", AgentTarget.KIMI)
|
|
from timmy.vassal.dispatch import dispatch_issue
|
|
|
|
record = await dispatch_issue(issue)
|
|
|
|
assert record.agent == AgentTarget.KIMI
|
|
assert record.label_applied is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DispatchRecord fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_dispatch_record_defaults():
|
|
r = DispatchRecord(
|
|
issue_number=5,
|
|
issue_title="Test issue",
|
|
agent=AgentTarget.TIMMY,
|
|
rationale="because",
|
|
)
|
|
assert r.label_applied is False
|
|
assert r.comment_posted is False
|
|
assert r.dispatched_at # has a timestamp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_or_create_label
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_HEADERS = {"Authorization": "token x"}
|
|
_BASE_URL = "http://gitea"
|
|
_REPO = "org/repo"
|
|
|
|
|
|
def _mock_response(status_code: int, json_data=None):
|
|
resp = MagicMock()
|
|
resp.status_code = status_code
|
|
resp.json.return_value = json_data or {}
|
|
return resp
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_label_finds_existing():
|
|
"""Returns the ID of an existing label without creating it."""
|
|
existing = [{"name": "claude-ready", "id": 42}, {"name": "other", "id": 7}]
|
|
client = AsyncMock()
|
|
client.get.return_value = _mock_response(200, existing)
|
|
|
|
result = await _get_or_create_label(client, _BASE_URL, _HEADERS, _REPO, "claude-ready")
|
|
|
|
assert result == 42
|
|
client.post.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_label_creates_when_missing():
|
|
"""Creates the label when it doesn't exist in the list."""
|
|
client = AsyncMock()
|
|
# GET returns empty list
|
|
client.get.return_value = _mock_response(200, [])
|
|
# POST creates label
|
|
client.post.return_value = _mock_response(201, {"id": 99})
|
|
|
|
result = await _get_or_create_label(client, _BASE_URL, _HEADERS, _REPO, "claude-ready")
|
|
|
|
assert result == 99
|
|
client.post.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_label_returns_none_on_get_error():
|
|
"""Returns None if the GET raises an exception."""
|
|
client = AsyncMock()
|
|
client.get.side_effect = Exception("network error")
|
|
|
|
result = await _get_or_create_label(client, _BASE_URL, _HEADERS, _REPO, "claude-ready")
|
|
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_label_returns_none_on_create_error():
|
|
"""Returns None if POST raises an exception."""
|
|
client = AsyncMock()
|
|
client.get.return_value = _mock_response(200, [])
|
|
client.post.side_effect = Exception("post failed")
|
|
|
|
result = await _get_or_create_label(client, _BASE_URL, _HEADERS, _REPO, "claude-ready")
|
|
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_label_uses_default_color_for_unknown():
|
|
"""Unknown label name uses '#cccccc' fallback color."""
|
|
client = AsyncMock()
|
|
client.get.return_value = _mock_response(200, [])
|
|
client.post.return_value = _mock_response(201, {"id": 5})
|
|
|
|
await _get_or_create_label(client, _BASE_URL, _HEADERS, _REPO, "unknown-label")
|
|
|
|
call_kwargs = client.post.call_args
|
|
assert call_kwargs.kwargs["json"]["color"] == "#cccccc"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _apply_label_to_issue
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_label_to_issue_success():
|
|
"""Returns True when label is found and applied."""
|
|
client = AsyncMock()
|
|
client.get.return_value = _mock_response(200, [{"name": "claude-ready", "id": 10}])
|
|
client.post.return_value = _mock_response(201)
|
|
|
|
result = await _apply_label_to_issue(client, _BASE_URL, _HEADERS, _REPO, 42, "claude-ready")
|
|
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_label_to_issue_returns_false_when_no_label_id():
|
|
"""Returns False when label ID cannot be obtained."""
|
|
client = AsyncMock()
|
|
client.get.side_effect = Exception("unavailable")
|
|
|
|
result = await _apply_label_to_issue(client, _BASE_URL, _HEADERS, _REPO, 42, "claude-ready")
|
|
|
|
assert result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_label_to_issue_returns_false_on_bad_status():
|
|
"""Returns False when the apply POST returns a non-2xx status."""
|
|
client = AsyncMock()
|
|
client.get.return_value = _mock_response(200, [{"name": "claude-ready", "id": 10}])
|
|
client.post.return_value = _mock_response(403)
|
|
|
|
result = await _apply_label_to_issue(client, _BASE_URL, _HEADERS, _REPO, 42, "claude-ready")
|
|
|
|
assert result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _post_dispatch_comment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_dispatch_comment_success():
|
|
"""Returns True on successful comment post."""
|
|
client = AsyncMock()
|
|
client.post.return_value = _mock_response(201)
|
|
|
|
issue = _make_triaged(7, "Some issue", AgentTarget.CLAUDE, priority=75)
|
|
result = await _post_dispatch_comment(client, _BASE_URL, _HEADERS, _REPO, issue, "claude-ready")
|
|
|
|
assert result is True
|
|
body = client.post.call_args.kwargs["json"]["body"]
|
|
assert "Claude" in body
|
|
assert "claude-ready" in body
|
|
assert "75" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_dispatch_comment_failure():
|
|
"""Returns False when comment POST returns a non-2xx status."""
|
|
client = AsyncMock()
|
|
client.post.return_value = _mock_response(500)
|
|
|
|
issue = _make_triaged(8, "Other issue", AgentTarget.KIMI)
|
|
result = await _post_dispatch_comment(client, _BASE_URL, _HEADERS, _REPO, issue, "kimi-ready")
|
|
|
|
assert result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _perform_gitea_dispatch — settings-level gate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_gitea_dispatch_skips_when_disabled():
|
|
"""Does not call Gitea when gitea_enabled is False."""
|
|
import config
|
|
from timmy.vassal.dispatch import _perform_gitea_dispatch
|
|
|
|
mock_settings = SimpleNamespace(gitea_enabled=False, gitea_token="tok")
|
|
with patch.object(config, "settings", mock_settings):
|
|
issue = _make_triaged(9, "Disabled", AgentTarget.CLAUDE)
|
|
record = DispatchRecord(
|
|
issue_number=9,
|
|
issue_title="Disabled",
|
|
agent=AgentTarget.CLAUDE,
|
|
rationale="r",
|
|
)
|
|
await _perform_gitea_dispatch(issue, record)
|
|
|
|
assert record.label_applied is False
|
|
assert record.comment_posted is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_gitea_dispatch_skips_when_no_token():
|
|
"""Does not call Gitea when gitea_token is empty."""
|
|
import config
|
|
from timmy.vassal.dispatch import _perform_gitea_dispatch
|
|
|
|
mock_settings = SimpleNamespace(gitea_enabled=True, gitea_token="")
|
|
with patch.object(config, "settings", mock_settings):
|
|
issue = _make_triaged(10, "No token", AgentTarget.CLAUDE)
|
|
record = DispatchRecord(
|
|
issue_number=10,
|
|
issue_title="No token",
|
|
agent=AgentTarget.CLAUDE,
|
|
rationale="r",
|
|
)
|
|
await _perform_gitea_dispatch(issue, record)
|
|
|
|
assert record.label_applied is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_gitea_dispatch_updates_record():
|
|
"""Record is mutated to reflect label/comment success."""
|
|
import config
|
|
from timmy.vassal.dispatch import _perform_gitea_dispatch
|
|
|
|
mock_settings = SimpleNamespace(
|
|
gitea_enabled=True,
|
|
gitea_token="tok",
|
|
gitea_url="http://gitea",
|
|
gitea_repo="org/repo",
|
|
)
|
|
|
|
mock_client = AsyncMock()
|
|
# GET labels → empty list, POST create label → id 1
|
|
mock_client.get.return_value = _mock_response(200, [])
|
|
mock_client.post.side_effect = [
|
|
_mock_response(201, {"id": 1}), # create label
|
|
_mock_response(201), # apply label
|
|
_mock_response(201), # post comment
|
|
]
|
|
|
|
with (
|
|
patch.object(config, "settings", mock_settings),
|
|
patch("httpx.AsyncClient") as mock_cls,
|
|
):
|
|
mock_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
issue = _make_triaged(11, "Full dispatch", AgentTarget.CLAUDE)
|
|
record = DispatchRecord(
|
|
issue_number=11,
|
|
issue_title="Full dispatch",
|
|
agent=AgentTarget.CLAUDE,
|
|
rationale="r",
|
|
)
|
|
await _perform_gitea_dispatch(issue, record)
|
|
|
|
assert record.label_applied is True
|
|
assert record.comment_posted is True
|