#!/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 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()