boxsim-fileformat-rev/tools/difftest.py

#!/usr/bin/env python3
"""Differentially test boxsim.a2f against the Boxsim.exe oracle.

Every container in the Boxsim distribution is decoded twice, once by the pure
Python codec and once by emulating Boxsim's own machine code, and the results
must be byte identical. Writes the oracle digests to a manifest so the fast
test suite can guard regressions without needing Unicorn or Boxsim.exe.

Needs the dev venv (unicorn, pefile):

    .venv/bin/python tools/difftest.py [--jobs N] [--limit N]
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

MANIFEST = ROOT / "tests" / "oracle_manifest.json"


def find_containers(dist: Path) -> list[Path]:
    files = sorted(
        p
        for p in dist.rglob("*")
        if p.is_file() and p.suffix.upper() in (".BPJ", ".BCH")
    )
    if not files:
        raise SystemExit(f"no .bpj/.bch files found under {dist}")
    return files


def check_one(path_str: str) -> dict:
    """Decode with both implementations and compare. Returns a result record."""
    from boxsim import _oracle
    from boxsim.container import decode

    path = Path(path_str)
    raw = path.read_bytes()
    rec = {"file": str(path.relative_to(ROOT)), "size": len(raw)}

    try:
        mine = decode(raw)
        rec["sha_python"] = hashlib.sha256(mine).hexdigest()
        rec["len_python"] = len(mine)
    except Exception as exc:  # noqa: BLE001 - report, do not mask
        rec["error_python"] = f"{type(exc).__name__}: {exc}"

    try:
        theirs = _oracle.decode_bpj(raw)
        rec["sha_oracle"] = hashlib.sha256(theirs).hexdigest()
        rec["len_oracle"] = len(theirs)
    except Exception as exc:  # noqa: BLE001
        rec["error_oracle"] = f"{type(exc).__name__}: {exc}"

    both_failed = "error_python" in rec and "error_oracle" in rec
    if both_failed and "not a MyIni2CK file" in rec["error_python"]:
        # Pre-2.0 Boxsim used a different container with no MyIni2CK magic.
        # Both implementations refusing it is agreement, not a pass: count it
        # separately so it can never masquerade as coverage.
        rec["verdict"] = "rejected"
    elif both_failed:
        rec["verdict"] = "both-errored"
    elif "error_python" in rec or "error_oracle" in rec:
        rec["verdict"] = "one-errored"
    elif rec["sha_python"] == rec["sha_oracle"]:
        rec["verdict"] = "identical"
    else:
        rec["verdict"] = "mismatch"
    rec["ok"] = rec["verdict"] in ("identical", "rejected")
    return rec


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument(
        "--dist",
        default=str(ROOT / "boxsim_dl"),
        help="Boxsim distribution directory to sweep",
    )
    ap.add_argument("--jobs", type=int, default=max(1, (os.cpu_count() or 2) - 1))
    ap.add_argument("--limit", type=int, default=0, help="only the first N files")
    ap.add_argument(
        "--self-test",
        action="store_true",
        help="corrupt one expectation to prove the comparison has teeth",
    )
    args = ap.parse_args()

    files = find_containers(Path(args.dist))
    if args.limit:
        files = files[: args.limit]
    print(f"{len(files)} containers, {args.jobs} workers", flush=True)

    started = time.monotonic()
    results = []
    with ProcessPoolExecutor(max_workers=args.jobs) as pool:
        for i, rec in enumerate(pool.map(check_one, [str(f) for f in files]), 1):
            results.append(rec)
            print(f"  [{i}/{len(files)}] {rec['verdict']:10s} {rec['file']}", flush=True)

    if args.self_test:
        # Feed the comparison a deliberately wrong expectation. If this does not
        # get reported as a mismatch, the gate is decorative.
        print("\nself-test: corrupting one expectation", flush=True)
        decoded = [r for r in results if r["verdict"] == "identical"]
        if not decoded:
            print("SELF-TEST INCONCLUSIVE: no decoded file to corrupt")
            return 2
        poisoned = dict(decoded[0])
        poisoned["sha_oracle"] = "0" * 64
        if poisoned["sha_python"] == poisoned["sha_oracle"]:
            print("SELF-TEST FAILED: corrupt expectation still compared equal")
            return 2
        print(
            f"self-test passed: a corrupt digest for {poisoned['file']} "
            "is reported as a mismatch"
        )

    failures = [r for r in results if not r["ok"]]
    elapsed = time.monotonic() - started
    counts: dict[str, int] = {}
    for r in results:
        counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
    print(f"\nverdicts after {elapsed:.0f}s: " + ", ".join(
        f"{v}={n}" for v, n in sorted(counts.items())))
    print(
        f"{counts.get('identical', 0)} byte-identical, "
        f"{counts.get('rejected', 0)} rejected by both as pre-2.0 containers"
    )

    if failures:
        print("\nfailures:")
        for r in failures:
            print(f"  {r['file']}")
            for key in ("error_python", "error_oracle"):
                if key in r:
                    print(f"     {key}: {r[key]}")
            if "sha_python" in r and "sha_oracle" in r:
                print(f"     python {r['sha_python'][:16]} len {r.get('len_python')}")
                print(f"     oracle {r['sha_oracle'][:16]} len {r.get('len_oracle')}")
        return 1

    decoded = {
        r["file"]: {"sha256": r["sha_oracle"], "len": r["len_oracle"]}
        for r in results
        if r["verdict"] == "identical"
    }
    rejected = sorted(r["file"] for r in results if r["verdict"] == "rejected")
    MANIFEST.parent.mkdir(exist_ok=True)
    MANIFEST.write_text(
        json.dumps(
            {
                "note": "oracle digests from boxsim._oracle, see tools/difftest.py",
                "files": decoded,
                "rejected_not_myini2ck": rejected,
            },
            indent=2,
            sort_keys=True,
        )
        + "\n"
    )
    print(
        f"wrote {MANIFEST.relative_to(ROOT)}: "
        f"{len(decoded)} digests, {len(rejected)} rejected files recorded"
    )
    return 0


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