felhom.eu/scripts/test_image_runtime.py

#!/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`.
"""
import io
import os
import sys
import tarfile
import tempfile

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])


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)
    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())