linuxcncWebWASM/scripts/generate_linuxcnc_source_unit_s5.py

#!/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())