828038ec02
Expand test_run_service.py from a single stub to a full test suite using an in-memory SQLite database (same pattern as test_blueprint_service.py). New test classes cover: - TestCreateRun: verifies field storage, blueprint_id, execution_mode, created_at - TestGetRun: existing run retrieval and None for unknown ID - TestListRuns: empty list, full listing, limit and offset pagination - TestUpdateRun: status changes, auto-setting completed_at on terminal statuses (completed/failed), not setting it for non-terminal statuses, storing final_draft/critic_score/iteration_count, and silently ignoring unknown field keys via the hasattr guard All tests use isolated fixtures that create and drop tables per test, ensuring no state leaks between cases. https://claude.ai/code/session_01MLq4Vjq9XWKPtu3iX3iqL3
312 lines
10 KiB
Python
312 lines
10 KiB
Python
"""
|
|
Tests for the run history service and CouncilRun model.
|
|
|
|
DB-backed tests use an in-memory SQLite database for full isolation —
|
|
the same approach as test_blueprint_service.py.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from models.blueprint import Base # CouncilRun shares this Base
|
|
from services.run_service import create_run, get_run, list_runs, update_run
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# In-memory SQLite test database
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
|
|
TestSessionLocal = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def session():
|
|
"""Create all tables and yield a fresh session. Tables are dropped after each test."""
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
async with TestSessionLocal() as sess:
|
|
yield sess
|
|
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CouncilRun model serialization (no DB required)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCouncilRunModel:
|
|
"""Tests for the CouncilRun SQLAlchemy model."""
|
|
|
|
def test_to_dict_serialization(self):
|
|
from models.council_run import CouncilRun
|
|
from datetime import datetime, timezone
|
|
|
|
run = CouncilRun(
|
|
id="test-id",
|
|
blueprint_id="bp-id",
|
|
input_topic="Test topic",
|
|
status="completed",
|
|
execution_mode="auto-pilot",
|
|
final_draft="Final text",
|
|
critic_score=8.5,
|
|
iteration_count=3,
|
|
active_node="done",
|
|
error=None,
|
|
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
|
completed_at=datetime(2026, 1, 1, 0, 5, tzinfo=timezone.utc),
|
|
)
|
|
|
|
d = run.to_dict()
|
|
assert d["id"] == "test-id"
|
|
assert d["blueprint_id"] == "bp-id"
|
|
assert d["status"] == "completed"
|
|
assert d["critic_score"] == 8.5
|
|
assert d["iteration_count"] == 3
|
|
assert d["created_at"] is not None
|
|
assert d["completed_at"] is not None
|
|
|
|
def test_to_dict_with_none_timestamps(self):
|
|
from models.council_run import CouncilRun
|
|
|
|
run = CouncilRun(
|
|
id="test-id",
|
|
input_topic="Test",
|
|
status="pending",
|
|
execution_mode="god-mode",
|
|
created_at=None,
|
|
completed_at=None,
|
|
)
|
|
|
|
d = run.to_dict()
|
|
assert d["created_at"] is None
|
|
assert d["completed_at"] is None
|
|
assert d["execution_mode"] == "god-mode"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCreateRun:
|
|
"""Tests for the create_run service function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_stores_correct_fields(self, session):
|
|
run = await create_run(session, run_id="run-001", input_topic="Test topic")
|
|
|
|
assert run.id == "run-001"
|
|
assert run.input_topic == "Test topic"
|
|
assert run.status == "pending"
|
|
assert run.execution_mode == "auto-pilot"
|
|
assert run.blueprint_id is None
|
|
assert run.completed_at is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_with_blueprint_id(self, session):
|
|
run = await create_run(
|
|
session,
|
|
run_id="run-002",
|
|
input_topic="Topic",
|
|
blueprint_id="bp-abc",
|
|
)
|
|
|
|
assert run.blueprint_id == "bp-abc"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_with_god_mode(self, session):
|
|
run = await create_run(
|
|
session,
|
|
run_id="run-003",
|
|
input_topic="Topic",
|
|
execution_mode="god-mode",
|
|
)
|
|
|
|
assert run.execution_mode == "god-mode"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_sets_created_at(self, session):
|
|
run = await create_run(session, run_id="run-004", input_topic="Topic")
|
|
|
|
assert run.created_at is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetRun:
|
|
"""Tests for the get_run service function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_run_returns_existing_run(self, session):
|
|
await create_run(session, run_id="run-100", input_topic="Get test")
|
|
fetched = await get_run(session, "run-100")
|
|
|
|
assert fetched is not None
|
|
assert fetched.id == "run-100"
|
|
assert fetched.input_topic == "Get test"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_run_returns_none_for_unknown_id(self, session):
|
|
result = await get_run(session, "nonexistent-id")
|
|
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_runs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListRuns:
|
|
"""Tests for the list_runs service function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_empty(self, session):
|
|
runs = await list_runs(session)
|
|
|
|
assert runs == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_returns_all(self, session):
|
|
await create_run(session, run_id="run-a", input_topic="Topic A")
|
|
await create_run(session, run_id="run-b", input_topic="Topic B")
|
|
runs = await list_runs(session)
|
|
|
|
assert len(runs) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_limit(self, session):
|
|
for i in range(5):
|
|
await create_run(session, run_id=f"run-{i}", input_topic=f"Topic {i}")
|
|
|
|
runs = await list_runs(session, limit=3)
|
|
|
|
assert len(runs) == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_offset(self, session):
|
|
for i in range(4):
|
|
await create_run(session, run_id=f"run-{i}", input_topic=f"Topic {i}")
|
|
|
|
runs_all = await list_runs(session)
|
|
runs_offset = await list_runs(session, offset=2)
|
|
|
|
# The offset skips the first 2 runs (most recent first)
|
|
assert len(runs_offset) == 2
|
|
assert runs_offset[0].id == runs_all[2].id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update_run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUpdateRun:
|
|
"""Tests for the update_run service function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_returns_none_for_unknown_id(self, session):
|
|
result = await update_run(session, "ghost-id", {"status": "running"})
|
|
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_changes_status(self, session):
|
|
await create_run(session, run_id="run-u1", input_topic="Topic")
|
|
updated = await update_run(session, "run-u1", {"status": "running"})
|
|
|
|
assert updated is not None
|
|
assert updated.status == "running"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_sets_completed_at_when_completed(self, session):
|
|
await create_run(session, run_id="run-u2", input_topic="Topic")
|
|
updated = await update_run(session, "run-u2", {"status": "completed"})
|
|
|
|
assert updated is not None
|
|
assert updated.status == "completed"
|
|
assert updated.completed_at is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_sets_completed_at_when_failed(self, session):
|
|
await create_run(session, run_id="run-u3", input_topic="Topic")
|
|
updated = await update_run(session, "run-u3", {"status": "failed", "error": "Timeout"})
|
|
|
|
assert updated is not None
|
|
assert updated.status == "failed"
|
|
assert updated.completed_at is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_does_not_set_completed_at_for_running(self, session):
|
|
await create_run(session, run_id="run-u4", input_topic="Topic")
|
|
updated = await update_run(session, "run-u4", {"status": "running"})
|
|
|
|
assert updated is not None
|
|
assert updated.completed_at is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_stores_final_draft_and_score(self, session):
|
|
await create_run(session, run_id="run-u5", input_topic="Topic")
|
|
updated = await update_run(
|
|
session,
|
|
"run-u5",
|
|
{
|
|
"status": "completed",
|
|
"final_draft": "Polished text",
|
|
"critic_score": 9.0,
|
|
"iteration_count": 2,
|
|
},
|
|
)
|
|
|
|
assert updated.final_draft == "Polished text"
|
|
assert updated.critic_score == 9.0
|
|
assert updated.iteration_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_run_ignores_unknown_fields(self, session):
|
|
"""Unknown keys in the updates dict are silently skipped via hasattr check."""
|
|
await create_run(session, run_id="run-u6", input_topic="Topic")
|
|
# "nonexistent_field" does not exist on CouncilRun — should not raise
|
|
updated = await update_run(
|
|
session, "run-u6", {"status": "running", "nonexistent_field": "value"}
|
|
)
|
|
|
|
assert updated is not None
|
|
assert updated.status == "running"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run history routes (lightweight — service is mocked)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRunHistoryRoutes:
|
|
"""Tests for the run history API routes."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_empty(self):
|
|
"""List runs returns empty list when no runs exist."""
|
|
from api.run_history_routes import list_all_runs
|
|
|
|
mock_session = AsyncMock()
|
|
|
|
mock_result = MagicMock()
|
|
mock_scalars = MagicMock()
|
|
mock_scalars.all.return_value = []
|
|
mock_result.scalars.return_value = mock_scalars
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
with patch("services.run_service.list_runs") as mock_list:
|
|
mock_list.return_value = []
|
|
result = await list_all_runs(limit=50, offset=0, session=mock_session)
|
|
assert result == []
|