Jason Woltje
88953fc998
Implements the Coordinator class with main orchestration loop: - Async loop architecture with configurable poll interval - process_queue() method gets next ready issue and spawns agent (stub) - Graceful shutdown handling with stop() method - Error handling that allows loop to continue after failures - Logging for all actions (start, stop, processing, errors) - Integration with QueueManager from #159 - Active agent tracking for future agent management Configuration settings added: - COORDINATOR_POLL_INTERVAL (default: 5.0s) - COORDINATOR_MAX_CONCURRENT_AGENTS (default: 10) - COORDINATOR_ENABLED (default: true) Tests: 27 new tests covering all acceptance criteria Coverage: 92% overall (100% for coordinator.py) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
123 lines
3.2 KiB
Python
123 lines
3.2 KiB
Python
"""Pytest fixtures for coordinator tests."""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def webhook_secret() -> str:
|
|
"""Return a test webhook secret."""
|
|
return "test-webhook-secret-12345"
|
|
|
|
|
|
@pytest.fixture
|
|
def gitea_url() -> str:
|
|
"""Return a test Gitea URL."""
|
|
return "https://git.mosaicstack.dev"
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_assigned_payload() -> dict[str, object]:
|
|
"""Return a sample Gitea 'assigned' issue webhook payload."""
|
|
return {
|
|
"action": "assigned",
|
|
"number": 157,
|
|
"issue": {
|
|
"id": 157,
|
|
"number": 157,
|
|
"title": "[COORD-001] Set up webhook receiver endpoint",
|
|
"state": "open",
|
|
"assignee": {
|
|
"id": 1,
|
|
"login": "mosaic",
|
|
"full_name": "Mosaic Bot",
|
|
},
|
|
},
|
|
"repository": {
|
|
"name": "stack",
|
|
"full_name": "mosaic/stack",
|
|
"owner": {"login": "mosaic"},
|
|
},
|
|
"sender": {
|
|
"id": 2,
|
|
"login": "admin",
|
|
"full_name": "Admin User",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_unassigned_payload() -> dict[str, object]:
|
|
"""Return a sample Gitea 'unassigned' issue webhook payload."""
|
|
return {
|
|
"action": "unassigned",
|
|
"number": 157,
|
|
"issue": {
|
|
"id": 157,
|
|
"number": 157,
|
|
"title": "[COORD-001] Set up webhook receiver endpoint",
|
|
"state": "open",
|
|
"assignee": None,
|
|
},
|
|
"repository": {
|
|
"name": "stack",
|
|
"full_name": "mosaic/stack",
|
|
"owner": {"login": "mosaic"},
|
|
},
|
|
"sender": {
|
|
"id": 2,
|
|
"login": "admin",
|
|
"full_name": "Admin User",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_closed_payload() -> dict[str, object]:
|
|
"""Return a sample Gitea 'closed' issue webhook payload."""
|
|
return {
|
|
"action": "closed",
|
|
"number": 157,
|
|
"issue": {
|
|
"id": 157,
|
|
"number": 157,
|
|
"title": "[COORD-001] Set up webhook receiver endpoint",
|
|
"state": "closed",
|
|
"assignee": {
|
|
"id": 1,
|
|
"login": "mosaic",
|
|
"full_name": "Mosaic Bot",
|
|
},
|
|
},
|
|
"repository": {
|
|
"name": "stack",
|
|
"full_name": "mosaic/stack",
|
|
"owner": {"login": "mosaic"},
|
|
},
|
|
"sender": {
|
|
"id": 2,
|
|
"login": "admin",
|
|
"full_name": "Admin User",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def client(webhook_secret: str, gitea_url: str, monkeypatch: pytest.MonkeyPatch) -> TestClient:
|
|
"""Create a FastAPI test client with test configuration."""
|
|
# Set test environment variables
|
|
monkeypatch.setenv("GITEA_WEBHOOK_SECRET", webhook_secret)
|
|
monkeypatch.setenv("GITEA_URL", gitea_url)
|
|
monkeypatch.setenv("ANTHROPIC_API_KEY", "test-anthropic-api-key")
|
|
monkeypatch.setenv("LOG_LEVEL", "debug")
|
|
|
|
# Force reload of settings
|
|
import importlib
|
|
|
|
from src import config
|
|
importlib.reload(config)
|
|
|
|
# Import app after settings are configured
|
|
from src.main import app
|
|
return TestClient(app)
|