linuxcncWebWASM/scripts/generate_test_entries.py

#!/usr/bin/env python3
"""Generate a deterministic index of LinuxCNC test and unit-test entrypoints."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
from collections import Counter
from pathlib import Path


TEST_ROOT = Path("linuxcnc/tests")
UNIT_ROOT = Path("linuxcnc/unit_tests")
MESON_PATHS = (Path("linuxcnc/meson.build"),)
ENTRY_NAMES = {"test.sh", "test.py", "runstreamer", "checkresult"}
HARDWARE_TOKENS = (
    "/dev/", "hm2", "hostmot2", "mesa", "parport", "gpio", "pci", "modbus",
    "serial", "rtai", "xenomai", "latency", "linuxcncrsh", "mqtt", "tcp", "usb",
)
REALTIME_TOKENS = ("$realtime", "realtime", "rtapi", "halcmd", "halrun", "loadrt")
CONFIG_SUFFIXES = {".ini", ".hal", ".tbl", ".var", ".ngc", ".nc", ".tap", ".py", ".tcl"}
CONFIG_TOKEN_RE = re.compile(r"(?<![A-Za-z0-9_./~-])([^\s\"'=;,]+)")


def sha256(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 rel(root: Path, path: Path) -> str:
    # Keep symlinked/shared test paths distinct; hashing still follows the link.
    return path.absolute().relative_to(root.absolute()).as_posix()


def read_text(path: Path) -> tuple[str | None, str | None]:
    try:
        return path.read_text(encoding="utf-8", errors="strict"), None
    except (OSError, UnicodeError) as exc:
        return None, f"{type(exc).__name__}: {exc}"


def classify(path: Path, text: str) -> tuple[str, str]:
    haystack = f"{path.as_posix()}\n{text}".lower()
    if any(token in haystack for token in HARDWARE_TOKENS):
        return "REAL-EXCLUDED", "hardware_or_external_transport"
    if any(token in haystack for token in REALTIME_TOKENS):
        return "needs_review", "realtime_or_linuxcnc_process_dependency"
    return "portable_candidate", "simulated_or_library_only"


def directory_context(root: Path, path: Path) -> tuple[list[str], list[str]]:
    directory = path.parent
    files = sorted(candidate for candidate in directory.iterdir() if candidate.is_file())
    inputs = [rel(root, candidate) for candidate in files if candidate != path]
    expected = [
        rel(root, candidate) for candidate in files
        if candidate.name.lower().startswith(("expected", "result", "golden"))
        or candidate.name.lower() in {"expected", "expected.txt", "result", "golden"}
    ]
    return inputs, expected


def config_refs(root: Path, path: Path, text: str) -> list[str]:
    refs: set[str] = set()
    for token in CONFIG_TOKEN_RE.findall(text):
        if Path(token).suffix.lower() not in CONFIG_SUFFIXES:
            continue
        candidate = (path.parent / token).resolve()
        try:
            candidate.relative_to(root.resolve())
        except ValueError:
            continue
        refs.add(rel(root, candidate))
    return sorted(refs)


def entry(root: Path, path: Path, test_kind: str, command: str) -> dict[str, object]:
    text, error = read_text(path)
    if text is None:
        text = ""
    portability, reason = classify(path, text)
    inputs, expected = directory_context(root, path)
    status = "blocked" if error else "parsed"
    row: dict[str, object] = {
        "test_source_id": f"test:{rel(root, path)}",
        "path": rel(root, path),
        "test_kind": test_kind,
        "command": command,
        "inputs": inputs,
        "expected": expected,
        "config_refs": config_refs(root, path, text),
        "portability": portability,
        "portability_reason": reason,
        "source_sha": sha256(path) if error is None else None,
        "parse_status": status,
    }
    if error:
        row["parse_error"] = error
    return row


def script_command(root: Path, path: Path) -> tuple[str, str]:
    name = path.name
    if name == "test.py":
        return "python_test", f"python3 {rel(root, path)}"
    if name == "runstreamer":
        return "stream_test", f"bash {rel(root, path)}"
    if name == "checkresult":
        text, _ = read_text(path)
        interpreter = "python3" if text and "python" in text.splitlines()[0].lower() else "bash"
        return "oracle_check", f"{interpreter} {rel(root, path)}"
    return "shell_test", f"bash {rel(root, path)}"


def meson_entries(root: Path) -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []
    paths = [root / path for path in MESON_PATHS]
    paths.extend(sorted((root / UNIT_ROOT).rglob("meson.build")))
    for path in sorted(set(paths)):
        if not path.is_file():
            continue
        text, error = read_text(path)
        if text is None:
            text = ""
        portability, reason = classify(path, text)
        source_path = rel(root, path)
        targets = sorted(set(re.findall(r"\btest\s*\(\s*['\"]([^'\"]+)", text)))
        targets.extend(sorted(set(re.findall(r"executable\s*\(\s*['\"](test_[^'\"]+)", text))))
        target_names = sorted(set(targets))
        command = "meson test -C build-native --suite unit_tests"
        row: dict[str, object] = {
            "test_source_id": f"test:{source_path}",
            "path": source_path,
            "test_kind": "unit_test_build",
            "command": command,
            "inputs": sorted(rel(root, child) for child in path.parent.iterdir() if child.is_file()),
            "expected": [],
            "config_refs": [],
            "meson_targets": target_names,
            "portability": portability,
            "portability_reason": reason,
            "source_sha": sha256(path) if error is None else None,
            "parse_status": "blocked" if error else ("parsed" if target_names else "needs_review"),
        }
        if error:
            row["parse_error"] = error
        rows.append(row)
    return rows


def unit_source_entries(root: Path) -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []
    if not (root / UNIT_ROOT).is_dir():
        return rows
    for path in sorted((root / UNIT_ROOT).rglob("*")):
        if not path.is_file() or path.suffix.lower() not in {".c", ".cc", ".cpp", ".cxx"}:
            continue
        if not (path.name.startswith("test_") or path.name.startswith("tests_")):
            continue
        source_path = rel(root, path)
        portability, reason = classify(path, read_text(path)[0] or "")
        rows.append({
            "test_source_id": f"test:{source_path}",
            "path": source_path,
            "test_kind": "unit_test_source",
            "command": f"meson test -C build-native {path.stem.removeprefix('test_')}",
            "inputs": [source_path],
            "expected": [],
            "config_refs": [],
            "portability": portability,
            "portability_reason": reason,
            "source_sha": sha256(path),
            "parse_status": "parsed",
        })
    return rows


def generate(root: Path, output: Path, report_path: Path) -> tuple[int, str]:
    roots = [root / TEST_ROOT, root / UNIT_ROOT]
    rows: list[dict[str, object]] = []
    parse_errors: list[dict[str, str]] = []
    for scan_root in roots:
        if not scan_root.is_dir():
            continue
        for path in sorted(scan_root.rglob("*")):
            if not path.is_file() or path.name not in ENTRY_NAMES:
                continue
            kind, command = script_command(root, path)
            row = entry(root, path, kind, command)
            rows.append(row)
            if row["parse_status"] == "blocked":
                parse_errors.append({"path": str(row["path"]), "error": str(row.get("parse_error", ""))})
    rows.extend(meson_entries(root))
    rows.extend(unit_source_entries(root))
    rows.sort(key=lambda row: str(row["test_source_id"]))
    output.parent.mkdir(parents=True, exist_ok=True)
    with output.open("w", encoding="utf-8", newline="\n") as handle:
        for row in rows:
            handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n")
    digest = sha256(output)
    portability = Counter(str(row["portability"]) for row in rows)
    kinds = Counter(str(row["test_kind"]) for row in rows)
    report = {
        "schema": "test-entries-report.v1",
        "entries": len(rows),
        "test_directories": len({str(Path(str(row["path"])).parent) for row in rows}),
        "test_kinds": dict(sorted(kinds.items())),
        "portability": dict(sorted(portability.items())),
        "parse_errors": parse_errors,
        "blocked": sum(1 for row in rows if row["parse_status"] == "blocked"),
        "output_sha256": digest,
    }
    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")
    return len(rows), digest


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument("--output", type=Path, default=Path("coverage/test-entries.jsonl"))
    parser.add_argument("--report", type=Path, default=Path("coverage/test-entries.report.json"))
    args = parser.parse_args()
    root = args.repo_root.resolve()
    count, digest = generate(root, root / args.output, root / args.report)
    print(f"test_entries={count} sha256={digest}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())