123 lines
6.9 KiB
Python
123 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Run real LinuxCNC sim cases in isolated temp directories and record conditions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
REPORT = Path("coverage/native-linuxcnc-case-conditions.report.json")
|
|
CASES = {
|
|
"minimal_xyz": ROOT / "linuxcnc/configs/sim/axis/minimal_xyz.ini",
|
|
"corexy_by_kins": ROOT / "linuxcnc/configs/sim/axis/corexy/corexy_by_kins.ini",
|
|
}
|
|
COREXY_SUPPORT = ("corexy_base.inc", "3joint_xyz.hal")
|
|
MARKERS = ("Starting LinuxCNC", "Starting LinuxCNC server program", "Starting TASK program", "Starting DISPLAY program", "task:")
|
|
|
|
|
|
def digest(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def canonical(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def run_case(name: str, source: Path, destination: Path, *, fixture_resources: bool) -> dict[str, Any]:
|
|
case_dir = destination / name
|
|
case_dir.mkdir(parents=True, exist_ok=True)
|
|
if name == "minimal_xyz":
|
|
shutil.copy2(source, case_dir / source.name)
|
|
else:
|
|
for filename in (source.name, *COREXY_SUPPORT):
|
|
origin = source.parent / filename
|
|
shutil.copy2(origin, case_dir / filename)
|
|
shutil.copy2(ROOT / "linuxcnc/linuxcnc.gif", case_dir / "linuxcnc.gif")
|
|
if fixture_resources:
|
|
(case_dir / "linuxcnc.var").write_text("", encoding="utf-8")
|
|
(case_dir / "tool.tbl").write_text("T1 P1 Z0 D1\n", encoding="utf-8")
|
|
environment = os.environ.copy()
|
|
command = ["timeout", "--signal=TERM", "5s", "linuxcnc", "-r", str(case_dir / source.name)]
|
|
result = subprocess.run(command, cwd=ROOT, env=environment, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
|
|
output = result.stdout.decode("utf-8", errors="replace")
|
|
marker_flags = {marker: marker in output for marker in MARKERS}
|
|
generated = sorted(path.name for path in case_dir.iterdir() if path.is_file() and path.name not in {source.name, *COREXY_SUPPORT, "linuxcnc.gif", "linuxcnc.var", "tool.tbl"})
|
|
return {
|
|
"case": name,
|
|
"fixture_resources": fixture_resources,
|
|
"command": command,
|
|
"returncode": result.returncode,
|
|
"expected_timeout_running": result.returncode == 124,
|
|
"started": marker_flags["Starting LinuxCNC"] and marker_flags["task:"],
|
|
"display_started": marker_flags["Starting DISPLAY program"],
|
|
"task_ticks_observed": marker_flags["task:"],
|
|
"marker_flags": marker_flags,
|
|
"generated_files": generated,
|
|
"stdout_sha256": digest(result.stdout),
|
|
"output_excerpt": "\n".join([line for line in output.splitlines() if any(token in line for token in ("Starting LinuxCNC", "Starting TASK", "Starting DISPLAY", "Falling back", "POSIX", "tool table", "No such", "link (updating", "task:"))][:24]),
|
|
}
|
|
|
|
|
|
def generate(output: Path) -> dict[str, Any]:
|
|
with tempfile.TemporaryDirectory(prefix="linuxcnc-case-conditions-") as directory:
|
|
temp_root = Path(directory)
|
|
minimal = run_case("minimal_xyz", CASES["minimal_xyz"], temp_root, fixture_resources=False)
|
|
corexy_missing = run_case("corexy_by_kins_missing", CASES["corexy_by_kins"], temp_root, fixture_resources=False)
|
|
corexy_fixture = run_case("corexy_by_kins_fixture", CASES["corexy_by_kins"], temp_root, fixture_resources=True)
|
|
lock = json.loads((ROOT / "context-lock.json").read_text(encoding="utf-8"))
|
|
package = subprocess.run(["dpkg-query", "-W", "-f=${Package} ${Version}", "linuxcnc-uspace"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
|
|
package_version = package.stdout.decode("utf-8", errors="replace").strip()
|
|
kernel = subprocess.run(["uname", "-a"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False).stdout.decode("utf-8", errors="replace").strip()
|
|
source_version = (ROOT / "linuxcnc/VERSION").read_text(encoding="utf-8").strip()
|
|
display = os.environ.get("DISPLAY", "")
|
|
passed = minimal["started"] and corexy_missing["started"] and corexy_fixture["started"] and corexy_fixture["expected_timeout_running"]
|
|
report: dict[str, Any] = {
|
|
"schema": "native-linuxcnc-case-conditions.v1",
|
|
"task_id": "REAL-CASE-CONDITIONS-01",
|
|
"context_id": lock.get("context_id"),
|
|
"status": "verified" if passed else "blocked",
|
|
"source_version": source_version,
|
|
"installed_package": package_version,
|
|
"kernel": kernel,
|
|
"environment": {"display": display, "display_available": bool(display), "xvfb": shutil.which("Xvfb") is not None, "linuxcnc": shutil.which("linuxcnc"), "halrun": shutil.which("halrun"), "axis": shutil.which("axis"), "qtvcp": shutil.which("qtvcp")},
|
|
"conditions": {
|
|
"simulation_requires_uspace_or_posix_fallback": True,
|
|
"current_kernel_realtime": "not_preempt_rt" if not any(token in kernel for token in ("PREEMPT_RT", "-rt", "-rtai", "xenomai")) else "realtime_capable",
|
|
"gui_requires_display": True,
|
|
"case_directory_must_be_writable": True,
|
|
"parameter_files_are_runtime_state": True,
|
|
"tool_table_must_exist_for_tool_cases": True,
|
|
"browser_equivalent_must_supply_explicit_workspace_files": True,
|
|
},
|
|
"cases": {"minimal_xyz": minimal, "corexy_missing_resources": corexy_missing, "corexy_with_temporary_fixture": corexy_fixture},
|
|
"boundaries": ["temporary fixture files are not source-owned inputs", "non-realtime POSIX run is simulation-only and cannot validate hardware timing", "installed package is 2.10.0~pre1 while locked source is 2.10.0~pre0", "this report does not promote browser runtime or release status"],
|
|
"inputs": [{"path": "context-lock.json", "sha256": digest((ROOT / "context-lock.json").read_bytes())}, {"path": "linuxcnc/VERSION", "sha256": digest((ROOT / "linuxcnc/VERSION").read_bytes())}, *[{"path": path.relative_to(ROOT).as_posix(), "sha256": digest(path.read_bytes())} for path in CASES.values()]],
|
|
}
|
|
report["report_sha256"] = hashlib.sha256(canonical({key: value for key, value in report.items() if key != "report_sha256"}).encode()).hexdigest()
|
|
target = output if output.is_absolute() else ROOT / output
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.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("--report", type=Path, default=REPORT)
|
|
args = parser.parse_args()
|
|
report = generate(args.report)
|
|
print(f"native_linuxcnc_case_conditions={report['status']} source={report['source_version']} installed={report['installed_package']} report_sha256={report['report_sha256']}")
|
|
return 0 if report["status"] == "verified" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|