Files
home-network-k3s-impl/conformance/lib/merge_junit.py
jakman85andClaude Opus 4.8 948c390272 conformance: serialize heavy L2/L3 tests + author alerting SMTP sink
TASK 1 — reliably single-pass green L2/L3:
Split run_chainsaw into a LIGHT pass (concurrent, config parallelism) and a
HEAVY pass (serial, parallel=1). The KubeVirt-CDI/Longhorn/DR tests
(vm-lifecycle, vm-data-survival, vm-default-deny, data-survival, dr-restore)
spike embedded-etcd on the small dev VMs; running them serially in their own
pass keeps their etcd spikes from ever coinciding with another test's setup —
the coincidence that let one heavy test's apiserver blip fail-close Kyverno
admission and abort every in-flight test at 0.01s. Heavy set is keyed by test
dir basename and overridable via CONF_HEAVY_TESTS. Each pass writes its own
JUnit; lib/merge_junit.py stitches them into the one stable report.junit.xml
CI collects, so the split is invisible downstream. Overall rc is the union of
both passes (no masking — real failures still surface).

Proof: full --layer l2,l3 run now completes single-pass green: 24/24 passed,
0 failures/errors/skipped. vpn-killswitch, which previously failed with an
apiserver EOF (etcd blip), passed; no mass environmental setup-aborts.

TASK 2 (T-4.1) — author conformance/tests/l3-fault-injection/alerting/
fixtures/smtp-sink.yaml: a hermetic Mailpit SMTP capture sink (Service
smtp-sink, SMTP 1025, HTTP API 8025) the alerting test applies as its mail
sink. restricted-PodSecurity compliant, in-memory (no PVC). Unblocks the
previously-aborting alerting scaffold; it now passes.

Touches conformance/ + ci-read content only — no apps/ or platform/ workload.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
2026-08-26 02:56:56 -05:00

85 lines
3.1 KiB
Python

#!/usr/bin/env python3
"""Merge several Chainsaw JUnit reports into one.
run.sh splits the L2/L3 run into a LIGHT pass (concurrent) and a HEAVY pass
(serial, parallel=1) so the storage/VM-heavy tests never spike embedded-etcd
while other tests are mid-setup (see run.sh `run_chainsaw`). Each pass writes its
own JUnit file; this stitches them back into the single stable artifact CI
collects (`_artifacts/report.junit.xml`) so the split is invisible downstream.
Chainsaw emits `<testsuites>` with one `<testsuite>` per test. Merging is a
concatenation of the child `<testsuite>` elements plus a recomputed roll-up of
tests/failures/errors/time on the root, so the combined report totals exactly the
sum of the passes (no double counting, no dropped cases). Inputs that don't exist
are skipped (a pass may legitimately select zero tests); at least one must exist.
"""
from __future__ import annotations
import argparse
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
def _int(el: ET.Element, attr: str) -> int:
try:
return int(el.get(attr, "0") or "0")
except ValueError:
return 0
def _float(el: ET.Element, attr: str) -> float:
try:
return float(el.get(attr, "0") or "0")
except ValueError:
return 0.0
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--out", type=Path, required=True, help="merged JUnit path")
ap.add_argument("--name", default="chainsaw-report", help="root testsuites name")
ap.add_argument("inputs", nargs="+", type=Path, help="JUnit files to merge (missing ones skipped)")
args = ap.parse_args()
merged = ET.Element("testsuites", {"name": args.name})
tests = failures = errors = skipped = 0
total_time = 0.0
found = 0
for path in args.inputs:
if not path.exists() or path.stat().st_size == 0:
continue
found += 1
root = ET.parse(path).getroot()
# Accept either a <testsuites> wrapper or a bare <testsuite>.
suites = [root] if root.tag == "testsuite" else list(root.findall("testsuite"))
for suite in suites:
merged.append(suite)
tests += _int(suite, "tests")
failures += _int(suite, "failures")
errors += _int(suite, "errors")
skipped += _int(suite, "skipped")
total_time += _float(suite, "time")
if found == 0:
print(f"merge_junit: none of the input reports exist: {args.inputs}", file=sys.stderr)
return 1
merged.set("tests", str(tests))
merged.set("failures", str(failures))
merged.set("errors", str(errors))
if skipped:
merged.set("skipped", str(skipped))
merged.set("time", f"{total_time:.6f}")
args.out.parent.mkdir(parents=True, exist_ok=True)
ET.ElementTree(merged).write(args.out, encoding="unicode", xml_declaration=False)
print(f"merge_junit: merged {found} report(s) -> {args.out} "
f"(tests={tests} failures={failures} errors={errors} skipped={skipped})")
return 0
if __name__ == "__main__":
raise SystemExit(main())