370 lines
14 KiB
Python
370 lines
14 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. Validation
|
|
against a real Sword of the Stars save is still pending.
|
|
"""
|
|
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)
|
|
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 schema float stored as a word the walker guessed 'int' is re-read from raw bytes."""
|
|
w = sw.SaveWriter()
|
|
w.begin("Tmrs")
|
|
w.int("tstl", 0x7F7FFFFF)
|
|
w.float("tctl", 60.0) # 0x42700000, walker may call it float; schema says int
|
|
w.int("tqtl", 0x7F7FFFFF)
|
|
w.int("tqtle", 0)
|
|
w.end()
|
|
schema = sr.Seq([sr.R("t", sr.Tmrs)])
|
|
res = sr.read_bytes(w.bytes(), padding="joint", schema=schema)
|
|
self.assertEqual(res.typed["t"]["tctl"], 0x42700000)
|
|
self.assertEqual(res.typed["t"]["tstl"], 0x7F7FFFFF)
|
|
|
|
|
|
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()
|