497 lines
26 KiB
Python
497 lines
26 KiB
Python
#!/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})
|
|
self.assertIn("| CfgVar_RegisterKey | 4 |", 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 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)
|