#!/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)