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

DECLARED SCOPE, AND WHY A SKIP IS NOT ALWAYS HARMLESS (--unwired-scope)
----------------------------------------------------------------------
Exit 0 here means "every selected check passed"; a check that exits 3 is SKIPPED and, by
default, does not fail the run (a scaffold with no check script, an audit whose subject
subtree does not exist yet). That default is WRONG for a WHOLE-REPO invariant. Those are the
checks ci/l0.sh used to run as hand-written "UNCONDITIONAL" stages, and each of those stages
mapped rc 3 to the GATE's rc 3 with the words "not a pass":

    3) skip "substitution check SKIPPED (clusters/ absent) — not a pass"
       summary "- subst-safety: **SKIPPED** (nothing to audit)"
       if [ "$rc" -eq 0 ]; then rc=3; fi

A whole-repo check reporting "nothing to audit" is reporting that the tree it audits has
VANISHED — which is a louder problem than the check failing, not a quieter one. So:

  --unwired-scope SCOPE   a test DECLARING `<prefix>/scope: SCOPE` whose check SKIPS (rc 3,
                          or no check script at all) makes this runner exit 3 instead of 0.
                          Hard failures still win (exit 1); rc 3 never masks one.

Exit codes: 0 all selected checks passed / 1 at least one FAILED / 3 nothing failed but a
scope-declared check was UNWIRED. conformance/run.sh propagates 3 verbatim; ci/l0.sh turns it
into the gate's own rc 3. This is the ONE place the "unwired, not green" rule lives now,
instead of once per hand-written stage.
"""
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 scope_of(test_yaml: Path, annotation: str) -> str:
    """The scope a test DECLARES, or "" when it declares none.

    Matched on a real YAML mapping line: a `#` comment cannot begin with the annotation key,
    so a scope named in the test's header prose is documentation, not a declaration — the
    same "reference, not declaration" rule ci/touched-requirements.sh applies to R-IDs.
    Mirrors conf_dirs_for_scope() in lib/common.sh; they must agree.
    """
    if not test_yaml.exists():
        return ""
    rx = re.compile(r"^\s*" + re.escape(annotation) + r"\s*:\s*[\"']?([A-Za-z0-9_.-]+)[\"']?\s*$")
    for line in test_yaml.read_text().splitlines():
        m = rx.match(line)
        if m:
            return m.group(1)
    return ""


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("--scope-annotation", default="conformance.home.k3s/scope",
                    help="annotation key a test uses to DECLARE its invariant's scope")
    ap.add_argument("--unwired-scope", action="append", default=[], metavar="SCOPE",
                    help="a test declaring this scope that SKIPS is UNWIRED, not green "
                         "(makes this runner exit 3); repeatable")
    ap.add_argument("--summary-out", type=Path, default=None,
                    help="write one Markdown line per case here, for the CI step summary")
    ap.add_argument("dirs", nargs="+", type=Path)
    args = ap.parse_args()

    strict_scopes = set(args.unwired_scope)

    cases = []  # (name, reqs, dir, rc, output, scope)
    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)
        scope = scope_of(d / "chainsaw-test.yaml", args.scope_annotation)
        if check is None:
            cases.append((name, reqs, d, 3,
                          f"no offline check script ({' or '.join(CHECK_SCRIPTS)}) in {d}",
                          scope))
            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, scope))

    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)
    # A SKIP of a scope-declared check is UNWIRED, not green (see the module docstring).
    unwired = [c for c in cases if c[3] == 3 and c[5] in strict_scopes]

    suite = ET.Element(
        "testsuite",
        {
            "name": "home-k3s-conformance L0/L1 (offline)",
            "tests": str(len(cases)),
            "failures": str(failures),
            "errors": "0",
            "skipped": str(skipped),
        },
    )
    summary_lines = []
    for name, reqs, d, rc, out, scope 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)})
        ET.SubElement(props, "property", {"name": "scope", "value": scope})
        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")
        mark = f" scope={scope}" if scope else ""
        print(f"[offline] {layer_of(d):3s} {status:4s} {name} "
              f"({','.join(reqs) or 'untagged'}){mark}",
              file=sys.stderr)
        if rc not in (0, 3):
            print(out, file=sys.stderr)
        # One step-summary line per case, naming what ran — the generated replacement for the
        # hand-written `summary "- subst-safety: **passed**"` line each deleted stage carried.
        tag = f" ({scope})" if scope else ""
        if rc == 0:
            verdict = "**passed**"
        elif rc == 3:
            verdict = ("**SKIPPED** (UNWIRED, not green)" if scope in strict_scopes
                       else "**SKIPPED** (nothing to audit)")
        else:
            verdict = f"**FAILED** (exit {rc})"
        summary_lines.append(f"- {layer_of(d).lower()}/{name}{tag}: {verdict}")
        if rc == 3 and scope in strict_scopes:
            print(f"[offline] {name}: SKIPPED, and it declares scope={scope!r} — "
                  f"UNWIRED, NOT GREEN. {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)
    if args.summary_out is not None:
        args.summary_out.parent.mkdir(parents=True, exist_ok=True)
        args.summary_out.write_text("\n".join(summary_lines) + ("\n" if summary_lines else ""))

    print(f"[offline] wrote JUnit ({len(cases)} case(s), {failures} failure(s), "
          f"{skipped} skip(s), {len(unwired)} unwired) -> {args.junit}", file=sys.stderr)
    if failures:
        return 1
    if unwired:
        print("[offline] UNWIRED (declared scope, SKIPPED — not a pass): "
              + ", ".join(c[0] for c in unwired), file=sys.stderr)
        return 3
    return 0


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