"""Durable last-run state. Job outcomes lived only in an in-memory dict, so every deploy wiped them and Admin → Jobs could only report "Active" with no indication a job had ever run. """ from datetime import datetime, timezone import pytest from sqlalchemy import select from app import scheduler as sched from app.models.job_run_state import JobRunState from app.services import job_run_store from tests.conftest import _test_session_factory @pytest.fixture def session_factory(): """A real, independently committing session. _persist_job_run opens its own session and commits, which is what production does; the shared db_session fixture holds an outer transaction that a commit would tear down. """ return _test_session_factory async def _rows(session) -> dict[str, JobRunState]: result = await session.execute(select(JobRunState)) return {row.job_name: row for row in result.scalars().all()} async def _committed_rows() -> dict[str, JobRunState]: async with _test_session_factory() as session: return await _rows(session) class TestRecordFinish: async def test_inserts_then_updates_one_row_per_job(self, db_session): await job_run_store.record_finish( db_session, "rr_scanner", {"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00", "processed": 5, "total": 5}, ) await db_session.flush() await job_run_store.record_finish( db_session, "rr_scanner", {"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"}, ) await db_session.flush() rows = await _rows(db_session) assert list(rows) == ["rr_scanner"], "upsert, not append-only history" assert rows["rr_scanner"].status == "error" assert rows["rr_scanner"].message == "boom" async def test_missing_finish_time_falls_back_to_now(self, db_session): await job_run_store.record_finish(db_session, "alerts", {"status": "completed"}) await db_session.flush() assert (await _rows(db_session))["alerts"].finished_at is not None class TestPipelinePersistence: """_run_pipeline is the only write path for steps: they are plain coroutine calls, so they emit no scheduler events for the listener to catch.""" @pytest.fixture(autouse=True) def _enabled(self, monkeypatch, session_factory): async def enabled(db, job_name): return True monkeypatch.setattr("app.scheduler.async_session_factory", session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) async def test_persists_both_the_step_and_the_orchestrator(self, monkeypatch): async def ok_step(): sched._runtime_finish("rr_scanner", "completed", processed=3, total=3) monkeypatch.setattr(sched, "ok_step", ok_step, raising=False) await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "ok_step")]) rows = await _committed_rows() assert rows["rr_scanner"].status == "completed" assert rows["near_close_pipeline"].status == "completed" async def test_a_failing_step_still_records_its_error(self, monkeypatch): """The persist sits after the except that swallows step errors — inside it, exactly the runs worth seeing would be skipped.""" async def boom(): sched._runtime_finish("rr_scanner", "error", processed=0, total=1, message="kaboom") raise RuntimeError("kaboom") monkeypatch.setattr(sched, "boom", boom, raising=False) await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "boom")]) rows = await _committed_rows() assert rows["rr_scanner"].status == "error" assert rows["rr_scanner"].message == "kaboom" # The pipeline itself survives a failing step. assert rows["near_close_pipeline"].status == "completed" async def test_disabled_pipeline_records_skipped(self, monkeypatch): async def disabled(db, job_name): return False monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) await sched._run_pipeline("daily_pipeline", []) assert (await _committed_rows())["daily_pipeline"].status == "skipped" async def test_persistence_failure_never_breaks_the_pipeline(self, monkeypatch): calls: list[str] = [] async def exploding_record(db, job_name, runtime): calls.append(job_name) raise RuntimeError("db down") async def ok_step(): sched._runtime_finish("rr_scanner", "completed", processed=1, total=1) monkeypatch.setattr(sched.job_run_store, "record_finish", exploding_record) monkeypatch.setattr(sched, "ok_step", ok_step, raising=False) await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "ok_step")]) assert calls, "persistence was attempted" assert sched.get_job_runtime_snapshot("near_close_pipeline")["status"] == "completed" async def test_a_job_that_never_finished_writes_nothing(self): sched._runtime_start("event_study", total=1) await sched._persist_job_run("event_study") assert "event_study" not in await _committed_rows() class TestListJobsSplitsLiveFromPersisted: async def test_a_stale_error_does_not_pin_the_status_chip(self, db_session): """runtime_* must stay live-only: the chip and the rate-limit banner read it, so a week-old error there would read as the current state forever.""" from app.scheduler import configure_scheduler, scheduler from app.services.admin_service import list_jobs scheduler.remove_all_jobs() configure_scheduler() # _job_runtime is module-global and survives across tests; pin the live # row to idle so the assertion is about the split, not about ordering. sched._job_runtime["rr_scanner"] = sched._idle_runtime() await job_run_store.record_finish( db_session, "rr_scanner", { "status": "error", "finished_at": datetime(2026, 8, 1, tzinfo=timezone.utc).isoformat(), "message": "old failure", }, ) await db_session.flush() job = {j["name"]: j for j in await list_jobs(db_session)}["rr_scanner"] assert job["last_run_status"] == "error" assert job["last_run_message"] == "old failure" assert job["runtime_status"] == "idle" assert job["running"] is False scheduler.remove_all_jobs()