140 lines
3.9 KiB
Python
140 lines
3.9 KiB
Python
"""Unit tests for timmy.vassal.orchestration_loop — VassalOrchestrator."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from timmy.vassal.orchestration_loop import VassalCycleRecord, VassalOrchestrator
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VassalCycleRecord
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cycle_record_healthy_when_no_errors():
|
|
r = VassalCycleRecord(
|
|
cycle_id=1,
|
|
started_at="2026-01-01T00:00:00+00:00",
|
|
)
|
|
assert r.healthy is True
|
|
|
|
|
|
def test_cycle_record_unhealthy_with_errors():
|
|
r = VassalCycleRecord(
|
|
cycle_id=1,
|
|
started_at="2026-01-01T00:00:00+00:00",
|
|
errors=["backlog: connection refused"],
|
|
)
|
|
assert r.healthy is False
|
|
|
|
|
|
def test_cycle_record_unhealthy_with_warnings():
|
|
r = VassalCycleRecord(
|
|
cycle_id=1,
|
|
started_at="2026-01-01T00:00:00+00:00",
|
|
house_warnings=["disk 90% full"],
|
|
)
|
|
assert r.healthy is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VassalOrchestrator state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_orchestrator_initial_state():
|
|
orch = VassalOrchestrator()
|
|
assert orch.cycle_count == 0
|
|
assert orch.is_running is False
|
|
assert orch.history == []
|
|
|
|
|
|
def test_orchestrator_get_status_no_cycles():
|
|
orch = VassalOrchestrator()
|
|
status = orch.get_status()
|
|
assert status["running"] is False
|
|
assert status["cycle_count"] == 0
|
|
assert status["last_cycle"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_cycle — integration (no Gitea, no Ollama in test env)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_cycle_completes_without_services():
|
|
"""run_cycle must complete and record even when external services are down."""
|
|
from timmy.vassal.dispatch import clear_dispatch_registry
|
|
|
|
clear_dispatch_registry()
|
|
orch = VassalOrchestrator(cycle_interval=300)
|
|
|
|
record = await orch.run_cycle()
|
|
|
|
assert isinstance(record, VassalCycleRecord)
|
|
assert record.cycle_id == 1
|
|
assert record.finished_at # was set
|
|
assert record.duration_ms >= 0
|
|
# No Gitea → fetched = 0, dispatched = 0
|
|
assert record.issues_fetched == 0
|
|
assert record.issues_dispatched == 0
|
|
# History updated
|
|
assert len(orch.history) == 1
|
|
assert orch.cycle_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_cycle_increments_cycle_count():
|
|
from timmy.vassal.dispatch import clear_dispatch_registry
|
|
|
|
clear_dispatch_registry()
|
|
orch = VassalOrchestrator()
|
|
|
|
await orch.run_cycle()
|
|
await orch.run_cycle()
|
|
|
|
assert orch.cycle_count == 2
|
|
assert len(orch.history) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_after_cycle():
|
|
from timmy.vassal.dispatch import clear_dispatch_registry
|
|
|
|
clear_dispatch_registry()
|
|
orch = VassalOrchestrator()
|
|
|
|
await orch.run_cycle()
|
|
status = orch.get_status()
|
|
|
|
assert status["cycle_count"] == 1
|
|
last = status["last_cycle"]
|
|
assert last is not None
|
|
assert last["cycle_id"] == 1
|
|
assert last["issues_fetched"] == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# start / stop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_orchestrator_stop_when_not_running():
|
|
"""stop() on an idle orchestrator must not raise."""
|
|
orch = VassalOrchestrator()
|
|
orch.stop() # should be a no-op
|
|
assert orch.is_running is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module-level singleton
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_module_singleton_exists():
|
|
from timmy.vassal import vassal_orchestrator, VassalOrchestrator
|
|
|
|
assert isinstance(vassal_orchestrator, VassalOrchestrator)
|