persistence sweep: 53 templates measured; gramps-web + wishlist fixed; runtime gate added
Campaign 10's R-156 found papra writing its database into the container's writable layer while the volume the template preserves stayed empty — a backup that completes, verifies, and contains nothing. papra was never the point: nothing anywhere checked that the folder a template preserves is the folder the app writes to. All 53 templates have now been measured live. 43 CLEAN / 3 BROKEN / 7 UNDETERMINED. UNDETERMINED is counted separately, each with its reason, and never folded into CLEAN. FIXED (neither app is deployed anywhere, so nothing was stranded): - gramps-web mounted /app/data, /app/media, /tmp — and /app/data is a path the application never writes. Its accounts database and ITS FAMILY TREE both landed in the writable layer while gramps_data was tarred nightly as an empty directory. Now persists the eight paths the image's own environment names, matching upstream's reference compose. Proven: users.sqlite and the family-tree files survive a redeploy byte-identical, same inode. - wishlist mounted wishlist_data:/data, another path the app never writes; prod.db landed in the ANONYMOUS volume from the image's VOLUME directive — absent from ResolveDockerVolumeNames, so never backed up, and orphaned by a redeploy. Now mounts /usr/src/app/data + /usr/src/app/uploads. Proven: prod.db byte-identical, same inode, across a redeploy. Every corrected path confirmed by two independent sources — the shipped image's own environment/Config.Volumes and upstream's reference compose — never inferred from a directory name. papra is NOT fixed. It is live on one box, and changing the mount target makes the next compose up recreate the container and destroy the writable layer its documents live in. The fix is prepared and proven in the scratch guest (current: db.sqlite differs after a redeploy, so a real account created via the API is lost; fixed: byte-identical, it survives). Referred to the operator with the two options; no migration written. NEW GATE scripts/check-volume-persistence.py — the third catalog gate and the only RUNTIME one. This class is invisible to static analysis, measured not assumed: a static audit of all 53 composes reports the catalog clean AND reports papra clean. Exit 0 clean / 1 REFUSED / 2 undecided. It refuses to report at all unless it has just re-proven itself in both directions against two canary templates that differ only in which path the volume mounts at, so every run carries a live demonstration of R-156 and of its fix. No docker exec anywhere (Campaign 7 §1.1). 44 fixture tests driving check(), the function __main__ calls; every rule red-proofed. Enforcement is convention, not CI — this repo has no CI. Stated plainly in the report; raising it is proposed as R-160. Report, per-app evidence, proofs and proposed register entries (R-158..R-161, NOT filed — felhom.eu is fenced this session): audits/persistence-sweep-2026-08-02/
This commit is contained in:
@@ -0,0 +1,451 @@
|
||||
#!/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/<uuid>/{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)
|
||||
Reference in New Issue
Block a user