admin
a32541684a
Moving a template out of templates/ un-offers it but also makes the controller's orphan detector see it as GONE for anyone already running the app - flagging their working install Elavult with a Torles button. Withdrawing an app must never take a working app away from a customer. Optional lifecycle: available|hidden|abandoned in .felhom.yml instead. plant-it returns to templates/ as the first abandoned app; retired/ removed. Resolvability gate skips (and reports) non-available apps.
203 lines
9.9 KiB
Python
203 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Fixture tests for check-image-resolvable.py. NO NETWORK — the resolver is injected.
|
|
|
|
Run: python3 scripts/test_check_image_resolvable.py
|
|
"""
|
|
import importlib.util
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"cir", Path(__file__).resolve().parent / "check-image-resolvable.py")
|
|
cir = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(cir)
|
|
|
|
# One ref that a fake registry "has", one that is knowingly invalid — `.invalid` is RFC-2606
|
|
# reserved, so this can never accidentally succeed even if someone runs the suite online.
|
|
GOOD = "getmeili/meilisearch:v1.49"
|
|
DEAD = "flomp-nonexistent.invalid/wanderer:0.16.0"
|
|
|
|
|
|
def fake_resolver(ref: str):
|
|
"""Resolves exactly one ref. Everything else — including the canary — is reported GONE."""
|
|
return (cir.OK, "") if ref == GOOD else (cir.ABSENT, "manifest unknown")
|
|
|
|
|
|
def throttled_resolver(ref: str):
|
|
"""Docker Hub throttling: rc=0 AND an error on stderr — the live 2026-07-21 failure."""
|
|
return (cir.INCONCLUSIVE, "toomanyrequests: You have reached your unauthenticated pull rate limit")
|
|
|
|
|
|
def make_catalog(tmp: Path, apps: dict[str, list[str]]) -> Path:
|
|
for app, images in apps.items():
|
|
d = tmp / "templates" / app
|
|
d.mkdir(parents=True)
|
|
body = "services:\n" + "".join(
|
|
f" svc{i}:\n image: {img}\n" for i, img in enumerate(images))
|
|
(d / "docker-compose.yml").write_text(body, encoding="utf-8")
|
|
return tmp
|
|
|
|
|
|
class TestResolvabilityGate(unittest.TestCase):
|
|
def test_dead_ref_fails_and_is_named(self):
|
|
"""The whole point: one dead pin must fail the gate, and say which app pins it."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"alive": [GOOD], "rotten": [DEAD]})
|
|
sites, _ = cir.collect_images(root)
|
|
absent, inconclusive = cir.check_images(sites, fake_resolver)
|
|
self.assertEqual(absent, [DEAD], "a dead image pin MUST fail the gate")
|
|
self.assertEqual(inconclusive, [])
|
|
self.assertEqual(sites[DEAD], ["rotten:3"],
|
|
"the failure must point at the app+line that pins it")
|
|
|
|
def test_all_resolvable_passes(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"alive": [GOOD], "also": [GOOD]})
|
|
self.assertEqual(cir.check_images(cir.collect_images(root)[0], fake_resolver), ([], []))
|
|
|
|
def test_same_ref_in_two_apps_is_resolved_once_but_both_sites_reported(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"a": [DEAD], "b": [DEAD]})
|
|
sites, _ = cir.collect_images(root)
|
|
calls = []
|
|
|
|
def counting(ref):
|
|
calls.append(ref)
|
|
return fake_resolver(ref)
|
|
|
|
absent, _ = cir.check_images(sites, counting)
|
|
self.assertEqual(calls, [DEAD], "each unique ref must be resolved exactly once")
|
|
self.assertEqual(sorted(sites[DEAD]), ["a:3", "b:3"])
|
|
self.assertEqual(absent, [DEAD])
|
|
|
|
def test_multi_service_app_collects_every_image(self):
|
|
"""A repo split (wanderer web+db) means one app pins several images — miss one and the
|
|
gate would pass an app that cannot start."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"wanderer": [GOOD, DEAD]})
|
|
sites, _ = cir.collect_images(root)
|
|
self.assertEqual(sorted(sites), sorted([GOOD, DEAD]))
|
|
absent, _ = cir.check_images(sites, fake_resolver)
|
|
self.assertEqual(absent, [DEAD])
|
|
|
|
def test_untrustworthy_resolver_refuses_to_report(self):
|
|
"""RED-PROOF companion for the false-all-clear mode: a resolver that says yes to
|
|
everything (broken docker, an intercepting proxy) must NOT yield a clean bill of health."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"rotten": [DEAD]})
|
|
rc = cir.check(root, resolver=lambda ref: (cir.OK, ""))
|
|
self.assertEqual(rc, 2, "a resolver that resolves the canary must abort, not pass")
|
|
|
|
def test_only_filter_restricts_to_named_apps(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"a": [GOOD], "b": [DEAD]})
|
|
self.assertEqual(sorted(cir.collect_images(root, only=["a"])[0]), [GOOD])
|
|
|
|
def test_check_returns_1_end_to_end_on_a_dead_pin(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"alive": [GOOD], "rotten": [DEAD]})
|
|
self.assertEqual(cir.check(root, resolver=fake_resolver), 1)
|
|
|
|
def test_empty_catalog_is_an_error_not_a_pass(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
(Path(td) / "templates").mkdir()
|
|
self.assertEqual(cir.check(Path(td), resolver=fake_resolver), 2)
|
|
|
|
|
|
class TestClassifyGuardsAgainstFalseAlarms(unittest.TestCase):
|
|
"""RED-PROOF for the defect the first live sweep actually had: Docker Hub throttling was
|
|
reported as 24 dead images (postgres:16-alpine, redis:7-alpine among them). A gate that
|
|
accuses healthy images gets ignored, so ambiguity must resolve to INCONCLUSIVE."""
|
|
|
|
def test_rate_limit_on_rc0_is_inconclusive_not_ok(self):
|
|
"""The live shape: rc=0 WITH an error on stderr. Trusting rc alone reports a false PASS."""
|
|
self.assertEqual(
|
|
cir.classify(0, "toomanyrequests: You have reached your unauthenticated pull rate limit"),
|
|
cir.INCONCLUSIVE)
|
|
|
|
def test_rate_limit_is_never_absent(self):
|
|
for rc in (0, 1):
|
|
self.assertEqual(
|
|
cir.classify(rc, "toomanyrequests: rate limit exceeded"), cir.INCONCLUSIVE,
|
|
"a throttle must NEVER be reported as a missing image")
|
|
|
|
def test_genuine_absence_is_absent(self):
|
|
self.assertEqual(cir.classify(1, "manifest unknown"), cir.ABSENT)
|
|
self.assertEqual(cir.classify(1, "errors:\n denied: requested access to the resource is denied"),
|
|
cir.INCONCLUSIVE, "an auth wall is not proof of absence")
|
|
|
|
def test_clean_success_is_ok(self):
|
|
self.assertEqual(cir.classify(0, ""), cir.OK)
|
|
|
|
def test_unrecognised_failure_is_inconclusive(self):
|
|
self.assertEqual(cir.classify(7, "something nobody has seen before"), cir.INCONCLUSIVE)
|
|
|
|
def test_throttled_sweep_reports_incomplete_not_failure(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = make_catalog(Path(td), {"a": [GOOD], "b": [DEAD]})
|
|
rc = cir.check(root, resolver=lambda ref: (
|
|
(cir.ABSENT, "manifest unknown") if ref == cir.CANARY_REF
|
|
else throttled_resolver(ref)))
|
|
self.assertEqual(rc, 2, "a fully-throttled sweep is INCOMPLETE (2), not a failure (1) "
|
|
"and not a pass (0)")
|
|
|
|
|
|
class TestLifecycleSkipping(unittest.TestCase):
|
|
"""An abandoned app's image is EXPECTED to be gone — it is not offered for new installs. Counting
|
|
it as a failure would leave the gate permanently red for something nobody intends to fix, and a
|
|
gate that is always red is a gate nobody reads."""
|
|
|
|
def _catalog(self, td):
|
|
root = make_catalog(Path(td), {"alive": [GOOD], "dead-app": [DEAD]})
|
|
(root / "templates" / "dead-app" / ".felhom.yml").write_text(
|
|
'slug: "dead-app"\nlifecycle: abandoned\n', encoding="utf-8")
|
|
(root / "templates" / "alive" / ".felhom.yml").write_text('slug: "alive"\n', encoding="utf-8")
|
|
return root
|
|
|
|
def test_abandoned_app_is_skipped_and_reported(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
sites, skipped = cir.collect_images(self._catalog(td))
|
|
self.assertEqual(sorted(sites), [GOOD], "an abandoned app's images must not be checked")
|
|
self.assertEqual(skipped, ["dead-app (abandoned)"],
|
|
"a skipped app must be REPORTED, never silently dropped")
|
|
|
|
def test_abandoned_app_does_not_fail_the_gate(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
self.assertEqual(cir.check(self._catalog(td), resolver=fake_resolver), 0)
|
|
|
|
def test_all_flag_includes_it_again(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = self._catalog(td)
|
|
sites, skipped = cir.collect_images(root, include_unavailable=True)
|
|
self.assertEqual(sorted(sites), sorted([GOOD, DEAD]))
|
|
self.assertEqual(skipped, [])
|
|
self.assertEqual(cir.check(root, resolver=fake_resolver, include_unavailable=True), 1,
|
|
"--all must surface the abandoned app's dead image again")
|
|
|
|
def test_everything_skipped_is_a_pass_not_an_error(self):
|
|
"""All-skipped is a legitimate outcome (e.g. `… plant-it`), not a broken catalog."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = self._catalog(td)
|
|
self.assertEqual(cir.check(root, only=["dead-app"], resolver=fake_resolver), 0)
|
|
|
|
def test_lifecycle_parsing(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
d = Path(td); d.mkdir(exist_ok=True)
|
|
def lc(body):
|
|
(d / ".felhom.yml").write_text(body, encoding="utf-8")
|
|
return cir.app_lifecycle(d)
|
|
self.assertEqual(lc('slug: "x"\n'), "available", "absent field ≡ available")
|
|
self.assertEqual(lc('lifecycle: hidden\n'), "hidden")
|
|
self.assertEqual(lc('lifecycle: "abandoned"\n'), "abandoned", "quoted value")
|
|
self.assertEqual(lc('lifecycle: bogus\n'), "available",
|
|
"an unknown value must degrade to available, never brick the template")
|
|
self.assertEqual(lc(' lifecycle: abandoned\n'), "available",
|
|
"indented => not a top-level key, must not match")
|
|
self.assertEqual(cir.app_lifecycle(Path(td) / "nope"), "available", "no file ≡ available")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|