home-network-k3s-impl/conformance/lib/run_l1_offline.py

#!/usr/bin/env python3
"""Run the CLUSTER-INDEPENDENT conformance checks OFFLINE and emit JUnit.

docs/14 puts two layers in CI with no cluster:
  * L0 static/policy  -- "manifests/ServiceDefinitions obey invariants" (tests/l0-static/,
                         each dir carrying a `check.sh`).
  * L1 rendering      -- ServiceDefinition -> rendered resources vs golden expectations
                         (tests/l1-rendering/, each dir carrying a `render-check.sh`).
Both are executed here so they report a REAL pass/fail instead of being skipped with the
cluster-gated layers. run.sh invokes this for `--layer l0`, `--layer l1`, and as the
offline leg of a full run.

Pure standard library (matches the harness's no-pip-dependency posture); the checks
themselves shell out (render checks to catalog/render.py, whose PyYAML dependency lives
there; L0 checks to git + PyYAML).

Usage: run_l1_offline.py --junit <out.xml> [--label-prefix P] <test-dir> [<test-dir> ...]
Exit 0 iff every selected offline check passes.

The per-dir check script is whichever of `render-check.sh` / `check.sh` the directory
carries; a dir with neither is reported SKIPPED (never a pass).
"""
from __future__ import annotations

import argparse
import re
import subprocess
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
from xml.dom import minidom

LABEL_LINE_RE_TMPL = r"^\s*{prefix}/(R-[A-Z0-9]+-[0-9]+)\s*:\s*[\"']?true"

# The offline check script a test dir may carry, in preference order.
CHECK_SCRIPTS = ("render-check.sh", "check.sh")


def layer_of(d: Path) -> str:
    """Layer for the JUnit properties, from the subtree the test lives in."""
    rel = str(d)
    if "l0-static" in rel:
        return "L0"
    if "l1-rendering" in rel:
        return "L1"
    return "OFFLINE"


def collect_meta(test_yaml: Path, prefix: str):
    text = test_yaml.read_text() if test_yaml.exists() else ""
    name_m = re.search(r"^\s*name:\s*(\S+)", text, re.M)
    name = name_m.group(1) if name_m else test_yaml.parent.name
    rx = re.compile(LABEL_LINE_RE_TMPL.format(prefix=re.escape(prefix)), re.I)
    reqs = sorted({m.group(1).upper() for m in (rx.match(l) for l in text.splitlines()) if m})
    return name, reqs


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--junit", type=Path, required=True)
    ap.add_argument("--label-prefix", default="conformance.home.k3s")
    ap.add_argument("dirs", nargs="+", type=Path)
    args = ap.parse_args()

    cases = []  # (name, reqs, dir, rc, output)
    for d in args.dirs:
        check = next((d / s for s in CHECK_SCRIPTS if (d / s).exists()), None)
        name, reqs = collect_meta(d / "chainsaw-test.yaml", args.label_prefix)
        if check is None:
            cases.append((name, reqs, d, 3,
                          f"no offline check script ({' or '.join(CHECK_SCRIPTS)}) in {d}"))
            continue
        proc = subprocess.run(
            ["bash", str(check)], capture_output=True, text=True
        )
        out = (proc.stdout + proc.stderr).strip()
        cases.append((name, reqs, d, proc.returncode, out))

    failures = sum(1 for c in cases if c[3] not in (0, 3))
    skipped = sum(1 for c in cases if c[3] == 3)

    suite = ET.Element(
        "testsuite",
        {
            "name": "home-k3s-conformance L0/L1 (offline)",
            "tests": str(len(cases)),
            "failures": str(failures),
            "errors": "0",
            "skipped": str(skipped),
        },
    )
    for name, reqs, d, rc, out in cases:
        tc = ET.SubElement(
            suite,
            "testcase",
            {"classname": f"{layer_of(d)}:{','.join(reqs) if reqs else 'UNTAGGED'}",
             "name": name},
        )
        props = ET.SubElement(tc, "properties")
        ET.SubElement(props, "property", {"name": "requirements", "value": ",".join(reqs)})
        ET.SubElement(props, "property", {"name": "layer", "value": layer_of(d)})
        ET.SubElement(props, "property", {"name": "path", "value": str(d)})
        if rc == 3:
            ET.SubElement(tc, "skipped", {"message": out or "no offline check script"})
        elif rc != 0:
            fail = ET.SubElement(tc, "failure", {"message": f"offline check exit {rc}"})
            fail.text = out
        status = "PASS" if rc == 0 else ("SKIP" if rc == 3 else "FAIL")
        print(f"[offline] {layer_of(d):3s} {status:4s} {name} "
              f"({','.join(reqs) or 'untagged'})",
              file=sys.stderr)
        if rc not in (0, 3):
            print(out, file=sys.stderr)

    root = ET.Element(
        "testsuites",
        {
            "name": "home-k3s-conformance",
            "tests": str(len(cases)),
            "failures": str(failures),
            "errors": "0",
            "skipped": str(skipped),
        },
    )
    root.append(suite)
    args.junit.parent.mkdir(parents=True, exist_ok=True)
    xml = minidom.parseString(ET.tostring(root, "utf-8")).toprettyxml(indent="  ")
    args.junit.write_text(xml)
    print(f"[offline] wrote JUnit ({len(cases)} case(s), {failures} failure(s)) -> "
          f"{args.junit}", file=sys.stderr)
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())