forked from Rockachopa/Timmy-time-dashboard
350e6f54ff
The httpx AsyncClient was cached across asyncio.run() boundaries. Each asyncio.run() creates and closes a new event loop, leaving the cached client's connections on a dead loop. Second+ calls would fail with "Event loop is closed". Fix: create a fresh client per request and close it in a finally block. No more cross-loop client reuse. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""Tests for the Gitea Hand.
|
|
|
|
Covers:
|
|
- GiteaResult dataclass defaults
|
|
- Token resolution (settings vs filesystem fallback)
|
|
- Availability checks
|
|
- Title similarity dedup
|
|
- HTTP request handling (mocked)
|
|
- Info summary
|
|
- Graceful degradation on failure
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GiteaResult dataclass
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_gitea_result_defaults():
|
|
"""GiteaResult should have sensible defaults."""
|
|
from infrastructure.hands.gitea import GiteaResult
|
|
|
|
r = GiteaResult(operation="GET /issues", success=True)
|
|
assert r.data == {}
|
|
assert r.error == ""
|
|
assert r.latency_ms == 0.0
|
|
|
|
|
|
def test_gitea_result_with_data():
|
|
"""GiteaResult should carry data."""
|
|
from infrastructure.hands.gitea import GiteaResult
|
|
|
|
r = GiteaResult(
|
|
operation="POST /issues",
|
|
success=True,
|
|
data={"number": 42, "html_url": "http://localhost:3000/issues/42"},
|
|
latency_ms=15.3,
|
|
)
|
|
assert r.data["number"] == 42
|
|
assert r.success is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Title similarity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_title_similar_matches():
|
|
"""Similar titles should be detected as duplicates."""
|
|
from infrastructure.hands.gitea import _title_similar
|
|
|
|
assert _title_similar("memory_forget tool fails", "Memory_forget tool fails") is True
|
|
assert (
|
|
_title_similar(
|
|
"MEMORY.md not updating after thoughts",
|
|
"MEMORY.md hasn't updated since March 8",
|
|
)
|
|
is True
|
|
)
|
|
|
|
|
|
def test_title_similar_rejects_different():
|
|
"""Different titles should not match."""
|
|
from infrastructure.hands.gitea import _title_similar
|
|
|
|
assert _title_similar("fix login page CSS", "add dark mode toggle") is False
|
|
assert _title_similar("memory bug", "completely different topic") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_resolve_token_from_settings():
|
|
"""Token from settings should be preferred."""
|
|
from infrastructure.hands.gitea import _resolve_token
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_token = "from-settings"
|
|
assert _resolve_token() == "from-settings"
|
|
|
|
|
|
def test_resolve_token_from_file(tmp_path):
|
|
"""Token should fall back to filesystem."""
|
|
from infrastructure.hands.gitea import _resolve_token
|
|
|
|
token_file = tmp_path / "token"
|
|
token_file.write_text("from-file\n")
|
|
|
|
with (
|
|
patch("infrastructure.hands.gitea.settings") as mock_settings,
|
|
patch("infrastructure.hands.gitea._TOKEN_FILE", token_file),
|
|
):
|
|
mock_settings.gitea_token = ""
|
|
assert _resolve_token() == "from-file"
|
|
|
|
|
|
def test_resolve_token_missing(tmp_path):
|
|
"""Empty string when no token available."""
|
|
|
|
from infrastructure.hands.gitea import _resolve_token
|
|
|
|
with (
|
|
patch("infrastructure.hands.gitea.settings") as mock_settings,
|
|
patch("infrastructure.hands.gitea._TOKEN_FILE", tmp_path / "nonexistent"),
|
|
):
|
|
mock_settings.gitea_token = ""
|
|
assert _resolve_token() == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Availability
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_available_when_configured():
|
|
"""Hand should be available when token and repo are set."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
assert hand.available is True
|
|
|
|
|
|
def test_not_available_without_token():
|
|
"""Hand should not be available without token."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
with (
|
|
patch("infrastructure.hands.gitea.settings") as mock_settings,
|
|
patch("infrastructure.hands.gitea._resolve_token", return_value=""),
|
|
):
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = ""
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="")
|
|
assert hand.available is False
|
|
|
|
|
|
def test_not_available_when_disabled():
|
|
"""Hand should not be available when disabled."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = False
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
assert hand.available is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP request handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_returns_error_when_unavailable():
|
|
"""_request should return error when not configured."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = False
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = ""
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="")
|
|
result = await hand._request("GET", "/test")
|
|
assert result.success is False
|
|
assert "not configured" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_issue_success():
|
|
"""create_issue should POST and return issue data."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_response.text = '{"number": 1, "html_url": "http://localhost:3000/issues/1"}'
|
|
mock_response.json.return_value = {
|
|
"number": 1,
|
|
"html_url": "http://localhost:3000/issues/1",
|
|
}
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.request = AsyncMock(return_value=mock_response)
|
|
mock_client.aclose = AsyncMock()
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
hand._get_client = MagicMock(return_value=mock_client)
|
|
|
|
result = await hand.create_issue("Test issue", "Body text")
|
|
assert result.success is True
|
|
assert result.data["number"] == 1
|
|
|
|
# Verify the API call
|
|
mock_client.request.assert_called_once()
|
|
call_args = mock_client.request.call_args
|
|
assert call_args[0] == ("POST", "/api/v1/repos/owner/repo/issues")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_issue_handles_http_error():
|
|
"""create_issue should handle HTTP errors gracefully."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 500
|
|
mock_response.text = "Internal Server Error"
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.request = AsyncMock(return_value=mock_response)
|
|
mock_client.aclose = AsyncMock()
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
hand._get_client = MagicMock(return_value=mock_client)
|
|
|
|
result = await hand.create_issue("Test issue")
|
|
assert result.success is False
|
|
assert "500" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_issue_handles_connection_error():
|
|
"""create_issue should handle connection errors gracefully."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.request = AsyncMock(side_effect=ConnectionError("refused"))
|
|
mock_client.aclose = AsyncMock()
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
hand._get_client = MagicMock(return_value=mock_client)
|
|
|
|
result = await hand.create_issue("Test issue")
|
|
assert result.success is False
|
|
assert "refused" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_duplicate_detects_match():
|
|
"""find_duplicate should detect similar open issues."""
|
|
from infrastructure.hands.gitea import GiteaHand, GiteaResult
|
|
|
|
existing_issues = [
|
|
{"number": 5, "title": "MEMORY.md not updating", "html_url": "http://example.com/5"},
|
|
{"number": 6, "title": "Add dark mode", "html_url": "http://example.com/6"},
|
|
]
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
|
|
# Mock list_issues
|
|
hand.list_issues = AsyncMock(
|
|
return_value=GiteaResult(
|
|
operation="GET",
|
|
success=True,
|
|
data=existing_issues,
|
|
)
|
|
)
|
|
|
|
dup = await hand.find_duplicate("MEMORY.md hasn't updated")
|
|
assert dup is not None
|
|
assert dup["number"] == 5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_duplicate_no_match():
|
|
"""find_duplicate should return None when no similar issue exists."""
|
|
from infrastructure.hands.gitea import GiteaHand, GiteaResult
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
|
|
hand.list_issues = AsyncMock(
|
|
return_value=GiteaResult(
|
|
operation="GET",
|
|
success=True,
|
|
data=[
|
|
{"number": 1, "title": "Completely unrelated issue"},
|
|
],
|
|
)
|
|
)
|
|
|
|
dup = await hand.find_duplicate("memory_forget tool throws error")
|
|
assert dup is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Info summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_info_returns_summary():
|
|
"""info() should return a dict with status."""
|
|
from infrastructure.hands.gitea import GiteaHand
|
|
|
|
with patch("infrastructure.hands.gitea.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.gitea_token = "test-token"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_timeout = 30
|
|
hand = GiteaHand(token="test-token")
|
|
info = hand.info()
|
|
assert "base_url" in info
|
|
assert "repo" in info
|
|
assert "available" in info
|
|
assert info["available"] is True
|