- The scheduler keeps one registry of the runs this process has open, for the interval jobs and the ad-hoc runs alike. A row the table still says is `running` but no run here has open (its outcome failed to write) reads as interrupted, as the next start's sweep will close it: no pulsing dot beside "No run recorded yet". - Migration 0109's downgrade sums a still-running run's quota events into its spend before the link to them is dropped, as the startup sweep would. - A test holds that a dialog closing onto the same address leaves an open log line open.
166 lines
6.1 KiB
Python
166 lines
6.1 KiB
Python
"""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
|