#!/usr/bin/env python3 """Generate the S5-03 LinuxCNC interp/TP unit adapter golden report.""" from __future__ import annotations import argparse import hashlib import json import shutil import subprocess from pathlib import Path from typing import Any CONTEXT_ID = "linuxcnc-2.10.0-pre0-wasm-m1-py-c1" SOURCE_ROOTS = (Path("linuxcnc/unit_tests/interp"), Path("linuxcnc/unit_tests/tp")) SOURCE_COMMAND = "meson test -C build-linuxcnc test_interp test_blendmath" NATIVE_COMMAND = ("build-native/sim_source_unit_adapter_test",) WASM_COMMAND = ("node", "build-wasm/sim_source_unit_adapter_wasm_test.js") def canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def digest_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def digest_file(path: Path) -> str: return digest_bytes(path.read_bytes()) def run(root: Path, command: tuple[str, ...]) -> dict[str, Any]: result = subprocess.run(command, cwd=root, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) return { "command": list(command), "returncode": result.returncode, "status": "PASSED" if result.returncode == 0 else "FAILED", "stdout_sha256": digest_bytes(result.stdout), "stderr_sha256": digest_bytes(result.stderr), "stdout": result.stdout.decode("utf-8", errors="replace"), "stderr": result.stderr.decode("utf-8", errors="replace"), } def source_files(root: Path) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for source_root in SOURCE_ROOTS: for path in sorted((root / source_root).rglob("*")): if path.is_file(): records.append({ "path": path.relative_to(root).as_posix(), "bytes": path.stat().st_size, "sha256": digest_file(path), }) return records def parse_adapter(result: dict[str, Any]) -> dict[str, Any]: parsed: dict[str, Any] = {key: value for key, value in result.items() if key not in {"stdout", "stderr"}} try: parsed["golden"] = json.loads(result["stdout"]) except json.JSONDecodeError as exc: parsed["parse_error"] = str(exc) parsed["golden"] = None return parsed def upstream_probe(root: Path) -> dict[str, Any]: meson = shutil.which("meson") if meson is None: return { "status": "SOURCE_BLOCKED", "command": SOURCE_COMMAND, "reason": "meson_not_installed", "required_executable": "meson", "native_oracle_claim": False, } build_dir = root / "build-linuxcnc" if not build_dir.is_dir(): return { "status": "SOURCE_BLOCKED", "command": SOURCE_COMMAND, "reason": "build-linuxcnc_missing", "required_executable": meson, "native_oracle_claim": False, } result = run(root, (meson, "test", "-C", str(build_dir), "test_interp", "test_blendmath")) result.pop("stdout", None) result.pop("stderr", None) result["native_oracle_claim"] = result["status"] == "PASSED" return result def generate(root: Path, output: Path) -> dict[str, Any]: native = parse_adapter(run(root, NATIVE_COMMAND)) wasm = parse_adapter(run(root, WASM_COMMAND)) native_golden = native.get("golden") wasm_golden = wasm.get("golden") golden_equal = native_golden is not None and native_golden == wasm_golden ctest = run(root, ("ctest", "--test-dir", "build-native", "-R", "^sim_source_unit_adapter_test$", "--output-on-failure")) ctest.pop("stdout", None) ctest.pop("stderr", None) sources = source_files(root) report: dict[str, Any] = { "schema": "linuxcnc-source-unit-execution.v1", "task_id": "S5-03", "context_id": CONTEXT_ID, "status": "verified" if all((native["status"] == "PASSED", wasm["status"] == "PASSED", golden_equal, ctest["status"] == "PASSED", len(sources) == 8)) else "blocked", "suites": [ { "source_id": "source-unit:interp", "target": "test_interp", "source_root": "linuxcnc/unit_tests/interp", "adapter_target": "sim_source_unit_adapter_test", "wasm_target": "sim_source_unit_adapter_wasm_test", "source_files": [record for record in sources if "/interp/" in record["path"]], }, { "source_id": "source-unit:tp", "target": "test_blendmath", "source_root": "linuxcnc/unit_tests/tp", "adapter_target": "sim_source_unit_adapter_test", "wasm_target": "sim_source_unit_adapter_wasm_test", "source_files": [record for record in sources if "/tp/" in record["path"]], }, ], "source_inventory": {"file_count": len(sources), "files": sources}, "adapter": { "native": native, "wasm": wasm, "golden_diff": { "status": "PASSED" if golden_equal else "FAILED", "byte_identical": native.get("stdout_sha256") == wasm.get("stdout_sha256"), "json_equal": golden_equal, "native_stdout_sha256": native.get("stdout_sha256"), "wasm_stdout_sha256": wasm.get("stdout_sha256"), "first_difference": None if golden_equal else "adapter JSON differs or could not be parsed", }, "case_count": len(native_golden.get("cases", [])) if isinstance(native_golden, dict) else 0, "case_counts": native_golden.get("counts", {}) if isinstance(native_golden, dict) else {}, "aggregate_hash": native_golden.get("aggregate_hash") if isinstance(native_golden, dict) else None, }, "ctest": ctest, "upstream_oracle": upstream_probe(root), "pending_adapters": [ { "source_id": "source-test:linuxcnc/tests/abort/g64/test.sh", "adapter": "WORKER_TASK_MOTION_PENDING", "reason": "G64 blend/tolerance and abort endpoint require Worker Task plus motion queue", }, { "source_id": "source-test:linuxcnc/tests/abort/feed-rate/test.sh", "adapter": "WORKER_TASK_MOTION_PENDING", "reason": "abort/restart timing is owned by Worker Task and fixed-tick motion", }, ], "boundaries": [ "upstream Meson test_interp/test_blendmath is independent source oracle and remains SOURCE_BLOCKED when Meson/dependencies are unavailable", "adapter golden is exact for the nine portable source assertions listed in the suite records", "G64 and abort are not promoted to interpreter parity; they remain Worker Task/motion work", ], } unsigned = {key: value for key, value in report.items() if key != "report_sha256"} report["report_sha256"] = digest_bytes(canonical(unsigned).encode("utf-8")) 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(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--report", type=Path, default=Path("coverage/linuxcnc-source-unit-execution-s5.report.json")) args = parser.parse_args() report = generate(args.repo_root.resolve(), args.repo_root.resolve() / args.report) print(f"linuxcnc_source_unit_s5={report['status']} cases={report['adapter']['case_count']} oracle={report['upstream_oracle']['status']}") return 0 if report["status"] == "verified" else 1 if __name__ == "__main__": raise SystemExit(main())