app-catalog-felhom.eu/scripts/check-test-record-move.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""check-test-record-move.py — catalog gate: an `image:` move must bring its own PROVEN test record.

    python3 scripts/check-test-record-move.py                        # diff origin/main..HEAD
    python3 scripts/check-test-record-move.py --range <A>..<B>       # what .githooks/pre-push passes
    python3 scripts/check-test-record-move.py --no-network ...       # skip the registry comparison
                                                                     # (tests only; says so)
    python3 scripts/check-test-record-move.py --digests-from F.json  # the "registry" is this file
                                                                     # {ref: digest} (decoy tests; says so)

`09-update-architecture.md` §3 decision 13: the catalog holds only tested steps, and an image move
with no test record is refused HERE, at push time. Night 2026-09-23 (§6.4 part 4).

For every template whose per-service images differ between A and B, the `.felhom.yml` at B must
carry, in an `update_ladder:` line that was NOT there at A, an entry that
  - is well-formed (ladder.check_entry) and NOT `backfilled` (a backfill cites an old record; a new
    move needs a new test),
  - has `verdict: "proven"`,
  - has `from` equal to the images at A and `to` equal to the images at B, service by service,
  - carries digests that the registry STILL serves for those refs right now (decision 17). A digest
    that moved since the test means the image that was tested is not the image a box would pull;
  - and, when the entry is marked `memory_tight`, the same range changes that app's memory limit
    (`09` RomM follow-up: a version move re-checks the limit — this is that check as a gate).

THE NETWORK, and why this "fast" gate uses it. The hook runs `--fast` gates only, and until tonight
fast meant "no network". This gate resolves ONLY the refs of templates whose images moved in the
range — zero requests on a push that moves nothing, a handful on a move. A registry that cannot be
asked is INCONCLUSIVE (exit 2), which the runner reports as never-a-pass: a move is refused until
the digest has been compared. Decided by CC unattended 2026-09-23 — operator may reverse.

SHALLOW CLONES: needs two commits, so on CI's `--depth 1` clone it is skipped out loud by
catalog_gates.py; its static twin `check-test-record.py` still runs there and catches a compose
that no longer matches its ladder's head. Exit 0 clean · 1 convicted · 2 inconclusive.
"""
import os
import re
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import ladder  # noqa: E402

TEMPLATE_RE = re.compile(r"^templates/([^/]+)/docker-compose\.ya?ml$")
ZERO_SHA_RE = re.compile(r"^0{40}$")
MEM_RE = re.compile(r"^\s+(memory|mem_limit):", re.M)


def git(*args):
    p = subprocess.run(["git"] + list(args), capture_output=True, text=True)
    return p.returncode, p.stdout, p.stderr


def resolve_range(spec):
    if not spec or ".." not in spec:
        return None, None, "range must be <A>..<B> (got %r)" % spec
    a, b = spec.split("..", 1)
    if ZERO_SHA_RE.match(a):
        a = "origin/main"
    for r in (a, b):
        rc, _, err = git("rev-parse", "--verify", "-q", r + "^{commit}")
        if rc != 0:
            return None, None, "cannot resolve %r (%s)" % (r, err.strip() or "not a commit")
    return a, b, ""


def show(rev, path):
    rc, out, _ = git("show", "%s:%s" % (rev, path))
    return out if rc == 0 else None


def mem_lines(text):
    return sorted(l.strip() for l in (text or "").splitlines() if MEM_RE.match(l))


def default_resolver(ref):
    import image_digest
    return image_digest.resolve(ref)


def judge_app(app, a, b, resolver, network=True):
    """(problems, inconclusive) for one template whose compose changed in A..B."""
    cpath, fpath = "templates/%s/docker-compose.yml" % app, "templates/%s/.felhom.yml" % app
    before_c, after_c = show(a, cpath), show(b, cpath)
    if after_c is None:
        return [], []  # removed at B: nothing is offered
    before = ladder.images_in(before_c) if before_c is not None else {}
    after = ladder.images_in(after_c)
    if before == after:
        return [], []  # the compose changed, but no image moved
    if before_c is None:
        return [], []  # a NEW template: its first images are not a move (the app is tested before it is listed)
    new_entries = []
    _e_a, raws_a, _ = ladder.parse(show(a, fpath) or "")
    ents_b, raws_b, errs_b = ladder.parse(show(b, fpath) or "")
    problems, inconclusive = [], []
    for err in errs_b:
        problems.append(err)
    old = set(raws_a)
    for e, raw in zip(ents_b, raws_b):
        if raw not in old:
            new_entries.append(e)
    moved = sorted(s for s in after if before.get(s) != after[s])
    if not new_entries:
        problems.append("images moved (%s) but this range adds NO update_ladder entry — an image move "
                        "needs its test record (decision 13); run the harness and let it write the entry"
                        % ", ".join("%s: %s -> %s" % (s, before.get(s), after[s]) for s in moved))
        return problems, inconclusive
    match = [e for e in new_entries if e.get("to") == after]
    if not match:
        problems.append("the new ladder entry names other refs than the compose now carries: compose %s"
                        % after)
        return problems, inconclusive
    e = match[-1]
    for p in ladder.check_entry(e):
        problems.append("new entry: " + p)
    if e.get("backfilled") is not None:
        problems.append("new entry is marked backfilled — a new move needs a new test, not an old record")
    if e.get("verdict") != "proven":
        problems.append("new entry's verdict is %r — only a PROVEN step may be published" % e.get("verdict"))
    if e.get("from") != before:
        problems.append("new entry's `from` %s is not the compose before the move %s" % (e.get("from"), before))
    if (e.get("marks") or {}).get("memory_tight"):
        if mem_lines(before_c) == mem_lines(after_c) and mem_lines(show(a, fpath)) == mem_lines(show(b, fpath)):
            problems.append("the entry is memory_tight (peak %s%%) and this range does not change the memory "
                            "limit — raise it and re-run the memory watch (RomM follow-up)" % e.get("memory_peak_pct"))
    if problems:
        return problems, inconclusive
    if not network:
        print("  %s: digest comparison SKIPPED (--no-network) — not a pass for a real push" % app)
        return problems, inconclusive
    for svc, ref in sorted(after.items()):
        want = (e.get("digest") or {}).get(svc)
        got, why = resolver(ref)
        if got is None:
            inconclusive.append("%s %s: the registry could not be asked (%s)" % (svc, ref, why))
        elif got != want:
            problems.append("%s %s: the registry serves %s, the test record says %s — the image that was "
                            "tested is not the image a box would pull" % (svc, ref, got, want))
    return problems, inconclusive


def main(argv, resolver=None):
    spec = "origin/main..HEAD"
    network = "--no-network" not in argv
    for i, arg in enumerate(argv):
        if arg.startswith("--range="):
            spec = arg[len("--range="):]
        elif arg == "--range" and i + 1 < len(argv):
            spec = argv[i + 1]
    rc, shallow, _ = git("rev-parse", "--is-shallow-repository")
    if rc == 0 and shallow.strip() == "true":
        print("TEST-RECORD-MOVE GATE INCONCLUSIVE: this clone is SHALLOW — no parent commit to diff "
              "(the R-452 gap). Enforced by the pre-push hook on the full clone; the static twin "
              "check-test-record.py still ran.")
        return 2
    a, b, why = resolve_range(spec)
    if a is None:
        print("TEST-RECORD-MOVE GATE INCONCLUSIVE: %s" % why)
        return 2
    rc, names, err = git("diff", "--name-only", a, b, "--", "templates")
    if rc != 0:
        print("TEST-RECORD-MOVE GATE INCONCLUSIVE: git diff failed: %s" % err.strip())
        return 2
    apps = sorted({TEMPLATE_RE.match(n).group(1) for n in names.split("\n") if TEMPLATE_RE.match(n)})
    for i, arg in enumerate(argv):
        if arg == "--digests-from" and i + 1 < len(argv):
            import json
            table = json.load(open(argv[i + 1]))
            print("  registry answers come from %s, NOT the registry (decoy tests only)" % argv[i + 1])
            resolver = lambda ref, _t=table: ((_t[ref], None) if _t.get(ref) else (None, "not in the table"))
    resolver = resolver or default_resolver
    convicted, undecided = {}, {}
    for app in apps:
        p, inc = judge_app(app, a, b, resolver, network)
        if p:
            convicted[app] = p
        if inc:
            undecided[app] = inc
    print("test-record-move gate — range %s..%s: %d compose file(s) changed" % (a, b, len(apps)))
    for app, ps in convicted.items():
        for p in ps:
            print("REFUSED %s: %s" % (app, p))
    for app, ps in undecided.items():
        for p in ps:
            print("INCONCLUSIVE %s: %s" % (app, p))
    if convicted:
        return 1
    if undecided:
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))