felhom.eu/documentation/audits/update-night-2026-09-21/run_edge.py

#!/usr/bin/env python3
"""run_edge.py <app> <from-ref> <to-ref> [--sub X] [--keep] [--no-remove]

One app's full walk on guest 9202. Writes everything under
`documentation/audits/update-night-2026-09-21/apps/<app>/`:

    log.txt        every line this run printed, as it printed it
    badges.json    the „Frissítés elérhető" badge in BOTH languages, before and after
    phases.json    every update phase with its timestamp
    observables.json  the four version observables side by side
    verdict.json   `09`'s verdict-record shape

`inconclusive` is never collapsed into `failed`.
"""
import argparse, json, os, sys, time
from datetime import datetime, timezone

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import walk as w            # noqa: E402
from fixtures import FIXTURES   # noqa: E402


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("app")
    ap.add_argument("frm")
    ap.add_argument("to")
    ap.add_argument("--sub", default=None)
    ap.add_argument("--no-remove", action="store_true")
    ap.add_argument("--class", dest="cls", default="other")
    a = ap.parse_args()

    app = a.app
    appdir = os.path.join(HERE, "apps", app)
    os.makedirs(appdir, exist_ok=True)
    fx = FIXTURES.get(app)
    sub = a.sub or (fx.sub if fx else app)

    t_start = time.time()
    w.say(f"==== {app}: {a.frm}  ->  {a.to}   (sub={sub}, class={a.cls})")
    w.login()

    rec = {"harness_version": 1, "app": app, "venue": "guest 9202 demo-hp-scratch, controller 0.261.0",
           "class": a.cls, "from": {}, "to": {}, "verdict": "inconclusive",
           "seed_read_before": False, "seed_read_after": False, "healthy_after": False,
           "migration_observed": None, "abort": "not-attempted", "abort_detail": None,
           "duration_s": 0, "measured_at": datetime.now(timezone.utc).isoformat(),
           "evidence": f"apps/{app}/", "notes": []}

    def bail(why, verdict="inconclusive"):
        rec["verdict"] = verdict
        rec["notes"].append(why)
        rec["duration_s"] = round(time.time() - t_start, 1)
        finish(rec, appdir)
        sys.exit(0 if verdict != "failed" else 0)

    # ---- 1 deploy at the LIVE pin -------------------------------------------------------
    if not w.deploy(app, sub):
        bail("deploy never reached running — nothing else could be measured")
    pre = w.observables(app)
    rec["from"] = pre["pinned_images"] or {}
    json.dump(pre, open(f"{appdir}/observables-before.json", "w"), indent=2, ensure_ascii=False)

    # ---- 2+3 seed and read it back (control C1) ------------------------------------------
    if fx is None:
        rec["notes"].append("no fixture: no non-browser seed route was written for this app tonight")
        bail("no fixture — recorded inconclusive rather than faked (R-156)")
    w.say("  [2] seeding through the app's own front door")
    tok = fx.seed(w, sub, w.say)
    if tok is None:
        rec["notes"].append("seed refused through the app's own route — see log.txt for what was tried")
        bail("seed route did not work tonight")
    w.say("  [3] control C1 — reading the seed back BEFORE the update")
    if not fx.verify(w, sub, tok, w.say):
        rec["notes"].append("C1 FAILED: the fixture could not prove itself before the update, "
                            "so it can prove nothing after")
        bail("C1 failed — a fixture that cannot prove itself first proves nothing after")
    rec["seed_read_before"] = True

    # ---- 4 „Mentés most" -----------------------------------------------------------------
    w.backup_now(app)

    # ---- 5 the drill bump, sync, rescan, badge -------------------------------------------
    b_before = w.badges(app)
    h = w.drill_bump(app, a.frm, a.to)
    if h is None:
        bail("the drill bump could not be committed — the FROM ref did not match the template")
    w.sync_rescan()
    b_after = w.badges(app)
    json.dump({"before": b_before, "after": b_after, "drill_commit": h},
              open(f"{appdir}/badges.json", "w"), indent=2, ensure_ascii=False)
    w.say(f"  [5] badge HU: {b_after['hu']}")
    w.say(f"  [5] badge EN: {b_after['en']}")
    st = w.stack(app)
    rec["to"] = st.get("catalog_images") or {}

    # ---- 6 the guarded Update ------------------------------------------------------------
    res = w.press_update(app)
    json.dump(res, open(f"{appdir}/phases.json", "w"), indent=2, ensure_ascii=False)
    rec["duration_s"] = res["duration_s"]
    if not res["accepted"]:
        rec["notes"].append(f"the Update was REFUSED before anything moved: {json.dumps(res['refusal'], ensure_ascii=False)[:400]}")
        bail("refused at the preflight — nothing moved")

    # ---- 7 read the seed back ------------------------------------------------------------
    w.say("  [7] reading the seed back AFTER the update")
    after_ok = fx.verify(w, sub, tok, w.say)
    rec["seed_read_after"] = after_ok

    # ---- migration line, quoted verbatim, never inferred from timing ---------------------
    logs = w.app_logs(app, 500)
    open(f"{appdir}/app-logs-after.txt", "w").write(logs)
    for pat in ("migrat", "Migrat", "MIGRAT", "upgrade", "Upgrade", "schema"):
        for line in logs.splitlines():
            if pat in line and len(line) < 400:
                rec["migration_observed"] = line.strip()
                break
        if rec["migration_observed"]:
            break

    # ---- 8 the four observables ----------------------------------------------------------
    post = w.observables(app)
    json.dump({"before": pre, "after": post},
              open(f"{appdir}/observables.json", "w"), indent=2, ensure_ascii=False)
    w.say(f"  [8] pinned    = {post['pinned_images']}")
    w.say(f"  [8] installed = {post['installed_images']}")
    w.say(f"  [8] compose   = {post['live_compose_image_lines']}")
    w.say(f"  [8] inspect   = {post['docker_inspect']}")
    rec["observables_after"] = post

    st = w.stack(app)
    rec["healthy_after"] = (st.get("state") == "running" and not st.get("hold_reason"))
    rec["final_phase"] = res["final_phase"]
    rec["hold_reason"] = res.get("hold_reason")
    rec["update_error"] = res.get("update_error")

    # ---- 9 the verdict -------------------------------------------------------------------
    moved = post["pinned_images"] != pre["pinned_images"]
    if res["final_phase"] == "done" and after_ok and rec["healthy_after"] and moved:
        rec["verdict"] = "proven"
    elif res.get("hold_reason") or res["final_phase"] == "failed":
        rec["verdict"] = "failed"
        rec["notes"].append("the edge ended HELD or failed — this is a RESULT, not an error of the run")
    elif not after_ok:
        rec["verdict"] = "failed"
        rec["notes"].append("the app came up but the seeded data did not read back")
    else:
        rec["verdict"] = "inconclusive"
        rec["notes"].append(f"final_phase={res['final_phase']} moved={moved} healthy={rec['healthy_after']}")

    finish(rec, appdir)

    if not a.no_remove and rec["verdict"] != "failed":
        w.remove(app)


def finish(rec, appdir):
    rec["duration_s"] = rec.get("duration_s", 0)
    w.write_verdict(rec, appdir)
    open(f"{appdir}/log.txt", "w").write("\n".join(w.LOG) + "\n")


if __name__ == "__main__":
    main()