#!/usr/bin/env python3 """Fixture tests for check-volume-persistence.py. NO DOCKER — the prober is injected. The tests drive `check()` — the function `__main__` calls — rather than `classify()` alone, so they cover the path that actually decides the exit code. A gate whose verdict logic is tested but whose entry point is not has been shipped inert in this project before (the seam-wiring rule). Run: python3 scripts/test_check_volume_persistence.py """ import importlib.util import io import sys import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path _spec = importlib.util.spec_from_file_location( "cvp", Path(__file__).resolve().parent / "check-volume-persistence.py") cvp = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(cvp) # --------------------------------------------------------------------- probe fixtures def _ctr(name="app", uid=999, gid=999, mounts=(), data_dirs=(), status="running", health=None): return {"name": name, "status": status, "health": health, "exit": 0, "restarts": 0, "uid": uid, "gid": gid, "mounts": list(mounts), "diff_total": 0, "diff_data_dirs": [{"dir": d, "files": ["db.sqlite"], "db_signature": True} for d in data_dirs], "diff_other_dirs": []} def _mount(target, cls="named-declared", files=0, writable="yes"): return {"target": target, "class": cls, "name": "v", "source": "/var/lib/docker/volumes/v/_data", "files": files, "sample": [], "writable_by_app": writable} # The measured papra shape (R-156 evidence, campaign10-evidence-2026-07-31/r156-papra-volume.txt): # volume mounted at /app/data, root-owned, app runs as 999, real DB in /app/app-data/db. PAPRA = {"app": "papra", "containers": [ _ctr("papra", 999, 999, mounts=[_mount("/app/data", files=0, writable="NO")], data_dirs=["/app/app-data/db"])]} # vaultwarden as measured: one declared volume at /data holding the app's own db.sqlite3, # writable by the app, and nothing data-classified in the writable layer. VAULTWARDEN = {"app": "vaultwarden", "containers": [ _ctr("vaultwarden", 0, 0, mounts=[_mount("/data", files=4, writable="yes")])]} # Started, healthy, and wrote nothing at all — uptime-kuma's measured shape. IDLE = {"app": "idle", "containers": [ _ctr("idle", 0, 0, health="healthy", mounts=[_mount("/app/data", files=0)])]} ANON = {"app": "anon", "containers": [ _ctr("anon", 0, 0, mounts=[_mount("/var/lib/mysql", cls="anonymous", files=120)])]} class TestClassify(unittest.TestCase): """The verdict logic, pure.""" def test_papra_signature_is_broken(self): """The whole reason this gate exists: it must reproduce the known instance.""" status, why = cvp.classify(PAPRA) self.assertEqual(status, cvp.BROKEN) joined = " ".join(why) self.assertIn("/app/app-data/db", joined, "must name where the data actually went") self.assertIn("NOT writable", joined, "must report R-156's second leg too") def test_known_good_is_clean(self): """A detector with no proven negative is a detector that flags everything.""" self.assertEqual(cvp.classify(VAULTWARDEN)[0], cvp.CLEAN) def test_wrote_nothing_is_undetermined_not_clean(self): """The load-bearing distinction. An app that wrote nothing has not been shown to be correct — folding it into CLEAN is how a sweep reports 53 clean results with 6 unexamined.""" status, why = cvp.classify(IDLE) self.assertEqual(status, cvp.UNDETERMINED) self.assertIn("Health is not data", " ".join(why)) def test_container_not_running_is_undetermined(self): probe = {"app": "x", "containers": [_ctr("x", status="exited")]} self.assertEqual(cvp.classify(probe)[0], cvp.UNDETERMINED) def test_anonymous_volume_with_data_is_broken(self): """An anonymous volume survives a restart, which is what makes it deceptive: it is absent from ResolveDockerVolumeNames, so it is never backed up, and down+up orphans it.""" status, why = cvp.classify(ANON) self.assertEqual(status, cvp.BROKEN) self.assertIn("ANONYMOUS", " ".join(why)) def test_an_anonymous_volume_at_a_RUNTIME_path_is_not_a_defect(self): """The measured privatebin shape: its image declares `VOLUME /run`, so docker created an anonymous volume there holding nginx.pid, php-fpm.sock and s6 supervision fifos. Losing /run costs a restart. The prefixes are written `/run/`, so a bare `/run` matched nothing and every mount rule was blind to it.""" c = _ctr("privatebin", 0, 0, mounts=[_mount("/run", cls="anonymous", files=14), _mount("/srv/data", files=3)]) status, why = cvp.classify({"app": "privatebin", "containers": [c]}) self.assertEqual(status, cvp.CLEAN) self.assertNotIn("ANONYMOUS", " ".join(why)) def test_an_anonymous_volume_holding_REAL_data_still_convicts(self): """…and the runtime exemption must not become a blanket one.""" self.assertEqual(cvp.classify(ANON)[0], cvp.BROKEN) def test_is_noise_dir_covers_the_runtime_directories(self): for p in ("/run", "/tmp", "/var/log", "/var/cache", "/var/lib/nginx/tmp"): self.assertTrue(cvp.is_noise_dir(p), p) for p in ("/srv/data", "/app/app-data", "/var/lib/postgresql/data", "/config"): self.assertFalse(cvp.is_noise_dir(p), p) def test_empty_declared_volume_alone_is_a_note_not_a_verdict(self): """An empty volume on an app that also wrote nothing is UNDETERMINED, not BROKEN — the accusation needs positive evidence of data landing elsewhere.""" status, _ = cvp.classify(IDLE) self.assertNotEqual(status, cvp.BROKEN) def test_no_containers_is_undetermined(self): self.assertEqual(cvp.classify({"app": "x", "containers": []})[0], cvp.UNDETERMINED) def test_nothing_landed_in_any_mount_is_caught_without_any_path_vocabulary(self): """The measured gramps-web miss. Its family tree goes to /root/.gramps/grampsdb//{database.txt,name.txt} — no database-signature filename, no data token — so every path heuristic in this file is silent on it. A rule that only recognises the shapes someone thought of will always have a next blind spot; this one asks a question that needs no vocabulary.""" c = _ctr("gramps-web", 0, 0, mounts=[_mount("/app/data", files=0), _mount("/app/media", files=0)]) c["diff_other_dirs"] = [{"dir": "/root/.gramps/grampsdb/uuid", "added": ["database.txt", "name.txt"]}, {"dir": "/app/thumbnail_cache", "added": ["x"]}] status, why = cvp.classify({"app": "gramps-web", "containers": [c]}) self.assertEqual(status, cvp.UNDETERMINED) self.assertIn("/root/.gramps/grampsdb/uuid", " ".join(why)) self.assertIn("NOTHING this app wrote landed", " ".join(why)) def test_the_structural_finding_SURVIVES_a_broken_verdict(self): """The measured gramps-web reporting bug. Its accounts DB (rule 2) convicted, and the structural finding — its FAMILY TREE, the entire point of the app, landing outside every mount — was dropped because `undet` is discarded whenever `broken` is non-empty. A finding that disappears because a different finding won is the same class as an absent log line read as health.""" c = _ctr("gramps-web", 0, 0, mounts=[_mount("/app/data", files=0), _mount("/app/media", files=0)], data_dirs=["/app/users"]) c["diff_other_dirs"] = [{"dir": "/root/.gramps/grampsdb/uuid", "added": ["database.txt", "name.txt"]}] status, why = cvp.classify({"app": "gramps-web", "containers": [c]}) self.assertEqual(status, cvp.BROKEN) joined = " ".join(why) self.assertIn("/app/users", joined, "the convicting leg must still be reported") self.assertIn("/root/.gramps/grampsdb/uuid", joined, "and the structural finding must NOT be swallowed by it") def test_a_mount_that_received_data_silences_the_structural_check(self): """It must not fire on every app with one empty volume — crafty-controller has three empty mounts and two populated ones, and is correct.""" c = _ctr("crafty", 0, 0, mounts=[_mount("/crafty/app/config", files=16), _mount("/crafty/backups", files=0)]) c["diff_other_dirs"] = [{"dir": "/crafty/app/classes", "added": ["x"]}] self.assertEqual(cvp.classify({"app": "crafty", "containers": [c]})[0], cvp.CLEAN) def test_a_SIBLING_container_holding_the_state_silences_it(self): """The measured docmost / immich / claper shape, and the reason the question is asked per APP: the app container's only volume is for user uploads and is legitimately empty on a fresh install, while every byte of real state sits in the database container's volume (1540 / 1833 / 1470 files). Asked per container this called three correct apps unclean.""" app = _ctr("docmost", 0, 0, mounts=[_mount("/app/data/storage", files=0)]) app["diff_other_dirs"] = [{"dir": "/app/apps/client/dist", "added": ["index.js"]}] db = _ctr("docmost-postgres", 0, 0, mounts=[_mount("/var/lib/postgresql/data", files=1540)]) status, why = cvp.classify({"app": "docmost", "containers": [app, db]}) self.assertEqual(status, cvp.CLEAN) self.assertNotIn("NOTHING this app wrote landed", " ".join(why)) self.assertIn("benign when a sibling container holds the state", " ".join(why), "the per-container observation must still be reported, not dropped") def test_structural_check_stays_silent_when_the_app_wrote_nothing_at_all(self): """An idle app is UNDETERMINED for the existing reason, not accused by this one.""" status, why = cvp.classify(IDLE) self.assertEqual(status, cvp.UNDETERMINED) self.assertNotIn("NOTHING this app wrote landed", " ".join(why)) class TestDiffRollup(unittest.TestCase): """`docker diff` is noisy; these are the rules that separate a database from a cache.""" def test_db_signature_beats_everything(self): data, token, suspect, other = cvp.rollup_diff([("A", "/opt/whatever/store.sqlite")]) self.assertEqual([d["dir"] for d in data], ["/opt/whatever"]) self.assertTrue(data[0]["db_signature"]) self.assertEqual((suspect, other), ([], [])) def test_logs_caches_and_pids_are_noise(self): self.assertEqual(cvp.rollup_diff([ ("A", "/var/log/app.log"), ("C", "/tmp/x"), ("A", "/root/.cache/pip/w"), ("A", "/run/nginx.pid"), ("A", "/app/__pycache__/m.pyc")]), ([], [], [], [])) def test_deleted_entries_are_not_writes(self): self.assertEqual(cvp.rollup_diff([("D", "/app/data/gone.sqlite")]), ([], [], [], [])) def test_unclassified_writes_are_reported_never_dropped(self): data, token, suspect, other = cvp.rollup_diff([("A", "/opt/zzz/thing")]) self.assertEqual((data, suspect), ([], [])) self.assertEqual([d["dir"] for d in other], ["/opt/zzz"], "an unrecognised write must still be visible for judgement") def test_a_chown_sweep_over_image_files_is_not_data(self): """The measured calibre-web shape: 92 `C` entries under `cps/static/css/images/**` from a linuxserver.io entrypoint re-owning the app tree. Scored as data, this calls a clean app BROKEN — it did, on the first pass of the sweep.""" entries = [("C", f"/app/cwa/cps/static/css/images/icomoon/x{i}.png") for i in range(92)] data, token, suspect, other = cvp.rollup_diff(entries) self.assertEqual(data, [], "changed image files are furniture, not customer data") self.assertEqual(suspect, []) self.assertTrue(other, "…but they must still be listed, not dropped") def test_created_file_in_a_data_path_IS_data(self): """The other direction — the rule must not become blind. papra's verb is `A`.""" data, token, _, _ = cvp.rollup_diff([("A", "/app/app-data/db/db.sqlite")]) self.assertEqual([d["dir"] for d in data], ["/app/app-data/db"]) def test_a_bytecode_cache_DIRECTORY_is_noise(self): """The measured crafty-controller shape. `docker diff` lists directories too, so the bytecode cache appears as a bare `…/config/__pycache__` entry while its `.pyc` children are filtered by suffix — leaving the directory as the only surviving entry under a path containing the token `config`. That called a correct app BROKEN four times over.""" entries = [("A", "/crafty/app/classes/web/routes/api/crafty/config/__pycache__"), ("A", "/crafty/app/classes/web/routes/api/crafty/config/__pycache__/x.pyc")] data, token, suspect, _ = cvp.rollup_diff(entries) self.assertEqual(data, [], "a bytecode cache is not customer data") self.assertEqual(suspect, []) def test_cache_directories_are_filtered_at_ENTRY_level(self): """Where the filtering happens matters, because it is the reason a second 'are all this dir's children noise?' rule would be dead code: nothing that reaches the per-directory scoring has survived `is_noise`. Pinning the mechanism keeps that true.""" data, token, _, other = cvp.rollup_diff([("A", "/srv/storage/node_modules"), ("A", "/srv/storage/.cache")]) self.assertEqual(data, []) self.assertEqual(other, [], "noise is dropped before scoring, not scored and then excused") def test_but_one_real_file_among_noise_still_convicts(self): """…and the rule must not become a blanket amnesty for any directory with a cache in it.""" data, token, _, _ = cvp.rollup_diff([("A", "/srv/storage/__pycache__"), ("A", "/srv/storage/library.sqlite")]) self.assertEqual([d["dir"] for d in data], ["/srv/storage"]) def test_a_path_token_alone_does_NOT_convict(self): """The measured onlyoffice shape: the document server unpacks its OWN static assets into the writable layer at first boot — plugin icons, slide-theme `media/`, `web-apps/apps/api/documents/api.js`, 2560 added entries — while its real data mount received data normally. Vocabulary is not evidence: `media/` holds customer photos in one app and shipped clip-art in the next.""" data, token, _, _ = cvp.rollup_diff([ ("A", "/var/www/onlyoffice/documentserver/sdkjs/slide/themes/theme12/media/image1.jpg"), ("A", "/var/www/onlyoffice/documentserver/web-apps/apps/api/documents/api.js")]) self.assertEqual(data, [], "a path token must not be enough to accuse") self.assertEqual(len(token), 2, "…but it must still be surfaced for judgement") def test_a_token_dir_is_reported_and_counts_as_the_app_having_written(self): """Demoted is not discarded. The note must reach the operator, and an app that wrote only token-classified things must not then be reported as idle.""" c = _ctr("oo", 0, 0, mounts=[_mount("/var/www/onlyoffice/Data", files=3)]) c["diff_token_dirs"] = [{"dir": "/var/www/oo/themes/media", "added": ["image1.jpg"]}] status, why = cvp.classify({"app": "oo", "containers": [c]}) self.assertEqual(status, cvp.CLEAN) self.assertIn("judgement needed", " ".join(why)) self.assertIn("/var/www/oo/themes/media", " ".join(why)) def test_a_database_signature_still_convicts_on_its_own(self): """The demotion must not weaken rule 2 — papra and gramps-web are both caught by it.""" data, token, _, _ = cvp.rollup_diff([("A", "/app/app-data/db/db.sqlite")]) self.assertEqual([d["dir"] for d in data], ["/app/app-data/db"]) self.assertEqual(token, []) def test_a_postgres_CONFIG_file_is_not_a_database_signature(self): """The measured immich shape: the postgres entrypoint writes /etc/postgresql/postgresql.conf at init while PGDATA sits correctly in its volume with 1831 files. Scoring a config file as a database called a correct app BROKEN.""" data, token, suspect, other = cvp.rollup_diff([("A", "/etc/postgresql/postgresql.conf")]) self.assertEqual(data, [], "postgresql.conf is configuration, not data") self.assertEqual(suspect, []) def test_a_genuinely_misplaced_PGDATA_is_still_caught(self): """…and removing it must not open a blind spot: PG_VERSION and pg_control are the real markers of a PGDATA directory.""" for marker in ("PG_VERSION", "pg_control"): data, token, _, _ = cvp.rollup_diff([("A", f"/opt/stray/{marker}")]) self.assertEqual([d["dir"] for d in data], ["/opt/stray"], marker) def test_changed_db_file_is_SUSPECT_not_a_verdict(self): """`C` on a database file is genuinely ambiguous: a chown produces it, and so does an app writing into a DB that ships in its image. It must be adjudicated, never guessed.""" data, token, suspect, _ = cvp.rollup_diff([("C", "/app/cwa/empty_library/metadata.db")]) self.assertEqual(data, [], "a `C` alone must not convict") self.assertEqual([s["dir"] for s in suspect], ["/app/cwa/empty_library"]) class TestSuspectAdjudication(unittest.TestCase): """A suspect is settled by BYTES. These drive `classify()` with each possible outcome.""" @staticmethod def _probe(**kw): c = _ctr("app", 0, 0, mounts=[_mount("/config", files=3)]) c.update(kw) return {"app": "x", "containers": [c]} def test_benign_touch_stays_clean_and_is_still_reported(self): status, why = cvp.classify(self._probe(diff_benign_db_touches=[ {"dir": "/app/empty_library", "why": "byte-identical"}])) self.assertEqual(status, cvp.CLEAN) self.assertIn("chown sweep", " ".join(why), "benign ≠ invisible") def test_confirmed_write_into_an_image_file_is_broken(self): status, why = cvp.classify(self._probe(diff_data_dirs=[ {"dir": "/app/empty_library", "files": ["metadata.db"], "db_signature": True, "why": "DIFFERS from the image copy (0 B -> 40960 B)"}])) self.assertEqual(status, cvp.BROKEN) self.assertIn("DIFFERS", " ".join(why)) def test_unresolved_suspect_is_undetermined_never_clean(self): status, why = cvp.classify(self._probe(diff_unresolved=[ {"dir": "/app/empty_library", "why": "could not read both copies"}])) self.assertEqual(status, cvp.UNDETERMINED) self.assertIn("could not decide", " ".join(why)) class TestCheckEntryPoint(unittest.TestCase): """Drive `check()` — the function `__main__` calls — so the exit codes are covered.""" @staticmethod def _catalog(tmp, apps, lifecycle=None): for a in apps: d = Path(tmp) / "templates" / a d.mkdir(parents=True) (d / "docker-compose.yml").write_text("services:\n s:\n image: alpine:3.22\n") lc = (lifecycle or {}).get(a) (d / ".felhom.yml").write_text(f"slug: {a}\n" + (f"lifecycle: {lc}\n" if lc else "")) return Path(tmp) @staticmethod def _prober(table): def p(app, app_dir, **kw): if app.startswith("canary-"): return PAPRA if app == "canary-broken" else VAULTWARDEN return table[app] return p def test_one_broken_app_refuses_with_rc1(self): with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["papra", "vaultwarden"]) buf = io.StringIO() with redirect_stdout(buf): rc = cvp.check(root, prober=self._prober( {"papra": PAPRA, "vaultwarden": VAULTWARDEN})) self.assertEqual(rc, 1, "the gate must REFUSE, not warn") self.assertIn("REFUSED", buf.getvalue()) self.assertIn("papra", buf.getvalue()) def test_all_clean_passes_with_rc0(self): with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["vaultwarden"]) with redirect_stdout(io.StringIO()) as buf: rc = cvp.check(root, prober=self._prober({"vaultwarden": VAULTWARDEN})) self.assertEqual(rc, 0) self.assertIn("gate OK", buf.getvalue()) def test_undetermined_is_rc2_not_rc0(self): """UNDETERMINED must never read as a clean bill of health.""" with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["idle"]) with redirect_stdout(io.StringIO()) as buf: rc = cvp.check(root, prober=self._prober({"idle": IDLE})) self.assertEqual(rc, 2) self.assertIn("not a clean bill of health", buf.getvalue()) def test_broken_wins_over_undetermined(self): with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["papra", "idle"]) with redirect_stdout(io.StringIO()): rc = cvp.check(root, prober=self._prober({"papra": PAPRA, "idle": IDLE})) self.assertEqual(rc, 1) def test_a_prober_that_never_flags_is_refused(self): """The self-test is the gate's own red-proof. A prober that calls the R-156 canary CLEAN must not be allowed to issue a clean bill of health for the catalog.""" blind = lambda app, app_dir, **kw: VAULTWARDEN # noqa: E731 — flags nothing with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["papra"]) err = io.StringIO() with redirect_stdout(io.StringIO()), redirect_stdout(io.StringIO()): rc = cvp.check(root, prober=blind) self.assertEqual(rc, 2, "a blind prober must yield rc=2, never rc=0") def test_a_prober_that_flags_everything_is_refused(self): """The other direction — a prober that cannot clear a correct template is equally useless.""" crying_wolf = lambda app, app_dir, **kw: PAPRA # noqa: E731 with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["vaultwarden"]) with redirect_stdout(io.StringIO()): rc = cvp.check(root, prober=crying_wolf) self.assertEqual(rc, 2) def test_out_of_circulation_apps_are_skipped_and_named(self): with tempfile.TemporaryDirectory() as td: root = self._catalog(td, ["vaultwarden", "old"], lifecycle={"old": "abandoned"}) with redirect_stdout(io.StringIO()) as buf: rc = cvp.check(root, prober=self._prober({"vaultwarden": VAULTWARDEN})) self.assertEqual(rc, 0) self.assertIn("old (abandoned)", buf.getvalue()) class TestEnvBuilding(unittest.TestCase): def test_every_compose_var_resolves(self): """A var resolving to "" binds a bogus root-owned dir at the container root (felhom-controller deploy.go:571) — the probe would then measure the harness, not the app.""" compose = "services:\n s:\n image: x\n environment:\n - A=${WEIRD_ONE}\n" env = cvp.build_env("app", "subdomain: app\n", compose) self.assertTrue(env.get("WEIRD_ONE")) def test_deploy_field_default_and_generate_are_honoured(self): felhom = ("subdomain: kuma\n" "deploy_fields:\n" " - env_var: SUBDOMAIN\n type: subdomain\n default: \"kuma\"\n" " - env_var: AUTH_SECRET\n type: secret\n generate: \"hex:32\"\n" " - env_var: HDD_PATH\n type: path\n" "app_info:\n tagline: x\n") env = cvp.build_env("kuma", felhom, "image: x ${AUTH_SECRET}") self.assertEqual(env["SUBDOMAIN"], "kuma") self.assertEqual(len(env["AUTH_SECRET"]), 64, "hex:32 is 32 bytes = 64 hex chars") self.assertEqual(env["HDD_PATH"], cvp.SCRATCH_HDD) def test_base64key_carries_the_controller_s_base64_prefix(self): """felhom-controller `deploy.go:904` returns "base64:"+b64. Without the prefix Laravel rejects APP_KEY and bookstack serves 500s — a harness bug that reads as an app defect. Campaign 7 §1.1 and Campaign 10 §4d are both records of a harness corrupting a matrix.""" felhom = "deploy_fields:\n - env_var: APP_KEY\n type: secret\n generate: \"base64key:32\"\n" v = cvp.build_env("bookstack", felhom, "${APP_KEY}")["APP_KEY"] self.assertTrue(v.startswith("base64:"), f"missing the controller's prefix: {v[:12]}…") import base64 self.assertEqual(len(base64.b64decode(v[len("base64:"):])), 32) def test_deploy_fields_block_ends_at_the_next_top_level_key(self): felhom = "deploy_fields:\n - env_var: A\n type: text\napp_info:\n tagline: x\n" self.assertEqual([f["env_var"] for f in cvp.parse_deploy_fields(felhom)], ["A"]) if __name__ == "__main__": unittest.main(verbosity=2)