d46b3f8391
Implements the 2026-09-09 fleet-id-order resolution, section 3 item 1. Given the
pre-turn save, compute the ids new in each post-turn save, match the client-minted
(node nibble != 0) new fleets by a key that does not mention the id -- (LocID or
FPlan destination, sorted ship-id set) -- build the bijection pi, rewrite every
fleet reference, compare the master id lists as sets, mask /Summary/Checksum with
its reason on the line, and print pi.
Acceptance, both halves:
bp-pinA vs bp-pinB IDENTICAL modulo pi = {1970<->1986} (35 leaves -> 0)
ad-oracle-A vs -B REFUSED, then DIVERGED: 94 leaves (unchanged)
Five guards, every one refusing rather than degrading: only ids absent from the
pre-turn save; only non-zero node nibbles; pi must permute one set; content keys
must correspond one-to-one and be unique per side; and no leaf anywhere may hold a
permuted id at an unmodelled site (matched on raw bytes, not the reader's typed
value). A refusal rewrites nothing and falls back to the ordinary comparison.
Three corrections to the specification from contact with the data, in
findings/subsystems/relabel-new-ids.md section 4: FtName is an id-attached label
and needs the same treatment as the id; relabelling the Flt[] keys is the wrong
operation (exchange the bodies -- the fleet table is id-ordered and identical in
both saves); a node's new-id set spans object kinds.
Default path proven unchanged: pre- and post-change modules agree on the root
digest, coverage, mask hits and every (path, digest) in the tree over all 43 saves
under two policies, and on 5,602 lines of CLI stdout across every mode.
Also fixes a pre-existing, unrelated test failure: the re-save localisation test
enumerated pairs over sorted filenames and hard-coded the direction 4 -> 0, which
a later corpus addition reversed. Suite 38 -> 62 tests, all passing.
806 lines
36 KiB
Python
806 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for state_checksum.py.
|
|
|
|
Synthetic tests always run. Real-save tests are skipped cleanly when no saves
|
|
are available: they look in $SOTS_SAVES_DIR first, then at the in-repo
|
|
verify/results/saves/. No save is ever written into either repo by this file.
|
|
|
|
uv run python3 -m unittest discover -s verify/state-checksum -v
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import gzip
|
|
import json
|
|
import math
|
|
import os
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
_VERIFY = os.path.dirname(_HERE)
|
|
for _p in (_HERE, os.path.join(_VERIFY, "save-reader")):
|
|
if _p not in sys.path:
|
|
sys.path.insert(0, _p)
|
|
|
|
import save_reader as sr # noqa: E402
|
|
import save_writer_stub as sw # noqa: E402
|
|
import state_checksum as sc # noqa: E402
|
|
|
|
|
|
# --- helpers ------------------------------------------------------------------
|
|
|
|
def tiny(status_a=4, status_b=4, checksum=-1000, extra_ids=(),
|
|
fval_a=1.5, fval_b=2.5) -> bytes:
|
|
"""A miniature save with the shapes the tool special-cases.
|
|
|
|
The `Sim` prefix fields (`KeyPath`..`NMnx`) are emitted because the reader
|
|
types `PlayerIDs` positionally off the Sim shape; without them the tag is
|
|
guessed and a 4-byte int is indistinguishable from a 2-byte string at the
|
|
same item size (SAVE_FORMAT.md section 2).
|
|
"""
|
|
w = sw.SaveWriter("joint")
|
|
w.begin("Summary")
|
|
w.string("GameName", "T")
|
|
w.int("Turn", 2)
|
|
w.int("Checksum", checksum)
|
|
w.end()
|
|
w.begin("Sim")
|
|
w.string("KeyPath", "")
|
|
w.int("NMSz", 16)
|
|
w.int("NMLc", 0)
|
|
w.int("NMnx", 109)
|
|
ids = (16, 32) + tuple(extra_ids)
|
|
w.int("PlayerIDs", len(ids))
|
|
for i in ids:
|
|
w.int(".", i)
|
|
w.int("NumPlrs", 2)
|
|
for pid, st, fv in ((16, status_a, fval_a), (32, status_b, fval_b)):
|
|
w.int("PlayerID", pid)
|
|
w.begin("Player")
|
|
w.string("PlryName", f"p{pid}")
|
|
w.int("Status", st)
|
|
w.float("IdealSuit", fv)
|
|
w.end()
|
|
w.end()
|
|
return w.bytes()
|
|
|
|
|
|
#: (id, name, LocID, FPlan destination or None, [ship ids])
|
|
PRE_FLEETS = [(258, "Alpha Fleet", 100, None, [16]),
|
|
(274, "Beta Fleet", 0, 700, [32, 48])]
|
|
|
|
|
|
def fleets_save(fleets, sys_flt=None, extra=None) -> bytes:
|
|
"""A miniature save carrying a master fleet table.
|
|
|
|
Shaped like the real one where this tool looks: `FleetIDs[]`/`ShipIDs[]`
|
|
inline master lists, `FltID`/`Flt` object pairs, `FtName`, `LocID`, an
|
|
`FPlan` with a `pnd` destination, and `ShipID`/`Ship` pairs whose `Ship`
|
|
frame carries the `FltID` back-reference.
|
|
"""
|
|
w = sw.SaveWriter("joint")
|
|
w.begin("Summary")
|
|
w.string("GameName", "T")
|
|
w.int("Turn", 28)
|
|
w.int("Checksum", sum(f[0] for f in fleets))
|
|
w.end()
|
|
w.begin("Sim")
|
|
w.string("KeyPath", "")
|
|
w.int("NMSz", 16)
|
|
w.int("NMLc", 0)
|
|
w.int("NMnx", 100)
|
|
w.int("PlayerIDs", 1)
|
|
w.int(".", 32)
|
|
w.int("FleetIDs", len(fleets))
|
|
for f in fleets:
|
|
w.int(".", f[0])
|
|
ships = [s for f in fleets for s in f[4]]
|
|
w.int("ShipIDs", len(ships))
|
|
for s in ships:
|
|
w.int(".", s)
|
|
if extra is not None:
|
|
w.int(extra[0], extra[1])
|
|
if sys_flt is not None:
|
|
w.int("Flt", sys_flt)
|
|
w.int("NumFlts", len(fleets))
|
|
for fid, name, loc, dest, fships in fleets:
|
|
w.int("FltID", fid)
|
|
w.begin("Flt")
|
|
w.string("FtName", name)
|
|
w.int("LocID", loc)
|
|
if dest is not None:
|
|
w.begin("FPlan")
|
|
w.int("pnd", dest)
|
|
w.end()
|
|
w.int("NumShips", len(fships))
|
|
for s in fships:
|
|
w.int("ShipID", s)
|
|
w.begin("Ship")
|
|
w.int("FltID", fid)
|
|
w.end()
|
|
w.end()
|
|
w.end()
|
|
return w.bytes()
|
|
|
|
|
|
def find_saves() -> list[str]:
|
|
d = os.environ.get("SOTS_SAVES_DIR") or os.path.join(_VERIFY, "results", "saves")
|
|
if not os.path.isdir(d):
|
|
return []
|
|
return sorted(os.path.join(d, f) for f in os.listdir(d) if f.endswith(".sav"))
|
|
|
|
|
|
REAL = find_saves()
|
|
needs_saves = unittest.skipUnless(REAL, "no saves in $SOTS_SAVES_DIR or verify/results/saves")
|
|
|
|
_CACHE: dict = {}
|
|
|
|
|
|
def real_ck(path: str, **kw):
|
|
"""Memoised checksum_save -- a real save costs ~6 s to parse."""
|
|
key = (path, tuple(sorted(kw.items())))
|
|
if key not in _CACHE:
|
|
_CACHE[key] = sc.checksum_save(path, **kw)
|
|
return _CACHE[key]
|
|
|
|
|
|
# --- coverage: the property that makes the digest evidence --------------------
|
|
|
|
class CoverageTest(unittest.TestCase):
|
|
def test_reconstruction_is_exact_on_a_synthetic_save(self):
|
|
data = tiny()
|
|
ck = sc.checksum_bytes(data)
|
|
self.assertTrue(ck.coverage["ok"])
|
|
self.assertEqual(ck.coverage["rebuiltBytes"], len(data))
|
|
|
|
def test_reconstruction_is_exact_on_the_schema_fixture(self):
|
|
data, _ = sw.build_fixture()
|
|
ck = sc.checksum_bytes(data)
|
|
self.assertTrue(ck.coverage["ok"], f"first diff at {ck.coverage['firstDiff']}")
|
|
|
|
def test_reconstruction_handles_gzip_and_bare_streams_alike(self):
|
|
data = tiny()
|
|
a = sc.checksum_bytes(data)
|
|
b = sc.checksum_bytes(gzip.compress(data, mtime=0))
|
|
self.assertEqual(a.digest, b.digest)
|
|
self.assertTrue(b.coverage["ok"])
|
|
|
|
def test_a_broken_reconstruction_is_reported_not_swallowed(self):
|
|
res = sr.read_bytes(tiny())
|
|
# corrupt the parse, not the file: a leaf whose raw no longer matches
|
|
for n in res.tree.children[0].children:
|
|
if n.name == "Turn":
|
|
n.raw = b"\xff\xff\xff\xff"
|
|
cov = sc.audit_coverage(res.tree, res.inflated)
|
|
self.assertFalse(cov["ok"])
|
|
self.assertIsNotNone(cov["firstDiff"])
|
|
|
|
|
|
# --- determinism and sensitivity ---------------------------------------------
|
|
|
|
class DigestTest(unittest.TestCase):
|
|
def test_same_bytes_same_digest(self):
|
|
data = tiny()
|
|
self.assertEqual(sc.checksum_bytes(data).digest, sc.checksum_bytes(data).digest)
|
|
|
|
def test_every_value_leaf_is_load_bearing(self):
|
|
"""Flip each leaf in turn; the root must move every single time.
|
|
|
|
This is the anti-'empty region set' test: a leaf the digest does not
|
|
actually consume would pass silently under any other check.
|
|
"""
|
|
base = sc.checksum_bytes(tiny())
|
|
seen = 0
|
|
for name, kw in (("status", dict(status_a=5)), ("checksum", dict(checksum=-1001)),
|
|
("float", dict(fval_a=1.5000001))):
|
|
with self.subTest(name):
|
|
other = sc.checksum_bytes(tiny(**kw))
|
|
self.assertNotEqual(base.digest, other.digest)
|
|
seen += 1
|
|
self.assertEqual(seen, 3)
|
|
|
|
def test_a_one_bit_float_change_moves_the_root(self):
|
|
a = sc.checksum_bytes(tiny(fval_a=1.5))
|
|
(bits,) = struct.unpack("<I", struct.pack("<f", 1.5))
|
|
(nudged,) = struct.unpack("<f", struct.pack("<I", bits + 1))
|
|
b = sc.checksum_bytes(tiny(fval_a=nudged))
|
|
self.assertNotEqual(a.digest, b.digest)
|
|
d = sc.diff(a.root, b.root)
|
|
self.assertEqual(len(d), 1, [repr(e) for e in d])
|
|
self.assertTrue(d[0].path.endswith("/IdealSuit"))
|
|
|
|
def test_a_float_leaf_diff_carries_its_ulp_distance(self):
|
|
"""Built from reader nodes directly: the fixture's positional type
|
|
inference would otherwise decide whether the leaf is a float at all."""
|
|
def leaf(v):
|
|
raw = struct.pack("<f", v)
|
|
n = sr.Node("F", "float", v, 0, 12, raw=raw)
|
|
return sc._build(n, "/F", "F", "bits", (), {})
|
|
(bits,) = struct.unpack("<I", struct.pack("<f", 1.5))
|
|
(nudged,) = struct.unpack("<f", struct.pack("<I", bits + 3))
|
|
d = sc.diff(leaf(1.5), leaf(nudged))
|
|
self.assertEqual(len(d), 1)
|
|
self.assertEqual(d[0].ulps, 3.0)
|
|
|
|
def test_the_digest_depends_on_the_inferred_kind_not_only_the_bytes(self):
|
|
"""Documented caveat: same 4 bytes, different reader typing, different
|
|
digest. Hence READER_FINGERPRINT is recorded with every chain."""
|
|
raw = struct.pack("<f", 1.5)
|
|
as_f = sc._build(sr.Node("F", "float", 1.5, 0, 12, raw=raw), "/F", "F", "bits", (), {})
|
|
as_i = sc._build(sr.Node("F", "int", 1069547520, 0, 12, raw=raw), "/F", "F", "bits", (), {})
|
|
self.assertNotEqual(as_f.digest, as_i.digest)
|
|
self.assertNotEqual(sc.READER_FINGERPRINT, "unknown")
|
|
|
|
def test_policy_is_domain_separated_into_the_root(self):
|
|
"""A strict root and a lenient root must never be confusable."""
|
|
data = tiny()
|
|
self.assertNotEqual(sc.checksum_bytes(data, floats="bits").digest,
|
|
sc.checksum_bytes(data, floats="canonical").digest)
|
|
|
|
|
|
# --- localisation -------------------------------------------------------------
|
|
|
|
class LocalisationTest(unittest.TestCase):
|
|
def test_diff_names_the_object_not_just_the_hash(self):
|
|
a = sc.checksum_bytes(tiny(status_a=4, status_b=4))
|
|
b = sc.checksum_bytes(tiny(status_a=0, status_b=0))
|
|
d = sc.diff(a.root, b.root)
|
|
paths = sorted(e.path for e in d)
|
|
self.assertEqual(paths, ['/Sim/players/Player[16 "p16"]/Status',
|
|
'/Sim/players/Player[32 "p32"]/Status'])
|
|
self.assertEqual([(e.a, e.b) for e in d], [(4, 0), (4, 0)])
|
|
|
|
def test_identical_trees_report_nothing(self):
|
|
a = sc.checksum_bytes(tiny())
|
|
b = sc.checksum_bytes(tiny())
|
|
self.assertEqual(sc.diff(a.root, b.root), [])
|
|
|
|
def test_an_inline_id_list_reports_as_one_list_not_a_shift_cascade(self):
|
|
a = sc.checksum_bytes(tiny())
|
|
b = sc.checksum_bytes(tiny(extra_ids=(48,)))
|
|
d = sc.diff(a.root, b.root)
|
|
lists = [e for e in d if e.kind == "list"]
|
|
self.assertEqual(len(lists), 1, [repr(e) for e in d])
|
|
self.assertEqual(lists[0].path, "/Sim/PlayerIDs[]")
|
|
self.assertEqual(lists[0].b, [48])
|
|
self.assertEqual(lists[0].a, [])
|
|
|
|
def test_sibling_indices_are_per_tag_so_insertions_do_not_renumber(self):
|
|
a = sc.checksum_bytes(tiny())
|
|
b = sc.checksum_bytes(tiny(extra_ids=(48,)))
|
|
# the players group keeps its labels even though ids were inserted before it
|
|
pa = {c.label for c in a.root.find("/Sim").find("/Sim/players").children}
|
|
pb = {c.label for c in b.root.find("/Sim").find("/Sim/players").children}
|
|
self.assertEqual(pa, pb)
|
|
|
|
def test_object_identity_folds_the_id_in(self):
|
|
"""Two objects with the same body but different ids must differ."""
|
|
w = sc.checksum_bytes(tiny())
|
|
p16 = w.root.find('/Sim/players/Player[16 "p16"]')
|
|
self.assertIsNotNone(p16)
|
|
self.assertEqual(p16.label, 'Player[16 "p16"]')
|
|
|
|
|
|
# --- float policy -------------------------------------------------------------
|
|
|
|
class FloatPolicyTest(unittest.TestCase):
|
|
NEG0 = struct.pack("<I", 0x80000000)
|
|
POS0 = struct.pack("<I", 0x00000000)
|
|
SNAN = struct.pack("<I", 0x7F800001)
|
|
QNAN = struct.pack("<I", 0x7FC00000)
|
|
FLT_MAX = struct.pack("<I", 0x7F7FFFFF)
|
|
|
|
def test_bits_policy_separates_signed_zero_and_nan_payloads(self):
|
|
self.assertNotEqual(sc.float_key(self.NEG0, "bits"), sc.float_key(self.POS0, "bits"))
|
|
self.assertNotEqual(sc.float_key(self.SNAN, "bits"), sc.float_key(self.QNAN, "bits"))
|
|
|
|
def test_canonical_policy_merges_them(self):
|
|
self.assertEqual(sc.float_key(self.NEG0, "canonical"), sc.float_key(self.POS0, "canonical"))
|
|
self.assertEqual(sc.float_key(self.SNAN, "canonical"), sc.float_key(self.QNAN, "canonical"))
|
|
|
|
def test_canonical_leaves_every_ordinary_value_alone(self):
|
|
for v in (0.0, 1.0, -1.0, 1e-30, 3.4028234663852886e38, 11.106206893920898):
|
|
raw = struct.pack("<f", v)
|
|
self.assertEqual(sc.float_key(raw, "canonical"), raw, v)
|
|
|
|
def test_flt_max_is_not_infinity(self):
|
|
"""events.md: the default EvPos is FLT_MAX (0x7f7fffff), not inf."""
|
|
(v,) = struct.unpack("<f", self.FLT_MAX)
|
|
self.assertFalse(math.isinf(v))
|
|
self.assertEqual(sc.float_key(self.FLT_MAX, "canonical"), self.FLT_MAX)
|
|
|
|
def test_ulps_apart(self):
|
|
one = struct.pack("<f", 1.0)
|
|
(b,) = struct.unpack("<I", one)
|
|
self.assertEqual(sc.ulps_apart(one, one), 0.0)
|
|
self.assertEqual(sc.ulps_apart(one, struct.pack("<I", b + 1)), 1.0)
|
|
self.assertEqual(sc.ulps_apart(one, struct.pack("<I", b + 5)), 5.0)
|
|
# crossing zero is continuous under the ordinal map
|
|
self.assertEqual(sc.ulps_apart(self.POS0, self.NEG0), 0.0)
|
|
self.assertEqual(sc.ulps_apart(one, struct.pack("<f", float("inf"))), math.inf)
|
|
self.assertEqual(sc.ulps_apart(self.QNAN, self.SNAN), 0.0)
|
|
|
|
def test_tolerance_is_not_available_as_a_hashing_policy(self):
|
|
"""Deliberate: a tolerant hash is a contradiction (see STATE_CHECKSUM.md 3.4)."""
|
|
self.assertEqual(set(sc.FLOAT_POLICIES), {"bits", "canonical"})
|
|
with self.assertRaises(ValueError):
|
|
sc.float_key(struct.pack("<f", 1.0), "tol:2")
|
|
|
|
|
|
# --- masking ------------------------------------------------------------------
|
|
|
|
class MaskTest(unittest.TestCase):
|
|
def test_resave_mask_absorbs_exactly_the_documented_delta(self):
|
|
a = sc.checksum_bytes(tiny(status_a=4, status_b=4, checksum=-1000), mask="resave")
|
|
b = sc.checksum_bytes(tiny(status_a=0, status_b=0, checksum=-1016), mask="resave")
|
|
self.assertEqual(a.digest, b.digest)
|
|
|
|
def test_mask_reports_what_it_hit(self):
|
|
ck = sc.checksum_bytes(tiny(), mask="resave")
|
|
self.assertEqual(ck.mask_hits, {"Status": 2, "Checksum": 1})
|
|
|
|
def test_mask_is_path_scoped(self):
|
|
"""A same-named tag outside the rule's path must NOT be masked."""
|
|
w = sw.SaveWriter("joint")
|
|
w.begin("Sim")
|
|
w.int("Status", 4) # not under /Sim/players/Player[...]
|
|
w.end()
|
|
ck = sc.checksum_bytes(w.bytes(), mask="resave")
|
|
self.assertEqual(ck.mask_hits, {})
|
|
other = sc.checksum_bytes(w.bytes().replace(struct.pack("<i", 4),
|
|
struct.pack("<i", 0)), mask="resave")
|
|
self.assertNotEqual(ck.digest, other.digest)
|
|
|
|
def test_no_mask_by_default(self):
|
|
a = sc.checksum_bytes(tiny(status_a=4))
|
|
b = sc.checksum_bytes(tiny(status_a=0))
|
|
self.assertNotEqual(a.digest, b.digest)
|
|
self.assertEqual(a.mask_hits, {})
|
|
|
|
|
|
# --- --relabel-new-ids --------------------------------------------------------
|
|
#
|
|
# The feature is small; the guards are the point. Every test below that asserts
|
|
# a REFUSAL is asserting that a real difference cannot be absorbed.
|
|
|
|
class RelabelTest(unittest.TestCase):
|
|
A_NEW = [(1970, "Sigma Fleet", 384, None, [6976]),
|
|
(1986, "Tau Fleet", 0, 80, [5264])]
|
|
#: the same two groups, visited in the other order: the id and the
|
|
#: id-attached name stay put, the contents swap
|
|
B_NEW = [(1970, "Sigma Fleet", 0, 80, [5264]),
|
|
(1986, "Tau Fleet", 384, None, [6976])]
|
|
|
|
def trees(self, a_new, b_new, pre=PRE_FLEETS, **kw):
|
|
p = sr.read_bytes(fleets_save(pre))
|
|
a = sr.read_bytes(fleets_save(pre + list(a_new), **kw))
|
|
b = sr.read_bytes(fleets_save(pre + list(b_new), **kw))
|
|
return p.tree, a.tree, b.tree, a, b
|
|
|
|
def compare(self, a_res, b_res, rl):
|
|
sc.apply_relabelling(a_res.tree, rl)
|
|
ca = sc.checksum_result(a_res, "A", {"ok": None},
|
|
extra_masks=sc.RELABEL_MASK, id_lists_as_sets=True)
|
|
cb = sc.checksum_result(b_res, "B", {"ok": None},
|
|
extra_masks=sc.RELABEL_MASK, id_lists_as_sets=True)
|
|
return ca, cb, sc.diff(ca.root, cb.root)
|
|
|
|
# -- the feature ---------------------------------------------------------
|
|
|
|
def test_a_pure_transposition_is_absorbed_and_pi_is_reported(self):
|
|
p, a, b, ares, bres = self.trees(self.A_NEW, self.B_NEW)
|
|
rl = sc.build_relabelling(p, a, b)
|
|
self.assertEqual(rl.pi, {1970: 1986, 1986: 1970})
|
|
self.assertEqual(sc.format_permutation(rl.pi), "{1970<->1986}")
|
|
ca, cb, d = self.compare(ares, bres, rl)
|
|
self.assertEqual(d, [], [repr(e) for e in d])
|
|
self.assertEqual(ca.digest, cb.digest)
|
|
|
|
def test_the_ship_to_fleet_back_reference_is_rewritten(self):
|
|
p, a, b, ares, bres = self.trees(self.A_NEW, self.B_NEW)
|
|
rl = sc.apply_relabelling(ares.tree, sc.build_relabelling(p, a, b))
|
|
self.assertEqual(rl.rewrites.get("FltID"), 2)
|
|
self.assertEqual(rl.moves, 2)
|
|
|
|
def test_a_system_s_fleet_reference_is_rewritten(self):
|
|
p, a, b, ares, bres = self.trees(self.A_NEW, self.B_NEW)
|
|
# the same physical group is at system 384 in both, so the system's
|
|
# fleet reference names 1970 in A and 1986 in B
|
|
_, a2, _, ares2, _ = self.trees(self.A_NEW, self.B_NEW, sys_flt=1970)
|
|
_, _, b2, _, bres2 = self.trees(self.A_NEW, self.B_NEW, sys_flt=1986)
|
|
rl = sc.build_relabelling(p, a2, b2)
|
|
ca, cb, d = self.compare(ares2, bres2, rl)
|
|
self.assertEqual(rl.rewrites.get("Flt"), 1)
|
|
self.assertEqual(d, [], [repr(e) for e in d])
|
|
|
|
def test_the_identity_case_leaves_everything_alone(self):
|
|
p, a, b, ares, bres = self.trees(self.A_NEW, self.A_NEW)
|
|
rl = sc.build_relabelling(p, a, b)
|
|
self.assertEqual(rl.moved, {})
|
|
self.assertFalse(bool(rl))
|
|
ca, cb, d = self.compare(ares, bres, rl)
|
|
self.assertEqual(d, [])
|
|
|
|
def test_pi_is_printed_as_a_cycle_when_it_is_not_a_transposition(self):
|
|
self.assertEqual(sc.format_permutation({1: 2, 2: 3, 3: 1}), "{1->2->3->1}")
|
|
self.assertIn("identity", sc.format_permutation({1970: 1970}))
|
|
|
|
def test_a_relabelled_root_is_domain_separated_from_a_plain_one(self):
|
|
"""A 'modulo pi' root must never be mistaken for a strict one."""
|
|
data = fleets_save(PRE_FLEETS)
|
|
plain = sc.checksum_bytes(data)
|
|
loose = sc.checksum_bytes(data, extra_masks=sc.RELABEL_MASK,
|
|
id_lists_as_sets=True)
|
|
self.assertNotEqual(plain.digest, loose.digest)
|
|
|
|
def test_the_default_path_is_untouched_by_the_new_parameters(self):
|
|
data = fleets_save(PRE_FLEETS)
|
|
self.assertEqual(sc.checksum_bytes(data).digest,
|
|
sc.checksum_bytes(data, extra_masks=(),
|
|
id_lists_as_sets=False).digest)
|
|
|
|
# -- G1: nothing present in the pre-turn save may be relabelled ----------
|
|
|
|
def test_ids_present_in_the_pre_turn_save_are_never_in_pi(self):
|
|
p, a, b, _, _ = self.trees(self.A_NEW, self.B_NEW)
|
|
rl = sc.build_relabelling(p, a, b)
|
|
pre_ids = {n.value for n, _ in sc._fleet_slots(p)}
|
|
self.assertTrue(pre_ids)
|
|
self.assertFalse(pre_ids & (set(rl.pi) | set(rl.pi.values())))
|
|
|
|
def test_two_pre_turn_fleets_that_swapped_contents_are_NOT_absorbed(self):
|
|
"""The strongest form of G1: the very shape this mode absorbs, but on
|
|
ids that already existed. That would be a real divergence."""
|
|
swapped = [(258, "Alpha Fleet", 0, 700, [32, 48]),
|
|
(274, "Beta Fleet", 100, None, [16])]
|
|
p = sr.read_bytes(fleets_save(PRE_FLEETS)).tree
|
|
a = sr.read_bytes(fleets_save(PRE_FLEETS)).tree
|
|
b = sr.read_bytes(fleets_save(swapped)).tree
|
|
rl = sc.build_relabelling(p, a, b)
|
|
self.assertEqual(rl.pi, {})
|
|
ares = sr.read_bytes(fleets_save(PRE_FLEETS))
|
|
bres = sr.read_bytes(fleets_save(swapped))
|
|
_, _, d = self.compare(ares, bres, rl)
|
|
self.assertTrue(d, "a swap among pre-existing fleets must still report")
|
|
|
|
# -- G2: only client-minted (non-zero node nibble) ids ------------------
|
|
|
|
def test_server_minted_ids_are_never_relabelled(self):
|
|
"""Node nibble 0 is the server's space (id-allocation.md section 1)."""
|
|
a_new = [(7120, "Phi Fleet", 384, None, [6976]),
|
|
(7136, "Chi Fleet", 0, 80, [5264])]
|
|
b_new = [(7120, "Phi Fleet", 0, 80, [5264]),
|
|
(7136, "Chi Fleet", 384, None, [6976])]
|
|
p, a, b, ares, bres = self.trees(a_new, b_new)
|
|
rl = sc.build_relabelling(p, a, b)
|
|
self.assertEqual(rl.pi, {})
|
|
self.assertEqual(sc.id_node(7120), 0)
|
|
_, _, d = self.compare(ares, bres, rl)
|
|
self.assertTrue(d, "a node-0 swap is not this residue and must report")
|
|
|
|
# -- G3: pi must permute one set of new ids -----------------------------
|
|
|
|
def test_different_new_id_sets_are_refused(self):
|
|
"""Lane AD's unpinned pair in miniature: one run minted a third fleet."""
|
|
b_new = self.B_NEW + [(2002, "Upsilon Fleet", 0, 384, [6992])]
|
|
p, a, b, _, _ = self.trees(self.A_NEW, b_new)
|
|
with self.assertRaises(sc.RelabelRefused) as cm:
|
|
sc.build_relabelling(p, a, b)
|
|
self.assertIn("only in B [2002]", str(cm.exception))
|
|
|
|
def test_a_refusal_leaves_both_trees_untouched(self):
|
|
b_new = self.B_NEW + [(2002, "Upsilon Fleet", 0, 384, [6992])]
|
|
p, a, b, ares, bres = self.trees(self.A_NEW, b_new)
|
|
before = sc.checksum_result(ares, "A", {"ok": None}).digest
|
|
with self.assertRaises(sc.RelabelRefused):
|
|
sc.build_relabelling(p, a, b)
|
|
self.assertEqual(sc.checksum_result(ares, "A", {"ok": None}).digest, before)
|
|
|
|
# -- G4: the content match must be a one-to-one correspondence ----------
|
|
|
|
def test_new_fleets_whose_contents_differ_are_refused(self):
|
|
b_new = [(1970, "Sigma Fleet", 0, 80, [5264]),
|
|
(1986, "Tau Fleet", 384, None, [9999])] # a different ship
|
|
p, a, b, _, _ = self.trees(self.A_NEW, b_new)
|
|
with self.assertRaises(sc.RelabelRefused) as cm:
|
|
sc.build_relabelling(p, a, b)
|
|
self.assertIn("do not correspond by content", str(cm.exception))
|
|
|
|
def test_an_ambiguous_content_key_is_refused(self):
|
|
dup = [(1970, "Sigma Fleet", 384, None, [6976]),
|
|
(1986, "Tau Fleet", 384, None, [6976])]
|
|
p, a, b, _, _ = self.trees(dup, dup)
|
|
with self.assertRaises(sc.RelabelRefused) as cm:
|
|
sc.build_relabelling(p, a, b)
|
|
self.assertIn("ambiguous", str(cm.exception))
|
|
|
|
def test_a_fleet_at_a_system_never_keys_the_same_as_one_bound_for_it(self):
|
|
at = sr.read_bytes(fleets_save([(1970, "S", 384, None, [1])])).tree
|
|
to = sr.read_bytes(fleets_save([(1970, "S", 0, 384, [1])])).tree
|
|
ka = sc._fleet_content_key(sc._fleet_slots(at)[0][1])
|
|
kb = sc._fleet_content_key(sc._fleet_slots(to)[0][1])
|
|
self.assertNotEqual(ka, kb)
|
|
|
|
# -- the name guard ------------------------------------------------------
|
|
|
|
def test_a_name_that_does_not_track_the_id_is_refused(self):
|
|
"""`FtName` is minted from the same per-fleet ordinal as the id, so it
|
|
is an id-attached label -- but only while it demonstrably tracks the
|
|
id in both saves. If it does not, something real differs."""
|
|
b_new = [(1970, "Tau Fleet", 0, 80, [5264]),
|
|
(1986, "Sigma Fleet", 384, None, [6976])]
|
|
p, a, b, _, _ = self.trees(self.A_NEW, b_new)
|
|
with self.assertRaises(sc.RelabelRefused) as cm:
|
|
sc.build_relabelling(p, a, b)
|
|
self.assertIn("does not track the id", str(cm.exception))
|
|
|
|
def test_the_name_stays_with_the_slot_not_with_the_body(self):
|
|
p, a, b, ares, bres = self.trees(self.A_NEW, self.B_NEW)
|
|
rl = sc.apply_relabelling(ares.tree, sc.build_relabelling(p, a, b))
|
|
got = {n.value: sc._fleet_name(f) for n, f in sc._fleet_slots(ares.tree)}
|
|
self.assertEqual(got[1970], "Sigma Fleet")
|
|
self.assertEqual(got[1986], "Tau Fleet")
|
|
|
|
# -- G5: the rewrite must be complete -----------------------------------
|
|
|
|
def test_an_unmodelled_reference_site_is_refused(self):
|
|
"""A leaf holding a permuted id at a tag this tool does not model would
|
|
keep the old label through the rewrite, so two different states could
|
|
compare equal. That is a refusal, not a warning."""
|
|
p, a, b, _, _ = self.trees(self.A_NEW, self.B_NEW,
|
|
extra=("SomeOtherRef", 1970))
|
|
with self.assertRaises(sc.RelabelRefused) as cm:
|
|
sc.build_relabelling(p, a, b)
|
|
self.assertIn("does not model", str(cm.exception))
|
|
self.assertIn("SomeOtherRef", str(cm.exception))
|
|
|
|
def test_the_completeness_scan_matches_raw_bytes_not_the_typed_value(self):
|
|
"""The digest is a function of (bytes, reader schema); a leaf the schema
|
|
types as a float still carries the id. A value-based scan would miss
|
|
it and the guard would be silently empty (rule 1)."""
|
|
node = sr.Node("Odd", "float", 2.76e-42, 0, 12,
|
|
raw=struct.pack("<i", 1970))
|
|
frame = sr.Node("Sim", "complex", None, 0, 0)
|
|
frame.children = [node]
|
|
root = sr.Node(None, "complex", None, 0, 0)
|
|
root.children = [frame]
|
|
found = sc._id_occurrences(root, {1970})
|
|
self.assertEqual([s for s, _, _ in found], ["UNMODELLED"])
|
|
|
|
# -- CLI -----------------------------------------------------------------
|
|
|
|
def test_the_flag_is_rejected_outside_a_two_save_comparison(self):
|
|
with self.assertRaises(SystemExit):
|
|
sc.main(["one.sav", "--relabel-new-ids", "pre.sav"])
|
|
|
|
|
|
# --- chain --------------------------------------------------------------------
|
|
|
|
class ChainTest(unittest.TestCase):
|
|
def _write(self, data: bytes) -> str:
|
|
f = tempfile.NamedTemporaryFile(suffix=".sav", delete=False)
|
|
f.write(gzip.compress(data, mtime=0))
|
|
f.close()
|
|
self.addCleanup(os.unlink, f.name)
|
|
return f.name
|
|
|
|
def test_record_then_verify_matches(self):
|
|
files = [self._write(tiny(checksum=-1000)), self._write(tiny(checksum=-2000))]
|
|
chain = sc.build_chain(files)
|
|
self.assertTrue(all(t["coverage"] for t in chain["turns"]))
|
|
ok, msgs = sc.verify_chain(chain, files)
|
|
self.assertTrue(ok, msgs)
|
|
|
|
def test_verify_stops_at_the_first_divergent_turn(self):
|
|
files = [self._write(tiny(checksum=-1000)), self._write(tiny(checksum=-2000))]
|
|
chain = sc.build_chain(files)
|
|
bad = [files[0], self._write(tiny(checksum=-2001))]
|
|
ok, msgs = sc.verify_chain(chain, bad)
|
|
self.assertFalse(ok)
|
|
self.assertIn("MATCH", msgs[0])
|
|
self.assertIn("DIVERGE", msgs[1])
|
|
self.assertTrue(any("/Summary" in m for m in msgs))
|
|
|
|
def test_chain_records_the_policy_it_was_built_under(self):
|
|
files = [self._write(tiny())]
|
|
chain = sc.build_chain(files, floats="canonical", mask="resave")
|
|
self.assertEqual(chain["policy"]["floats"], "canonical")
|
|
self.assertEqual(chain["policy"]["mask"], "resave")
|
|
self.assertEqual(chain["policy"]["readerFingerprint"], sc.READER_FINGERPRINT)
|
|
json.dumps(chain) # must stay serialisable
|
|
|
|
def test_verifying_under_a_different_policy_is_refused(self):
|
|
files = [self._write(tiny())]
|
|
chain = sc.build_chain(files, floats="bits")
|
|
ok, msgs = sc.verify_chain(chain, files, floats="canonical")
|
|
self.assertFalse(ok)
|
|
self.assertTrue(any("not comparable" in m for m in msgs), msgs)
|
|
|
|
def test_a_stale_reader_fingerprint_is_called_out(self):
|
|
files = [self._write(tiny())]
|
|
chain = sc.build_chain(files)
|
|
chain["policy"]["readerFingerprint"] = "0" * 16
|
|
_, msgs = sc.verify_chain(chain, files)
|
|
self.assertTrue(any("save_reader" in m for m in msgs), msgs)
|
|
|
|
|
|
# --- real saves ---------------------------------------------------------------
|
|
|
|
@needs_saves
|
|
class RealSaveTest(unittest.TestCase):
|
|
def test_coverage_is_proved_on_every_available_save(self):
|
|
for p in REAL:
|
|
with self.subTest(os.path.basename(p)):
|
|
ck = real_ck(p)
|
|
self.assertTrue(ck.coverage["ok"],
|
|
f"reconstruction diverged at {ck.coverage['firstDiff']}")
|
|
self.assertEqual(ck.coverage["rebuiltBytes"], ck.coverage["inflatedBytes"])
|
|
|
|
def test_identical_bytes_give_identical_digests(self):
|
|
by_hash: dict = {}
|
|
for p in REAL:
|
|
with open(p, "rb") as f:
|
|
data = f.read()
|
|
by_hash.setdefault(data, []).append(p)
|
|
for data, paths in by_hash.items():
|
|
if len(paths) < 2:
|
|
continue
|
|
digs = {real_ck(p).digest for p in paths}
|
|
self.assertEqual(len(digs), 1, paths)
|
|
|
|
def test_repeated_runs_are_stable(self):
|
|
p = REAL[0]
|
|
self.assertEqual(real_ck(p).digest, real_ck(p).digest)
|
|
|
|
def test_reader_is_clean_on_every_save(self):
|
|
for p in REAL:
|
|
with self.subTest(os.path.basename(p)):
|
|
ck = real_ck(p, audit=False)
|
|
self.assertEqual(ck.res.count("error"), 0)
|
|
self.assertEqual(ck.res.count("warn"), 0)
|
|
|
|
def test_canonical_float_policy_is_a_noop_on_this_corpus(self):
|
|
"""No save holds a NaN, a -0.0 or an infinity, so 'canonical' changes
|
|
nothing below the root. Recorded so a future save that breaks it fails
|
|
here rather than silently."""
|
|
for p in REAL:
|
|
with self.subTest(os.path.basename(p)):
|
|
a = real_ck(p, audit=False, floats="bits")
|
|
b = real_ck(p, audit=False, floats="canonical")
|
|
self.assertEqual(sc.diff(a.root, b.root), [])
|
|
|
|
def test_known_resave_delta_localises_to_five_named_leaves(self):
|
|
"""determinism-oracle.md: loading a post-turn autosave and re-saving
|
|
changes exactly Player.Status (4 -> 0) on the four turn-participating
|
|
players, plus the derived Summary.Checksum. The tree must name them."""
|
|
pairs = []
|
|
for a in REAL:
|
|
for b in REAL:
|
|
if a >= b:
|
|
continue
|
|
ca, cb = real_ck(a, audit=False), real_ck(b, audit=False)
|
|
d = sc.diff(ca.root, cb.root)
|
|
if d and all(e.kind == "value" for e in d) and \
|
|
{e.path.rsplit("/", 1)[-1] for e in d} == {"Status", "Checksum"}:
|
|
pairs.append((a, b, d))
|
|
if not pairs:
|
|
self.skipTest("no re-save pair among the available saves")
|
|
for a, b, d in pairs:
|
|
self.assertEqual(len(d), 5, [repr(e) for e in d])
|
|
statuses = [e for e in d if e.path.endswith("/Status")]
|
|
self.assertEqual(len(statuses), 4)
|
|
# `pairs` is enumerated over sorted filenames, so the re-saved file
|
|
# can be either side; the claim is symmetric, so assert it that way
|
|
# rather than assuming which name sorts first (2026-09-09: adding
|
|
# `cb-turn2to3-endturn.sav` to the corpus put one pair the other
|
|
# way round and this test failed on a corpus fact, not a defect).
|
|
direction = (statuses[0].a, statuses[0].b)
|
|
self.assertIn(direction, ((4, 0), (0, 4)))
|
|
for e in statuses:
|
|
self.assertIn("/Sim/players/Player[", e.path)
|
|
self.assertEqual((e.a, e.b), direction)
|
|
chk = [e for e in d if e.path == "/Summary/Checksum"]
|
|
self.assertEqual(len(chk), 1)
|
|
self.assertEqual(chk[0].b - chk[0].a,
|
|
-16 if direction == (4, 0) else 16)
|
|
|
|
def test_resave_mask_makes_that_pair_identical(self):
|
|
found = False
|
|
for a in REAL:
|
|
for b in REAL:
|
|
if a >= b:
|
|
continue
|
|
if real_ck(a, audit=False, mask="resave").digest == \
|
|
real_ck(b, audit=False, mask="resave").digest and \
|
|
real_ck(a, audit=False).digest != \
|
|
real_ck(b, audit=False).digest:
|
|
found = True
|
|
if not found:
|
|
self.skipTest("no re-save pair among the available saves")
|
|
self.assertTrue(found)
|
|
|
|
# -- --relabel-new-ids acceptance, both halves -------------------------
|
|
|
|
def _saves(self, *names):
|
|
d = os.path.dirname(REAL[0])
|
|
paths = [os.path.join(d, n) for n in names]
|
|
if not all(os.path.exists(p) for p in paths):
|
|
self.skipTest(f"need {names}")
|
|
return paths
|
|
|
|
def _relabelled(self, pre, a, b):
|
|
return sc._relabelled_pair(pre, a, b, floats="bits", mask="none", audit=False)
|
|
|
|
def test_POSITIVE_bp_pinned_pair_is_identical_modulo_one_transposition(self):
|
|
"""Lane BP's two processes with identical pinned AI seeds differ in 35
|
|
of 61,147 leaves; all 35 are fleets 1970 and 1986 exchanging contents.
|
|
Under pi they must read IDENTICAL, and pi must be exactly that swap."""
|
|
pre, a, b = self._saves("bp-turn28-pre.sav", "bp-pinA-turn28.sav",
|
|
"bp-pinB-turn28.sav")
|
|
ca, cb, rl, refusal = self._relabelled(pre, a, b)
|
|
self.assertIsNone(refusal)
|
|
self.assertEqual(rl.moved, {1970: 1986, 1986: 1970})
|
|
self.assertEqual(rl.new_ids, [1970, 1986, 2002])
|
|
self.assertEqual(rl.nodes, [2])
|
|
self.assertEqual(sc.diff(ca.root, cb.root), [])
|
|
self.assertEqual(ca.digest, cb.digest)
|
|
|
|
def test_POSITIVE_the_same_pair_still_diverges_without_the_flag(self):
|
|
pre, a, b = self._saves("bp-turn28-pre.sav", "bp-pinA-turn28.sav",
|
|
"bp-pinB-turn28.sav")
|
|
d = sc.diff(real_ck(a, audit=False).root, real_ck(b, audit=False).root)
|
|
self.assertEqual(len(d), 35, [repr(e) for e in d])
|
|
|
|
def test_NEGATIVE_ad_unpinned_pair_is_refused_and_still_diverges(self):
|
|
"""Lane AD's pair differs because the seeds differ -- real decisions.
|
|
A tool that made this look identical would be worthless."""
|
|
pre, a, b = self._saves("ad-turn27-two-raiders.sav", "ad-oracle-A-post.sav",
|
|
"ad-oracle-B-post.sav")
|
|
ca, cb, rl, refusal = self._relabelled(pre, a, b)
|
|
self.assertIsNone(rl, "AD's pair must not be relabelled")
|
|
self.assertIn("2002", refusal)
|
|
self.assertNotEqual(ca.digest, cb.digest)
|
|
d = sc.diff(ca.root, cb.root, limit=500)
|
|
self.assertEqual(len(d), 94, [repr(e) for e in d])
|
|
|
|
def test_NEGATIVE_even_forcing_a_partial_relabelling_cannot_hide_it(self):
|
|
"""Belt and braces: hand the applier the transposition it *would* have
|
|
found if G3 were absent. The pair must still diverge -- the guard is
|
|
not the only thing standing between AD's pair and a green verdict."""
|
|
pre, a, b = self._saves("ad-turn27-two-raiders.sav", "ad-oracle-A-post.sav",
|
|
"ad-oracle-B-post.sav")
|
|
with open(a, "rb") as f:
|
|
ares = sr.read_bytes(f.read())
|
|
with open(b, "rb") as f:
|
|
bres = sr.read_bytes(f.read())
|
|
forced = sc.Relabelling({1970: 1986, 1986: 1970}, [2], [1970, 1986],
|
|
{1970: None, 1986: None})
|
|
sc.apply_relabelling(ares.tree, forced)
|
|
ca = sc.checksum_result(ares, a, {"ok": None},
|
|
extra_masks=sc.RELABEL_MASK, id_lists_as_sets=True)
|
|
cb = sc.checksum_result(bres, b, {"ok": None},
|
|
extra_masks=sc.RELABEL_MASK, id_lists_as_sets=True)
|
|
self.assertNotEqual(ca.digest, cb.digest)
|
|
|
|
def test_a_real_turn_transition_localises_to_named_objects(self):
|
|
by_turn = {}
|
|
for p in REAL:
|
|
ck = real_ck(p, audit=False)
|
|
t = (ck.res.typed.get("summary") or {}).get("Turn")
|
|
by_turn.setdefault(t, ck)
|
|
if not {2, 3} <= set(by_turn):
|
|
self.skipTest("need a turn-2 and a turn-3 save")
|
|
d = sc.diff(by_turn[2].root, by_turn[3].root, limit=500)
|
|
self.assertGreater(len(d), 10)
|
|
self.assertTrue(any(e.path == "/Summary/Turn" for e in d))
|
|
# every difference must be attributed to a path, never to a bare hash
|
|
for e in d:
|
|
self.assertTrue(e.path.startswith("/"), repr(e))
|
|
named = [e for e in d if "/Sim/players/Player[" in e.path]
|
|
self.assertTrue(named, "a turn transition must move at least one player")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|