sots-re/verify/save-reader/test_save_reader.py
lane-w 2308b6edb9 lane W: SvSctOb variant factories; fix the four save_reader.py defects openly
Part 1 (notes side): findings/objects/svsctob-variants.md records the two maps
that are nowhere on the wire and were read out of the game -- EncID -> class
from a 23-entry dword jump table at 0x0052bf60 indexed by EncID-1 (0x0052bf00),
and xscn -> class from an exhaustive four-way _stricmp chain at 0x005a7050.
Twelve live EncIDs, four scenario names, and "indsys" =
Game::SVSOIndependentSystems whose Read and Write are both the shared `ret 4`
stub at 0x005f8ac0, so its empty frame is correct output rather than a
truncation. Also: SVSOSots::Read accepts NPCPlr and hastraps, which Write never
emits -- read-only backward compatibility, not a hole in the recovery. And a
correction to the recovery itself: SVSOCrowDefenders writes `dsys` INSIDE the
ndsys loop; layouts.json calls it a plain member and no save can settle it
because both counts are 0 everywhere. 13 addresses in ghidra/addresses.d/
lane-w.json; gen_addresses.py merges to 640 with no duplicate name.

Part 2: lane G found four defects in both readers and deliberately did not
patch the oracle mid-campaign. Fixed now, with tests, and byte-neutral.
  1. Game::SystemParams field 1 is a string, not an int (empty string == four
     zero bytes == int 0, so it round-tripped by luck).
  2. ObservedTech/ObservedWeapon odet is a bool, not an int (byte-safe only
     because a 4-char tag makes both items 12 bytes; 3 chars would not).
  3. SpeciesRatios nv is a count, not a field.
  4. ShipRecords srbd is a count, not a field -- and this one is behaviourally
     confirmed, not inferred: srbd takes 0, 1, 3 and 4 across the players and
     every non-zero count is followed by exactly srbd x 5 scalars.
Note that 3 and 4 were an ABSENCE in save_reader.py, not an error: ShipRecs and
civr were both A(..., "any"), so the fix had to add the shapes rather than
retype a field.

Byte-neutrality: every item's inflated offset is unchanged on all four saves
(38,933 / 39,843 / 40,300 / 35,771 offsets, sequences identical), so no item
boundary moved. state_checksum.py still reports coverage: PROVED on all four
with the same rebuilt byte counts. The /CreateParams and /Sim/players digests do
change, because they hash typed VALUES and two fixes change what a value is --
and the value-byte deltas balance exactly: odet items x 3 plus p1 items x 4.
--strict exit 0 on all four saves; tests 36 -> 48.

findings/objects/wire-schema-closeout.md carries the whole account, including
the proof that CD/TurnCommands_v5 cannot be typed without a save that has
issued orders.
2026-09-08 07:05:09 -04:00

832 lines
39 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""Unit tests for save_reader.py using the synthetic fixture only.
/usr/bin/python3 -m unittest -v test_save_reader
NOTE: these prove the reader against its OWN framing assumptions (the synthetic
writer). The real-save check is `save_reader.py verify/results/saves/turn*-state.sav
--strict` (exit 0 on all three as of the schema-gaps-resolved.md patch).
"""
from __future__ import annotations
import gzip
import io
import json
import os
import struct
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import save_reader as sr
import save_writer_stub as sw
def norm(x):
"""Drop offsets / guess annotations so typed output compares to expectations."""
if isinstance(x, dict):
return {k: norm(v) for k, v in x.items() if k not in ("_off", "off", "guessed", "alt")}
if isinstance(x, list):
return [norm(v) for v in x]
return x
def walk(data, padding="joint", schema=None):
w = sr.Walker(data, padding, schema)
return w, w.walk()
class PrimitivesTest(unittest.TestCase):
def check_prims(self, padding):
w = sw.SaveWriter(padding)
# names of every length mod 4, every primitive width
w.int("Idx", 7)
w.int("Size", -3)
w.float("Suit", 0.75)
w.bool("NPC", True) # 3-char name + 1 byte: the padding-mode-sensitive case
w.bool("Abdn", False)
w.int64("Bats2", (5 << 33) + 9)
w.string("Name", "Sol")
w.string("PlryName", "")
w.string("Bdg", "café") # cp1252 byte 0xE9
w.int("TRA", 100)
data = w.bytes()
self.assertEqual(len(data) % 4, 0)
wk, root = walk(data, padding)
kinds = [(n.name, n.kind, n.value) for n in root.children]
self.assertEqual(kinds, [
("Idx", "int", 7), ("Size", "int", -3), ("Suit", "float", 0.75),
("NPC", "bool", True), ("Abdn", "bool", False), ("Bats2", "int64", (5 << 33) + 9),
("Name", "string", "Sol"), ("PlryName", "string", ""), ("Bdg", "string", "café"),
("TRA", "int", 100)])
self.assertEqual(wk.stats["resyncs"], 0)
self.assertEqual(wk.issues, [])
# offsets are contiguous and 4-aligned
pos = 0
for n in root.children:
self.assertEqual(n.offset, pos)
self.assertEqual(pos % 4, 0)
pos += n.size
self.assertEqual(pos, len(data))
def test_joint(self):
self.check_prims("joint")
def test_split(self):
self.check_prims("split")
def test_padding_layout_joint(self):
w = sw.SaveWriter("joint")
w.bool("NPC", True)
# [03 00 00 00]['N''P''C'][01] -> 8 bytes, no padding needed
self.assertEqual(w.bytes(), b"\x03\x00\x00\x00NPC\x01")
w = sw.SaveWriter("joint")
w.int("Idx", 1)
# 4 + 3 + 4 = 11 -> pad 1
self.assertEqual(w.bytes(), b"\x03\x00\x00\x00Idx\x01\x00\x00\x00\x00")
def test_padding_layout_split(self):
w = sw.SaveWriter("split")
w.bool("NPC", True)
self.assertEqual(w.bytes(), b"\x03\x00\x00\x00NPC\x00\x01\x00\x00\x00")
def test_unknown_names_are_guessed(self):
w = sw.SaveWriter("joint")
w.int("zzq", 12) # 3-char names: bool/int/int64 sizes all differ
w.bool("zzb", True)
w.int64("zzl", 1 << 40)
w.string("zzs", "hello")
w.float("zzf", 3.5)
w.int("zzz", 99)
_, root = walk(w.bytes())
got = [(n.name, n.kind, n.value, n.hinted) for n in root.children]
self.assertEqual(got, [("zzq", "int", 12, False), ("zzb", "bool", True, False),
("zzl", "int64", 1 << 40, False), ("zzs", "string", "hello", False),
("zzf", "float", 3.5, False), ("zzz", "int", 99, False)])
def test_word_ambiguity_is_reported_as_alt(self):
w = sw.SaveWriter("joint")
w.float("zzzz", 1.0) # 0x3F800000: unknown name -> float with int alt
_, root = walk(w.bytes())
n = root.children[0]
self.assertEqual((n.kind, n.value, n.alt), ("float", 1.0, 0x3F800000))
def test_catalog_hint_overrides_guess(self):
w = sw.SaveWriter("joint")
w.float("Suit", 0.0) # bits 0 would be guessed int; catalog says float
w.bool("Elim", True) # 4-char name + bool is size-ambiguous with int
_, root = walk(w.bytes())
self.assertEqual([(n.kind, n.value, n.hinted) for n in root.children],
[("float", 0.0, True), ("bool", True, True)])
class FramingTest(unittest.TestCase):
def test_nested_frames_and_markers(self):
w = sw.SaveWriter()
w.begin("Outer")
w.int("a", 1)
w.begin("Inner")
w.string("s", "x")
w.end()
w.begin(None) # tagless frame
w.int("t", 2)
w.end()
w.end()
data = w.bytes()
self.assertTrue(data.startswith(b"\x05\x00\x00\x00Outer\x00\x00\x00" + sr.BEGIN_BYTES))
self.assertTrue(data.endswith(sr.END_BYTES + sr.END_BYTES))
wk, root = walk(data)
outer = root.children[0]
self.assertEqual((outer.name, outer.kind, outer.size), ("Outer", "complex", len(data)))
names = [(c.name, c.kind) for c in outer.children]
self.assertEqual(names, [("a", "int"), ("Inner", "complex"), (None, "complex")])
self.assertEqual(outer.children[2].children[0].value, 2)
self.assertEqual(wk.issues, [])
def test_unnamed_vec3_body(self):
w = sw.SaveWriter()
w.vec3("Pos", 1.5, -2.0, 1e6)
w.vec3("Pos", 1.5, -2.0, 1e6, named=True)
wk, root = walk(w.bytes())
raw = root.children[0].children
self.assertEqual(len(raw), 1)
self.assertEqual(raw[0].kind, "raw")
self.assertEqual(struct.unpack("<3f", raw[0].raw), (1.5, -2.0, 1e6))
self.assertEqual([c.kind for c in root.children[1].children], ["float"] * 3)
self.assertTrue(all(i.level == "info" for i in wk.issues))
ap = sr.Applier()
self.assertEqual(ap.vec3(root.children[0], ""), [1.5, -2.0, 1e6])
self.assertEqual(ap.vec3(root.children[1], ""), [1.5, -2.0, 1e6])
def test_resync_through_garbage(self):
"""A frame whose body is unknown binary must not derail the parse."""
import random
rnd = random.Random(7)
blob = bytes(rnd.getrandbits(8) for _ in range(2500))
w = sw.SaveWriter()
w.int("before", 1)
w.begin("Mystery")
w.raw("State", blob) # named blob, unknown name -> no layout fits
w.end()
w.begin("Junk")
w.buf += blob[:301] # untagged garbage, unaligned length
w._pad()
w.end()
w.int("after", 2)
wk, root = walk(w.bytes())
self.assertEqual([c.name for c in root.children], ["before", "Mystery", "Junk", "after"])
self.assertEqual(root.children[3].value, 2)
self.assertEqual(root.children[1].children[0].kind, "raw")
self.assertEqual(root.children[2].children[0].kind, "raw")
self.assertEqual(wk.stats["resyncs"], 2)
self.assertTrue(any(i.level == "warn" for i in wk.issues))
def test_raw_frame_hint(self):
"""RNG is declared opaque: its body is read to the END marker without warnings."""
w = sw.SaveWriter()
w.begin("RNG")
w.raw("State", bytes(range(256)) * 9 + b"\x00" * 196)
w.end()
wk, root = walk(w.bytes())
n = root.children[0].children[0]
self.assertEqual((n.name, n.kind, n.hinted), ("State", "raw", True))
self.assertEqual(n.value["len"], 2500 + 3) # joint padding bytes are included
self.assertEqual(wk.stats["resyncs"], 0)
def test_truncated_stream(self):
w = sw.SaveWriter()
w.begin("Summary")
w.int("turn", 5)
w.string("gameName", "abcdef")
w.end()
data = w.bytes()[:-9]
wk, root = walk(data)
self.assertEqual(root.children[0].children[0].value, 5)
self.assertTrue(any("not terminated" in i.msg for i in wk.issues))
self.assertTrue(all(i.offset <= len(data) for i in wk.issues))
def test_damaged_gzip(self):
data = gzip.compress(b"\x03\x00\x00\x00Idx\x01\x00\x00\x00\x00")[:-6]
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(data)
def test_inflate_passthrough(self):
raw = b"\x03\x00\x00\x00Idx\x01\x00\x00\x00\x00"
self.assertEqual(sr.inflate(raw), raw)
self.assertEqual(sr.inflate(gzip.compress(raw)), raw)
class SchemaRoundTripTest(unittest.TestCase):
def roundtrip(self, padding, request="auto"):
data, expected = sw.build_fixture(padding)
res = sr.read_bytes(gzip.compress(data), padding=request, strict=True)
self.assertEqual(res.padding, padding)
self.assertEqual(res.count("error"), 0)
self.assertEqual(res.count("warn"), 0)
self.assertEqual(res.stats["resyncs"], 0)
self.assertEqual(norm(res.typed), expected)
return res
def test_roundtrip_joint(self):
res = self.roundtrip("joint")
sim = res.typed["sim"]
self.assertEqual(len(sim["players"]), 2)
self.assertEqual(len(sim["systems"]), 2)
sysd = sim["systems"][0]["Sys"]
self.assertIsInstance(sysd["Bats2"], int)
self.assertGreater(sysd["Bats2"], 1 << 33)
self.assertIsInstance(sysd["Abdn"], bool)
self.assertEqual(len(sysd["Pos"]), 3)
self.assertIn("indi", sysd) # hindi gate honoured
ply = sim["players"][0]["Player"]
self.assertIsInstance(ply["pswd"], str)
self.assertIsInstance(ply["TRM"], float)
self.assertEqual(len(ply["nexp"]), 2)
self.assertEqual(set(ply["nexp"][0]), {"xid", "xmin", "xmax", "xper"})
ship = sim["fleets"][0]["Flt"]["ships"][0]["Ship"]
self.assertIsInstance(ship["RefCap"], float)
self.assertIn("BQ2", ship)
self.assertIn("pop", ship)
self.assertEqual(len(res.typed["cdTable"]["ids"]), 2)
self.assertEqual(len(res.typed["customData"]), 2) # Repeat(CD) at the root
self.assertEqual(len(sim["species"]), 2) # Repeat(ISsp/ISsu) without a count
self.assertEqual(len(sim["turnstats"]["players"][0]["hist"]["stats"]), 2)
colors = [p["Slot"]["FxCrID"] for p in res.typed["summary"]["Players"]]
custom = [c for c in colors if c["idx"] == -1]
self.assertTrue(custom and all("r" in c for c in custom)) # -1 -> RGB follows
self.assertTrue(all("r" not in c for c in colors if c["idx"] != -1))
def test_roundtrip_split(self):
self.roundtrip("split")
def test_explicit_padding_mode(self):
self.roundtrip("joint", "joint")
self.roundtrip("split", "split")
def test_wrong_padding_mode_is_noisy(self):
data, _ = sw.build_fixture("split")
res = sr.read_bytes(data, padding="joint")
self.assertGreater(res.stats["resyncs"] + res.stats["hint_failures"] + res.count("error"), 0)
def test_json_serializable(self):
data, _ = sw.build_fixture("joint")
res = sr.read_bytes(data)
json.dumps(res.typed)
json.dumps(sr.plain(res.tree))
json.dumps([i.as_dict() for i in res.issues])
def test_strict_rejects_missing_authoritative_field(self):
w = sw.SaveWriter()
w.begin("Sys")
w.vec3("Pos", 0, 0, 0)
w.float("R", 1); w.float("G", 1); w.float("B", 1); w.float("A", 1)
w.int("Idx", 3)
w.int("Suit", 0) # Size missing; Suit written as int
w.end()
data = w.bytes()
schema = sr.Seq([sr.R("sys", sr.Sys)])
res = sr.read_bytes(data, padding="joint", schema=schema)
self.assertIsNone(res.typed["sys"]["Size"])
self.assertTrue(any(i.level == "error" and "Size" in i.msg for i in res.issues))
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(data, padding="joint", schema=schema, strict=True)
def test_lenient_skips_unexpected_items(self):
w = sw.SaveWriter()
w.begin("PopG")
w.int("PopT", 1)
w.int("Legacy", 5) # not in the schema
w.int("PopS", 2)
w.int64("PopC", 3)
w.end()
schema = sr.Seq([sr.R("g", sr.PopG)])
res = sr.read_bytes(w.bytes(), padding="joint", schema=schema)
g = res.typed["g"]
self.assertEqual((g["PopT"], g["PopS"], g["PopC"]), (1, 2, 3))
self.assertEqual(g["_unexpected"][0]["name"], "Legacy")
self.assertTrue(any(i.level == "warn" for i in res.issues))
def test_coerce_retypes_words(self):
"""A 4-byte word the walker guessed one way is re-read from its raw bytes
when the schema (matched positionally, so the walker had no hint) says otherwise."""
w = sw.SaveWriter()
w.begin("Wd")
w.float("zzza", 60.0) # 0x42700000: guessed float; schema says int
w.int("zzzb", 0x7F7FFFFF) # guessed int; schema says float (FLT_MAX)
w.end()
shape = sr.Shape("Wd", [sr.R("a", "int"), sr.R("b", "float"), sr.Rest()])
res = sr.read_bytes(w.bytes(), padding="joint", schema=sr.Seq([sr.R("t", shape)]))
self.assertEqual(res.typed["t"]["a"], 0x42700000)
self.assertEqual(res.typed["t"]["b"], 3.4028234663852886e+38)
self.assertEqual(res.count("error"), 0)
def test_non_ascii_string_value_never_vetoes_layout(self):
"""Gap 1+2: a 1-byte bool followed by a bool and a cp1252 string with a
byte in 0x80-0x9f (0x92 = right single quote) must parse exactly like an
ASCII name -- the text test applies to tags, never to string values."""
for name in ("Kor’Voth", "Kaa’Vaalu", "Plain"):
w = sw.SaveWriter()
w.begin("Tst") # unknown frame: hints come from the catalog by name
w.int("haltc", 1)
w.int("haltt", 0)
w.bool("haltv", False) # 5-char tag + 1 byte -> 2 pad bytes (joint)
w.bool("vnh", False)
w.string("Name", name) # encodes 0x92 for the quote
w.int("VFlags", 0)
w.end()
data = w.bytes()
if "’" in name:
self.assertIn(b"\x92", data)
wk, root = walk(data)
got = [(c.name, c.kind, c.value) for c in root.children[0].children]
self.assertEqual(got, [("haltc", "int", 1), ("haltt", "int", 0), ("haltv", "bool", False),
("vnh", "bool", False), ("Name", "string", name), ("VFlags", "int", 0)])
self.assertEqual(wk.stats["resyncs"], 0)
self.assertEqual(wk.stats["hint_failures"], 0)
# same guarantee for an unhinted (guessed) tag holding non-ASCII text
w = sw.SaveWriter()
w.bool("zzq", True)
w.string("zzs", "Kor’Voth")
w.int("zzz", 5)
wk, root = walk(w.bytes())
self.assertEqual([(c.kind, c.value) for c in root.children],
[("bool", True), ("string", "Kor’Voth"), ("int", 5)])
self.assertEqual(wk.stats["resyncs"], 0)
def test_empty_string_values(self):
"""Gap 5: an empty string is 4 zero bytes, byte-identical to int 0. It
must be read as "" both when the tag is hinted (walker) and when the
walker had no hint and guessed int (applier coerce)."""
w = sw.SaveWriter()
w.begin("CreateParams")
w.string("Name", "g")
w.int("ID", 1); w.int("RSeed", 2); w.int("AID", 3)
w.string("Key", "") # 03 00 00 00 'Key' 00 00 00 00 + 1 pad
w.end()
data = w.bytes()
self.assertIn(b"\x03\x00\x00\x00Key\x00\x00\x00\x00\x00", data)
wk, root = walk(data) # hinted: CreateParams is in the catalog
key = root.children[0].children[4]
self.assertEqual((key.name, key.kind, key.value, key.hinted), ("Key", "string", "", True))
self.assertEqual(wk.issues, [])
# unhinted: an unknown frame name -> walker guesses int 0; positional schema wants a string
w = sw.SaveWriter()
w.begin("Unknown")
w.string("zzk", "")
w.string("zzm", "")
w.end()
shape = sr.Shape("Unknown", [sr.R("k", "string"), sr.R("m", "string"), sr.Rest()])
res = sr.read_bytes(w.bytes(), padding="joint", schema=sr.Seq([sr.R("u", shape)]), strict=True)
self.assertEqual((res.typed["u"]["k"], res.typed["u"]["m"]), ("", ""))
self.assertEqual(res.count("error"), 0)
def test_prisoner_hold_gating(self):
"""Gap 4: PrNSp and the (PrSp, PrNum) pairs are written only when PrMax > 0."""
def prish(pr_max, pairs):
w = sw.SaveWriter()
w.begin("PrisH")
w.int("PrMax", pr_max)
if pr_max > 0:
w.int("PrNSp", len(pairs))
for sp, num in pairs:
w.int("PrSp", sp)
w.int("PrNum", num)
w.end()
return sr.read_bytes(w.bytes(), padding="joint", strict=True,
schema=sr.Seq([sr.R("p", sr.PrisonerHold)])).typed["p"]
empty = prish(0, [])
self.assertEqual(empty["PrMax"], 0)
self.assertNotIn("prisoners", empty)
full = prish(40, [(1, 12), (3, 5)])
self.assertEqual(full["PrMax"], 40)
self.assertEqual(full["prisoners"], [{"PrSp": 1, "PrNum": 12}, {"PrSp": 3, "PrNum": 5}])
# a PrMax>0 frame with the pairs missing is a real schema error
w = sw.SaveWriter()
w.begin("PrisH"); w.int("PrMax", 40); w.end()
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(w.bytes(), padding="joint", strict=True, schema=sr.Seq([sr.R("p", sr.PrisonerHold)]))
class TagNameCorrectionsTest(unittest.TestCase):
"""SAVE_FORMAT.md §10: tags the C++ round-trip writer had to emit to reproduce
the real saves byte-for-byte, which the spec/reader had wrong or unnamed.
Positional R() matching used to hide the misspellings; the fields are now
A() and a wrong spelling is a strict failure."""
REAL = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "results", "saves")
@staticmethod
def field_names(shape):
return [(f.name, f.auth) for f in shape.fields if isinstance(f, sr.Field)]
def test_schema_declares_the_real_tags(self):
for shape in (sr.Odes, sr.Owep, sr.Otch):
names = self.field_names(shape)
self.assertEqual(names[0], ("otnF", True)) # not R1's "ontF"
self.assertTrue(all(auth for _, auth in names), names)
self.assertEqual(self.field_names(sr.Design),
[("FAIDes", True), ("DHide", True), ("DWep", True), ("DName", True)])
# typed-dict keys keep R1's spelling (verify/design-rules/stock_designs.py reads them)
self.assertEqual([f.key for f in sr.Design.fields if isinstance(f, sr.Field)],
["faiDes", "dHide", "dWep", "dName"])
self.assertEqual(self.field_names(sr.NodeGrid), [("paths", True), ("nextid", True)])
self.assertEqual(self.field_names(sr.BuildQueue), [("ords", True)])
self.assertEqual(self.field_names(sr.FlightPlan)[0], ("wpts", True))
for f in (sr.BuildQueue.fields[0], sr.FlightPlan.fields[0], sr.NodeGrid.fields[0]):
self.assertFalse(f.flex, f) # framed VectorHelper, no inline hedge
def test_catalog_knows_only_the_real_spellings(self):
for wrong in ("ontF", "nextId", "faiDes", "dHide", "dWep", "dName"):
self.assertNotIn(wrong, sr.GLOBAL_KINDS)
self.assertEqual(sr.GLOBAL_KINDS["otnF"], "int")
self.assertEqual(sr.GLOBAL_KINDS["nextid"], "int")
self.assertEqual((sr.GLOBAL_KINDS["FAIDes"], sr.GLOBAL_KINDS["DHide"],
sr.GLOBAL_KINDS["DWep"], sr.GLOBAL_KINDS["DName"]),
("bool", "bool", "int", "string"))
for name in ("ords", "wpts", "paths"):
self.assertIsInstance(sr.GLOBAL_SHAPES[name], sr.CArr)
@staticmethod
def objectives(lead):
"""Player-level odes/owep/otch arrays as the real saves lay them out."""
w = sw.SaveWriter()
w.begin("odes"); w.int(".", 1)
w.begin("."); w.int(lead, 2); w.int("otnL", 2); w.int("odid", 18); w.int("opid", 32); w.end()
w.end()
w.begin("owep"); w.int(".", 0); w.end()
w.begin("otch"); w.int(".", 1)
w.begin("."); w.int(lead, 1); w.int("otnL", 1); w.int("odet", 0)
w.string("otch", "WEP_RedLas"); w.int("owith", 1); w.end()
w.end()
return w.bytes()
def test_objective_records_use_otnF(self):
schema = sr.Seq([sr.A("odes", sr.CArr(sr.Odes)), sr.A("owep", sr.CArr(sr.Owep)),
sr.A("otch", sr.CArr(sr.Otch))])
res = sr.read_bytes(self.objectives("otnF"), padding="joint", strict=True, schema=schema)
self.assertEqual(norm(res.typed["odes"]), [{"otnF": 2, "otnL": 2, "odid": 18, "opid": 32}])
self.assertEqual(res.typed["owep"], [])
self.assertEqual(norm(res.typed["otch"]),
[{"otnF": 1, "otnL": 1, "odet": 0, "otch": "WEP_RedLas", "owith": 1}])
self.assertFalse([i for i in res.issues if "read positionally" in i.msg])
# R1's misspelling is no longer accepted silently
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(self.objectives("ontF"), padding="joint", strict=True, schema=schema)
@staticmethod
def design(tags):
w = sw.SaveWriter()
w.begin("Des")
w.bool(tags[0], True); w.bool(tags[1], False); w.int(tags[2], 1); w.string(tags[3], "Honor Lance")
w.begin("DSec"); w.int("ga", 1); w.end() # sections stay generic
w.end()
return w.bytes()
def test_design_header_tags(self):
schema = sr.Seq([sr.A("Des", sr.Design)])
data = self.design(("FAIDes", "DHide", "DWep", "DName"))
wk, root = walk(data)
self.assertTrue(all(c.hinted for c in root.children[0].children[:4])) # typed from the Des shape
res = sr.read_bytes(data, padding="joint", strict=True, schema=schema)
d = res.typed["Des"]
self.assertEqual((d["faiDes"], d["dHide"], d["dWep"], d["dName"]), (True, False, 1, "Honor Lance"))
self.assertEqual([s["_name"] for s in d["sections"]], ["DSec"])
self.assertNotIn("FAIDes", d) # key stays R1's; only the tag changed
with self.assertRaises(sr.SaveFormatError): # camel-case R1 spelling is not on disk
sr.read_bytes(self.design(("faiDes", "dHide", "dWep", "dName")),
padding="joint", strict=True, schema=schema)
@staticmethod
def node_grid(tag):
w = sw.SaveWriter()
w.begin("NdGr2")
w.begin("paths"); w.int(".", 1)
w.begin(".")
for i, n in enumerate(("npt", "npid", "npfr", "npto", "npctm", "npcby",
"npdtn", "npdtf", "npenp", "npuse", "nptf")):
w.int(n, i)
w.end()
w.end()
w.int(tag, 44)
w.end()
return w.bytes()
def test_node_grid_nextid(self):
schema = sr.Seq([sr.A("NdGr2", sr.NodeGrid)])
res = sr.read_bytes(self.node_grid("nextid"), padding="joint", strict=True, schema=schema)
g = res.typed["NdGr2"]
self.assertEqual(g["nextid"], 44)
self.assertNotIn("nextId", g)
self.assertEqual((g["paths"][0]["npt"], g["paths"][0]["nptf"]), (0, 10))
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(self.node_grid("nextId"), padding="joint", strict=True, schema=schema)
def test_build_queue_ords_and_flight_plan_wpts(self):
w = sw.SaveWriter()
w.begin("BQ"); w.begin("ords"); w.int(".", 1)
w.begin("."); w.int("desID", 5); w.int("con", 1); w.int("conleft", 2); w.int("sav", 3); w.int("ordID", 7); w.end()
w.end(); w.end()
w.begin("FPlan"); w.begin("wpts"); w.int(".", 1)
w.begin("."); w.int("Wpt", 272); w.int("Tp", 1)
w.begin("nrt"); w.int("nrp", -1); w.int("nrf", 0); w.int("nrt", 0); w.end()
w.end(); w.end()
w.float("FPsp2", 2.5); w.int("FPeta2", 3)
w.vec3("FPogn2", 1.0, 2.0, 3.0, named=True); w.vec3("FPdpos", 4.0, 5.0, 6.0, named=True)
w.int("pnd", 0)
w.end()
schema = sr.Seq([sr.A("BQ", sr.BuildQueue), sr.A("FPlan", sr.FlightPlan)])
res = sr.read_bytes(w.bytes(), padding="joint", strict=True, schema=schema)
self.assertEqual(norm(res.typed["BQ"]["ords"]),
[{"desID": 5, "con": 1, "conleft": 2, "sav": 3, "ordID": 7}])
fp = res.typed["FPlan"]
self.assertEqual((fp["wpts"][0]["Wpt"], fp["wpts"][0]["Tp"], fp["wpts"][0]["nrt"]["nrp"]), (272, 1, -1))
self.assertEqual((fp["FPsp2"], fp["FPeta2"], fp["FPogn2"], fp["pnd"]), (2.5, 3, [1.0, 2.0, 3.0], 0))
# an empty queue is a framed count of 0 (what turn1..3 carry)
w = sw.SaveWriter()
w.begin("BQ"); w.begin("ords"); w.int(".", 0); w.end(); w.end()
res = sr.read_bytes(w.bytes(), padding="joint", strict=True, schema=sr.Seq([sr.A("BQ", sr.BuildQueue)]))
self.assertEqual(res.typed["BQ"]["ords"], [])
# the inline (unframed) hedge the old flex R() allowed is not what the game writes
w = sw.SaveWriter()
w.begin("BQ"); w.int("ords", 0); w.end()
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(w.bytes(), padding="joint", strict=True, schema=sr.Seq([sr.A("BQ", sr.BuildQueue)]))
@unittest.skipUnless(os.path.exists(os.path.join(REAL, "turn3-state.sav")), "real saves not present")
def test_real_saves_carry_the_corrected_tags(self):
res = sr.read_save(os.path.join(self.REAL, "turn2-state.sav"), padding="joint", strict=True)
players = [p["Player"] for p in res.typed["sim"]["players"]]
first = players[0]["designs"][0]["Des"]
self.assertEqual((first["dName"], first["faiDes"], first["dHide"], first["dWep"]), ("Armor", False, False, 0))
self.assertEqual((players[0]["otch"][0]["otnF"], players[0]["otch"][0]["otch"]), (1, "WEP_RedLas"))
self.assertTrue(any(o["odid"] == 18 and o["opid"] == 32 for p in players for o in p["odes"]))
self.assertEqual(res.typed["sim"]["NdGr2"]["nextid"], 44)
self.assertEqual(len(res.typed["sim"]["NdGr2"]["paths"]), 43)
# every remaining positional match is a NULL-named "." item
self.assertFalse([i for i in res.issues if "read positionally" in i.msg and "tag '.'" not in i.msg])
# turn3 is the only save with an active flight plan
res3 = sr.read_save(os.path.join(self.REAL, "turn3-state.sav"), padding="joint", strict=True)
plans = [f["Flt"]["FPlan"] for f in res3.typed["sim"]["fleets"] if f["Flt"]["HFPlan"]]
self.assertEqual(len(plans), 1)
self.assertEqual((plans[0]["wpts"][0]["Wpt"], plans[0]["wpts"][0]["nrt"]["nrp"]), (272, -1))
class WireSchemaDefectsTest(unittest.TestCase):
"""The four defects lane G's SchemaProbe found in BOTH readers, and which no
round-trip test could catch (findings/objects/wire-schema-channel.md §3).
Each `test_*_would_have_caught_it` builds the case the real saves happen not
to contain. Each `test_*_is_byte_neutral_here` shows why the defect stayed
invisible: on the data we actually hold the wrong type occupies the same
bytes, so the round trip was clean and wrong at the same time.
"""
REAL = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "results", "saves")
# --- 1. Game::SystemParams field 1 is a string, not an int ------------------
@staticmethod
def system_params(name):
"""One MapP frame with a single SystemParams element carrying `name`."""
w = sw.SaveWriter()
w.begin("MapP")
w.int(".", 0) # mapType
w.begin(".") # VectorHelper<SystemParams>
w.int(".", 1)
w.begin(".") # the element, all tags "."
w.vec3(".", 1.0, 2.0, 3.0)
w.string(".", name) # <- the field in question
w.int(".", 7)
w.int(".", 8)
w.float(".", 0.5)
w.end()
w.end()
w.int(".", 0) # players: bare count, no groups
w.begin("."); w.int(".", 0); w.end() # nodePaths: framed count
w.end()
return w.bytes()
def test_system_params_p1_is_a_string(self):
schema = sr.Seq([sr.A("MapP", sr.MapP)])
res = sr.read_bytes(self.system_params(""), padding="joint", strict=True, schema=schema)
planet = res.typed["MapP"]["planets"][0]
self.assertEqual(planet["p1"], "")
self.assertIsInstance(planet["p1"], str)
def test_a_named_system_reads_back_intact(self):
"""The case no save we hold contains. A non-empty name is a string of a
length the int reading cannot express, and every following field still has
to land -- this is the read the campaign had no answer for."""
data = self.system_params("Beta Hydri")
res = sr.read_bytes(data, padding="joint", strict=True, schema=sr.Seq([sr.A("MapP", sr.MapP)]))
planet = res.typed["MapP"]["planets"][0]
self.assertEqual((planet["p1"], planet["p2"], planet["p3"], planet["p4"]),
("Beta Hydri", 7, 8, 0.5))
def test_the_old_int_type_disagrees_with_these_bytes(self):
"""The catching assertion: with `p1` typed int, applying the schema to a
SystemParams element is a hard error, named or not. (The reader survived
the real saves only because its own kind catalog also said int there, so
schema and walker agreed with each other and both were wrong.)"""
old = sr.Shape(None, [sr.R("pos", "vec3"), sr.R("p1", "int"), sr.R("p2", "int"),
sr.R("p3", "int"), sr.R("p4", "float"), sr.Rest()])
old_map = sr.Shape("MapP", [sr.R("mapType", "int"), sr.R("planets", sr.CArr(old)),
sr.R("players", sr.NArr(sr.CArr("int"))),
sr.R("nodePaths", sr.CArr("any")), sr.Rest()])
for name in ("", "Beta Hydri"):
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(self.system_params(name), padding="joint", strict=True,
schema=sr.Seq([sr.A("MapP", old_map)]))
def test_empty_name_is_why_it_stayed_hidden(self):
"""An empty string is four zero bytes -- byte-identical to the int 0 -- so
on every save we hold, the wrong type cost nothing."""
s = sw.SaveWriter(); s.string(".", "")
i = sw.SaveWriter(); i.int(".", 0)
self.assertEqual(s.bytes(), i.bytes())
# only the empty one is: a string payload is [int32 len][bytes], so a name
# is 4 + len bytes where the int is 4, and the item stops agreeing as soon
# as the name pushes it past the next multiple of four.
named = sw.SaveWriter(); named.string(".", "Beta Hydri")
self.assertNotEqual(len(named.bytes()), len(i.bytes()))
# --- 2. ObservedTech / ObservedWeapon `odet` is a bool, not an int ----------
def test_odet_is_declared_bool(self):
for shape in (sr.Owep, sr.Otch):
kinds = {f.name: f.type for f in shape.fields if isinstance(f, sr.Field)}
self.assertEqual(kinds["odet"], "bool")
self.assertEqual(sr.GLOBAL_KINDS["odet"], "bool")
def test_odet_reads_back_as_a_bool_not_an_int(self):
w = sw.SaveWriter()
w.begin("otch"); w.int(".", 1)
w.begin("."); w.int("otnF", 1); w.int("otnL", 1); w.bool("odet", True)
w.string("otch", "WEP_RedLas"); w.int("owith", 1); w.end()
w.end()
res = sr.read_bytes(w.bytes(), padding="joint", strict=True,
schema=sr.Seq([sr.A("otch", sr.CArr(sr.Otch))]))
v = res.typed["otch"][0]["odet"]
self.assertIsInstance(v, bool) # `assertEqual(v, 1)` passes either way
self.assertIs(v, True)
def test_a_four_char_tag_is_the_only_reason_odet_was_byte_safe(self):
"""bool and int items are the same size ONLY when the tag length makes the
joint padding agree. 4 chars: 4+4+1 -> 12 and 4+4+4 = 12. 3 chars:
4+3+1 -> 8 but 4+3+4 -> 12, and the parse desyncs."""
four = sw.SaveWriter(); four.bool("odet", True); four.int("owith", 1)
as_int = sw.SaveWriter(); as_int.int("odet", 1); as_int.int("owith", 1)
self.assertEqual(len(four.bytes()), len(as_int.bytes())) # the coincidence
three = sw.SaveWriter(); three.bool("det", True); three.int("owith", 1)
three_int = sw.SaveWriter(); three_int.int("det", 1); three_int.int("owith", 1)
self.assertNotEqual(len(three.bytes()), len(three_int.bytes())) # not a property
# --- 3. Game::SpeciesRatios `nv` is a count -------------------------------
@staticmethod
def civr(pairs):
w = sw.SaveWriter()
w.begin("civr"); w.float("smx", 0.5)
w.begin("spe"); w.int("nv", len(pairs))
for sp, va2 in pairs:
w.int("sp", sp); w.int("va2", va2)
w.end(); w.end()
return w.bytes()
def test_nv_is_a_count_not_a_field(self):
schema = sr.Seq([sr.A("civr", sr.CivilianRatios)])
res = sr.read_bytes(self.civr([(0, 100)]), padding="joint", strict=True, schema=schema)
self.assertEqual(norm(res.typed["civr"]["spe"]["ratios"]), [{"sp": 0, "va2": 100}])
res0 = sr.read_bytes(self.civr([]), padding="joint", strict=True, schema=schema)
self.assertEqual(res0.typed["civr"]["spe"]["ratios"], [])
def test_two_species_would_have_broken_the_field_reading(self):
"""nv is 0 or 1 in every save we hold, so a field reading survives. With
two pairs the second is unexplained and strict parsing fails."""
data = self.civr([(0, 60), (5, 40)])
res = sr.read_bytes(data, padding="joint", strict=True,
schema=sr.Seq([sr.A("civr", sr.CivilianRatios)]))
self.assertEqual(norm(res.typed["civr"]["spe"]["ratios"]),
[{"sp": 0, "va2": 60}, {"sp": 5, "va2": 40}])
wrong = sr.Shape("spe", [sr.A("nv", "int"), sr.A("sp", "int"), sr.A("va2", "int")])
wrong_civr = sr.Shape("civr", [sr.A("smx", "float"), sr.A("spe", wrong)])
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(data, padding="joint", strict=True,
schema=sr.Seq([sr.A("civr", wrong_civr)]))
# --- 4. Game::ShipRecords `srbd` is a count -------------------------------
@staticmethod
def ship_recs(recs, designs):
w = sw.SaveWriter()
w.begin("ShipRecs")
w.int("srnc", len(recs))
for srb, srl, srk, sri in recs:
w.int("srb", srb); w.int("srl", srl); w.int("srk", srk); w.int("sri", sri)
w.int("srbd", len(designs))
for srd, src, srb, srl, sri in designs:
w.int("srd", srd); w.int("src", src); w.int("srb", srb)
w.int("srl", srl); w.int("sri", sri)
w.end()
return w.bytes()
def test_srbd_is_a_count_not_a_field(self):
data = self.ship_recs([(0, 0, 0, 0)], [(18, 0, 2, 0, 2)])
res = sr.read_bytes(data, padding="joint", strict=True,
schema=sr.Seq([sr.A("ShipRecs", sr.ShipRecords)]))
sr_ = res.typed["ShipRecs"]
self.assertEqual(norm(sr_["records"]), [{"srb": 0, "srl": 0, "srk": 0, "sri": 0}])
self.assertEqual(norm(sr_["designRecords"]),
[{"srd": 18, "src": 0, "srb": 2, "srl": 0, "sri": 2}])
def test_srbd_as_a_field_cannot_explain_the_trailing_records(self):
data = self.ship_recs([(0, 0, 0, 0)], [(18, 0, 2, 0, 2)])
wrong = sr.Shape("ShipRecs", [
sr.A("srnc", sr.NArr(sr.Seq([sr.A("srb", "int"), sr.A("srl", "int"),
sr.A("srk", "int"), sr.A("sri", "int")]))),
sr.A("srbd", "int"),
])
with self.assertRaises(sr.SaveFormatError):
sr.read_bytes(data, padding="joint", strict=True,
schema=sr.Seq([sr.A("ShipRecs", wrong)]))
# --- the four corrections against the real saves ---------------------------
@unittest.skipUnless(os.path.exists(os.path.join(REAL, "turn3-state.sav")), "real saves not present")
def test_real_saves_agree_with_the_corrected_types(self):
res = sr.read_save(os.path.join(self.REAL, "turn3-state.sav"), padding="joint", strict=True)
# 1. every SystemParams name is the empty STRING (never the int 0)
planets = res.typed["createParams"]["MapP"]["planets"]
self.assertTrue(planets)
for p in planets:
self.assertIsInstance(p["p1"], str)
self.assertEqual(p["p1"], "")
players = [p["Player"] for p in res.typed["sim"]["players"]]
# 2. every odet is a bool
odets = [o["odet"] for pl in players for key in ("otch", "owep") for o in pl.get(key, [])]
self.assertTrue(odets)
for v in odets:
self.assertIsInstance(v, bool)
# 3. nv counts (sp, va2) pairs: 1 -> one pair, 0 -> none
seen = set()
for pl in players:
ratios = pl["civr"]["spe"]["ratios"]
seen.add(len(ratios))
for r in ratios:
self.assertEqual(set(r) - {"_off"}, {"sp", "va2"})
self.assertEqual(seen, {0, 1})
# 4. srbd counts 5-scalar design records; turn3 exercises 0, 1 and 4
counts = sorted({len(pl["shipRecs"]["designRecords"]) for pl in players})
self.assertEqual(counts, [0, 1, 4])
for pl in players:
for d in pl["shipRecs"]["designRecords"]:
self.assertEqual(set(d) - {"_off"}, {"srd", "src", "srb", "srl", "sri"})
class CliTest(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
data, _ = sw.build_fixture("joint")
self.path = os.path.join(self.tmp.name, "fixture.sav")
with open(self.path, "wb") as f:
f.write(gzip.compress(data))
def tearDown(self):
self.tmp.cleanup()
def run_cli(self, *args):
buf = io.StringIO()
with redirect_stdout(buf):
rc = sr.main([self.path, *args])
return rc, buf.getvalue()
def test_summary(self):
rc, out = self.run_cli()
self.assertEqual(rc, 0)
self.assertIn("padding: joint", out)
self.assertIn("summary: game=", out)
def test_dump(self):
rc, out = self.run_cli("--dump")
self.assertEqual(rc, 0)
self.assertIn("@00000000 Summary {", out)
self.assertIn("Bats2 int64", out)
def test_json_and_inflate(self):
outp = os.path.join(self.tmp.name, "out.json")
infl = os.path.join(self.tmp.name, "inflated.bin")
rc, _ = self.run_cli("--json", "--strict", "--out", outp, "--inflate", infl)
self.assertEqual(rc, 0)
with open(outp) as f:
doc = json.load(f)
self.assertIn("summary", doc["data"])
self.assertEqual(doc["padding"], "joint")
with open(infl, "rb") as f:
self.assertTrue(f.read().startswith(b"\x07\x00\x00\x00Summary"))
def test_dump_json(self):
rc, out = self.run_cli("--dump", "--json")
self.assertEqual(rc, 0)
doc = json.loads(out)
self.assertEqual(doc["tree"][0]["_name"], "Summary")
if __name__ == "__main__":
unittest.main()