alex/sots-re
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

board: claim VM140 for lane M; MoveFleet ULP + ObservedTech rows

Browse source
This commit is contained in:
alex 2026-09-08 03:09:37 -04:00
parent 05095e93da
commit 5b74b98778
3 changed files with 1215 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
campaign/board.md
View file

@ -75,3 +75,5 @@ Status flow: `backlog → in-progress → mapped → verified` (or `blocked`).
| ReVa MCP link drop (workaround) | meta | verified | high | 100% | 2026-09-08 | The ReVa MCP client link dropped mid-session while the CT111 server stayed healthy (systemd active, :8080 listening, valid key -> 200). `tools/reva_call.py <tool> '<json>'` calls the same server over plain HTTP (initialize -> notifications/initialized -> tools/call; replies are SSE with a leading `id:` line, initialize is plain JSON). Key is NEVER stored in the repo: $REVA_KEY, else ~/.claude.json, else ssh to the CT properties file. Use this whenever mcp__plugin_ReVa_ReVa__* is unavailable |
| event posting API | subsystem | mapped | high | 90% | 2026-09-08 | RECOVERED (lane E, `findings/subsystems/events.md`). Container: `EventStorage` embedded at `ServerPlayer+0x29c` (0x1c), `EvNxID` at +0x14 = player+0x2b0 — exactly the guard's byte run. Nested `vector<TurnEvents{int EvTurn; vector<PlayerEvent>}>`, record 0x74 B, tags `EvEID EvDsc EvMsg EvImg EvLoc EvPos EvAct EvCID`; layout confirmed field-by-field against turn3-state.sav, which CONTAINS the overbudget record. Entry point `int __thiscall EventStorage::PostEvent(this, string BYVAL, string BYVAL, obj*, Vector3*, turn, const char* img, int act)` 0x008862b0 RET 0x4c — **161 call sites in 113 functions, the whole sim's event API**. B3 defect fully explained: 0x00587b97, in the completion-roll-FAILED branch under `!wasDone && nowDone && owner`. 3 note corrections (EvPos is FLT_MAX not inf; the save array is turn-bucketed not flat; TECHS_UNLOCKED has no parent clause). 56 entries in addresses.json; 11 prototypes + 13 labels + 12 comments + 2 structs written back to Ghidra. Engine: `sots-engine` branch `wip/events` a7348be, `src/game/events` + 112 checks, ctest 32/32. NOT YET WIRED INTO A HOOK — see `docs/E-events.md` for the proposed region/Coverage change |
| state-checksum replay harness | verify | in-progress | — | 0% | 2026-09-08 | Lane C: whole-state per-turn checksum as the COMPLEMENT to per-function compares (no region-declaration mistake can hide from it). Diagnostic tree that localises WHICH object moved, explicit float-parity policy (float32 + fpu_cw 0x127f + fistp ties-to-even; matters for the future x64/SSE port). Replay loop designed but unrun - lane R holds VM140 |
| MoveFleet position ULP divergence | phase2 | in-progress | — | 0% | 2026-09-08 | Lane M. FIRST arithmetic divergence caught by BEHAVIOURAL compare rather than static reading: 8 of 45 calls differ by 1 ULP on a position component (worst 64 ULP at a near-zero result; absolute error ~1.2e-7 everywhere = half an ULP of the INPUTS) => one rounding too many/few in the position update. Step length and all ship ranges match. B4 saw only 1 moving call and it still matches 4 of 5 moves, which is exactly why 1-call coverage is not evidence. All 15 moving calls are the same straight-run waypoint type; types 2-5 never occurred |
| ObservedTech append (undeclared) | verify | backlog | — | 0% | 2026-09-08 | Lane R's guards caught a vector<ObservedTech> append at `player+0x274` during SetResearched. It is SERIALIZED state and appears in NO coverage note anywhere - found only because guards localise rather than just flag a moved hash. Needs a declared region + a model in ours |

810
verify/state-checksum/state_checksum.py Normal file
View file

@ -0,0 +1,810 @@
#!/usr/bin/env python3
"""state_checksum.py -- whole-state, diagnostic checksum of a SOTS1 save.
Why this exists
---------------
Every other verification in this project is *per-function*: hook one routine,
compare the regions it declares, print a verdict. That is only as good as the
region declaration, and region declarations have been wrong (B4: three hooks
that reported "0 diverged" while comparing nothing; the harness audit: 23
undeclared side effects). A green verdict over an empty region set is a
failure, not a pass.
This tool is the complement. It does not ask "did the declared outputs match".
It asks **"is the entire simulation state still identical"**, computed from the
one artefact the game emits as a pure function of state -- the save file, which
`findings/subsystems/determinism-oracle.md` proved byte-identical across runs
and processes.
Two properties make it evidence rather than a comforting number:
1. **Provable coverage.** The checksum is computed from a tree that is
re-serialised and compared byte-for-byte against the inflated save
(`--audit`, on by default). If the reconstruction reproduces the stream,
then every byte of the save is a function of the checksum's inputs, so *no*
state change can be invisible to it. Coverage is asserted by construction,
not by a hand-written region list that someone can forget to fill in.
2. **It localises.** The root digest is the fold of a tree of per-subsystem /
per-object digests, and objects are named (`Player[16 "re"]`,
`Sys[112 "Gamma Cephei"]`, `Events/[turn=3]/[id=3]`), so a divergence
reports *which object moved*, not just that the hash moved.
Float parity: see STATE_CHECKSUM.md section 3. Short version -- the digest is
always exact (raw IEEE-754 bits, or bits with -0.0/NaN normalised under
`--floats canonical`). Tolerance is deliberately *not* a hashing mode; it
exists only in the differ (`--ulps`), because a tolerant hash is a
contradiction: quantisation just moves the cliff, it does not remove it.
Usage
-----
state_checksum.py SAVE # root + subsystem digests
state_checksum.py SAVE --tree --depth 3 # digest tree
state_checksum.py A B # localise divergence
state_checksum.py A B --ulps 2 # classify float diffs
state_checksum.py --chain chain.json S1 S2 ... # verify a recorded chain
state_checksum.py SAVE --json # machine-readable
Stdlib only (plus save_reader.py, which is stdlib only).
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import struct
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
_READER = os.path.join(os.path.dirname(_HERE), "save-reader")
if _READER not in sys.path:
sys.path.insert(0, _READER)
import save_reader as sr # noqa: E402
__all__ = [
"FLOAT_POLICIES", "MASK_PRESETS", "CkNode", "checksum_save", "checksum_bytes",
"reconstruct", "audit_coverage", "diff", "DiffEntry", "float_key",
]
DIGEST_BYTES = 16 # blake2b-128; 2**-64 collision floor at our object counts
SHORT = 16 # hex chars shown in text output
BEGIN = struct.pack("<I", 0xBEEFBEEF)
END = struct.pack("<I", 0x41104110)
# --- policy -------------------------------------------------------------------
FLOAT_POLICIES = ("bits", "canonical")
#: Documented canonicalisation presets (see determinism-oracle.md). A rule is
#: (leaf tag, required path prefix, replacement kind, replacement value) -- the
#: path prefix keeps a mask from silently swallowing a same-named field
#: somewhere else in the tree, and the run reports how many leaves each rule
#: actually hit so an over-broad mask is visible rather than comfortable.
#:
#: Masking is opt-in. By default nothing is masked, so the known re-save delta
#: is *localised* rather than absorbed.
MASK_PRESETS = {
"none": (),
# load -> save of a post-turn autosave: the server clears the per-player
# "turn ended" flag (4 -> 0) and the derived top-level Checksum follows.
"resave": (
("Status", "/Sim/players/Player[", "int", 0),
("Checksum", "/Summary/", "derived", None),
),
}
def float_key(raw: bytes, policy: str) -> bytes:
"""Hashing key for a 4-byte IEEE-754 single under `policy`."""
if policy == "bits":
return raw
if policy != "canonical":
raise ValueError(f"unknown float policy {policy!r}")
(bits,) = struct.unpack("<I", raw)
if bits == 0x80000000: # -0.0 -> +0.0
return b"\0\0\0\0"
if (bits & 0x7F800000) == 0x7F800000 and (bits & 0x007FFFFF):
return struct.pack("<I", 0x7FC00000) # any NaN -> one quiet NaN
return raw
def ulps_apart(a: bytes, b: bytes) -> float:
"""Distance in float32 ULPs between two raw singles. inf if not comparable."""
(x,) = struct.unpack("<f", a)
(y,) = struct.unpack("<f", b)
if math.isnan(x) or math.isnan(y):
return 0.0 if (math.isnan(x) and math.isnan(y)) else math.inf
if math.isinf(x) or math.isinf(y):
return 0.0 if x == y else math.inf
def ordinal(v: bytes) -> int:
(u,) = struct.unpack("<i", v)
return u if u >= 0 else -0x80000000 - u # monotone map across the sign
return float(abs(ordinal(a) - ordinal(b)))
# --- naming -------------------------------------------------------------------
#: lead-int tag -> (frame tag, group name). The save stores object tables as a
#: flat run of (id, frame) sibling pairs; folding each run into a named group
#: turns `Sim`'s 259 children into a subsystem tree.
PAIR_GROUPS = {
"PlayerID": ("Player", "players"),
"SysID": ("Sys", "systems"),
"FltID": ("Flt", "fleets"),
"ShipID": ("Ship", "ships"),
"DesID": ("Des", "designs"),
"TradeID": ("Trade", "trades"),
"ply": ("hist", "history"),
}
#: frame tag -> child tags to use as a human label, first match wins
NAME_FIELDS = {
"Player": ("PlryName",),
"Sys": ("Name",),
"Flt": ("FltNm", "Name", "FName"),
"Ship": ("ShpNm", "Name", "SName"),
"Des": ("DName",),
}
#: child tags that identify a NULL-named ('.') element frame
ELEM_KEYS = ("EvTurn", "EvEID", "ordID", "npid", "desID", "Wpt", "seno2", "ID", "set")
#: NonComplexArray count tags whose elements are a run of `.` scalars in the
#: *same* frame (SAVE_FORMAT.md section 4, "inline count"). Left flat, a single
#: id insertion shifts every following sibling and the differ reports a hundred
#: spurious moves; folded, it reports one list that gained one id.
INLINE_ID_LISTS = ("PlayerIDs", "DesignIDs", "SystemIDs", "FleetIDs",
"ShipIDs", "TradeIDs")
#: top-level digest rows printed by the default (non-tree) view. Paths that do
#: not exist in a given save are silently skipped.
SUBSYSTEM_PATHS = (
"/Summary",
"/CreateParams",
"/Sim/RNG",
"/Sim/Attrib",
"/Sim/turnstats",
"/Sim/players",
"/Sim/systems",
"/Sim/fleets",
"/Sim/NdGr2",
"/Sim/trdmgr",
"/Sim/spymgr",
"/Sim/SvSctOb",
"/CDT",
)
#: per-player sub-rollups, appended under each player row in --subsystems
PLAYER_PARTS = ("TechTree", "Events", "designs", "Diplo", "spy2", "civr", "comms", "Ojvs")
# --- the digest tree ----------------------------------------------------------
class CkNode:
"""One node of the checksum tree."""
__slots__ = ("path", "label", "kind", "digest", "offset", "size",
"leaves", "value_bytes", "children", "value", "raw", "masked",
"values")
def __init__(self, path, label, kind, digest, offset, size,
leaves, value_bytes, children, value=None, raw=b"", masked=False,
values=None):
self.values = values # inline id list contents, for set-diffing
self.path = path
self.label = label
self.kind = kind
self.digest = digest
self.offset = offset
self.size = size
self.leaves = leaves # number of scalar leaves below (or 1)
self.value_bytes = value_bytes # bytes of *value* payload below (framing excluded)
self.children = children
self.value = value
self.raw = raw
self.masked = masked
@property
def hexd(self) -> str:
return self.digest.hex()
def short(self) -> str:
return self.digest.hex()[:SHORT]
def find(self, path: str):
if self.path == path:
return self
for c in self.children:
if path == c.path or path.startswith(c.path + "/"):
return c.find(path)
return None
def walk(self):
yield self
for c in self.children:
yield from c.walk()
def as_dict(self, depth: int = 99, _d: int = 0) -> dict:
out = {"path": self.path, "kind": self.kind, "digest": self.hexd,
"offset": self.offset, "size": self.size,
"leaves": self.leaves, "valueBytes": self.value_bytes}
if self.masked:
out["masked"] = True
if self.kind != "complex" and self.value is not None:
out["value"] = self.value
if self.children and _d < depth:
out["children"] = [c.as_dict(depth, _d + 1) for c in self.children]
elif self.children:
out["childCount"] = len(self.children)
return out
def _h(*parts: bytes) -> bytes:
d = hashlib.blake2b(digest_size=DIGEST_BYTES)
for p in parts:
d.update(struct.pack("<I", len(p)))
d.update(p)
return d.digest()
def _label(node: sr.Node, index: int, lead: sr.Node | None,
unique: bool = False) -> str:
"""Human, stable, greppable label for one item.
A key is appended only when it carries information: an object id, an object
name, or an element key. A bare positional index is appended only when the
tag is ambiguous among its siblings (`unique=False`), so a uniquely-named
field reads as `Status`, not `Status[36]`.
"""
base = node.name if node.name is not None else "<frame>"
key = None
if lead is not None and isinstance(lead.value, int):
key = str(lead.value)
if node.kind == "complex":
want = NAME_FIELDS.get(node.name or "")
if want:
for c in node.children:
if c.name in want and isinstance(c.value, str):
key = f'{key} "{c.value}"' if key else f'"{c.value}"'
break
if key is None and (node.name in (".", None)):
for c in node.children:
if c.name in ELEM_KEYS:
key = f"{c.name}={c.value}"
break
if key is None:
if unique:
return base
key = str(index)
return f"{base}[{key}]"
def _name_counts(nodes) -> dict:
counts: dict = {}
for n in nodes:
counts[n.name] = counts.get(n.name, 0) + 1
return counts
def _group_children(nodes: list[sr.Node]):
"""Fold object runs and inline id lists into synthetic groups.
Returns a list of entries:
("item", node, None) -- a plain child
("group", name, [(id_node, frame_node), ...]) -- an object table
("list", name, count_node, [value_nodes]) -- an inline id list
Order and byte coverage are preserved exactly; nothing is dropped, so the
reconstruction audit still proves total coverage.
"""
out, i, seen = [], 0, {}
n = len(nodes)
while i < n:
cur = nodes[i]
spec = PAIR_GROUPS.get(cur.name or "")
if (spec and i + 1 < n and nodes[i + 1].name == spec[0]
and nodes[i + 1].kind == "complex" and cur.kind in ("int", "float")):
frame_tag, gname = spec
pairs = []
while (i + 1 < n and nodes[i].name == cur.name
and nodes[i + 1].name == frame_tag and nodes[i + 1].kind == "complex"):
pairs.append((nodes[i], nodes[i + 1]))
i += 2
seen[gname] = seen.get(gname, 0) + 1
label = gname if seen[gname] == 1 else f"{gname}({seen[gname]})"
out.append(("group", label, pairs))
continue
if (cur.name in INLINE_ID_LISTS and cur.kind == "int"
and isinstance(cur.value, int) and 0 <= cur.value <= n - i - 1
and all(nodes[i + 1 + k].name == "." and nodes[i + 1 + k].kind != "complex"
for k in range(cur.value))):
vals = nodes[i + 1:i + 1 + cur.value]
out.append(("list", cur.name, cur, vals))
i += 1 + cur.value
continue
out.append(("item", cur, None))
i += 1
return out
def _value_key(node: sr.Node, path: str, policy: str, masks, hits: dict):
"""(kind, key_bytes, masked) for a scalar/raw leaf."""
kind = node.kind
for tag, under, mkind, mval in masks:
if node.name != tag or not path.startswith(under):
continue
hits[tag] = hits.get(tag, 0) + 1
if mkind == "derived":
return kind, b"<derived>", True
if mkind == "int":
return kind, struct.pack("<i", mval), True
raise ValueError(f"bad mask spec {(tag, under, mkind, mval)!r}")
raw = node.raw
if kind == "float":
return kind, float_key(raw, policy), False
if kind == "bool" and policy == "canonical":
return kind, b"\x01" if any(raw) else b"\x00", False
return kind, raw, False
def _build(node: sr.Node, path: str, label: str, policy: str, masks, hits: dict) -> CkNode:
if node.kind == "complex":
entries = _group_children(node.children)
counts = _name_counts(e[1] for e in entries if e[0] == "item")
# positional index is the ordinal among *same-named* siblings, so one
# insertion elsewhere in the frame does not renumber everything after it
ordinals: dict = {}
kids: list[CkNode] = []
for ent in entries:
if ent[0] == "item":
child = ent[1]
k = ordinals.get(child.name, 0)
ordinals[child.name] = k + 1
lab = _label(child, k, None, unique=counts.get(child.name) == 1)
kids.append(_build(child, f"{path}/{lab}", lab, policy, masks, hits))
elif ent[0] == "list":
_, lname, cnode, vals = ent
lpath = f"{path}/{lname}[]"
keys = [_value_key(v, lpath, policy, masks, hits)[1] for v in vals]
ldig = _h(b"list", lname.encode("ascii"), cnode.raw, *keys)
off = cnode.offset
last = vals[-1] if vals else cnode
kids.append(CkNode(lpath, f"{lname}[]", "list", ldig, off,
last.offset + last.size - off,
1 + len(vals),
len(cnode.raw) + sum(len(v.raw) for v in vals),
[], values=[v.value for v in vals]))
else:
_, gname, pairs = ent
gpath = f"{path}/{gname}"
gkids = []
for j, (idn, frame) in enumerate(pairs):
lab = _label(frame, j, idn)
ck = _build(frame, f"{gpath}/{lab}", lab, policy, masks, hits)
# the id int is part of the object's identity; fold it in
idk = _h(b"id", (idn.name or "").encode("ascii"), idn.raw)
ck.digest = _h(b"obj", idk, ck.digest)
ck.offset = idn.offset
ck.size = frame.offset + frame.size - idn.offset
ck.value_bytes += len(idn.raw)
ck.leaves += 1
gkids.append(ck)
gdig = _h(b"group", gname.encode("ascii"),
*[k.digest for k in gkids])
off = gkids[0].offset if gkids else node.offset
size = (gkids[-1].offset + gkids[-1].size - off) if gkids else 0
kids.append(CkNode(gpath, gname, "group", gdig, off, size,
sum(k.leaves for k in gkids),
sum(k.value_bytes for k in gkids), gkids))
dig = _h(b"frame", (node.name or "").encode("ascii"), *[k.digest for k in kids])
return CkNode(path, label, "complex", dig, node.offset, node.size,
sum(k.leaves for k in kids) or 0,
sum(k.value_bytes for k in kids), kids)
kind, key, masked = _value_key(node, path, policy, masks, hits)
dig = _h(b"leaf", (node.name or "").encode("ascii"), kind.encode("ascii"), key)
return CkNode(path, label, kind, dig, node.offset, node.size, 1, len(node.raw),
[], value=node.value, raw=node.raw, masked=masked)
# --- coverage proof -----------------------------------------------------------
def reconstruct(node: sr.Node, out: bytearray, base: int) -> None:
"""Re-serialise a reader Node back into `out`.
`base` is the inflated offset `out` starts at, so joint padding (which is
computed against the absolute 4-byte grid) lands where it did originally.
"""
def pad():
while (base + len(out)) & 3:
out.append(0)
if node.name is None and node.kind == "raw":
out += node.raw # unnamed payload / stray bytes
return
if node.kind == "complex":
if node.name is not None:
b = node.name.encode("ascii")
out += struct.pack("<i", len(b)) + b
pad()
out += BEGIN
for c in node.children:
reconstruct(c, out, base)
out += END
return
b = (node.name or "").encode("ascii")
out += struct.pack("<i", len(b)) + b
if node.kind == "string":
out += struct.pack("<i", len(node.raw)) # reader's raw excludes the length prefix
out += node.raw
pad()
def audit_coverage(tree: sr.Node, inflated: bytes) -> dict:
"""Re-serialise the whole tree and compare with the inflated stream."""
out = bytearray()
for c in tree.children:
reconstruct(c, out, 0)
ok = bytes(out) == inflated
first = None
if not ok:
n = min(len(out), len(inflated))
for i in range(n):
if out[i] != inflated[i]:
first = i
break
if first is None:
first = n
return {"ok": ok, "rebuiltBytes": len(out), "inflatedBytes": len(inflated),
"firstDiff": first}
# --- top level ----------------------------------------------------------------
class SaveChecksum:
def __init__(self, path, root: CkNode, coverage: dict, res: sr.SaveResult,
policy: str, mask: str, mask_hits: dict | None = None):
self.path = path
self.root = root
self.coverage = coverage
self.res = res
self.policy = policy
self.mask = mask
#: leaves each mask rule actually replaced. Printed on every masked run
#: so an over-broad mask is visible rather than comfortable.
self.mask_hits = mask_hits or {}
@property
def digest(self) -> str:
return self.root.hexd
def subsystems(self) -> list[tuple[str, CkNode]]:
rows = []
for p in SUBSYSTEM_PATHS:
n = self._by_tag(p)
if n is not None:
rows.append((p, n))
return rows
def _by_tag(self, tagpath: str):
"""Resolve a '/Sim/players'-style path against labelled nodes."""
cur = self.root
for seg in tagpath.strip("/").split("/"):
nxt = None
for c in cur.children:
base = c.label.split("[")[0]
if base == seg or c.label == seg:
nxt = c
break
if nxt is None:
return None
cur = nxt
return cur
def as_dict(self, depth: int = 2) -> dict:
return {
"file": self.path,
"root": self.digest,
"policy": {"floats": self.policy, "mask": self.mask,
"digest": f"blake2b-{DIGEST_BYTES * 8}"},
"coverage": self.coverage,
"readerIssues": {lvl: self.res.count(lvl) for lvl in ("error", "warn", "info")},
"maskHits": self.mask_hits,
"subsystems": [{"path": p, "digest": n.hexd, "leaves": n.leaves,
"valueBytes": n.value_bytes} for p, n in self.subsystems()],
"tree": self.root.as_dict(depth),
}
def checksum_bytes(data: bytes, path: str = "<bytes>", floats: str = "bits",
mask: str = "none", audit: bool = True,
strict: bool = False) -> SaveChecksum:
if floats not in FLOAT_POLICIES:
raise ValueError(f"floats must be one of {FLOAT_POLICIES}")
if mask not in MASK_PRESETS:
raise ValueError(f"mask must be one of {tuple(MASK_PRESETS)}")
res = sr.read_bytes(data, strict=strict)
masks = MASK_PRESETS[mask]
hits: dict = {}
kids = []
counts = _name_counts(res.tree.children)
ordinals: dict = {}
for c in res.tree.children:
k = ordinals.get(c.name, 0)
ordinals[c.name] = k + 1
lab = _label(c, k, None, unique=counts.get(c.name) == 1)
kids.append(_build(c, "/" + lab, lab, floats, masks, hits))
root_dig = _h(b"save", floats.encode("ascii"), mask.encode("ascii"),
*[k.digest for k in kids])
root = CkNode("", "<save>", "complex", root_dig, 0, len(res.inflated),
sum(k.leaves for k in kids), sum(k.value_bytes for k in kids), kids)
cov = audit_coverage(res.tree, res.inflated) if audit else {"ok": None}
return SaveChecksum(path, root, cov, res, floats, mask, hits)
def checksum_save(path: str, **kw) -> SaveChecksum:
with open(path, "rb") as f:
return checksum_bytes(f.read(), path=path, **kw)
# --- diffing ------------------------------------------------------------------
class DiffEntry:
__slots__ = ("path", "kind", "a", "b", "ulps", "note")
def __init__(self, path, kind, a=None, b=None, ulps=None, note=""):
self.path, self.kind, self.a, self.b = path, kind, a, b
self.ulps, self.note = ulps, note
def __repr__(self):
if self.kind == "value":
u = f" [{self.ulps:g} ulp]" if self.ulps is not None else ""
return f"{self.path}: {self.a!r} -> {self.b!r}{u}{self.note}"
if self.kind == "list":
return (f"{self.path}: removed {self.a!r}, added {self.b!r}{self.note}")
return f"{self.path}: {self.kind}{self.note}"
def as_dict(self):
d = {"path": self.path, "kind": self.kind}
if self.kind in ("value", "list", "list-reordered"):
d["a"], d["b"] = self.a, self.b
if self.ulps is not None:
d["ulps"] = self.ulps
if self.note:
d["note"] = self.note.strip()
return d
def diff(a: CkNode, b: CkNode, out: list | None = None, limit: int = 200) -> list:
"""Descend two digest trees, reporting only where they differ."""
out = [] if out is None else out
if len(out) >= limit:
return out
if a.digest == b.digest:
return out
if a.children or b.children:
an = {c.label: c for c in a.children}
bn = {c.label: c for c in b.children}
# positional first: order is part of the state
if [c.label for c in a.children] != [c.label for c in b.children]:
for lab in an:
if lab not in bn:
out.append(DiffEntry(a.path + "/" + lab, "only-in-A"))
for lab in bn:
if lab not in an:
out.append(DiffEntry(b.path + "/" + lab, "only-in-B"))
if not (set(an) - set(bn)) and not (set(bn) - set(an)):
out.append(DiffEntry(a.path, "reordered"))
for ca in a.children:
cb = bn.get(ca.label)
if cb is not None:
diff(ca, cb, out, limit)
return out
if a.kind == "list" and b.kind == "list":
av, bv = a.values or [], b.values or []
added = [v for v in bv if v not in av]
removed = [v for v in av if v not in bv]
if added or removed:
out.append(DiffEntry(a.path, "list", removed, added,
note=f" ({len(av)} -> {len(bv)} entries)"))
else:
out.append(DiffEntry(a.path, "list-reordered", av, bv))
return out
ulps = None
if a.kind == "float" and b.kind == "float" and len(a.raw) == 4 == len(b.raw):
ulps = ulps_apart(a.raw, b.raw)
note = ""
if a.kind != b.kind:
note = f" (kind {a.kind} -> {b.kind})"
av, bv = a.value, b.value
if a.kind == "raw" or b.kind == "raw":
# the reader's raw value is a truncated hex preview; two different blobs
# can preview identically, so report a digest of the whole blob instead
av = f"<raw {len(a.raw)} B {_h(a.raw).hex()[:SHORT]}>"
bv = f"<raw {len(b.raw)} B {_h(b.raw).hex()[:SHORT]}>"
out.append(DiffEntry(a.path, "value", av, bv, ulps, note))
return out
# --- chain --------------------------------------------------------------------
def build_chain(files: list[str], **kw) -> dict:
"""Record a chain: one entry per save, in turn order."""
entries = []
for f in files:
ck = checksum_save(f, **kw)
t = ck.res.typed.get("summary") or {}
entries.append({
"name": os.path.basename(f),
"turn": t.get("Turn"),
"root": ck.digest,
"coverage": bool(ck.coverage.get("ok")),
"subsystems": {p: n.hexd for p, n in ck.subsystems()},
})
return {"format": "sots-state-chain/1",
"policy": {"floats": kw.get("floats", "bits"),
"mask": kw.get("mask", "none"),
"digest": f"blake2b-{DIGEST_BYTES * 8}"},
"turns": entries}
def verify_chain(chain: dict, files: list[str], **kw) -> tuple[bool, list[str]]:
msgs, ok = [], True
rec = chain["turns"]
if len(rec) != len(files):
msgs.append(f"chain has {len(rec)} turns, {len(files)} saves given")
ok = False
for i, f in enumerate(files):
if i >= len(rec):
break
ck = checksum_save(f, **kw)
want = rec[i]
if ck.digest == want["root"]:
msgs.append(f"turn {want.get('turn')} MATCH {ck.digest[:SHORT]} {os.path.basename(f)}")
continue
ok = False
msgs.append(f"turn {want.get('turn')} DIVERGE {ck.digest[:SHORT]} != {want['root'][:SHORT]}"
f" {os.path.basename(f)}")
for p, n in ck.subsystems():
w = want["subsystems"].get(p)
if w and w != n.hexd:
msgs.append(f" subsystem {p}: {w[:SHORT]} -> {n.hexd[:SHORT]}")
msgs.append(" (re-run with the recorded save to localise to an object)")
break # first divergent turn is the one that matters
return ok, msgs
# --- CLI ----------------------------------------------------------------------
def _mask_note(ck: "SaveChecksum") -> str:
"""Always say what a mask actually hit -- a mask nobody audits is a hiding place."""
if ck.mask == "none":
return ""
if not ck.mask_hits:
return " [mask matched NOTHING -- check the rule paths]"
return " [masked: " + ", ".join(f"{k}x{v}" for k, v in sorted(ck.mask_hits.items())) + "]"
def _print_tree(n: CkNode, depth: int, out, indent: int = 0, top: bool = True):
if not top:
pad = " " * indent
extra = f" {n.leaves} leaves, {n.value_bytes} B" if n.kind in ("complex", "group") else ""
val = "" if n.kind in ("complex", "group") else f" = {n.value!r}"
mk = " [masked]" if n.masked else ""
out.append(f"{pad}{n.short()} {n.label}{val}{extra}{mk}")
if indent < depth:
for c in n.children:
_print_tree(c, depth, out, indent + 1, top=False)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(
description="whole-state diagnostic checksum for SOTS1 saves")
ap.add_argument("saves", nargs="*", help="one save (report), two (diff), or N with --chain")
ap.add_argument("--floats", choices=FLOAT_POLICIES, default="bits",
help="float hashing policy (default: bits = exact IEEE-754)")
ap.add_argument("--mask", choices=tuple(MASK_PRESETS), default="none",
help="canonicalisation preset (default: none, so deltas localise)")
ap.add_argument("--tree", action="store_true", help="print the digest tree")
ap.add_argument("--depth", type=int, default=2, help="tree depth (default 2)")
ap.add_argument("--json", action="store_true")
ap.add_argument("--no-audit", action="store_true",
help="skip the byte-for-byte coverage proof (faster, weaker)")
ap.add_argument("--strict", action="store_true", help="reader --strict")
ap.add_argument("--ulps", type=float, default=None,
help="in a diff, flag float differences at or below N ULPs")
ap.add_argument("--limit", type=int, default=200, help="max diff entries")
ap.add_argument("--chain", metavar="FILE", help="chain JSON to verify against")
ap.add_argument("--record-chain", metavar="FILE", help="write a chain JSON instead")
args = ap.parse_args(argv)
kw = dict(floats=args.floats, mask=args.mask,
audit=not args.no_audit, strict=args.strict)
if not args.saves:
ap.error("no saves given")
if args.record_chain:
chain = build_chain(args.saves, **kw)
with open(args.record_chain, "w") as f:
json.dump(chain, f, indent=1)
print(f"recorded {len(chain['turns'])} turns -> {args.record_chain}")
for t in chain["turns"]:
cov = "cov-ok" if t["coverage"] else "COV-FAIL"
print(f" turn {t['turn']} {t['root'][:SHORT]} {cov} {t['name']}")
return 0 if all(t["coverage"] for t in chain["turns"]) else 1
if args.chain:
with open(args.chain) as f:
chain = json.load(f)
ok, msgs = verify_chain(chain, args.saves, **kw)
print("\n".join(msgs))
return 0 if ok else 1
if len(args.saves) == 1:
ck = checksum_save(args.saves[0], **kw)
if args.json:
print(json.dumps(ck.as_dict(args.depth if args.tree else 2), indent=1))
return 0 if ck.coverage.get("ok") is not False else 1
cov = ck.coverage
covs = ("coverage: PROVED (%d bytes rebuilt == inflated)" % cov["rebuiltBytes"]
if cov.get("ok") else
"coverage: SKIPPED" if cov.get("ok") is None else
"coverage: FAILED at inflated offset 0x%x" % (cov.get("firstDiff") or 0))
print(f"file: {ck.path}")
print(f"root: {ck.digest}")
print(f"policy: floats={ck.policy} mask={ck.mask} digest=blake2b-{DIGEST_BYTES * 8}"
f"{_mask_note(ck)}")
print(f"{covs}; {ck.root.leaves} leaves, {ck.root.value_bytes} value bytes")
print(f"reader: {ck.res.count('error')} error, {ck.res.count('warn')} warn")
if args.tree:
lines: list[str] = []
_print_tree(ck.root, args.depth, lines)
print("\n".join(lines))
else:
print("subsystems:")
for p, n in ck.subsystems():
print(f" {n.short()} {p:<22} {n.leaves:>6} leaves {n.value_bytes:>8} B")
return 0 if cov.get("ok") is not False else 1
if len(args.saves) != 2:
ap.error("give one save, two saves, or use --chain / --record-chain")
a = checksum_save(args.saves[0], **kw)
b = checksum_save(args.saves[1], **kw)
entries = diff(a.root, b.root, limit=args.limit)
if args.ulps is not None:
for e in entries:
if e.kind == "value" and e.ulps is not None and e.ulps <= args.ulps:
e.note += f" <= {args.ulps:g} ULP"
if args.json:
print(json.dumps({"a": a.as_dict(0), "b": b.as_dict(0),
"identical": a.digest == b.digest,
"diffs": [e.as_dict() for e in entries]}, indent=1))
return 0 if a.digest == b.digest else 1
print(f"A {a.digest} {a.path}")
print(f"B {b.digest} {b.path}")
print(f"policy: floats={a.policy} mask={a.mask}{_mask_note(a)}")
for ck, nm in ((a, "A"), (b, "B")):
if ck.coverage.get("ok") is False:
print(f"!! {nm}: coverage FAILED at 0x{ck.coverage.get('firstDiff') or 0:x} "
f"-- the digest does not bound this file")
if a.digest == b.digest:
print("IDENTICAL")
return 0
print(f"DIVERGED: {len(entries)} leaf difference(s)"
f"{' (limit reached)' if len(entries) >= args.limit else ''}")
for e in entries:
print(f" {e}")
if args.ulps is not None:
flt = [e for e in entries if e.kind == "value" and e.ulps is not None]
near = [e for e in flt if e.ulps <= args.ulps]
print(f"floats: {len(flt)} differ, {len(near)} within {args.ulps:g} ULP")
return 1
if __name__ == "__main__":
sys.exit(main())

403
verify/state-checksum/test_state_checksum.py Normal file
View file

@ -0,0 +1,403 @@
#!/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=1.5) -> bytes:
"""A miniature save with the shapes the tool special-cases."""
w = sw.SaveWriter("joint")
w.begin("Summary")
w.string("GameName", "T")
w.int("Turn", 2)
w.int("Checksum", checksum)
w.end()
w.begin("Sim")
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 in ((16, status_a), (32, status_b)):
w.int("PlayerID", pid)
w.begin("Player")
w.string("PlryName", f"p{pid}")
w.int("Status", st)
w.float("IdealSuit", fval)
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")
# --- 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=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=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=nudged))
self.assertNotEqual(a.digest, b.digest)
d = sc.diff(a.root, b.root)
self.assertEqual(len(d), 1)
self.assertEqual(d[0].ulps, 1.0)
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, {})
# --- 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")
json.dumps(chain) # must stay serialisable
# --- 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 = sc.checksum_save(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 = {sc.checksum_save(p).digest for p in paths}
self.assertEqual(len(digs), 1, paths)
def test_repeated_runs_are_stable(self):
p = REAL[0]
self.assertEqual(sc.checksum_save(p).digest, sc.checksum_save(p).digest)
def test_reader_is_clean_on_every_save(self):
for p in REAL:
with self.subTest(os.path.basename(p)):
ck = sc.checksum_save(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 = sc.checksum_save(p, audit=False, floats="bits")
b = sc.checksum_save(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 = sc.checksum_save(a, audit=False), sc.checksum_save(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)
for e in statuses:
self.assertIn("/Sim/players/Player[", e.path)
self.assertEqual((e.a, e.b), (4, 0))
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) # 4 x (4 -> 0)
def test_resave_mask_makes_that_pair_identical(self):
found = False
for a in REAL:
for b in REAL:
if a >= b:
continue
if sc.checksum_save(a, audit=False, mask="resave").digest == \
sc.checksum_save(b, audit=False, mask="resave").digest and \
sc.checksum_save(a, audit=False).digest != \
sc.checksum_save(b, audit=False).digest:
found = True
if not found:
self.skipTest("no re-save pair among the available saves")
self.assertTrue(found)
def test_a_real_turn_transition_localises_to_named_objects(self):
by_turn = {}
for p in REAL:
ck = sc.checksum_save(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)