app-catalog-felhom.eu/scripts/test_check_image_resolvable.py

#!/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), 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"])), [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)")


if __name__ == "__main__":
    unittest.main(verbosity=2)