309 lines
14 KiB
Python
309 lines
14 KiB
Python
#!/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] [--jobs N] <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.
|
|
|
|
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())
|