admin
5b1972b7f9
gates / gates (push) Successful in 2s
- fixtures (upgrade_fixtures_box.py): calibre-web (Upload form -> OPDS readback + the served EPUB), wger (web login -> weight entry API), crafty-controller (API v2 roles), uptime-kuma (socket.io polling: setup, login, addStatusPage -> public /api/status-page/<slug>); gitea posts its own first-run installer form (R-624's fixable case) and keeps the admin CLI for an installed one. calibre-web and wger run the template's own after_install on the bench, which has none. - R-735: the bench's `password:N:special` now has the controller's shape (randomWithSpecial); test seen failing first (length 32, no special), then 45/45. - upgrade_boxport / the memory watch: a backend traefik reaches over https (loadbalancer.server.scheme) is reached over https on the bench too (crafty-controller). - upgrade-test: files-before/after-detail.json and `files_changed_detail` NAME the files behind a files_may_change mark (R-734's method, now in the harness); test ChangedFiles. - test_catalog_gates: the gate count was stale (9, the runner has 10) and red on main; now 10. Evidence: felhom.eu/documentation/audits/more-night-apps-2026-09-30/ Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
468 lines
26 KiB
Python
468 lines
26 KiB
Python
#!/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_password_special_carries_the_controllers_shape(self):
|
|
"""R-735: `password:N:special` must be what the controller's randomWithSpecial mints (deploy.go L1335):
|
|
N long, a letter or digit first, at least one lower, upper, digit and one of `-_.!@#%+=`. Without the
|
|
special character calibre-web's own rule refuses the password and the bench cannot seed the app.
|
|
COMPANION RED-PROOF: before the fix `_gen` returned letters and digits only; this test failed."""
|
|
felhom = "deploy_fields:\n - env_var: ADMIN_PASSWORD\n type: password\n generate: \"password:24:special\"\n"
|
|
for _ in range(50):
|
|
v = cvp.build_env("calibre-web", felhom, "${ADMIN_PASSWORD}")["ADMIN_PASSWORD"]
|
|
self.assertEqual(len(v), 24, v)
|
|
self.assertTrue(any(c in "-_.!@#%+=" for c in v), f"no special character: {v}")
|
|
self.assertNotIn(v[0], "-_.!@#%+=")
|
|
for cls in ("abcdefghijklmnopqrstuvwxyz", "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "0123456789"):
|
|
self.assertTrue(any(c in cls for c in v), f"missing one of {cls[:3]}…: {v}")
|
|
plain = "deploy_fields:\n - env_var: P\n type: password\n generate: \"password:16\"\n"
|
|
self.assertTrue(cvp.build_env("x", plain, "${P}")["P"].isalnum()) # the plain form is unchanged
|
|
|
|
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)
|