Michael
147 lines
5.6 KiB
Python
147 lines
5.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
|
|
from core.orchestrator import AgentOrchestrator
|
|
from gitea.models import IssueModel, PullRequestModel, RepositoryModel
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_state_file(tmp_path: Path) -> Path:
|
|
"""Fixture to mock state file path."""
|
|
state_file = tmp_path / "agent_state.json"
|
|
return state_file
|
|
|
|
|
|
@patch("core.orchestrator.AgentOrchestrator._get_state_file_path")
|
|
@patch("core.orchestrator.AgentDispatcher")
|
|
@patch("core.orchestrator.WorkspaceManager")
|
|
@patch("core.orchestrator.NotificationReaderAgent")
|
|
async def test_poll_and_dispatch_no_notifications(
|
|
mock_notification_reader_class: MagicMock,
|
|
mock_workspace_class: MagicMock,
|
|
mock_dispatcher_class: MagicMock,
|
|
mock_get_path: MagicMock,
|
|
temp_state_file: Path
|
|
) -> None:
|
|
mock_get_path.return_value = temp_state_file
|
|
mock_client = MagicMock()
|
|
|
|
# Return no notifications
|
|
mock_client.notifications.list_unread_notifications.return_value = []
|
|
|
|
orchestrator = AgentOrchestrator(
|
|
mock_client, MagicMock(), MagicMock(), MagicMock(), MagicMock()
|
|
)
|
|
await orchestrator.poll_and_dispatch()
|
|
|
|
mock_client.notifications.list_unread_notifications.assert_called_once_with(since=None)
|
|
assert not temp_state_file.exists()
|
|
|
|
|
|
@patch("core.orchestrator.AgentOrchestrator._get_state_file_path")
|
|
@patch("core.orchestrator.AgentDispatcher")
|
|
@patch("core.orchestrator.WorkspaceManager")
|
|
@patch("core.orchestrator.NotificationReaderAgent")
|
|
async def test_poll_and_dispatch_with_notifications(
|
|
mock_notification_reader_class: MagicMock,
|
|
mock_workspace_class: MagicMock,
|
|
mock_dispatcher_class: MagicMock,
|
|
mock_get_path: MagicMock,
|
|
temp_state_file: Path
|
|
) -> None:
|
|
mock_get_path.return_value = temp_state_file
|
|
|
|
mock_reader = MagicMock()
|
|
async def mock_decide_notification(mission: str, inspection_tools: list, notification_tools) -> str:
|
|
if "issue" in mission or "42" in mission:
|
|
notification_tools.process_issue("meeks", "repo1", 42, "Needs processing")
|
|
elif "pull" in mission or "10" in mission:
|
|
notification_tools.process_pr("meeks", "repo1", 10, "Needs processing")
|
|
else:
|
|
notification_tools.skip_notification("Unrelated")
|
|
return "Decided"
|
|
mock_reader.decide_notification = AsyncMock(side_effect=mock_decide_notification)
|
|
mock_notification_reader_class.return_value = mock_reader
|
|
mock_client = MagicMock()
|
|
|
|
# Set up mock Gitea notifications
|
|
notifications = [
|
|
{
|
|
"id": 101,
|
|
"updated_at": "2026-06-30T10:00:00Z",
|
|
"subject": {
|
|
"type": "issue",
|
|
"url": "https://gitea.meeks.freeddns.org/api/v1/repos/meeks/repo1/issues/42"
|
|
},
|
|
"repository": {
|
|
"name": "repo1",
|
|
"full_name": "meeks/repo1",
|
|
"owner": {"login": "meeks"}
|
|
}
|
|
},
|
|
{
|
|
"id": 102,
|
|
"updated_at": "2026-06-30T11:00:00Z",
|
|
"subject": {
|
|
"type": "pull",
|
|
"url": "https://gitea.meeks.freeddns.org/api/v1/repos/meeks/repo1/pulls/10"
|
|
},
|
|
"repository": {
|
|
"name": "repo1",
|
|
"full_name": "meeks/repo1",
|
|
"owner": {"login": "meeks"}
|
|
}
|
|
}
|
|
]
|
|
mock_client.notifications.list_unread_notifications.return_value = notifications
|
|
|
|
# Mock issue and PR get methods on client
|
|
issue_model = IssueModel(number=42, title="Bug issue", repository=RepositoryModel(name="repo1", full_name="meeks/repo1"))
|
|
pr_model = PullRequestModel(number=10, title="Fix PR", repository=RepositoryModel(name="repo1", full_name="meeks/repo1"))
|
|
|
|
mock_client.issues.get_issue.return_value = issue_model
|
|
mock_client.prs.get_pull_request.return_value = pr_model
|
|
|
|
# Mock dispatcher and workspace path
|
|
mock_dispatcher_instance = MagicMock()
|
|
mock_dispatcher_instance.dispatch = AsyncMock(return_value=["Issue comment posted", "PR verified"])
|
|
mock_dispatcher_class.return_value = mock_dispatcher_instance
|
|
|
|
mock_workspace_instance = MagicMock()
|
|
mock_workspace_instance.get_repo_path.return_value.exists.return_value = True
|
|
mock_workspace_class.return_value = mock_workspace_instance
|
|
|
|
# Create orchestrator and poll
|
|
orchestrator = AgentOrchestrator(
|
|
mock_client, MagicMock(), MagicMock(), MagicMock(), MagicMock()
|
|
)
|
|
await orchestrator.poll_and_dispatch()
|
|
|
|
# Assert notifications were checked with None (first execution)
|
|
mock_client.notifications.list_unread_notifications.assert_called_once_with(since=None)
|
|
|
|
# Assert work was processed by dispatcher
|
|
mock_dispatcher_instance.dispatch.assert_called_once()
|
|
work_items = mock_dispatcher_instance.dispatch.call_args[0][1]
|
|
assert len(work_items) == 2
|
|
assert work_items[0].task_number == 42
|
|
assert work_items[0].notification_id == 101
|
|
assert work_items[1].task_number == 10
|
|
assert work_items[1].notification_id == 102
|
|
|
|
# Assert notifications were marked as read
|
|
mock_client.notifications.mark_notification_as_read.assert_any_call(101)
|
|
mock_client.notifications.mark_notification_as_read.assert_any_call(102)
|
|
assert mock_client.notifications.mark_notification_as_read.call_count == 2
|
|
|
|
# Assert checkpoint date was persisted
|
|
assert temp_state_file.exists()
|
|
with open(temp_state_file, "r") as f:
|
|
state = json.load(f)
|
|
# Checkpoint should match latest updated_at
|
|
assert state["last_checked"] == "2026-06-30T11:00:00Z"
|