"""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() class TestConcurrentWrites: """Pipelines are separate scheduler jobs that can overlap, and they share step ids — data_collector belongs to all four.""" async def test_interleaved_first_writes_do_not_collide(self): """Both sessions SELECT before either INSERTs: select-then-insert lost this race with an IntegrityError, and the caller swallows it.""" async with _test_session_factory() as a, _test_session_factory() as b: await job_run_store.record_finish( a, "data_collector", {"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00"}, ) await job_run_store.record_finish( b, "data_collector", {"status": "completed", "finished_at": "2026-08-08T10:00:01+00:00"}, ) await a.commit() await b.commit() # must not raise rows = await _committed_rows() assert rows["data_collector"].status == "completed" async def test_an_older_finish_never_rewinds_the_row(self): """A slower pipeline finishing an older run last must not overwrite a newer outcome with a stale one.""" async with _test_session_factory() as s: await job_run_store.record_finish( s, "alerts", {"status": "completed", "finished_at": "2026-08-08T12:00:00+00:00"}, ) await s.commit() await job_run_store.record_finish( s, "alerts", {"status": "error", "finished_at": "2026-08-08T09:00:00+00:00", "message": "stale"}, ) await s.commit() row = (await _committed_rows())["alerts"] assert row.finished_at.isoformat().startswith("2026-08-08T12:00") assert row.status == "completed" assert row.message is None async def test_a_newer_finish_still_wins(self): async with _test_session_factory() as s: await job_run_store.record_finish( s, "rr_scanner", {"status": "completed", "finished_at": "2026-08-08T09:00:00+00:00"}, ) await s.commit() await job_run_store.record_finish( s, "rr_scanner", {"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"}, ) await s.commit() row = (await _committed_rows())["rr_scanner"] assert row.status == "error" assert row.message == "boom" class TestShutdownDrain: async def test_drains_a_task_queued_after_the_flush_starts(self): """scheduler.shutdown(wait=False) returns before APScheduler dispatches its completion events, so writes can appear mid-drain. Snapshotting the task set once would miss them and dispose the engine underneath.""" import asyncio done: list[str] = [] async def slow_first(): await asyncio.sleep(0.02) done.append("first") # Queued only once the first write is already finishing. sched._persist_tasks.add(asyncio.get_running_loop().create_task(late())) async def late(): await asyncio.sleep(0.02) done.append("late") sched._persist_tasks.clear() sched._persist_tasks.add(asyncio.get_running_loop().create_task(slow_first())) await sched.flush_job_run_persists(timeout=2.0) assert done == ["first", "late"] sched._persist_tasks.clear() async def test_returns_promptly_when_there_is_nothing_to_drain(self): sched._persist_tasks.clear() await sched.flush_job_run_persists(timeout=2.0) async def test_gives_up_rather_than_hanging_shutdown(self): import asyncio async def never(): await asyncio.sleep(30) sched._persist_tasks.clear() task = asyncio.get_running_loop().create_task(never()) sched._persist_tasks.add(task) await sched.flush_job_run_persists(timeout=0.15) # returns, does not hang task.cancel() sched._persist_tasks.clear()