#!/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 [--label-prefix P] [ ...] 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 `/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())