#!/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] [--jobs N] [ ...] 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. PARALLELISM (--jobs, env CONF_OFFLINE_JOBS) ------------------------------------------- The checks are independent processes, so they run in a bounded thread pool (one `bash` child per worker). Default: min(8, usable CPUs); `--jobs 1` is the old serial runner exactly. Each check's stdout+stderr is CAPTURED and only reported after every check has finished, in the ORDER THE DIRS WERE GIVEN, so the log, the JUnit and the step summary are byte-for-byte independent of which check finished first (only the trailing "[N.Ns]" wall-time on each status line varies). The 0/1/3 aggregation runs over the same ordered case list as before. Parallel-safety is a property of the CHECKS, not of this runner: every check must keep its scratch state in a private mktemp/mkdtemp dir and must not write into the tree or a fixed path. The audit that established that for the current set is recorded in decisions/2026-11-08-parallel-l0-checks.md; a new check that needs shared state must say so. Nested pools: the keycloak reconciler checks run their own mutation suites in a pool (mutations.py, env MUTATION_JOBS). When this runner is itself parallel it exports a SMALLER MUTATION_JOBS (unless the caller already set one), so the two suites share the CPUs the other jobs-2 checks are not using instead of each taking min(8, CPUs). See inner_jobs(). """ from __future__ import annotations import argparse import os import re import subprocess import sys import time from concurrent.futures import ThreadPoolExecutor 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 usable_cpus() -> int: """CPUs this process may run on (the cgroup/affinity view, not the host's count).""" try: return max(1, len(os.sched_getaffinity(0))) except (AttributeError, OSError): return max(1, os.cpu_count() or 1) def positive_int(text: str) -> int: n = int(text) if n < 1: raise argparse.ArgumentTypeError(f"must be >= 1, got {n}") return n def inner_jobs(cpus: int, outer: int) -> int: """MUTATION_JOBS to hand the checks when this runner runs `outer` of them at once. Two checks (the keycloak client-secret and passkey reconcilers) fan out internally; the other outer-2 workers are single processes. Give the two suites an equal share of the CPUs those others leave free, never fewer than 2 each (a 1-wide pool would make the 40-mutation passkey suite the critical path of the whole stage), never more than 8. Peak process count is then about (outer-2) + 2*inner, i.e. <= cpus on any machine with cpus >= outer + 2, and at worst ~1.25-1.5x cpus on a small runner — CPU-bound oversubscription of that size costs latency per process, not throughput. """ if outer <= 1: return min(8, cpus) return max(2, min(8, (cpus - outer + 2) // 2)) def run_check(check: Path, env: dict) -> tuple[int, str, float]: t0 = time.monotonic() proc = subprocess.run( ["bash", str(check)], capture_output=True, text=True, stdin=subprocess.DEVNULL, env=env, ) return proc.returncode, (proc.stdout + proc.stderr).strip(), time.monotonic() - t0 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("--jobs", type=positive_int, default=None, help="checks to run at once (default: env CONF_OFFLINE_JOBS, else " "min(8, usable CPUs)); 1 = serial") ap.add_argument("dirs", nargs="+", type=Path) args = ap.parse_args() strict_scopes = set(args.unwired_scope) cpus = usable_cpus() if args.jobs is not None: jobs = args.jobs elif os.environ.get("CONF_OFFLINE_JOBS", "").strip(): try: jobs = positive_int(os.environ["CONF_OFFLINE_JOBS"].strip()) except (ValueError, argparse.ArgumentTypeError) as e: ap.error(f"CONF_OFFLINE_JOBS={os.environ['CONF_OFFLINE_JOBS']!r}: {e}") else: jobs = min(8, cpus) child_env = dict(os.environ) if not child_env.get("MUTATION_JOBS", "").strip(): child_env["MUTATION_JOBS"] = str(inner_jobs(cpus, jobs)) print(f"[offline] running {len(args.dirs)} check(s), jobs={jobs} " f"(cpus={cpus}, MUTATION_JOBS={child_env['MUTATION_JOBS']})", file=sys.stderr) # Metadata first, in the given order; then the checks themselves through the pool. The # results are consumed in SUBMISSION order, so `cases` is identical to the serial runner's. planned = [] # (name, reqs, dir, scope, check-or-None) 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) planned.append((name, reqs, d, scope, check)) cases = [] # (name, reqs, dir, rc, output, scope) elapsed = [] # wall seconds per case, same order (None: nothing ran) with ThreadPoolExecutor(max_workers=jobs) as pool: futures = [pool.submit(run_check, check, child_env) if check is not None else None for (_n, _r, _d, _s, check) in planned] for (name, reqs, d, scope, check), fut in zip(planned, futures): if fut is None: cases.append((name, reqs, d, 3, f"no offline check script ({' or '.join(CHECK_SCRIPTS)}) in {d}", scope)) elapsed.append(None) continue rc, out, secs = fut.result() cases.append((name, reqs, d, rc, out, scope)) elapsed.append(secs) 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), secs in zip(cases, elapsed): 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 "" took = f" [{secs:.1f}s]" if secs is not None else "" print(f"[offline] {layer_of(d):3s} {status:4s} {name} " f"({','.join(reqs) or 'untagged'}){mark}{took}", 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())