catalog: lifecycle field replaces the retired/ directory move
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.
This commit is contained in:
@@ -11,7 +11,8 @@ this repo, so nothing in a change-triggered gate would ever notice.
|
||||
This resolves each unique `image:` pin against its registry with
|
||||
`docker manifest inspect <ref>` and exits non-zero listing everything that did not resolve.
|
||||
|
||||
python3 scripts/check-image-resolvable.py # whole catalog
|
||||
python3 scripts/check-image-resolvable.py # every AVAILABLE app
|
||||
python3 scripts/check-image-resolvable.py --all # include hidden/abandoned apps too
|
||||
python3 scripts/check-image-resolvable.py wanderer … # only these app dirs
|
||||
|
||||
THE TRAP THIS SCRIPT IS BUILT AROUND: gate on EACH `docker manifest inspect`'s OWN exit code, one
|
||||
@@ -37,18 +38,47 @@ IMAGE_RE = re.compile(r"^\s*image:\s*[\"']?([^\s\"'#]+)") # same shape as check
|
||||
CANARY_REF = "felhom-nonexistent.invalid/no/such:image"
|
||||
|
||||
|
||||
def collect_images(root: Path, only: list[str] | None = None) -> dict[str, list[str]]:
|
||||
"""Map each unique image ref -> the ['app:line'] sites that pin it. Pure; no network."""
|
||||
# `lifecycle:` at the top level of .felhom.yml. Matched with a line regex rather than a YAML parse so
|
||||
# this gate keeps working with no dependencies (the repo ships no requirements file).
|
||||
LIFECYCLE_RE = re.compile(r"""^lifecycle:\s*["']?([a-z]+)""", re.MULTILINE)
|
||||
|
||||
|
||||
def app_lifecycle(app_dir: Path) -> str:
|
||||
"""available / hidden / abandoned. Absent, empty or unknown ≡ available (see CLAUDE.md)."""
|
||||
f = app_dir / ".felhom.yml"
|
||||
if not f.is_file():
|
||||
return "available"
|
||||
m = LIFECYCLE_RE.search(f.read_text(encoding="utf-8"))
|
||||
if not m:
|
||||
return "available"
|
||||
v = m.group(1)
|
||||
return v if v in ("available", "hidden", "abandoned") else "available"
|
||||
|
||||
|
||||
def collect_images(root: Path, only: list[str] | None = None,
|
||||
include_unavailable: bool = False) -> tuple[dict[str, list[str]], list[str]]:
|
||||
"""Map each unique image ref -> the ['app:line'] sites that pin it. Pure; no network.
|
||||
|
||||
Returns (sites, skipped_apps). Apps whose `lifecycle:` is not `available` are SKIPPED by default:
|
||||
they are not offered for new installs, so a dead upstream image is the expected end state, not a
|
||||
finding. Including them would leave the gate permanently red for a reason nobody intends to fix —
|
||||
and a gate that is always red is a gate nobody reads. They are reported, never silently dropped.
|
||||
"""
|
||||
sites: dict[str, list[str]] = {}
|
||||
skipped: list[str] = []
|
||||
for f in sorted(root.glob("templates/*/docker-compose.yml")):
|
||||
app = f.parent.name
|
||||
if only and app not in only:
|
||||
continue
|
||||
lc = app_lifecycle(f.parent)
|
||||
if lc != "available" and not include_unavailable:
|
||||
skipped.append(f"{app} ({lc})")
|
||||
continue
|
||||
for lineno, line in enumerate(f.read_text(encoding="utf-8").splitlines(), 1):
|
||||
m = IMAGE_RE.match(line)
|
||||
if m:
|
||||
sites.setdefault(m.group(1), []).append(f"{app}:{lineno}")
|
||||
return sites
|
||||
return sites, skipped
|
||||
|
||||
|
||||
OK, ABSENT, INCONCLUSIVE = "ok", "absent", "inconclusive"
|
||||
@@ -129,9 +159,18 @@ def check_images(sites: dict[str, list[str]], resolver) -> tuple[list[str], list
|
||||
return absent, inconclusive
|
||||
|
||||
|
||||
def check(root: Path, only: list[str] | None = None, resolver=docker_resolver) -> int:
|
||||
sites = collect_images(root, only)
|
||||
def check(root: Path, only: list[str] | None = None, resolver=docker_resolver,
|
||||
include_unavailable: bool = False) -> int:
|
||||
sites, skipped = collect_images(root, only, include_unavailable)
|
||||
if skipped:
|
||||
print(f"skipping {len(skipped)} app(s) not offered for new installs: {', '.join(skipped)}")
|
||||
print(" (their images are not expected to resolve; re-run with --all to check them anyway)")
|
||||
if not sites:
|
||||
if skipped:
|
||||
# Everything in scope was deliberately skipped. That is a clean result, not a broken
|
||||
# catalog — saying "ERROR: no images found" here would be a false alarm of its own.
|
||||
print("nothing to check — every app in scope is out of circulation")
|
||||
return 0
|
||||
print(f"ERROR: no images found under {root}/templates/", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
@@ -174,4 +213,8 @@ def check(root: Path, only: list[str] | None = None, resolver=docker_resolver) -
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(check(Path(__file__).resolve().parent.parent, only=sys.argv[1:] or None))
|
||||
argv = sys.argv[1:]
|
||||
all_apps = "--all" in argv
|
||||
argv = [a for a in argv if a != "--all"]
|
||||
sys.exit(check(Path(__file__).resolve().parent.parent, only=argv or None,
|
||||
include_unavailable=all_apps))
|
||||
|
||||
@@ -45,7 +45,7 @@ class TestResolvabilityGate(unittest.TestCase):
|
||||
"""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)
|
||||
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, [])
|
||||
@@ -55,12 +55,12 @@ class TestResolvabilityGate(unittest.TestCase):
|
||||
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), fake_resolver), ([], []))
|
||||
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)
|
||||
sites, _ = cir.collect_images(root)
|
||||
calls = []
|
||||
|
||||
def counting(ref):
|
||||
@@ -77,7 +77,7 @@ class TestResolvabilityGate(unittest.TestCase):
|
||||
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)
|
||||
sites, _ = cir.collect_images(root)
|
||||
self.assertEqual(sorted(sites), sorted([GOOD, DEAD]))
|
||||
absent, _ = cir.check_images(sites, fake_resolver)
|
||||
self.assertEqual(absent, [DEAD])
|
||||
@@ -93,7 +93,7 @@ class TestResolvabilityGate(unittest.TestCase):
|
||||
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"])), [GOOD])
|
||||
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:
|
||||
@@ -144,5 +144,59 @@ class TestClassifyGuardsAgainstFalseAlarms(unittest.TestCase):
|
||||
"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)
|
||||
|
||||
Reference in New Issue
Block a user