app-catalog-felhom.eu/scripts/test_upgrade_bench.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""test_upgrade_bench.py — the test bench's two night-of-2026-09-23 faults (R-653, R-656). No Docker.

    python3 scripts/test_upgrade_bench.py
"""
import importlib.util
import os
import shutil
import tempfile
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
spec = importlib.util.spec_from_file_location("upgrade_test_mod", os.path.join(HERE, "upgrade-test.py"))
ut = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ut)


class LoadVerdict(unittest.TestCase):
    """R-653. RED-PROOF (REPORT.md): make load_verdict always return "reached" — the ghost case fails."""

    def test_every_request_errored_is_inconclusive(self):
        # ghost's first watch, 2026-09-23 night: 11 797 requests, all `err`
        self.assertEqual(ut.load_verdict(11797, {"err": 11797}), "inconclusive")

    def test_answers_of_any_code_are_the_app_answering(self):
        self.assertEqual(ut.load_verdict(11429, {"200": 5712, "401": 5717}), "reached")
        self.assertEqual(ut.load_verdict(10, {"302": 5, "err": 5}), "reached")

    def test_under_half_is_inconclusive_and_none_is_inconclusive(self):
        self.assertEqual(ut.load_verdict(10, {"200": 4, "err": 6}), "inconclusive")
        self.assertEqual(ut.load_verdict(0, {}), "inconclusive")


class ClearScratch(unittest.TestCase):
    """R-656. RED-PROOF (REPORT.md): make clear_scratch_folders return [] without removing — the first
    test fails with the last run's config still there."""

    def setUp(self):
        self.root = tempfile.mkdtemp(prefix="bench-scratch-")
        self.hdd = os.path.join(self.root, "hdd")
        self.env = {"HDD_PATH": self.hdd, "USERDATA_PATH": self.hdd + "/userdata"}
        self.said = []

    def tearDown(self):
        shutil.rmtree(self.root, ignore_errors=True)

    def plant(self, rel):
        p = os.path.join(self.hdd, rel, "config", "config.php")
        os.makedirs(os.path.dirname(p), exist_ok=True)
        open(p, "w").write("last run")
        return p

    def test_the_apps_own_folders_are_cleared_and_said(self):
        mine = self.plant("appdata/nextcloud")
        other = self.plant("appdata/immich")
        comp = "services:\n  nextcloud:\n    volumes:\n      - ${HDD_PATH}/appdata/nextcloud:/var/www/html\n"
        got = ut.clear_scratch_folders(comp, self.env, self.said.append)
        self.assertFalse(os.path.exists(mine), "the last run's files are still there")
        self.assertTrue(os.path.exists(other), "another app's folder was touched")
        self.assertEqual(len(got), 1)
        self.assertTrue(any("cleared before FROM" in s for s in self.said))

    def test_never_a_bare_root_never_outside(self):
        keep = self.plant("appdata/x")
        comp = ("    volumes:\n      - ${HDD_PATH}:/data\n      - ${HDD_PATH}/:/d2\n"
                "      - ${HDD_PATH}/../escape:/e\n")
        got = ut.clear_scratch_folders(comp, self.env, self.said.append)
        self.assertEqual(got, [])
        self.assertTrue(os.path.exists(keep))


class ChangedFiles(unittest.TestCase):
    """2026-09-30: a `files_may_change` mark names its files (R-734's method, now in the harness)."""

    def test_names_changed_added_removed_and_nothing_else(self):
        before = {"/d": {"a.txt": "1:aa", "b.txt": "1:bb", "gone": "1:cc"}}
        after = {"/d": {"a.txt": "1:aa", "b.txt": "2:bd", "new": "1:dd"}}
        self.assertEqual(ut.changed_files(before, after),
                         ["/d/b.txt (changed)", "/d/gone (removed)", "/d/new (added)"])
        self.assertEqual(ut.changed_files(before, before), [])


class MarkerIgnore(unittest.TestCase):
    """R-734 (`09` §3 decision 145): app-owned marker files do not make the `files_may_change` mark; any other
    changed file still does. The data is the bench's own measurement (audits/more-night-apps-2026-09-30/D/evidence/
    D-immich-0b82-768/files-{before,after}-detail.json). RED-PROOF (REPORT): make classify_changes ignore every
    changed file of an app with a list — test_an_unlisted_changed_file_still_marks fails."""

    SRC = "/srv/felhom-gate/hdd/appdata/immich"
    BEFORE = {SRC: {"profile/.immich": "13:7cb1f8a19d944d7e", "upload/.immich": "13:67848ba791bd7ef6",
                    "encoded-video/.immich": "13:68bdf71e87fabadc", "backups/.immich": "13:7cb1f8a19d944d7e",
                    "thumbs/.immich": "13:7cb1f8a19d944d7e", "library/.immich": "13:67848ba791bd7ef6"},
              "/srv/felhom-gate/hdd/userdata/media/photos": {}}
    AFTER = {SRC: {"profile/.immich": "13:0bacc6528e07d8cb", "upload/.immich": "13:a68db79a3fbe6725",
                   "encoded-video/.immich": "13:2cb994a10bfe243c", "backups/.immich": "13:648fe4b9664fadff",
                   "thumbs/.immich": "13:0bacc6528e07d8cb", "library/.immich": "13:a68db79a3fbe6725"},
             "/srv/felhom-gate/hdd/userdata/media/photos": {}}

    def mark(self, app, before, after):
        counted, ignored = ut.classify_changes(app, before, after)
        tree = sorted({c.split(" (")[0].rsplit("/", 2)[0] for c in counted + [i["file"] for i in ignored]})
        return ut.files_may_change(tree, counted, ignored), counted, ignored

    def test_the_six_measured_markers_do_not_mark_and_each_carries_a_reason(self):
        marked, counted, ignored = self.mark("immich", self.BEFORE, self.AFTER)
        self.assertFalse(marked)
        self.assertEqual(counted, [])
        self.assertEqual(len(ignored), 6)
        self.assertTrue(all(i["reason"] for i in ignored))

    def test_an_unlisted_changed_file_still_marks(self):
        after = {k: dict(v) for k, v in self.AFTER.items()}
        after[self.SRC]["library/admin/2026/IMG_0001.jpg"] = "2048:abcd"       # added: a household photo
        before = {k: dict(v) for k, v in self.BEFORE.items()}
        before[self.SRC]["upload/upload/x.jpg"] = "10:aa"
        after[self.SRC]["upload/upload/x.jpg"] = "10:bb"                        # changed, same size as a marker
        marked, counted, ignored = self.mark("immich", before, after)
        self.assertTrue(marked)
        self.assertIn(self.SRC + "/library/admin/2026/IMG_0001.jpg (added)", counted)
        self.assertIn(self.SRC + "/upload/upload/x.jpg (changed)", counted)
        self.assertEqual(len(ignored), 6)

    def test_a_listed_name_that_grew_or_vanished_still_marks(self):
        after = {k: dict(v) for k, v in self.AFTER.items()}
        after[self.SRC]["thumbs/.immich"] = "4096:ffff"                          # no longer a 13-byte marker
        del after[self.SRC]["upload/.immich"]                                    # removed
        marked, counted, _ = self.mark("immich", self.BEFORE, after)
        self.assertTrue(marked)
        self.assertIn(self.SRC + "/thumbs/.immich (changed)", counted)
        self.assertIn(self.SRC + "/upload/.immich (removed)", counted)

    def test_the_list_belongs_to_its_app(self):
        # the same files under another app (and another bind) count: the list is per app
        marked, counted, ignored = self.mark("calibre-web", self.BEFORE, self.AFTER)
        self.assertTrue(marked)
        self.assertEqual((len(counted), ignored), (6, []))

    def test_a_moved_tree_the_file_walk_cannot_name_keeps_the_mark(self):
        self.assertTrue(ut.files_may_change(["/d"], [], []))
        self.assertFalse(ut.files_may_change([], [], []))

    def test_every_entry_has_a_reason(self):
        for app, rules in ut.MARKER_IGNORE.items():
            for suffix, path, max_size, reason in rules:
                self.assertTrue(suffix and path and max_size > 0 and len(reason) > 40, (app, path))


# --- R-624 (`09` §3 decision 146): the bench-only admin seed, and the run's secrets -----------------------------------
import sys  # noqa: E402
from unittest import mock  # noqa: E402
sys.path.insert(0, HERE)
import upgrade_fixtures_box as fxbox  # noqa: E402

ADMIN = "a" * 8 + "0123456789abcdef0123456789abcdef0123456789abcdef01234567"   # stands in for a generated hex:32


class FakeVenue:
    """walk.py's interface without a network: records every call; answers like vaultwarden 1.36 did on 9202
    (audits/evidence-p1fixes-2026-09-15/E1-vaultwarden-spike.txt): a stranger's register 400, POST /admin with the
    right token sets VW_ADMIN, /admin/invite 200 with that cookie, an invited address registers 200."""

    def __init__(self, bench):
        if bench:
            self.VENUE = "bench"
        self.GENERATED = {"vaultwarden": {"ADMIN_TOKEN": ADMIN, "DOMAIN": "gate.invalid"}}
        self.calls, self.invited = [], set()

    def wait_app(self, *a, **k):
        return True

    def app_curl(self, sub, path, *extra, method=None, data=None, timeout=45):
        headers = []
        for i, x in enumerate(extra):
            if x == "-H" and extra[i + 1].startswith("@"):
                headers.append(open(extra[i + 1][1:]).read().strip())
                self.header_file = extra[i + 1][1:]
        self.calls.append({"path": path, "argv": list(extra), "data": data, "file_headers": headers})
        if path == "/identity/accounts/register":
            email = json.loads(data)["email"]
            return 0, ("200" if email in self.invited else "400"), '{"message":"Registration not allowed"}'
        if path == "/admin":
            if data == "token=" + ADMIN:
                return 0, "200", "HTTP/1.1 200 OK\r\nset-cookie: VW_ADMIN=jwt.admin.session; Path=/admin\r\n\r\n<html>"
            return 0, "401", "HTTP/1.1 401\r\n\r\nInvalid admin token"
        if path == "/admin/invite":
            if "Cookie: VW_ADMIN=jwt.admin.session" in headers:
                self.invited.add(json.loads(data)["email"])
                return 0, "200", "{}"
            return 0, "401", ""
        return 0, "404", ""


import json  # noqa: E402


class BenchAdminSeedGuard(unittest.TestCase):
    """The admin seed runs on the bench ONLY. RED-PROOF (REPORT): make bench_admin_seed_allowed return (True, "") —
    test_the_box_walk_never_signs_in_as_admin and each single-condition refusal fail."""

    def setUp(self):
        self.tmp = tempfile.mkdtemp(prefix="bench-guard-")
        self.nobox = os.path.join(self.tmp, "no-such-stacks")
        self.said = []

    def tearDown(self):
        shutil.rmtree(self.tmp, ignore_errors=True)

    def allowed(self, venue, env, box_marker):
        with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(fxbox, "BOX_MARKER", box_marker):
            if "FELHOM_BENCH_ADMIN_SEED" not in env:
                os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None)
            return fxbox.bench_admin_seed_allowed(venue)

    def test_all_three_conditions_allow(self):
        self.assertEqual(self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox), (True, ""))

    def test_each_condition_alone_refuses(self):
        ok, why = self.allowed(FakeVenue(False), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox)
        self.assertFalse(ok)
        self.assertIn("not the bench venue", why)
        ok, why = self.allowed(FakeVenue(True), {}, self.nobox)
        self.assertFalse(ok)
        self.assertIn("FELHOM_BENCH_ADMIN_SEED", why)
        ok, why = self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "yes"}, self.nobox)
        self.assertFalse(ok)
        ok, why = self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.tmp)   # a box: stacks exists
        self.assertFalse(ok)
        self.assertIn("Felhom box", why)

    def test_the_bench_venue_names_itself(self):
        import upgrade_boxport
        self.assertEqual(upgrade_boxport.Venue.VENUE, "bench")

    def seed(self, venue, env, box_marker):
        with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(fxbox, "BOX_MARKER", box_marker):
            if "FELHOM_BENCH_ADMIN_SEED" not in env:
                os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None)
            fx_ = fxbox.Vaultwarden()
            return fx_, fx_.seed(venue, "vault", self.said.append)

    def test_the_box_walk_never_signs_in_as_admin(self):
        v = FakeVenue(False)
        fx_, got = self.seed(v, {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox)
        self.assertIsNone(got)
        self.assertEqual([c["path"] for c in v.calls], ["/identity/accounts/register"])
        self.assertIn("NOT tried", fx_.tried)

    def test_the_bench_seeds_through_the_admin_invite_and_never_shows_the_token(self):
        v = FakeVenue(True)
        _, got = self.seed(v, {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox)
        self.assertIsNotNone(got)
        self.assertEqual([c["path"] for c in v.calls], ["/identity/accounts/register", "/admin", "/admin/invite",
                                                        "/identity/accounts/register"])
        for c in v.calls:
            self.assertFalse(any(ADMIN in a or "VW_ADMIN" in a for a in c["argv"]), "a secret on the command line")
        self.assertEqual(v.calls[2]["file_headers"], ["Cookie: VW_ADMIN=jwt.admin.session"])
        self.assertFalse(os.path.exists(v.header_file), "the cookie header file was not shredded")
        self.assertFalse(any(ADMIN in s or "jwt.admin.session" in s for s in self.said), "a secret was printed")


class SecretHygiene(unittest.TestCase):
    """The run's secrets: .env 0600 and shredded, every evidence file redacted. RED-PROOF (REPORT): make redact_tree
    return [] without rewriting — test_evidence_files_are_redacted fails."""

    FELHOM = ("deploy_fields:\n  - env_var: DOMAIN\n    type: domain\n  - env_var: ADMIN_TOKEN\n    type: secret\n"
              "    generate: \"hex:32\"\n  - env_var: SIGNUPS_ALLOWED\n    type: text\n    default: \"false\"\n")

    def setUp(self):
        self.tmp = tempfile.mkdtemp(prefix="bench-secrets-")

    def tearDown(self):
        shutil.rmtree(self.tmp, ignore_errors=True)

    def test_secret_values_are_the_generated_fields_and_the_seed_password(self):
        if ut.cvp is None:
            import importlib.util as iu
            sp = iu.spec_from_file_location("cvp", os.path.join(HERE, "check-volume-persistence.py"))
            m = iu.module_from_spec(sp)
            sp.loader.exec_module(m)
            ut.cvp = m
        env = {"DOMAIN": "gate.invalid", "ADMIN_TOKEN": ADMIN, "SIGNUPS_ALLOWED": "false"}
        got = ut.secret_values(self.FELHOM, env, {"email": "drill-1@gate.invalid", "pw": "Drill-0011223344"})
        self.assertEqual(set(got), {ADMIN, "Drill-0011223344"})

    def test_env_is_0600_and_shredded(self):
        p = ut.Path(self.tmp) / ".env"
        ut.write_secret_file(p, "ADMIN_TOKEN=%s\n" % ADMIN)
        self.assertEqual(os.stat(p).st_mode & 0o777, 0o600)
        ut.write_secret_file(p, "ADMIN_TOKEN=%s\n" % ADMIN)            # a re-render replaces it, still 0600
        self.assertEqual(os.stat(p).st_mode & 0o777, 0o600)
        ut.shred_file(p)
        self.assertFalse(p.exists())
        ut.shred_file(p)                                                 # a missing file is fine

    def test_evidence_files_are_redacted(self):
        ev = ut.Path(self.tmp) / "evidence" / "MV-vaultwarden"
        (ev / "sub").mkdir(parents=True)
        (ev / "run.log").write_text("token=%s ok\n" % ADMIN)
        (ev / "sub" / "to-full.log").write_text("clean line\n")
        (ev / "verdict.json").write_text(json.dumps({"abort_detail": "pw Drill-0011223344"}))
        held = ut.redact_tree(ev, [ADMIN, "Drill-0011223344"])
        self.assertEqual(len(held), 2)
        for p in ev.rglob("*"):
            if p.is_file():
                t = p.read_text()
                self.assertNotIn(ADMIN, t)
                self.assertNotIn("Drill-0011223344", t)
        self.assertIn(ut.REDACTED, (ev / "run.log").read_text())
        self.assertEqual((ev / "sub" / "to-full.log").read_text(), "clean line\n")

    def test_redact_longest_first(self):
        self.assertEqual(ut.redact("x abcdefgh123 y", ["abcdefgh123", "abcdefgh"]), "x <redacted> y")


if __name__ == "__main__":
    unittest.main(verbosity=2)