#!/usr/bin/env python3 """Run one indexed test in a fixed native environment and emit an oracle record.""" from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import shlex import subprocess import sys import time from typing import Any SCHEMA = "native-runner-result.v1" GOLDEN_SCHEMA = "native-golden.v1" DEFAULT_INDEX = Path("coverage/test-entries.jsonl") DEFAULT_REPORT = Path("coverage/native-runner.report.json") DEFAULT_ARTIFACTS = Path("build-native/native-runner-artifacts") MAX_OUTPUT_BYTES = 4 * 1024 * 1024 PORTABLE_ENV = { "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "HOME": ".runner-home", "LANG": "C", "LC_ALL": "C", "TZ": "UTC", "PYTHONHASHSEED": "0", } def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def read_entries(path: Path) -> list[dict[str, object]]: entries: list[dict[str, object]] = [] with path.open("r", encoding="utf-8") as handle: for line_number, line in enumerate(handle, 1): if not line.strip(): continue row = json.loads(line) if not isinstance(row, dict): raise ValueError(f"{path}:{line_number}: expected object") entries.append(row) return entries def environment(root: Path) -> dict[str, str]: result = dict(PORTABLE_ENV) result["PWD"] = str(root) result["LINUXCNC_RUNNER_ROOT"] = str(root) return result def command_for_workdir(root: Path, entry: dict[str, object], workdir: Path) -> list[str]: command = entry.get("command") if not isinstance(command, str) or not command.strip(): raise ValueError("entry has no executable command") tokens = shlex.split(command) entry_path = root / str(entry.get("path", "")) for index, token in enumerate(tokens): candidate = (root / token).resolve() if not token.startswith("/") else Path(token) try: candidate.relative_to(workdir.resolve()) except ValueError: continue if candidate.is_file(): tokens[index] = os.path.relpath(candidate, workdir) if entry_path.is_file() and tokens and tokens[-1] == str(entry.get("path")): tokens[-1] = entry_path.name return tokens def result_digest(result: dict[str, object]) -> str: payload = {key: value for key, value in result.items() if key not in {"duration_ms", "stdout", "stderr"}} return sha256_bytes(canonical_json(payload).encode("utf-8")) def compare_golden(actual: dict[str, object], golden: dict[str, object]) -> dict[str, object]: if golden.get("schema") != GOLDEN_SCHEMA: return {"status": "blocked", "reason": "unsupported_golden_schema"} expected = golden.get("expected", {}) if not isinstance(expected, dict): return {"status": "blocked", "reason": "golden_expected_must_be_object"} differences: list[dict[str, object]] = [] for field, expected_value in sorted(expected.items()): if actual.get(field) != expected_value: differences.append({"field": field, "expected": expected_value, "actual": actual.get(field)}) if differences: return {"status": "different", "differences": differences} return {"status": "passed", "differences": []} def run_entry(root: Path, entry: dict[str, object], artifacts_dir: Path, timeout: float, allow_real_excluded: bool = False, allow_needs_review: bool = False, golden: dict[str, object] | None = None) -> dict[str, object]: portability = str(entry.get("portability", "needs_review")) if portability == "REAL-EXCLUDED" and not allow_real_excluded: return { "schema": SCHEMA, "status": "REAL-EXCLUDED", "test_source_id": entry.get("test_source_id"), "path": entry.get("path"), "portability": portability, "reason": "hardware_or_external_transport_requires_substitute", } if portability == "needs_review" and not allow_needs_review: return { "schema": SCHEMA, "status": "blocked", "test_source_id": entry.get("test_source_id"), "path": entry.get("path"), "portability": portability, "reason": "needs_review_requires_explicit_override", } if entry.get("parse_status") == "blocked": return {"schema": SCHEMA, "status": "blocked", "test_source_id": entry.get("test_source_id"), "path": entry.get("path"), "reason": "entry_parse_blocked"} entry_path = root / str(entry.get("path", "")) workdir = entry_path.parent if entry_path.is_file() else root try: command = command_for_workdir(root, entry, workdir) except (ValueError, OSError) as exc: return {"schema": SCHEMA, "status": "blocked", "test_source_id": entry.get("test_source_id"), "path": entry.get("path"), "reason": str(exc)} try: completed = subprocess.run( command, cwd=workdir, env=environment(root), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, check=False, ) timed_out = False except subprocess.TimeoutExpired as exc: completed = subprocess.CompletedProcess(command, 124, exc.stdout or b"", exc.stderr or b"") timed_out = True stdout = bytes(completed.stdout or b"") stderr = bytes(completed.stderr or b"") artifact_root = artifacts_dir / str(entry.get("test_source_id", "unknown")).replace("/", "_").replace(":", "_") artifact_root.mkdir(parents=True, exist_ok=True) stdout_path = artifact_root / "stdout.bin" stderr_path = artifact_root / "stderr.bin" stdout_path.write_bytes(stdout[:MAX_OUTPUT_BYTES]) stderr_path.write_bytes(stderr[:MAX_OUTPUT_BYTES]) status = "timeout" if timed_out else ("passed" if completed.returncode == 0 else "failed") result: dict[str, object] = { "schema": SCHEMA, "status": status, "test_source_id": entry.get("test_source_id"), "path": entry.get("path"), "command": command, "cwd": str(workdir.relative_to(root)), "portability": portability, "exit_code": completed.returncode, "stdout_bytes": len(stdout), "stderr_bytes": len(stderr), "stdout_sha256": sha256_bytes(stdout), "stderr_sha256": sha256_bytes(stderr), "stdout_artifact": str(stdout_path.relative_to(root)), "stderr_artifact": str(stderr_path.relative_to(root)), "source_sha": entry.get("source_sha"), } if len(stdout) <= MAX_OUTPUT_BYTES: result["stdout"] = stdout.decode("utf-8", errors="replace") if len(stderr) <= MAX_OUTPUT_BYTES: result["stderr"] = stderr.decode("utf-8", errors="replace") result["result_sha256"] = result_digest(result) if golden is not None: result["oracle"] = compare_golden(result, golden) return result def self_test(root: Path, report_path: Path) -> int: completed = subprocess.run( [sys.executable, "-c", "print('native-runner-self-test')"], cwd=root, env=environment(root), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) passed = completed.returncode == 0 and completed.stdout == b"native-runner-self-test\n" and completed.stderr == b"" report = { "schema": "native-runner-report.v1", "status": "passed" if passed else "failed", "self_test": True, "result": { "exit_code": completed.returncode, "stdout_sha256": sha256_bytes(completed.stdout), "stderr_sha256": sha256_bytes(completed.stderr), "expected_stdout_sha256": sha256_bytes(b"native-runner-self-test\n"), }, } report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") print(f"native_runner=self-test status={report['status']}") return 0 if passed else 1 def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--index", type=Path, default=DEFAULT_INDEX) parser.add_argument("--test-source-id") parser.add_argument("--path") parser.add_argument("--timeout", type=float, default=60.0) parser.add_argument("--allow-real-excluded", action="store_true") parser.add_argument("--allow-needs-review", action="store_true") parser.add_argument("--golden", type=Path) parser.add_argument("--artifacts-dir", type=Path, default=DEFAULT_ARTIFACTS) parser.add_argument("--report", type=Path, default=DEFAULT_REPORT) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() root = args.repo_root.resolve() if args.self_test: return self_test(root, root / args.report) entries = read_entries(root / args.index) matches = [entry for entry in entries if (not args.test_source_id or entry.get("test_source_id") == args.test_source_id) and (not args.path or entry.get("path") == args.path)] if len(matches) != 1: parser.error(f"expected exactly one test entry, found {len(matches)}") golden = None if args.golden: golden = json.loads((root / args.golden).read_text(encoding="utf-8")) result = run_entry(root, matches[0], root / args.artifacts_dir, args.timeout, args.allow_real_excluded, args.allow_needs_review, golden) report = {"schema": "native-runner-report.v1", "status": result.get("status"), "result": result} (root / args.report).parent.mkdir(parents=True, exist_ok=True) (root / args.report).write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") print(canonical_json(result)) return 0 if result.get("status") == "passed" else 1 if __name__ == "__main__": raise SystemExit(main())