railiance-cluster/tests/test_knative_render.py

"""Offline proof that install/knative renders the declared CPU requests.

Runs install/knative/render.sh (download + SHA-256 verification + kustomize;
no cluster contact) and compares the result with the upstream assets: the six
declared CPU requests change, the Kourier Service is ClusterIP, the gateway
image is the pinned Envoy digest, and nothing else differs. Skips when the
upstream assets cannot be fetched or kubectl is missing.
"""
from __future__ import annotations

import copy
import hashlib
import os
import re
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path

try:
    import yaml
except ImportError:  # pragma: no cover
    yaml = None

ROOT = Path(__file__).resolve().parents[1]
KNATIVE = ROOT / "install" / "knative"
RAIL_KNATIVE = Path(os.environ.get("RAIL_KNATIVE_DIR", ROOT.parent / "rail-knative"))
DECLARED = {  # (namespace, deployment, container) -> cpu request
    ("knative-serving", "activator", "activator"): "50m",
    ("knative-serving", "autoscaler", "autoscaler"): "30m",
    ("knative-serving", "controller", "controller"): "30m",
    ("knative-serving", "webhook", "webhook"): "30m",
    ("knative-serving", "net-kourier-controller", "controller"): "30m",
    ("kourier-system", "3scale-kourier-gateway", "kourier-gateway"): "50m",
}


def lock() -> dict[str, str]:
    lines = (KNATIVE / "release-lock.env").read_text().splitlines()
    return dict(line.split("=", 1) for line in lines if "=" in line)


def load(path: Path) -> list[dict]:
    return [d for d in yaml.safe_load_all(path.read_text()) if d]


def key(doc: dict) -> tuple:
    meta = doc["metadata"]
    return (doc["apiVersion"], doc["kind"], meta.get("namespace"), meta["name"])


def patch_requests(path: Path) -> dict[tuple, str]:
    out = {}
    for doc in load(path):
        for c in doc["spec"]["template"]["spec"]["containers"]:
            out[(doc["metadata"]["namespace"], doc["metadata"]["name"], c["name"])] = c["resources"]["requests"]["cpu"]
    return out


@unittest.skipIf(yaml is None, "PyYAML not installed")
class DeclarationTests(unittest.TestCase):
    def test_overlays_carry_the_declared_requests(self) -> None:
        got = {}
        for overlay in ("serving-core", "kourier"):
            got.update(patch_requests(KNATIVE / "overlays" / overlay / "cpu-requests.patch.yaml"))
        self.assertEqual(DECLARED, got)

    def test_verify_checks_the_declared_requests(self) -> None:
        rows = re.search(r"<<'LIST'\n(.*?)\nLIST", (KNATIVE / "verify.sh").read_text(), re.S).group(1)
        got = {tuple(r.split()[:3]): r.split()[3] for r in rows.splitlines()}
        self.assertEqual(DECLARED, got)

    def test_matches_rail_knative_declaration(self) -> None:
        source = RAIL_KNATIVE / "substrate" / "v1.22.0" / "cpu-requests.patch.yaml"
        if not source.exists():
            self.skipTest(f"rail-knative not checked out at {RAIL_KNATIVE}")
        self.assertEqual(DECLARED, patch_requests(source))


@unittest.skipIf(yaml is None or shutil.which("kubectl") is None, "needs PyYAML and kubectl")
class RenderTests(unittest.TestCase):
    stage: Path

    @classmethod
    def setUpClass(cls) -> None:
        cls.stage = Path(os.environ.get("KNATIVE_STAGE_DIR") or tempfile.mkdtemp(prefix="knative-render-"))
        result = subprocess.run([str(KNATIVE / "render.sh"), str(cls.stage)], capture_output=True, text=True)
        if result.returncode != 0:
            if "checksum mismatch" in result.stderr:
                raise AssertionError(result.stderr)
            raise unittest.SkipTest(f"render.sh failed (offline?): {result.stderr.strip()[:200]}")

    def compare(self, upstream_file: str, rendered_file: str) -> dict[tuple, dict]:
        upstream = {key(d): d for d in load(self.stage / upstream_file)}
        rendered = {key(d): d for d in load(self.stage / rendered_file)}
        self.assertEqual(set(upstream), set(rendered))
        return {k: (upstream[k], rendered[k]) for k in upstream}

    def check(self, upstream_file: str, rendered_file: str) -> set:
        seen = set()
        envoy = lock()["ENVOY_IMAGE"]
        for k, (up, out) in self.compare(upstream_file, rendered_file).items():
            expected = copy.deepcopy(up)
            if k[1] == "Deployment":
                for c in expected["spec"]["template"]["spec"]["containers"]:
                    ident = (k[2], k[3], c["name"])
                    if ident in DECLARED:
                        c["resources"]["requests"]["cpu"] = DECLARED[ident]
                        seen.add(ident)
                    if c["image"].startswith("docker.io/envoyproxy/envoy:"):
                        c["image"] = envoy
            if k[1:] == ("Service", "kourier-system", "kourier"):
                expected["spec"]["type"] = "ClusterIP"
            self.assertEqual(expected, out, f"{k} differs beyond the declared changes")
        return seen

    def test_crds_are_the_pinned_upstream_file(self) -> None:
        digest = hashlib.sha256((self.stage / "crds.yaml").read_bytes()).hexdigest()
        self.assertEqual(lock()["SERVING_CRDS_SHA256"], digest)
        for doc in load(self.stage / "crds.yaml"):
            self.assertEqual("CustomResourceDefinition", doc["kind"])

    def test_rendered_carries_declared_requests_and_nothing_else_changes(self) -> None:
        seen = self.check("core.yaml", "serving-core.rendered.yaml")
        seen |= self.check("kourier.yaml", "kourier.rendered.yaml")
        self.assertEqual(set(DECLARED), seen)


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