generated from john/python-template
76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
"""Tests for maintenance run persistence and execution lifecycle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.models import MaintenanceJobType
|
|
from transcription.db.models import MaintenanceRunStatus
|
|
from transcription.services.maintenance import MaintenanceExecution
|
|
from transcription.services.maintenance import MaintenanceService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_and_list_runs(default_session_factory, default_settings):
|
|
service = MaintenanceService(session_factory=default_session_factory, settings=default_settings)
|
|
|
|
first = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test")
|
|
second = await service.enqueue_run(job_type=MaintenanceJobType.STORAGE_RECONCILIATION, triggered_by="test")
|
|
runs = await service.list_runs(limit=10)
|
|
|
|
assert len(runs) == 2
|
|
assert runs[0].id == second.id
|
|
assert runs[1].id == first.id
|
|
assert runs[0].status == MaintenanceRunStatus.QUEUED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_next_queued_run_persists_terminal_result(
|
|
default_session_factory,
|
|
default_settings,
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
settings = default_settings.model_copy(update={"log_dir": tmp_path / "logs"})
|
|
settings = Settings.model_validate(settings.model_dump())
|
|
service = MaintenanceService(session_factory=default_session_factory, settings=settings)
|
|
queued = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test")
|
|
|
|
async def _fake_execute(_run):
|
|
return MaintenanceExecution(
|
|
status=MaintenanceRunStatus.SUCCEEDED,
|
|
summary="Synthetic success",
|
|
output="stdout line\nstderr line",
|
|
)
|
|
|
|
monkeypatch.setattr(service, "_execute_run", _fake_execute)
|
|
|
|
processed = await service.process_next_queued_run()
|
|
assert processed is True
|
|
|
|
runs = await service.list_runs(limit=10)
|
|
updated = next(run for run in runs if run.id == queued.id)
|
|
assert updated.status == MaintenanceRunStatus.SUCCEEDED
|
|
assert updated.summary == "Synthetic success"
|
|
assert updated.log_path is not None
|
|
assert (settings.log_dir / Path(updated.log_path)).is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_recreates_missing_table(default_session_factory, default_settings):
|
|
service = MaintenanceService(session_factory=default_session_factory, settings=default_settings)
|
|
async with default_session_factory() as session:
|
|
await session.exec(text("DROP TABLE maintenance_run"))
|
|
await session.commit()
|
|
|
|
runs = await service.list_runs(limit=10)
|
|
assert runs == []
|
|
|
|
created = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test")
|
|
listed = await service.list_runs(limit=10)
|
|
assert any(run.id == created.id for run in listed)
|