#!/usr/bin/env python3 """test_image_runtime.py — the image reader's parsers (R-1327) and the DooPlex list maker (R-1328). No network. Pinned by consequence: an Alpine and a Debian package database read back the packages the runtime table matches; Node's version header gives the version; an image reference splits the way the watch's registry_of does (lscr.io is Docker Hub; a bare name is library/); a tar layer streamed top-down returns the TOPMOST copy of a file and the Python `.dist-info` versions; the DooPlex list keeps one line per image:tag with every file that names it, and skips templated and variable images. Red-proof: a scan that keeps the FIRST (bottom) copy instead of the topmost fails `topmost`. ZSTD (R-1327, 2026-10-28): a zstd layer above a gzip one is read with NO `zstd` program on the PATH (the CI runner has none) — through the pinned `zstandard` wheel; a zstd layer that does not decode, and one with no decoder at all, are NOTES (the watch's NOT CHECKED), never a pass; the wheel loader opens a wheel only when its sha256 is the pinned one. Where no decoder can be had (no wheel in the cache, FELHOM_WHEEL_FETCH unset) the read check says NOT CHECKED out loud and is not counted; with FELHOM_WHEEL_FETCH=1 (CI sets it) a missing decoder FAILS. Red-proof: the 2026-10-27 module (the `zstd` program or nothing) fails `zstd: a zstd layer is read with no zstd program`. """ import base64 import hashlib import io import os import sys import tarfile import tempfile import zipfile HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import image_runtime as ir # noqa: E402 import dooplex_inventory as di # noqa: E402 fails, ran = [], 0 def check(name, cond, detail=""): global ran ran += 1 print(" %s %s%s" % ("ok" if cond else "XX", name, "" if cond else " %s" % (detail,))) if not cond: fails.append(name) def tar_bytes(files): buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tf: for name, data in files.items(): ti = tarfile.TarInfo(name) ti.size = len(data) tf.addfile(ti, io.BytesIO(data)) return buf.getvalue() class FakeReg: def __init__(self, blobs): self.blobs = blobs def blob(self, digest, stream=False): return io.BytesIO(self.blobs[digest]) # A tar holding an Alpine package database (nodejs 24.22.1, brotli-libs 1.2.1), compressed by `zstd -19` (2026-10-28). ZSTD_LAYER = base64.b64decode( "KLUv/QRoLQMAsgUSEJB9hC/xJSHf5yhLWm0eYqY+iotURKotdmoB8FiUYcw4RjGQJXy2quX/jpvO2z94QaiU9/dP/+/npPcHKW0610jcI/M24iI" "NCiAwMT3M5Vx0jvfWN08dUPKASwQMGmAZDXf7YaaA") ZSTD_MT = "application/vnd.oci.image.layer.v1.tar+zstd" GZ_MT = "application/vnd.oci.image.layer.v1.tar+gzip" def zstd_checks(): gz = tar_bytes({"usr/local/lib/python3.12/site-packages/brotli-1.2.1.dist-info/RECORD": b"x", "lib/apk/db/installed": b"P:nodejs\nV:20.0.0-r0\n"}) layers = [{"digest": "sha256:gz", "mediaType": GZ_MT, "size": len(gz)}, {"digest": "sha256:zst", "mediaType": ZSTD_MT, "size": len(ZSTD_LAYER)}] reg = FakeReg({"sha256:gz": gz, "sha256:zst": ZSTD_LAYER}) path = os.environ.get("PATH", "") os.environ["PATH"] = tempfile.mkdtemp() # no `zstd` program: the CI runner's case ir._ZSTANDARD.clear() # the decoder is decided once per process: decide it HERE try: found, notes = ir.scan_layers(reg, layers) finally: os.environ["PATH"] = path ir._ZSTANDARD.clear() no_decoder = [n for n in notes if "no zstd decoder" in n] if no_decoder and os.environ.get("FELHOM_WHEEL_FETCH") != "1": print(" NOT CHECKED zstd: a zstd layer is read with no zstd program — %s" % no_decoder[0]) else: check("zstd: a zstd layer is read with no zstd program (the top layer's apk db wins)", (found.get("apk") or {}).get("nodejs") == "24.22.1-r0" and not notes, (found.get("apk"), notes)) check("zstd: the gzip layer under it is still read", (found.get("pydist") or {}).get("brotli") == "1.2.1", found.get("pydist")) found, notes = ir.scan_layers(reg, [{"digest": "sha256:zst", "mediaType": ZSTD_MT, "size": 9}], decoder=(None, "no zstd decoder: test")) ir._ZSTANDARD.clear() how = ir.zstd_decoder_how() check("zstd: the log names the decoder in words", isinstance(how, str) and ("zstandard" in how or "zstd program" in how or how.startswith("NONE")), how) ir._ZSTANDARD.clear() check("zstd: no decoder is a NOTE (NOT CHECKED), never a silent pass", notes and "no zstd decoder" in notes[0] and not found, (found, notes)) mod, _how = ir.zstandard_module() if mod: bad = ZSTD_LAYER[:40] + b"\x00" * 30 + ZSTD_LAYER[70:] found, notes = ir.scan_layers(FakeReg({"sha256:bad": bad}), [{"digest": "sha256:bad", "mediaType": ZSTD_MT, "size": len(bad)}], decoder=("module", mod)) check("zstd: a layer that does not decode is a NOTE (NOT CHECKED), never a pass", any("unreadable" in n for n in notes), (found, notes)) found, notes = ir.scan_layers(FakeReg({"sha256:t": ZSTD_LAYER[:60]}), [{"digest": "sha256:t", "mediaType": ZSTD_MT, "size": 60}], decoder=("module", mod)) check("zstd: a cut-off layer is a NOTE (NOT CHECKED), never a pass", any("unreadable" in n for n in notes), (found, notes)) def wheel_checks(): """The loader opens a wheel only when its sha256 is the pinned one. A fake `zstandard` wheel, served by file://.""" d = tempfile.mkdtemp() whl = os.path.join(d, "zstandard-0.0-fake.whl") with zipfile.ZipFile(whl, "w") as z: z.writestr("zstandard/__init__.py", "__version__ = 'fake-for-test'\n") good = hashlib.sha256(open(whl, "rb").read()).hexdigest() key = ("%d.%d" % sys.version_info[:2], ir._libc()) url = "file://" + whl mod, how = ir.zstandard_module(fetch=True, wheels={key: (url, good)}, cache=os.path.join(d, "c1"), try_import=False) check("wheel: the pinned wheel is fetched, hash-checked and loaded", mod is not None and getattr(mod, "__version__", "") == "fake-for-test" and "sha256 checked" in how, how) mod, how = ir.zstandard_module(fetch=True, wheels={key: (url, "0" * 64)}, cache=os.path.join(d, "c2"), try_import=False) check("wheel: a wheel whose sha256 is not the pinned one is never loaded", mod is None and "not the pinned" in how, how) c3 = os.path.join(d, "c3") os.makedirs(c3) open(os.path.join(c3, os.path.basename(whl)), "wb").write(b"changed") mod, how = ir.zstandard_module(fetch=False, wheels={key: (url, good)}, cache=c3, try_import=False) check("wheel: a changed wheel in the cache is not loaded, and without FETCH nothing is fetched", mod is None and "not in" in how, how) mod, how = ir.zstandard_module(fetch=True, wheels={}, cache=c3, try_import=False) check("wheel: no pin for this Python is a reason, never a guess", mod is None and "no pinned" in how, how) sys.modules.pop("zstandard", None) def retry_checks(): """A registry 429 (ghcr.io refused blob reads in two back-to-back CI dry runs, 2026-10-28) is waited out and tried again; the answer is a refusal only after the last try.""" import io, time, urllib.error, urllib.request calls, slept = [], [] real_open, real_sleep = urllib.request.urlopen, time.sleep class Resp(io.BytesIO): status = 200 headers = {} def fake(answers): def op(req, timeout=60): calls.append(req.full_url) a = answers.pop(0) if len(answers) > 1 else answers[0] if a == 429: raise urllib.error.HTTPError(req.full_url, 429, "Too Many Requests", {"Retry-After": "3"}, io.BytesIO(b"")) return Resp(b"ok") return op try: time.sleep = lambda s: slept.append(s) urllib.request.urlopen = fake([429, 200]) st, _h, body = ir._get("https://ghcr.io/v2/x/blobs/sha256:a") check("retry: a 429 then 200 -> the 200, after waiting its Retry-After", st == 200 and body == b"ok" and len(calls) == 2 and 3 in slept, (st, calls, slept)) calls.clear() urllib.request.urlopen = fake([429]) st, _h, body = ir._get("https://ghcr.io/v2/x/blobs/sha256:a") check("retry: a 429 on every try -> 429 after %d tries, never a pass" % ir.RETRY_TRIES, st == 429 and len(calls) == ir.RETRY_TRIES, (st, len(calls))) finally: urllib.request.urlopen, time.sleep = real_open, real_sleep def main(): apk = ir.parse_apk("P:musl\nV:1.2.5-r10\n\nP:nodejs\nV:24.18.1-r0\n\nP:brotli-libs\nV:1.2.0-r1\n") check("apk database -> nodejs and brotli-libs", apk.get("nodejs") == "24.18.1-r0" and apk.get("brotli-libs") == "1.2.0-r1", apk) dpkg = ir.parse_dpkg("Package: libbrotli1\nStatus: install ok installed\nVersion: 1.0.9-2+b6\n\nPackage: nodejs\nVersion: 20.20.0-1nodesource1\n") check("dpkg database -> libbrotli1 and nodejs", dpkg == {"libbrotli1": "1.0.9-2+b6", "nodejs": "20.20.0-1nodesource1"}, dpkg) h = "#define NODE_MAJOR_VERSION 24\n#define NODE_MINOR_VERSION 22\n#define NODE_PATCH_VERSION 1\n" check("node_version.h -> 24.22.1", ir.parse_node_h(h) == "24.22.1", ir.parse_node_h(h)) check("split_ref: lscr.io is Docker Hub, tag kept", ir.split_ref("lscr.io/linuxserver/qbittorrent:5.3.0") == ("docker.io", "linuxserver/qbittorrent", "5.3.0")) check("split_ref: a bare name is library/, no tag = latest", ir.split_ref("redis") == ("docker.io", "library/redis", "latest")) check("split_ref: index.docker.io is Docker Hub", ir.split_ref("index.docker.io/plexinc/pms-docker:1.43") == ("docker.io", "plexinc/pms-docker", "1.43")) check("split_ref: ghcr.io with a port-free path", ir.split_ref("ghcr.io/home-assistant/home-assistant:2026.9.4") == ("ghcr.io", "home-assistant/home-assistant", "2026.9.4")) # two layers: the bottom one has an OLD apk db and brotli 1.1.0; the top one replaces the db and adds brotli 1.2.1 bottom = tar_bytes({"lib/apk/db/installed": b"P:nodejs\nV:24.18.1-r0\n", "usr/lib/python3/site-packages/Brotli-1.1.0.dist-info/METADATA": b"x"}) top = tar_bytes({"lib/apk/db/installed": b"P:nodejs\nV:24.22.1-r0\n", "usr/local/lib/python3.12/site-packages/brotli-1.2.1.dist-info/RECORD": b"x"}) layers = [{"digest": "sha256:bottom", "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "size": len(bottom)}, {"digest": "sha256:top", "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "size": len(top)}] found, notes = ir.scan_layers(FakeReg({"sha256:bottom": bottom, "sha256:top": top}), layers) check("topmost: the top layer's apk db wins", (found.get("apk") or {}).get("nodejs") == "24.22.1-r0", found.get("apk")) check("topmost: the top layer's Brotli dist-info wins", (found.get("pydist") or {}).get("brotli") == "1.2.1", found.get("pydist")) check("no note for readable gzip layers", notes == [], notes) found, notes = ir.scan_layers(FakeReg({}), [{"digest": "sha256:z", "mediaType": "application/vnd.oci.image.layer.v1.tar+zstd", "size": 10}], budget=0) check("a layer over the byte budget is a NOTE, never a silent pass", notes and ("budget" in notes[0] or "zstd" in notes[0]), notes) zstd_checks() retry_checks() wheel_checks() d = tempfile.mkdtemp() open(os.path.join(d, "a.yaml"), "w").write("spec:\n containers:\n - image: rommapp/romm:5.4.0\n name: x\n" " - image: '{{ .Values.image }}'\n - image: $IMG\n") open(os.path.join(d, "b.yml"), "w").write("image: rommapp/romm:5.4.0\nimage: busybox\n") got = di.collect(d) check("dooplex: one line per image:tag with every file", got.get("rommapp/romm:5.4.0") == {"a.yaml", "b.yml"}, got) check("dooplex: a bare image becomes :latest; templated and variable images are skipped", set(got) == {"rommapp/romm:5.4.0", "busybox:latest"}, set(got)) print() if fails: print("image_runtime tests FAILED: %s" % ", ".join(fails)) return 1 print("image_runtime tests OK — %d checks" % ran) return 0 if __name__ == "__main__": sys.exit(main())