"""Migration 0109's data step, DB-lane: the audit log's run rows move into `job_runs` with their outcome intact, and leave the audit log; an admin action stays where it was. The downgrade moves them back.""" import json from datetime import timedelta from alembic import command from alembic.config import Config from sqlalchemy import text from app.db import engine def _audit(conn, actor, action, target, after): conn.execute( text( "INSERT INTO audit_log (actor_id, action, target_type, target_id, summary, after," " created_at) VALUES (:a, :act, 'job', :t, 's', CAST(:after AS json)," " CAST('2026-10-01T03:00:00+00:00' AS timestamptz))" ), {"a": actor, "act": action, "t": target, "after": json.dumps(after)}, ) def test_the_audit_logs_runs_move_into_the_run_table_and_back(db, make_user): uid = make_user().id cfg = Config("alembic.ini") command.downgrade(cfg, "0108_folder_links") try: with engine.begin() as conn: _audit( conn, None, "job_run", "rss_poll", {"status": "ok", "summary": "new=3", "result": {"new": 3}}, ) _audit( conn, uid, "job_run", "rss_poll", {"status": "error", "summary": "boom", "result": {"failure": "rss_all_failed"}}, ) _audit( conn, uid, "job_run", "backfill", {"status": "skipped", "summary": "sync paused"} ) _audit(conn, uid, "config_set", None, {"key": "x"}) finally: command.upgrade(cfg, "head") with engine.connect() as conn: runs = conn.execute( text( "SELECT job, trigger, actor_id, status, summary, result, error, failure," " to_char(started_at AT TIME ZONE 'UTC', 'MM-DD HH24'), finished_at = started_at" " FROM job_runs ORDER BY id" ) ).all() left = conn.execute(text("SELECT action FROM audit_log")).scalars().all() assert [tuple(r) for r in runs] == [ ("rss_poll", "scheduled", None, "ok", "new=3", {"new": 3}, None, None, "10-01 03", True), ( "rss_poll", "manual", uid, "error", None, None, "boom", {"failure": "rss_all_failed"}, "10-01 03", True, ), ("backfill", "manual", uid, "skipped", "sync paused", None, None, None, "10-01 03", True), ] assert left == ["config_set"] command.downgrade(cfg, "0108_folder_links") try: with engine.connect() as conn: back = conn.execute( text( "SELECT actor_id, target_id, after::jsonb FROM audit_log" " WHERE action = 'job_run' ORDER BY id" ) ).all() finally: command.upgrade(cfg, "head") # What the audit row has no column for rides along, so the next upgrade restores it. def run(trigger: str) -> dict: return {"trigger": trigger, "quota_units": 0, "started_at": "2026-10-01T03:00:00+00:00"} assert [tuple(r) for r in back] == [ ( None, "rss_poll", {"status": "ok", "summary": "new=3", "result": {"new": 3}, **run("scheduled")}, ), ( uid, "rss_poll", { "status": "error", "summary": "boom", "result": {"failure": "rss_all_failed"}, **run("manual"), }, ), (uid, "backfill", {"status": "skipped", "summary": "sync paused", **run("manual")}), ] def test_a_down_and_up_round_trip_keeps_every_run(db, make_user): """A rollback then a redeploy: an interrupted run, a run caught mid-way and a user's run of one channel come back as runs — not stranded in the audit log under an action no code knows.""" uid = make_user().id with engine.begin() as conn: conn.execute( text( "INSERT INTO job_runs (job, trigger, actor_id, target, status, started_at," " finished_at, quota_units) VALUES" " ('backfill', 'scheduled', NULL, NULL, 'interrupted', now(), now(), 0)," " ('maintenance', 'manual', :u, NULL, 'running', now(), NULL, 0)," " ('channel_backfill', 'ad_hoc', :u, 'UC_x', 'ok', now(), now(), 0)," # Its admin is gone (SET NULL): only the row itself still says it was manual. " ('enrich', 'manual', NULL, NULL, 'ok'," " '2026-10-01 10:00+00', '2026-10-01 10:30+00', 42)" ), {"u": uid}, ) # What the run caught mid-way had spent: only its quota rows know it yet. conn.execute( text( "INSERT INTO quota_events (action, units, job_run_id)" " SELECT 'videos.list', 7, id FROM job_runs WHERE job = 'maintenance'" ) ) cfg = Config("alembic.ini") command.downgrade(cfg, "0108_folder_links") command.upgrade(cfg, "head") with engine.connect() as conn: runs = conn.execute( text("SELECT job, trigger, actor_id, target, status FROM job_runs ORDER BY id") ).all() kept = conn.execute( text("SELECT quota_units, finished_at - started_at FROM job_runs WHERE job = 'enrich'") ).one() stopped = conn.execute( text("SELECT quota_units FROM job_runs WHERE job = 'maintenance'") ).scalar() stranded = conn.execute( text("SELECT count(*) FROM audit_log WHERE action = 'job_run'") ).scalar() assert [tuple(r) for r in runs] == [ ("backfill", "scheduled", None, None, "interrupted"), ("maintenance", "manual", uid, None, "interrupted"), # the rollback stopped it ("channel_backfill", "ad_hoc", uid, "UC_x", "ok"), ("enrich", "manual", None, None, "ok"), ] assert tuple(kept) == (42, timedelta(minutes=30)) assert stopped == 7 assert stranded == 0