a4aba6a9fb
Blocker #4 on lane Y's path to a byte-matching turn. HOW DESIGNS PERSIST, correcting a published finding. Game::ShipDesign::Write is 0x008325e0 and makes four stream calls. The recorded address 0x008747a0 is in NO vftable at all (lane V2's inversion), so "ShipDesign::Write makes no stream call" was a misattribution, not a fact about the class. Game::ShipDesign derives from Game::ShipDesignDef and inherits IStreamable second, so its writer is reached through an adjustor thunk -- which is what the slot-indexed serializer sweep found instead. A design persists as two serializers, base and derived: ShipDesignDef::Write 0x00827390 emits FAIDes/DHide/DWep/DName then exactly three DSec frames (+0x4c command, +0x24 mission, +0x74 engine), ShipDesign::Write appends Dtc, the Dwgv flag and a conditional Dwg frame. THREE sections, not five. The campaign's "slots 3-4 reserved and always empty" was save_reader.py's trailing Rest("sections") sweeping Dtc and Dwgv into the section list, and stock_designs.py decoding them as two empty sections -- rule 8 in its exact form, reader and consumer agreeing with each other and both wrong. Two independent enumerations say three: the writer's straight-line body, and the ctor's eh_vector_constructor_iterator(this+0x24, 0x28, 3) closing at 0x9c = sizeof(ShipDesignDef). DWep and Dwgv are BOOLs, not ints -- byte-indistinguishable from ints at a four-character tag, the same class of defect as ObservedTech.odet. THE 0x400 FLAG IS `defence_platform`, read off the .shipsection parser's own bit setter at 0x005749b7. NOT lane B5's 0x400: that one is a fleet flag, on the wire as FtFlg. The full role-flag table is in the finding. HULL SIZE is section_class through a three-name stricmp table (Destroyer/Cruiser/Dreadnought -> 0/1/2), absent or unrecognised meaning 0 with a log line rather than an error. Both words are recomputed from the data files by ShipDesign::UpdateDerivedStats 0x0087e7c0 and neither is on the wire. Corroborated by the default hull-health table the same bit picks: 500/3000/15000 without it, 100/500/1000 with. MEASURED: the census rebuilt from each save's own state matches the record the game archived, 480 leaves / 0 mismatched over 11 saves and 503 designs, computed independently in Python and in C++. COVERAGE REPORTED AS LOUDLY: only 32 of the 480 leaves are nonzero, and three of the six census leaves (both cruiser rows, dreadnought platforms) are unexercised by every save in the corpus. Closed 0 / regressed 0 against the standalone's divergence list, reported separately: the census leaves live in src/app's turn record, which lane A2 holds this cycle, so this lane evaluated and reported rather than writing. Oracles fixed openly (rule 12): save_reader.py's Des shape, 49/49 with three corrected tests and one added that pins "exactly three DSec" against real saves; stock_designs.json regenerated, whose diff is only raw_slots 5->3 and dWep int->bool across all 127 designs with every other field identical; test_design_rules.py still 32/32 with the same ground truth. 19 addresses in ghidra/addresses.d/lane-d2.json, no collision; the generated header was validated to a scratch path, never written in place.
868 lines
42 KiB
Python
868 lines
42 KiB
Python
#!/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<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"})
|
||
|
||
# --- 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()
|