85 lines
5.1 KiB
Python
85 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute a bounded representative subset through LinuxCNC scripts/runtests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MATRIX = ROOT / "coverage/linuxcnc-source-test-s5.report.json"
|
|
OUTPUT = ROOT / "coverage/linuxcnc-source-test-execution-s5.report.json"
|
|
SUMMARY_RE = re.compile(r"Runtest: (\d+) tests run, (\d+) successful, (\d+) failed \+ (\d+) expected, (\d+) skipped")
|
|
|
|
|
|
def digest(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def canonical(value: object) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
|
|
def choose(entries: list[dict[str, object]], limit: int) -> list[dict[str, object]]:
|
|
pure = [entry for entry in entries if entry["adapter"] in {"PURE_REGRESSION", "PURE_FUNCTIONAL"}]
|
|
selected: list[dict[str, object]] = []
|
|
for adapter in ("PURE_REGRESSION", "PURE_FUNCTIONAL"):
|
|
selected.extend([entry for entry in pure if entry["adapter"] == adapter][: max(1, limit // 2)])
|
|
return selected[:limit]
|
|
|
|
|
|
def execute(entry: dict[str, object], timeout: int) -> dict[str, object]:
|
|
command = ["linuxcnc/scripts/runtests", "-u", str(entry["test_directory"])]
|
|
try:
|
|
result = subprocess.run(command, cwd=ROOT, capture_output=True, timeout=timeout, check=False)
|
|
status = "PASSED" if result.returncode == 0 else "FAILED"
|
|
summary = None
|
|
match = SUMMARY_RE.search(result.stdout.decode("utf-8", errors="replace"))
|
|
if match: summary = {"tests_run": int(match.group(1)), "successful": int(match.group(2)), "failed": int(match.group(3)), "expected": int(match.group(4)), "skipped": int(match.group(5))}
|
|
return {"source_id": entry["source_id"], "test_directory": entry["test_directory"], "adapter": entry["adapter"], "command": command, "status": status, "returncode": result.returncode, "summary": summary, "stdout_sha256": digest(result.stdout), "stderr_sha256": digest(result.stderr), "expected_hash": next((item["sha256"] for item in entry["files"] if item["path"].endswith("/expected")), None), "source_hash": entry["source_hash"], "wasm_worker_status": "ADAPTER_PENDING"}
|
|
except subprocess.TimeoutExpired as error:
|
|
return {"source_id": entry["source_id"], "test_directory": entry["test_directory"], "adapter": entry["adapter"], "command": command, "status": "TIMEOUT", "returncode": None, "summary": None, "stdout_sha256": digest(error.stdout or b""), "stderr_sha256": digest(error.stderr or b""), "expected_hash": next((item["sha256"] for item in entry["files"] if item["path"].endswith("/expected")), None), "source_hash": entry["source_hash"], "wasm_worker_status": "ADAPTER_PENDING"}
|
|
|
|
|
|
def generate(output: Path = OUTPUT, limit: int = 12, timeout: int = 60) -> dict[str, object]:
|
|
matrix = json.loads(MATRIX.read_text(encoding="utf-8"))
|
|
selected = choose(matrix["source_tests"], limit)
|
|
results = [execute(entry, timeout) for entry in selected]
|
|
counts = {status: sum(item["status"] == status for item in results) for status in ("PASSED", "FAILED", "TIMEOUT")}
|
|
report: dict[str, object] = {
|
|
"schema": "linuxcnc-source-test-execution.v1",
|
|
"task_id": "S5-02",
|
|
"context_id": matrix["context_id"],
|
|
"status": "verified" if results and counts["TIMEOUT"] == 0 else "blocked",
|
|
"runner": {"command": "linuxcnc/scripts/runtests -u <test_directory>", "source_runner": "linuxcnc/scripts/runtests", "timeout_seconds": timeout, "locale": "C"},
|
|
"selection": {"requested": limit, "selected": len(selected), "policy": "sorted PURE_REGRESSION and PURE_FUNCTIONAL, equal adapter share"},
|
|
"counts": counts,
|
|
"results": results,
|
|
"wasm_worker_pending": [item["source_id"] for item in results if item["wasm_worker_status"] == "ADAPTER_PENDING"],
|
|
"boundaries": ["this task establishes native LinuxCNC source baseline; WASM/Worker adapter execution is S5-03", "HOST_EXCLUDED and SOURCE_BLOCKED entries are not counted as native candidates", "no source test result is promoted to official oracle without exact adapter parity"],
|
|
"report_sha256": None,
|
|
}
|
|
report["report_sha256"] = digest(canonical({key: value for key, value in report.items() if key != "report_sha256"}))
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
|
return report
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--limit", type=int, default=12)
|
|
parser.add_argument("--timeout", type=int, default=60)
|
|
parser.add_argument("--output", type=Path, default=OUTPUT)
|
|
args = parser.parse_args()
|
|
report = generate(args.output if args.output.is_absolute() else ROOT / args.output, args.limit, args.timeout)
|
|
print(f"linuxcnc_source_test_execution={report['status']} selected={report['selection']['selected']} passed={report['counts']['PASSED']} failed={report['counts']['FAILED']}")
|
|
return 0 if report["status"] == "verified" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|