#!/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) # Lane D2: the derived writer's Dtc/Dwgv are named fields, not part of the # section run; the run itself is a Repeat and holds exactly the three DSec frames. self.assertEqual(self.field_names(sr.Design), [("FAIDes", True), ("DHide", True), ("DWep", True), ("DName", True), ("Dtc", True), ("Dwgv", 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", "dtc", "dwgv"]) rep = [f for f in sr.Design.fields if isinstance(f, sr.Repeat)] self.assertEqual([(r.lead, r.key) for r in rep], [("DSec", "sections")]) 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", "bool", "string")) # DWep is WriteBool, not WriteInt self.assertEqual(sr.GLOBAL_KINDS["Dwgv"], "bool") # likewise 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.bool(tags[2], True); w.string(tags[3], "Honor Lance") w.begin("DSec"); w.int("ga", 1); w.end() # sections stay generic w.int("Dtc", 1); w.bool("Dwgv", False) # the derived writer's tail 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, True, "Honor Lance")) # the section run stops at Dtc: the tail is NOT two more "reserved slots" self.assertEqual([s["_name"] for s in d["sections"]], ["DSec"]) self.assertEqual((d["dtc"], d["dwgv"]), (1, False)) 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 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"}) # --- lane D2: the Des record has THREE sections, and DWep/Dwgv are bools ---- @unittest.skipUnless(os.path.exists(os.path.join(REAL, "turn3-state.sav")), "real saves not present") def test_design_records_hold_exactly_three_sections(self): """The campaign's "five slots, 3 and 4 reserved" was this reader's own `Rest()` sweeping Dtc and Dwgv into the section list. Both design writers are recovered: the base emits three DSec frames and nothing else, the derived one appends Dtc and the Dwgv flag.""" seen = 0 for name in ("turn1-state.sav", "turn2-state.sav", "turn3-state.sav"): path = os.path.join(self.REAL, name) if not os.path.exists(path): continue res = sr.read_save(path, padding="joint", strict=True) for pe in res.typed["sim"]["players"]: pl = pe["Player"] for key in ("designs", "legacyDesigns"): for d in pl.get(key, []): des = d["Des"] self.assertEqual([x["_name"] for x in des["sections"]], ["DSec", "DSec", "DSec"], des["dName"]) self.assertIsInstance(des["dWep"], bool) self.assertIsInstance(des["dwgv"], bool) self.assertFalse(des["dwgv"]) # never set in any save available self.assertNotIn("weaponGroups", des) seen += 1 self.assertGreater(seen, 100) 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()