admin/app-catalog-felhom.eu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
app-catalog-felhom.eu/scripts/test_upgrade_bench.py
T

522 lines
26 KiB
Python
Raw Permalink Blame History

#!/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
from pathlib import Path
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 FakeBoxWalk(FakeVenue):
"""box_walk's interface on a test box: no VENUE, HP/GUEST, the run's installs, and guest() that answers like the
in-box admin invite script would. Records every script sent into the box."""
def __init__(self, hp="demo-hp", guest="9202", deployed=("vaultwarden",), drill=True, invite="RESULT signin=200 session=yes invite=200"):
FakeVenue.__init__(self, False)
self.GENERATED = {}
self.HP, self.GUEST, self.DEPLOYED_THIS_RUN = hp, guest, set(deployed)
self.drill, self.invite, self.scripts = drill, invite, []
def guest(self, script, timeout=600):
self.scripts.append(script)
if "controller.yaml" in script:
return ("git:\n repo_url: https://gitea.dooplex.hu/admin/%s.git\n"
% ("app-catalog-drill" if self.drill else "app-catalog-felhom.eu"))
if "/admin/invite" in script:
m = __import__("re").search(r'"email":"([^"]+)"', script)
if self.invite.endswith("invite=200"):
self.invited.add(m.group(1))
return "some noise\n" + self.invite + "\n"
return ""
class BoxAdminSeedGuard(unittest.TestCase):
"""R-890 (`09` §3 decision 149): the admin seed on a TEST BOX — scratch 9202 on demo-hp, an app this run installed,
the box on the drill catalog, the run opted in. RED-PROOF (REPORT): make box_admin_seed_allowed return (True, "") —
test_each_condition_alone_refuses and test_a_household_shaped_box_is_never_seeded fail."""
ENV = {"FELHOM_BOX_ADMIN_SEED": "1"}
def allowed(self, w, env):
with mock.patch.dict(os.environ, env, clear=False):
if "FELHOM_BOX_ADMIN_SEED" not in env:
os.environ.pop("FELHOM_BOX_ADMIN_SEED", None)
return fxbox.box_admin_seed_allowed(w, "vaultwarden")
def test_all_conditions_allow(self):
self.assertEqual(self.allowed(FakeBoxWalk(), self.ENV), (True, ""))
def test_each_condition_alone_refuses(self):
cases = [(FakeVenue(True), self.ENV, "bench venue"),
(FakeBoxWalk(), {}, "FELHOM_BOX_ADMIN_SEED"),
(FakeBoxWalk(), {"FELHOM_BOX_ADMIN_SEED": "yes"}, "FELHOM_BOX_ADMIN_SEED"),
(FakeBoxWalk(guest="9201"), self.ENV, "not a test box"),
(FakeBoxWalk(hp="demo-felhom"), self.ENV, "not a test box"),
(FakeBoxWalk(deployed=()), self.ENV, "did not install"),
(FakeBoxWalk(drill=False), self.ENV, "drill catalog")]
for w, env, want in cases:
ok, why = self.allowed(w, env)
self.assertFalse(ok, want)
self.assertIn(want, why)
def seed(self, w, env):
with mock.patch.dict(os.environ, env, clear=False):
if "FELHOM_BOX_ADMIN_SEED" not in env:
os.environ.pop("FELHOM_BOX_ADMIN_SEED", None)
os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None)
fx_ = fxbox.Vaultwarden()
return fx_, fx_.seed(w, "vault", lambda *a: None)
def test_the_test_box_seeds_through_the_invite_inside_the_box(self):
w = FakeBoxWalk()
_, got = self.seed(w, self.ENV)
self.assertIsNotNone(got)
self.assertEqual([c["path"] for c in w.calls], ["/identity/accounts/register", "/identity/accounts/register"])
inv = [x for x in w.scripts if "/admin/invite" in x]
self.assertEqual(len(inv), 1)
# the token is read inside the box and handed to curl on stdin; the cookie file is shredded
self.assertIn("printenv ADMIN_TOKEN", inv[0])
self.assertIn("token@-", inv[0])
self.assertIn('shred -u "$H"', inv[0])
self.assertIn(got["email"], inv[0])
def test_a_household_shaped_box_is_never_seeded(self):
for w in (FakeBoxWalk(guest="9201"), FakeBoxWalk(deployed=()), FakeBoxWalk(drill=False)):
fx_, got = self.seed(w, self.ENV)
self.assertIsNone(got)
self.assertFalse([x for x in w.scripts if "/admin/invite" in x], "the admin route was tried")
self.assertIn("NOT tried", fx_.tried)
def test_a_refused_invite_is_inconclusive(self):
w = FakeBoxWalk(invite="RESULT signin=401 session=no")
fx_, got = self.seed(w, self.ENV)
self.assertIsNone(got)
self.assertIn("test-box admin invite", fx_.tried)
def test_the_tester_1_box_is_a_test_box_and_demo_hp_9201_is_not(self):
self.assertEqual(self.allowed(FakeBoxWalk(hp="tester-1", guest="9201"), self.ENV), (True, ""))
ok, why = self.allowed(FakeBoxWalk(hp="demo-hp", guest="9201"), self.ENV)
self.assertFalse(ok)
self.assertIn("not a test box", why)
def test_only_a_drill_address_is_invited(self):
with self.assertRaises(ValueError):
fxbox.box_admin_invite(FakeBoxWalk(), 'x"}; rm -rf /; {"@gate.invalid')
class DefinitionEdge(unittest.TestCase):
"""R-762: `--move-to <app> <dir>` — a step that ADDS a service. The TO side is the whole definition in <dir>, and the
TO render reads it (render() alone cannot add a service). RED-PROOF (REPORT): drop `to_template` from the edge — the
render test reads the FROM template and fails."""
def setUp(self):
self.root = Path(tempfile.mkdtemp(prefix="bench-defedge-"))
for d, extra in (("app1", ""), ("app1@new", " side:\n image: nginx:1.30.5-alpine\n")):
(self.root / d).mkdir()
(self.root / d / "docker-compose.yml").write_text("services:\n app1:\n image: app/one:1.0\n" + extra)
self.p = mock.patch.object(ut, "TEMPLATES", self.root)
self.p.start()
def tearDown(self):
self.p.stop()
shutil.rmtree(self.root, ignore_errors=True)
def test_the_edge_names_the_new_service(self):
eid = ut.add_definition_edge("app1", "app1@new")
e = ut.EDGES[eid]
self.assertEqual(e["frm"], {"app1": "app/one:1.0"})
self.assertEqual(e["to"], {"app1": "app/one:1.0", "side": "nginx:1.30.5-alpine"})
self.assertEqual(e["to_template"], "app1@new")
def test_the_to_render_reads_the_new_definition(self):
eid = ut.add_definition_edge("app1", "app1@new")
e = ut.EDGES[eid]
wd = self.root / "work"
with mock.patch.object(ut, "apply_bench_overrides", lambda app, t: t), mock.patch.object(ut, "pg_mounts_for", lambda t: t):
ut.render("app1", e["to"], wd, {}, e.get("to_template") or e.get("template"))
self.assertIn("nginx:1.30.5-alpine", (wd / "docker-compose.yml").read_text())
def test_nothing_new_is_refused(self):
(self.root / "same").mkdir()
(self.root / "same" / "docker-compose.yml").write_text((self.root / "app1" / "docker-compose.yml").read_text())
with self.assertRaises(SystemExit):
ut.add_definition_edge("app1", "same")
class BoxWalkTargets(unittest.TestCase):
"""R-892: box_walk's target table. Each test imports box_walk afresh under its own TARGET. RED-PROOF (REPORT): drop
the `-J` from ssh_args — test_the_tester_1_box_is_reached_through_the_hp_box fails."""
def load(self, env):
import importlib
with mock.patch.dict(os.environ, env, clear=False):
for k in ("TARGET", "GUEST", "BASE", "DOMAIN"):
if k not in env:
os.environ.pop(k, None)
sys.modules.pop("box_walk", None)
return importlib.import_module("box_walk")
def test_every_row_is_complete(self):
bw = self.load({})
for name, row in bw.TARGETS.items():
self.assertEqual(set(row), {"hp", "ssh", "jump", "guest", "base", "domain"}, name)
self.assertTrue(row["base"].startswith("https://"), name)
def test_the_default_stays_scratch_9202(self):
bw = self.load({})
self.assertEqual((bw.HP, bw.GUEST, bw.BASE), ("demo-hp", "9202", "https://192.168.0.114"))
self.assertEqual(bw.ssh_args()[-1], "demo-hp")
self.assertNotIn("-J", bw.ssh_args())
def test_the_old_guest_switch_still_selects_demo_hp_9201(self):
bw = self.load({"GUEST": "9201"})
self.assertEqual((bw.HP, bw.GUEST, bw.BASE), ("demo-hp", "9201", "https://192.168.0.155"))
def test_the_tester_1_box_is_reached_through_the_hp_box(self):
bw = self.load({"TARGET": "tester-1"})
self.assertEqual((bw.HP, bw.GUEST, bw.DOMAIN), ("tester-1", "9201", "enkicsifelhom.hu"))
a = bw.ssh_args()
self.assertEqual(a[a.index("-J") + 1], "demo-hp")
self.assertEqual(a[-1], "root@192.168.0.154")
self.assertIn(("tester-1", "9201"), fxbox.BOX_ADMIN_SEED_GUESTS)
def test_an_unknown_target_is_refused(self):
with self.assertRaises(SystemExit):
self.load({"TARGET": "ep0"})
def tearDown(self):
sys.modules.pop("box_walk", None)
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")
class SwapRecorded(unittest.TestCase):
"""R-733: the memory record carries the venue's swap and each container's swap peak.
COMPANION RED-PROOF (observed 2026-10-06): make venue_swap_bytes return None always → test_venue_swap_read
fails; make swap_peak_of ignore the kernel's swap_peak → test_kernel_peak_counts fails. Restored."""
def test_venue_swap_read(self):
self.assertEqual(ut.venue_swap_bytes("MemTotal: 1 kB\nSwapTotal: 524288 kB\nSwapFree: 1 kB\n"), 512 * 2**20)
self.assertEqual(ut.venue_swap_bytes("SwapTotal: 0 kB\n"), 0)
def test_venue_swap_unreadable_is_none_not_zero(self):
self.assertIsNone(ut.venue_swap_bytes("MemTotal: 1 kB\n"))
def test_kernel_peak_counts(self):
samples = [{"containers": {"app": {"swap": 10}}}, {"containers": {"app": {"swap": 30}}}]
self.assertEqual(ut.swap_peak_of(samples, "app", {"swap": 5, "swap_peak": 99}), 99)
self.assertEqual(ut.swap_peak_of(samples, "app", {"swap": 5, "swap_peak": None}), 30)
def test_nothing_readable_is_none(self):
self.assertIsNone(ut.swap_peak_of([{"containers": {}}], "app", {"swap": None, "swap_peak": None}))
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in New Issue View Git Blame Copy Permalink