"""scheduled_jobs UUID-seed guard. Hard CI invariant preventing a FOURTH recurrence of the no-dash ``scheduled_jobs.id`` defect that migrations 059 / 162 / 248 each had to repair: a seed migration that writes the primary key as ``REPLACE(UUID(), '-', '')`` produces a 32-char no-dash id in the CHAR(36) column. The ``GUID`` TypeDecorator then normalises it to the dashed form on read, so the worker schedules APScheduler with the dashed string but the run-wrapper's ``session.get(ScheduledJob, uuid.UUID(...))`` queries the dashed value, never matches the stored no-dash row, and the cron silently never runs (``run_count`` stays 0 forever). The fix is always the same: seed ``scheduled_jobs.id`` with a plain dashed ``UUID()``. This guard rejects any NEW migration that reintroduces the ``REPLACE(UUID(), '-', '')`` pattern anywhere near a ``scheduled_jobs`` INSERT, turning a thrice-recurring production incident class into a hard stop at PR time. Runs without a database connection — pure text scan of the version files. Exits 0 on success, 1 with a human-readable diagnostic on failure. """ from __future__ import annotations import re import sys from pathlib import Path # The exact offending pattern (whitespace-tolerant). We only care about # the no-dash form: ``REPLACE(UUID(), '-', '')``. The dashed no-op # ``REPLACE(UUID(), '-', '-')`` and plain ``UUID()`` are fine. _BAD_REPLACE = re.compile(r"REPLACE\(\s*UUID\(\)\s*,\s*'-'\s*,\s*''\s*\)") # Migrations that pre-date this guard and legitimately carry the pattern # for NON-scheduled_jobs tables (auth_roles / email-group seeds, etc.), # or are the historical fix migrations whose DOCSTRINGS quote the pattern. # These are grandfathered: the guard's job is to stop NEW scheduled_jobs # seeds, not to rewrite history. The eight scheduled_jobs offenders # (041/064/114/115/134/135/155) were corrected in the 248 PR and are NOT # on this list, so a regression in any of them re-trips the guard. _GRANDFATHERED = { "001_initial_mysql.py", "040_drop_cloud_admin_role.py", "043_hr_finance_roles_and_email_groups.py", "092_normalize_auth_role_ids.py", "059_fix_scheduled_jobs_id_dashes.py", # docstring quote only "162_fix_scheduled_jobs_id_dashes_2.py", # docstring quote only "248_fix_scheduled_jobs_id_dashes_3.py", # docstring quote only } def _versions_dir() -> Path: return Path(__file__).resolve().parent.parent / "alembic" / "versions" def main() -> int: versions = _versions_dir() if not versions.is_dir(): print(f"[scheduled-jobs-guard] versions dir not found: {versions}") return 1 offenders: list[str] = [] for path in sorted(versions.glob("*.py")): if path.name in _GRANDFATHERED: continue text = path.read_text(encoding="utf-8") if not _BAD_REPLACE.search(text): continue # Only flag when the pattern co-occurs with a scheduled_jobs INSERT # in the same file — the defect is specific to that table's id. if "scheduled_jobs" not in text: continue # Drive the gating decision off the SAME whole-text matcher as the # pre-filter (line numbers via finditer). A per-line re-scan would # miss a match split across physical lines (``\s`` spans newlines), # silently passing the guard — so iterate the full text instead. for m in _BAD_REPLACE.finditer(text): line_no = text.count("\n", 0, m.start()) + 1 offenders.append(f" {path.name}:{line_no}: {m.group(0)}") if offenders: print( "[scheduled-jobs-guard] FAIL — REPLACE(UUID(), '-', '') found in a " "migration that touches scheduled_jobs. Seed scheduled_jobs.id with " "a plain dashed UUID() instead; the no-dash 32-char form strands the " "cron (see migration 248 docstring).\n" + "\n".join(offenders) ) return 1 print("[scheduled-jobs-guard] OK — no no-dash UUID seeds in scheduled_jobs.") return 0 if __name__ == "__main__": sys.exit(main())