import hashlib import importlib.util import json import tempfile import unittest from pathlib import Path MODULE_PATH = Path(__file__).with_name("materialize-catalog-replay.py") SPEC = importlib.util.spec_from_file_location("materialize_catalog", MODULE_PATH) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC.loader SPEC.loader.exec_module(MODULE) class MaterializeCatalogReplayTests(unittest.TestCase): def fixture(self, root: Path, *, base="classic-1285-tft-open2-v1", engine="1.28.5", played_differs=False, map_path=None): source = hashlib.sha1(b"source-id").hexdigest() replay = b"converted" if played_differs else b"source-id" played = hashlib.sha1(replay).hexdigest() map_bytes = b"exact-map" map_sha1 = hashlib.sha1(map_bytes).hexdigest() map_path = map_path or "Maps\\FrozenThrone\\Arena.w3x" index = root / "index.json" index.write_text(json.dumps({"schema": 2, "replays": {source: { "playedReplaySha1": played, "baseId": base, "engine": engine, "mapPath": map_path, "mapContentSha1": map_sha1, "mapObjectKey": "maps/classic/Arena.w3x", }}})) objects = { MODULE.replay_object_key(source, played, "native-1140" if base.startswith("classic-1140-") else "native-1285"): replay, "maps/classic/Arena.w3x": map_bytes, } def fetch(key, destination): destination.write_bytes(objects[key]) return index, source, played, map_sha1, fetch def test_materializes_native_profile_with_exact_map(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) index, source, played, map_sha1, fetch = self.fixture(root) result = MODULE.materialize(index, source, root / "out", fetch) self.assertEqual("native-1285", result["engineProfile"]) self.assertEqual(played, result["replaySha1"]) self.assertEqual(map_sha1, result["mapContentSha1"]) self.assertTrue(Path(result["metadataPath"]).is_file()) def test_converted_replay_uses_c28_object(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) index, source, _played, _map, fetch = self.fixture( root, played_differs=True) result = MODULE.materialize(index, source, root / "out", fetch) self.assertEqual( f"replays/archive/{source}-c28.w3g", result["replayObjectKey"]) def test_native_114_allows_expired_patch_map(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) index, source, _played, _map, fetch = self.fixture( root, base="classic-1140-tft-open2-audio-v4", engine="1.14", map_path="Maps\\ExpiredPatchMap.tmp") value = json.loads(index.read_text()) entry = value["replays"][source] entry["mapObjectKey"] = ( "worker-image/classic/replaykit/1.14/v3/maps/" + entry["mapContentSha1"] + ".tmp") index.write_text(json.dumps(value)) objects = {} original_fetch = fetch def remapped(key, destination): if key == entry["mapObjectKey"]: destination.write_bytes(b"exact-map") else: original_fetch(key, destination) result = MODULE.materialize(index, source, root / "out", remapped) self.assertEqual("native-1140", result["engineProfile"]) self.assertTrue(result["allowExpiredPatchMap"]) self.assertEqual( "worker-image/classic/replaykit/1.14/v3/replays/" + result["replaySha1"] + ".w3g", result["replayObjectKey"]) def test_rejects_unknown_replay_and_unsafe_map(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) index, source, _played, _map, fetch = self.fixture(root) with self.assertRaises(MODULE.CatalogMaterializeError): MODULE.materialize(index, "f" * 40, root / "out", fetch) value = json.loads(index.read_text()) value["replays"][source]["mapObjectKey"] = "maps/../secret.w3x" index.write_text(json.dumps(value)) with self.assertRaises(MODULE.CatalogMaterializeError): MODULE.materialize(index, source, root / "out", fetch) if __name__ == "__main__": unittest.main() class R2MirrorIndexTests(unittest.TestCase): """Schema 3: the content-addressed R2 mirror read through the CDN.""" def fixture(self, root: Path, *, profile="native-1285", mode="tft", map_ext="w3x", map_path=None, players=("A", "B")): source = hashlib.sha1(b"source-id").hexdigest() replay = b"converted-bytes" played = hashlib.sha1(replay).hexdigest() map_bytes = b"exact-map" map_sha1 = hashlib.sha1(map_bytes).hexdigest() map_path = map_path or f"Maps\\Frozen\\Arena.{map_ext}" index = root / "index.json" index.write_text(json.dumps({"schema": 3, "replays": {source: { "played": played, "mapSha1": map_sha1, "mapExt": map_ext, "mapPath": map_path, "profile": profile, "mode": mode, "family": mode.upper(), "players": list(players), "durationMs": 1234}}})) objects = {f"replays/corpus/{played}.w3g": replay, f"maps/{map_sha1}.{map_ext}": map_bytes} fetched = [] def fetch(key, destination): fetched.append(key) destination.write_bytes(objects[key]) return index, source, played, map_sha1, fetch, fetched def test_materializes_from_content_addressed_keys(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) index, source, played, map_sha1, fetch, fetched = self.fixture(root) result = MODULE.materialize(index, source, root / "out", fetch) self.assertEqual(sorted(fetched), sorted([ f"replays/corpus/{played}.w3g", f"maps/{map_sha1}.w3x"])) self.assertEqual(result["engineProfile"], "native-1285") self.assertEqual(result["patchFamily"], "tft") self.assertEqual(result["displayName"], "A vs B") self.assertEqual(result["mapFile"], f"maps/{map_sha1}.w3x") self.assertEqual(result["replaySha1"], played) sidecar = json.loads((root / "out" / f"{source}.json").read_text()) self.assertEqual(sidecar["engineProfile"], "native-1285") self.assertTrue((root / "out" / f"{source}.w3g").is_file()) def test_rejects_unknown_profile_and_inconsistent_extension(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) index, source, *_rest, fetch, _ = self.fixture(root, profile="native-9999") with self.assertRaises(MODULE.CatalogMaterializeError): MODULE.materialize(index, source, root / "out", fetch) index, source, *_rest, fetch, _ = self.fixture( root, map_ext="w3x", map_path="Maps\\Arena.w3m") with self.assertRaises(MODULE.CatalogMaterializeError): MODULE.materialize(index, source, root / "out", fetch) def test_https_fetcher_only_accepts_content_addressed_keys(self): fetch = MODULE.https_fetcher("https://cdn.example") with tempfile.TemporaryDirectory() as tmp: with self.assertRaises(MODULE.CatalogMaterializeError): fetch("replays/archive/anything.w3g", Path(tmp) / "x") with self.assertRaises(MODULE.CatalogMaterializeError): fetch("maps/../etc/passwd", Path(tmp) / "x") with self.assertRaises(MODULE.CatalogMaterializeError): MODULE.https_fetcher("http://cdn.example")