#!/usr/bin/env python3 """Tests for tracecmp.py / mkfixture.py / oracle_parsers.py (synthetic data only). /usr/bin/python3 -m unittest -v test_tracecmp /usr/bin/python3 test_tracecmp.py Covers: the reference emitter round-trips through json; record/tv validation and exit code 2; exact and tolerant (abs/rel/ulp, CLI + meta policy, per-hook) diffs; set/unordered/ptr/bytes/json rules; replay (missing/extra/wrong); --hook, --json-out; the oracle bridge's canonical form and a bridge->replay round trip. """ from __future__ import annotations import io import json import os import shutil import sys import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import mkfixture as mk # noqa: E402 import oracle_parsers as op # noqa: E402 import tracecmp as tc # noqa: E402 TMP = tempfile.mkdtemp(prefix="tracecmp-test-") FX = mk.build_all(os.path.join(TMP, "fx"), n=12, seed=7) P = FX["paths"] def run(*argv) -> tuple[int, dict, str]: """tracecmp.main with --json-out; -> (exit, report, stdout).""" out = os.path.join(TMP, "rep%d.json" % (abs(hash(argv)) % 10**8)) buf = io.StringIO() with redirect_stdout(buf): rc = tc.main(list(argv) + ["--json-out", out]) with open(out) as f: return rc, json.load(f), buf.getvalue() def write_log(name: str, recs: list[dict], meta: dict | None = None) -> str: p = os.path.join(TMP, name) mk.write_log(p, meta if meta is not None else mk.meta(), recs) return p def rec(cid: int, ret, ours_ret=None, side=None, ours_side=None, hook="H", **kw) -> dict: r = {"ts": cid, "hook": hook, "mode": "compare", "call_id": cid, "thread": 1, "args": [], "ret": ret, "side": side or {}, "ours": {"ret": ret if ours_ret is None else ours_ret, "side": ours_side if ours_side is not None else {k: {"after": v["after"]} for k, v in (side or {}).items()}}, "diverged": False, "diff": []} r.update(kw) return r def diffs_of(*recs, **policy) -> list[dict]: pol = tc.Policy().merged(policy) out = [] for r in recs: d, _ = tc.compare_record(r, pol) out.append(d) return out class EmitterTest(unittest.TestCase): def test_roundtrip_every_fixture_record(self): def f32norm(x): """%.9g round-trips the float32, not the widened double: normalize.""" if isinstance(x, dict): if x.get("t") == "f32" and isinstance(x.get("v"), float): return {**x, "v": mk.round_f32(x["v"])} return {k: f32norm(v) for k, v in x.items()} if isinstance(x, list): return [f32norm(v) for v in x] return x for r in FX["trace"] + FX["compare_bad"]: line = mk.emit_record(r) self.assertTrue(line.endswith("}\n")) self.assertTrue(line.isascii()) back = json.loads(line) self.assertEqual(f32norm(back), f32norm(r)) self.assertNotEqual(back, None) def test_string_escaping_matches_python_ensure_ascii(self): for s in ['plain', 'Résumé', 'Tab\tKey', 'Quote"d', 'Back\\slash', 'nl\nx', 'cr\rx', '\x01\x7f\xff', '€']: u = s.encode("cp1252").decode("latin-1") self.assertEqual(json.loads(mk.esc(u)), u) self.assertEqual(mk.esc(u), json.dumps(u, ensure_ascii=True)) self.assertEqual(mk.esc("\x08\x0c"), '"\\u0008\\u000c"') # long form is also accepted self.assertEqual(json.loads(mk.esc("\x08\x0c")), "\x08\x0c") def test_float_and_bigint_forms(self): self.assertEqual(mk.emit_tv(mk.f32(1.5)), '{"t":"f32","v":1.5}') self.assertEqual(mk.emit_tv(mk.f32(float("nan"))), '{"t":"f32","v":"nan"}') self.assertEqual(mk.emit_tv(mk.f32(float("-inf"))), '{"t":"f32","v":"-inf"}') self.assertEqual(mk.emit_tv(mk.u64(2**64 - 1)), '{"t":"u64","v":"18446744073709551615"}') self.assertEqual(mk.emit_tv(mk.null()), '{"t":"null","v":null}') b = mk.by(b"\x00\xff") self.assertEqual(json.loads(mk.emit_tv(b)), b) self.assertEqual(b["hex"], "00ff") self.assertEqual(mk.by(b"")["sha256"], "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855") big = mk.by(bytes(range(256)) * 2) self.assertNotIn("hex", big) self.assertEqual(len(big["head"]), 64) class ValidationTest(unittest.TestCase): def test_invalid_log_exit_2(self): rc, rep, _ = run(P["invalid"]) self.assertEqual(rc, 2) self.assertEqual(rep["totals"]["invalid_records"], 7) self.assertEqual(rep["totals"]["calls"], 2) errs = " | ".join(e["error"] for e in rep["invalid"]) for needle in ("not JSON", "missing required field", "i32.v must be", "duplicate call_id 900", "needs ours", "sha256", "u8.v=300 out of range"): self.assertIn(needle, errs) self.assertTrue(any("unknown field 'bogus'" in w for w in rep["warnings"])) def test_skip_invalid(self): rc, rep, _ = run(P["invalid"], "--skip-invalid") self.assertEqual(rc, 0) def test_meta_format_version(self): p = write_log("badmeta.jsonl", [], meta={"format": 2}) rc, rep, _ = run(p) self.assertEqual(rc, 2) self.assertIn("unsupported format", rep["invalid"][0]["error"]) def test_tv_validation_details(self): def errs(tv): e = [] tc.validate_tv(tv, "x", e) return e self.assertEqual(errs(mk.i32(5)), []) self.assertTrue(errs({"t": "i32"})) self.assertTrue(errs({"t": "nope", "v": 1})) self.assertTrue(errs({"t": "str", "v": "€"})) # code point > U+00FF self.assertEqual(errs({"t": "wstr", "v": "€"}), []) self.assertTrue(errs({"t": "ptr", "v": "a36fd0"})) self.assertTrue(errs({"t": "bytes", "n": 2, "sha256": "a" * 64, "hex": "00"})) # hex len != 2n self.assertTrue(errs({"t": "bool", "v": 1})) self.assertEqual(errs({"t": "i64", "v": "-9223372036854775808"}), []) self.assertTrue(errs({"t": "i64", "v": "9223372036854775808"})) self.assertTrue(errs({"t": "f32", "v": "huge"})) self.assertEqual(errs({"t": "f32", "v": "inf"}), []) self.assertTrue(errs({"t": "list", "v": [{"t": "i32", "v": "x"}]})) self.assertTrue(errs({"t": "struct", "v": [1]})) def test_replace_and_trace_records_not_compared(self): r = rec(1, mk.i32(1), mk.i32(2)) r["mode"] = "replace" del r["ours"], r["diverged"], r["diff"] t = rec(2, mk.i32(1), mk.i32(2)) t["mode"] = "trace" t["err"] = "capture failed" rc, rep, _ = run(write_log("modes.jsonl", [r, t])) self.assertEqual(rc, 0) self.assertEqual(rep["hooks"]["H"]["modes"], {"replace": 1, "trace": 1}) self.assertEqual(rep["hooks"]["H"]["errors"], 1) self.assertEqual(rep["hooks"]["H"]["compared"], 0) class CompareTest(unittest.TestCase): def test_clean_log_is_clean(self): rc, rep, out = run(P["compare_clean"]) self.assertEqual(rc, 0) self.assertEqual(rep["totals"], {"calls": 12, "compared": 12, "diverged": 0, "invalid_records": 0, "guarded_calls": 0, "undeclared_calls": 0, "undeclared_writes": 0, "coverage_unstated": 0, "coverage_contradicted": 0}) self.assertIn("| CfgVar_RegisterKey | 4 |", out) # a clean run still states what it did not check self.assertIn("### coverage", out) self.assertIn("appends to the owner's event list", out) def test_injected_divergences_exact(self): rc, rep, out = run(P["compare_bad"], "--first", "20") self.assertEqual(rc, 1) got = {cid for s in rep["hooks"].values() for cid in s["diverged_call_ids"]} want = {e["call_id"] for e in FX["expect"] if e["diverges_exact"]} self.assertEqual(got, want) first = {d["call_id"]: d["diff"] for s in rep["hooks"].values() for d in s["diffs"]} for e in FX["expect"]: if not e["diverges_exact"]: continue d0 = first[e["call_id"]][0] self.assertEqual(d0["why"], e["why"], (e, d0)) self.assertTrue(d0["path"].startswith(e["path"]), (e, d0)) lie = [e for e in FX["expect"] if e.get("warn")][0] self.assertTrue(any(f"call {lie['call_id']}: shim said diverged=True" in w for w in rep["warnings"])) self.assertEqual(len(rep["warnings"]), 1) # truthful verdicts everywhere else def test_injected_divergences_tolerant(self): rc, rep, _ = run(P["compare_bad"], "--tolerance", "1e-3") self.assertEqual(rc, 1) got = {cid for s in rep["hooks"].values() for cid in s["diverged_call_ids"]} want = {e["call_id"] for e in FX["expect"] if e["diverges_tol"]} self.assertEqual(got, want) def test_per_hook_tolerance_and_meta_policy(self): nudged = [e["call_id"] for e in FX["expect"] if e["diverges_exact"] and not e["diverges_tol"]][0] # per-hook CLI on the wrong hook: still diverges rc, rep, _ = run(P["compare_bad"], "--tolerance", "Manifest_Load=1e-3", "--hook", "Mars::ParseBlock") self.assertIn(nudged, rep["hooks"]["Mars::ParseBlock"]["diverged_call_ids"]) # per-hook CLI on the right hook: passes rc, rep, _ = run(P["compare_bad"], "--tolerance", "Mars::ParseBlock=1e-3", "--hook", "Mars::ParseBlock") self.assertNotIn(nudged, rep["hooks"]["Mars::ParseBlock"]["diverged_call_ids"]) # meta policy in the log itself, no CLI bad = FX["compare_bad"] p = write_log("metapol.jsonl", bad, meta=mk.meta(**{"Mars::ParseBlock": {"ftol": 1e-3}})) rc, rep, _ = run(p, "--hook", "Mars::ParseBlock") self.assertNotIn(nudged, rep["hooks"]["Mars::ParseBlock"]["diverged_call_ids"]) # CLI default overrides meta rc, rep, _ = run(p, "--hook", "Mars::ParseBlock", "--tolerance", "0") self.assertIn(nudged, rep["hooks"]["Mars::ParseBlock"]["diverged_call_ids"]) def test_float_kinds(self): a, b = 1000.0, 1000.001 self.assertFalse(tc.floats_equal(a, b, tc.Policy())) self.assertTrue(tc.floats_equal(a, b, tc.Policy(ftol=1e-2))) self.assertTrue(tc.floats_equal(a, b, tc.Policy(ftol=1e-5, ftol_kind="rel"))) self.assertFalse(tc.floats_equal(a, b, tc.Policy(ftol=1e-7, ftol_kind="rel"))) x = mk.round_f32(1.0) y = mk.round_f32(1.0 + 2 ** -23) # 1 ulp of f32 self.assertTrue(tc.floats_equal(x, y, tc.Policy(ftol=1, ftol_kind="ulp"), "f32")) self.assertFalse(tc.floats_equal(x, y, tc.Policy(ftol=0.5, ftol_kind="ulp"), "f32")) self.assertFalse(tc.floats_equal(x, y, tc.Policy(ftol=1, ftol_kind="ulp"), "f64")) nan, inf = float("nan"), float("inf") self.assertTrue(tc.floats_equal(nan, nan, tc.Policy(ftol=1e9))) self.assertFalse(tc.floats_equal(nan, 0.0, tc.Policy(ftol=1e9))) self.assertFalse(tc.floats_equal(inf, -inf, tc.Policy(ftol=1e9))) self.assertTrue(tc.floats_equal(inf, inf, tc.Policy())) # via records: f32 vs f64 is a type divergence; "nan" strings compare equal d = diffs_of(rec(1, {"t": "f32", "v": 1.0}, {"t": "f64", "v": 1.0}), rec(2, {"t": "f32", "v": "nan"}, {"t": "f32", "v": "nan"})) self.assertEqual([x["why"] for x in d[0]], ["type"]) self.assertEqual(d[1], []) def test_ints_strings_enums(self): d = diffs_of(rec(1, mk.u64(2**60), {"t": "u64", "v": str(2**60)}), rec(2, mk.s("Résumé"), mk.s("Resume")), rec(3, mk.enum(1, "A"), mk.enum(1, "B")), rec(4, mk.enum(1), mk.enum(2)), rec(5, mk.boolean(True), mk.i32(1))) self.assertEqual(d[0], []) self.assertEqual(d[1][0]["why"], "exact") self.assertEqual(d[2], []) # symbolic name is informational self.assertEqual(d[3][0]["why"], "exact") self.assertEqual(d[4][0]["why"], "type") def test_ptr_policy(self): r = rec(1, mk.ptr(0x1000), mk.ptr(0x2000)) self.assertEqual(diffs_of(r)[0], []) self.assertEqual(diffs_of(r, ptr="exact")[0][0]["why"], "exact") self.assertEqual(diffs_of(rec(2, mk.ptr(0), mk.ptr(0x2000)))[0][0]["why"], "exact") rc, rep, _ = run(write_log("ptr.jsonl", [r]), "--ptr", "exact") self.assertEqual(rc, 1) def test_set_and_unordered(self): a = mk.sset([mk.i32(1), mk.i32(1), mk.i32(2)]) self.assertEqual(diffs_of(rec(1, a, mk.sset([mk.i32(2), mk.i32(1), mk.i32(1)])))[0], []) d = diffs_of(rec(2, a, mk.sset([mk.i32(1), mk.i32(2), mk.i32(2)])))[0] self.assertEqual(d[0]["why"], "exact") # multiset: multiplicity matters d = diffs_of(rec(3, a, mk.sset([mk.i32(1), mk.i32(2)])))[0] self.assertEqual(d[0]["why"], "len") l1, l2 = mk.lst([mk.i32(1), mk.i32(2)]), mk.lst([mk.i32(2), mk.i32(1)]) self.assertEqual(diffs_of(rec(4, l1, l2))[0][0]["why"], "exact") self.assertEqual(diffs_of(rec(4, l1, l2), unordered=["ret"])[0], []) rc, rep, _ = run(write_log("unord.jsonl", [rec(4, l1, l2, hook="Hk")]), "--unordered", "Hk=ret") self.assertEqual(rc, 0) rc, rep, _ = run(write_log("unord2.jsonl", [rec(4, l1, l2, hook="Hk")]), "--unordered", "Other=ret") self.assertEqual(rc, 1) def test_struct_and_side(self): s1 = mk.struct_({"a": mk.i32(1), "b": mk.s("x")}) s2 = mk.struct_({"a": mk.i32(1), "c": mk.s("x")}) d = diffs_of(rec(1, s1, s2))[0] self.assertEqual([(x["path"], x["why"]) for x in d], [("ret.v.b", "missing"), ("ret.v.c", "extra")]) side = {"r1": {"before": mk.i32(0), "after": mk.i32(1)}, "r2": {"after": mk.i32(5)}} ours = {"r1": {"after": mk.i32(1)}, "r3": {"after": mk.i32(5)}} d = diffs_of(rec(2, None, None, side=side, ours_side=ours))[0] self.assertEqual([(x["path"], x["why"]) for x in d], [("side.r2.after", "missing"), ("side.r3.after", "extra")]) # 'before' mismatch is a warning, not a divergence ours = {"r1": {"before": mk.i32(9), "after": mk.i32(1)}, "r2": {"after": mk.i32(5)}} dd, warns = tc.compare_record(rec(3, None, None, side=side, ours_side=ours), tc.Policy()) self.assertEqual(dd, []) self.assertTrue(any("snapshot" in w for w in warns)) # void vs value self.assertEqual(diffs_of(rec(4, None, mk.i32(1)))[0][0]["why"], "extra") def test_bytes(self): a, b = mk.by(b"abcdef"), mk.by(b"abcXef") d = diffs_of(rec(1, a, b))[0] self.assertEqual(d[0]["why"], "hash") self.assertEqual(d[0]["first_diff_offset"], 3) self.assertEqual(diffs_of(rec(2, a, mk.by(b"abcdefg")))[0][0]["why"], "len") big1, big2 = mk.by(bytes(300)), mk.by(bytes(299) + b"\x01") d = diffs_of(rec(3, big1, big2))[0] self.assertEqual(d[0]["why"], "hash") self.assertNotIn("first_diff_offset", d[0]) self.assertEqual(diffs_of(rec(4, mk.by(b""), mk.by(b"")))[0], []) def test_json_values(self): j = lambda v: mk.jsonv(v) # noqa: E731 self.assertEqual(diffs_of(rec(1, j({"a": [1, 2.5, "x", True, None]}), j({"a": [1, 2.5, "x", True, None]})))[0], []) d = diffs_of(rec(2, j({"a": 1}), j({"a": 1.0})))[0] self.assertEqual(d[0]["why"], "type") self.assertEqual(diffs_of(rec(2, j({"a": 1}), j({"a": 1.0})), numeric=True)[0], []) d = diffs_of(rec(3, j({"a": {"b": [1, 2]}}), j({"a": {"b": [1]}, "c": 0})))[0] self.assertEqual([(x["path"], x["why"]) for x in d], [("ret.c", "extra"), ("ret.a.b", "len")]) d = diffs_of(rec(4, j({"f": 1.0}), j({"f": 1.0 + 1e-7})))[0] self.assertEqual(d[0]["why"], "exact") self.assertEqual(diffs_of(rec(4, j({"f": 1.0}), j({"f": 1.0 + 1e-7})), ftol=1e-6)[0], []) self.assertEqual(diffs_of(rec(5, j({"x": True}), j({"x": 1})))[0][0]["why"], "exact") d = diffs_of(rec(6, j([3, 1, 2]), j([1, 2, 3])), unordered=["ret"])[0] self.assertEqual(d, []) def test_hook_filter_and_json_out(self): rc, rep, out = run(P["compare_bad"], "--hook", "Manifest_Load") self.assertEqual(list(rep["hooks"]), ["Manifest_Load"]) self.assertEqual(rep["totals"]["calls"], 4) self.assertEqual(rep["kind"], "report") self.assertEqual(rep["format"], 1) rc, rep, out = run(P["compare_bad"], "--hook", "NoSuchHook") self.assertEqual(rc, 0) self.assertEqual(rep["totals"]["calls"], 0) def test_first_n(self): rc, rep, out = run(P["compare_bad"], "--first", "1") for s in rep["hooks"].values(): self.assertLessEqual(len(s["diffs"]), 1) self.assertIn("other divergent call_ids", out) def test_multiple_logs(self): rc, rep, _ = run(P["compare_clean"], P["compare_bad"]) self.assertEqual(rep["totals"]["calls"], 24) self.assertEqual(rc, 1) class CoverageTest(unittest.TestCase): """The audit mechanism: a clean compare must state what it did not check, and a descriptor that claims complete coverage must be contradicted when a guard region says otherwise. See sots-engine/docs/harness-audit.md.""" def test_meta_coverage_reaches_the_report(self): _, rep, out = run(P["compare_clean"]) cov = rep["hooks"]["CfgVar_RegisterKey"]["coverage"] self.assertEqual(cov["verdict"], "partial") self.assertEqual(cov["checked_regions"], ["cfg_table"]) self.assertEqual(len(cov["unmodelled"]), 1) self.assertEqual(cov["unmodelled"][0]["risk"], "high") self.assertIn("| CfgVar_RegisterKey | partial | cfg_table |", out) def test_guard_findings_are_reported_but_do_not_fail_an_honest_hook(self): rc, rep, out = run(P["coverage_guarded"]) self.assertEqual(rc, 0) # ours matched every declared region: no divergence cov = rep["hooks"]["CfgVar_RegisterKey"]["coverage"] self.assertEqual(cov["verdict"], "partial") self.assertEqual(cov["guards"], ["player"]) self.assertEqual(cov["guarded_calls"], 4) self.assertEqual(cov["undeclared_calls"], FX["guard_calls"]) self.assertEqual(cov["undeclared_writes"], FX["guard_calls"]) self.assertEqual(cov["spans"]["compare"], ["player+0x2b0:4"]) self.assertEqual(rep["totals"]["undeclared_writes"], FX["guard_calls"]) self.assertIn("guard hits in compare mode: player+0x2b0:4", out) def test_strict_coverage_fails_on_any_undeclared_write(self): self.assertEqual(run(P["coverage_guarded"], "--strict-coverage")[0], 1) self.assertEqual(run(P["compare_clean"], "--strict-coverage")[0], 0) def test_a_false_completeness_claim_is_a_divergence(self): rc, rep, out = run(P["coverage_lying"]) self.assertEqual(rc, 1) self.assertEqual(rep["coverage_contradicted"], ["CfgVar_RegisterKey"]) self.assertEqual(rep["totals"]["coverage_contradicted"], 1) self.assertEqual(rep["totals"]["diverged"], 0) # not a per-call diff: a false claim self.assertEqual(rep["hooks"]["CfgVar_RegisterKey"]["coverage"]["verdict"], "contradicted") self.assertEqual(rep["hooks"]["Manifest_Load"]["coverage"]["verdict"], "complete") self.assertTrue(any("claims complete coverage but a guard saw" in w for w in rep["warnings"])) def test_a_log_with_no_coverage_statement_is_called_out(self): rc, rep, out = run(P["coverage_unstated"]) self.assertEqual(rc, 0) # legacy logs still pass... self.assertEqual(rep["coverage_unstated"], ["CfgVar_RegisterKey", "Manifest_Load", "Mars::ParseBlock"]) self.assertTrue(all(s["coverage"]["verdict"] == "unstated" for s in rep["hooks"].values())) self.assertTrue(any("no coverage statement in meta" in w for w in rep["warnings"])) self.assertIn("| CfgVar_RegisterKey | unstated |", out) self.assertEqual(run(P["coverage_unstated"], "--strict-coverage")[0], 1) # ...but not here def test_record_coverage_block_is_validated(self): good = {"guards": ["p"], "undeclared": [{"region": "p", "off": 4, "len": 4}], "n": 1} for bad in ({"guards": "p", "undeclared": []}, {"guards": [], "undeclared": [{"region": "p", "off": "x", "len": 4}], "n": 1}, {"guards": [], "undeclared": [{"region": "p", "off": 4, "len": 4}], "n": 0}, {"guards": [], "undeclared": [], "n": -1}, {"guards": []}, "not-an-object"): errs: list = [] tc.validate_coverage(bad, errs) self.assertTrue(errs, bad) errs = [] tc.validate_coverage(good, errs) self.assertEqual(errs, []) def test_a_bad_coverage_block_invalidates_the_record(self): path = os.path.join(TMP, "cov_invalid.jsonl") rec = dict(mk.to_compare(FX["trace"])[0]) rec["coverage"] = {"guards": [1], "undeclared": []} mk.write_log(path, mk.meta(), [rec]) rc, rep, _ = run(path) self.assertEqual(rc, 2) self.assertIn("coverage.guards", rep["invalid"][0]["error"]) def test_compare_vs_replace_asymmetry_is_named(self): """The B3 shape: the original writes something in compare mode that ours never writes in replace mode. Both logs are individually clean; only the two together show the gap.""" path = os.path.join(TMP, "cov_asym.jsonl") recs = mk.to_compare(FX["trace"])[:2] a, b = dict(recs[0]), dict(recs[1]) a["coverage"] = mk.guard_block(["player"], [{"region": "player", "off": 0x2b0, "len": 4}]) b["hook"] = a["hook"] # the same hook in both modes: that is what makes them comparable b["mode"] = "replace" b.pop("ours", None) b.pop("diverged", None) b.pop("diff", None) b["coverage"] = mk.guard_block(["player"]) mk.write_log(path, mk.meta(), [a, b]) rc, _, out = run(path) self.assertEqual(rc, 0) self.assertIn("ONLY the original writes these: player+0x2b0:4", out) class ReplayTest(unittest.TestCase): def test_replay_ok(self): rc, rep, _ = run(P["trace_clean"], "--replay", P["replay_impl_ok"]) self.assertEqual(rc, 0) self.assertEqual(rep["kind"], "replay") self.assertEqual(rep["totals"]["compared"], 12) def test_replay_bad(self): rc, rep, out = run(P["trace_clean"], "--replay", P["replay_impl_bad"]) self.assertEqual(rc, 1) rb = FX["replay_bad"] got = {cid for s in rep["hooks"].values() for cid in s["diverged_call_ids"]} self.assertEqual(got, {rb["wrong"], rb["missing"]}) whys = {d["call_id"]: d["diff"][0]["why"] for s in rep["hooks"].values() for d in s["diffs"]} self.assertEqual(whys[rb["wrong"]], "exact") self.assertEqual(whys[rb["missing"]], "missing") self.assertTrue(any(str(rb["extra"]) in w for w in rep["warnings"])) def test_replay_side_shapes_and_hook_mismatch(self): g = write_log("g.jsonl", [{"ts": 0, "hook": "H", "mode": "trace", "call_id": 5, "thread": 0, "args": [], "ret": mk.i32(1), "side": {"r": {"after": mk.i32(2)}}}]) p = os.path.join(TMP, "impl_shapes.jsonl") with open(p, "w") as f: f.write(json.dumps({"call_id": 5, "hook": "Other", "ret": mk.i32(1), "side": {"r": {"after": mk.i32(2)}}}) + "\n") rc, rep, _ = run(g, "--replay", p) self.assertEqual(rc, 0) self.assertTrue(any("hook 'Other'" in w for w in rep["warnings"])) with open(p, "w") as f: f.write(json.dumps({"call_id": 5, "ret": mk.i32(1), "side": {"r": mk.i32(3)}}) + "\n") f.write("garbage\n") rc, rep, _ = run(g, "--replay", p) self.assertEqual(rc, 2) # invalid impl line wins over the divergence self.assertEqual(rep["totals"]["diverged"], 1) rc, rep, _ = run(g, "--replay", p, "--skip-invalid") self.assertEqual(rc, 1) def test_replay_tolerance(self): g = write_log("gt.jsonl", [{"ts": 0, "hook": "H", "mode": "trace", "call_id": 1, "thread": 0, "args": [], "ret": mk.f32(2.0), "side": {}}]) p = os.path.join(TMP, "impl_tol.jsonl") with open(p, "w") as f: f.write(json.dumps({"call_id": 1, "ret": mk.f32(2.0 + 1e-6)}) + "\n") self.assertEqual(run(g, "--replay", p)[0], 1) self.assertEqual(run(g, "--replay", p, "--tolerance", "H=1e-5")[0], 0) class OracleBridgeTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.root = os.path.join(TMP, "gob") files = { "Data/globals.txt": 'FOO 1\nBAR 1.0\nNAME "Résumé"\nCOL "1 2 3"\nNEG -0.0\nBIG 7e+8\nFLAG TRUE // c\n', "Weapons/w.weapon": 'weapon {\n name "Laser" // x\n dam .5\n dam 2\n mount { size small }\n "bare"\n}\n', "Weapons/_weapons.txt": "1 w.weapon\n// DELETED - 2\n3 Other.WEAPON\n", "Weapons/_turrets.txt": 'a 1 "two words" 3.25\nb 2 x 4\n', "Data/x.csv": '# a,b\n1,"x,y"\n\n2,z\n', "Effects/e.effect": 'TXT\nNAME "New Emitter"\nMODIFIER\nBEGIN\n TYPE 1\n K 0.1\nEND\nMODIFIER\nBEGIN\n TYPE 2\nEND\n', "Weapons/broken.weapon": 'weapon { name "x" "unterminated\n', "Weapons/_shipsections.txt": "1 a.shipsection\n1 b.shipsection\n", "shaders/x.fx": "float4 main() {}\n", } for rel, txt in files.items(): p = os.path.join(cls.root, rel) os.makedirs(os.path.dirname(p), exist_ok=True) with open(p, "wb") as f: f.write(txt.encode("cp1252")) def test_canonical_floats_and_strings(self): self.assertEqual(op.canon_float(1.0), "1.0") self.assertEqual(op.canon_float(-0.0), "-0.0") self.assertEqual(op.canon_float(0.1), "0.100000001") # f32-rounded, %.9g self.assertEqual(op.canon_float(7e8), "700000000.0") self.assertEqual(op.canon_float(1e20), "1.00000002e+20") self.assertEqual(op.canon_float(float("nan")), '"nan"') self.assertEqual(op.canon_str("Résumé"), '"R\\u00e9sum\\u00e9"') self.assertEqual(op.canon_str("€"), '"\\u0080"') # cp1252 byte, not U+20AC self.assertEqual(op.canon_str('a"b\\c\n\x7f'), '"a\\"b\\\\c\\n\\u007f"') self.assertEqual(op.dumps(op.canonical({"b": [1, 2.5, (3, "x")], "a": {"z": True, "y": None}})), '{"a":{"y":null,"z":true},"b":[1,2.5,[3,"x"]]}') # no-float objects equal json.dumps canonical settings obj = {"k": ["a", {"n": 1, "m": "é"}]} self.assertEqual(op.dumps(op.canonical(obj)), json.dumps(obj, sort_keys=True, ensure_ascii=True, separators=(",", ":"))) def test_per_kind_output(self): r = self.root c, notes, err = op.oracle_file(os.path.join(r, "Data/globals.txt"), "kv") self.assertIsNone(err) self.assertEqual(op.dumps(c), '{"BAR":1.0,"BIG":700000000.0,"COL":"1 2 3","FLAG":true,"FOO":1,"NAME":"R\\u00e9sum\\u00e9","NEG":-0.0}') c, notes, err = op.oracle_file(os.path.join(r, "Weapons/w.weapon"), "brace:weapon") self.assertEqual(op.dumps(c), '{"weapon":{"_items":"bare","dam":[0.5,2],"mount":{"size":"small"},"name":"Laser"}}') c, notes, err = op.oracle_file(os.path.join(r, "Weapons/_weapons.txt"), "manifest") self.assertEqual(op.dumps(c), '{"deleted":[2],"entries":[[1,"w.weapon"],[3,"Other.WEAPON"]]}') c, notes, err = op.oracle_file(os.path.join(r, "Weapons/_turrets.txt"), "rows") self.assertEqual(op.dumps(c), '[["a",1,"two words",3.25],["b",2,"x",4]]') c, notes, err = op.oracle_file(os.path.join(r, "Data/x.csv"), "csv") self.assertEqual(op.dumps(c), '[["1","x,y"],["2","z"]]') c, notes, err = op.oracle_file(os.path.join(r, "Effects/e.effect"), "effect") self.assertEqual(op.dumps(c), '[["NAME","New Emitter"],["MODIFIER",[["TYPE",1],["K",0.100000001]]],["MODIFIER",[["TYPE",2]]]]') c, notes, err = op.oracle_file(os.path.join(r, "Weapons/broken.weapon"), "brace:weapon") self.assertEqual(c, {"_error": True}) self.assertIn("MarsSyntaxError", err) c, notes, err = op.oracle_file(os.path.join(r, "Weapons/_shipsections.txt"), "manifest") self.assertIn("duplicate id", err) def test_cli_and_replay_round_trip(self): r = self.root golden = os.path.join(TMP, "oracle", "golden.jsonl") outdir = os.path.join(TMP, "oracle", "canon") err = io.StringIO() with redirect_stderr(err), redirect_stdout(io.StringIO()): rc = op.main(["--root", r, "--all", "--jsonl", golden, "--out-dir", outdir]) self.assertEqual(rc, 1) # two files fail to parse self.assertIn("8 file(s), 2 parse error(s)", err.getvalue()) self.assertFalse(os.path.exists(os.path.join(outdir, "shaders/x.fx.json"))) # skipped kind with open(os.path.join(outdir, "Data/globals.txt.json")) as f: self.assertTrue(f.read().endswith("}\n")) # golden validates and self-replays clean rc, rep, _ = run(golden) self.assertEqual(rc, 0) self.assertEqual(rep["hooks"]["parse:brace"]["errors"], 1) self.assertEqual(rep["hooks"]["parse:manifest"]["errors"], 1) rc, rep, _ = run(golden, "--replay", golden) self.assertEqual(rc, 0) self.assertEqual(rep["totals"]["compared"], 8) # an "implementation" that gets one float wrong and one file's shape wrong impl = os.path.join(TMP, "oracle", "impl.jsonl") with open(golden) as f, open(impl, "w") as g: for line in f: d = json.loads(line) if "meta" in d: continue if d["args"][0]["v"] == "Data/globals.txt": d["ret"]["v"]["BAR"] = 1.0000001 if d["args"][0]["v"] == "Weapons/w.weapon": d["ret"]["v"]["weapon"]["dam"] = 2 g.write(json.dumps({"call_id": d["call_id"], "ret": d["ret"], "side": {}}) + "\n") rc, rep, _ = run(golden, "--replay", impl) self.assertEqual(rc, 1) self.assertEqual(rep["totals"]["diverged"], 2) rc, rep, _ = run(golden, "--replay", impl, "--tolerance", "1e-6") self.assertEqual(rep["totals"]["diverged"], 1) paths = [d["diff"][0]["path"] for s in rep["hooks"].values() for d in s["diffs"]] self.assertEqual(paths, ["ret.weapon.dam"]) # stdout mode prints one canonical line per file buf = io.StringIO() with redirect_stdout(buf), redirect_stderr(io.StringIO()): op.main(["--root", r, os.path.join(r, "Data/globals.txt"), "--kind", "kv"]) self.assertEqual(buf.getvalue().count("\n"), 1) def test_kind_dispatch_matches_verify(self): self.assertEqual(op.kind_of("Weapons/_weapons.txt"), "manifest") self.assertEqual(op.kind_of("Data/globals.txt"), "kv") self.assertEqual(op.kind_of("Weapons/x.weapon"), "brace:weapon") self.assertEqual(op.kind_of("Locale/EN/Desc_x.txt"), "prose") def tearDownModule(): shutil.rmtree(TMP, ignore_errors=True) if __name__ == "__main__": unittest.main(verbosity=1)