#!/usr/bin/env python3 """Large-file cache downloads. Two runs fetching the same asset used to stream into one shared scratch path, interleaving their writes into a full-size file with mixed content that then replaced the cache entry. """ from __future__ import annotations import hashlib import io import json import os import shutil import sys import tempfile import threading import unittest import urllib.error from pathlib import Path from unittest import mock REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "scripts")) import common # noqa: E402 import largefiles import check_release_assets # noqa: E402 PAYLOAD_A = b"A" * (256 * 1024) PAYLOAD_B = b"B" * (256 * 1024) class _SlowResponse(io.BytesIO): """Serve a payload in chunks, yielding between them.""" def __init__(self, data: bytes, barrier: threading.Barrier | None = None): super().__init__(data) self._barrier = barrier self._first = True def read(self, size: int = -1) -> bytes: chunk = super().read(min(size, 4096) if size and size > 0 else 4096) if self._first and self._barrier is not None: self._first = False self._barrier.wait(timeout=10) return chunk def __enter__(self): return self def __exit__(self, *exc): self.close() return False class LargeFileCacheTest(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp() self._urlopen = largefiles.urllib.request.urlopen def tearDown(self): largefiles.urllib.request.urlopen = self._urlopen def test_concurrent_fetches_do_not_mix(self): barrier = threading.Barrier(2, timeout=10) payloads = [PAYLOAD_A, PAYLOAD_B] index = iter(range(2)) lock = threading.Lock() def fake_urlopen(req, timeout=None): with lock: i = next(index) return _SlowResponse(payloads[i], barrier) largefiles.urllib.request.urlopen = fake_urlopen results: list[str | None] = [None, None] def worker(slot: int): results[slot] = common.fetch_large_file("asset.bin", dest_dir=self.dir) threads = [threading.Thread(target=worker, args=(i,)) for i in range(2)] for t in threads: t.start() for t in threads: t.join(timeout=30) cached = Path(self.dir) / "asset.bin" self.assertTrue(cached.exists()) digest = hashlib.sha1(cached.read_bytes()).hexdigest() accepted = {hashlib.sha1(p).hexdigest() for p in payloads} # Whichever run lands last wins, but never a blend of the two self.assertIn(digest, accepted) def test_no_scratch_file_survives_a_successful_fetch(self): largefiles.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( PAYLOAD_A ) common.fetch_large_file("asset.bin", dest_dir=self.dir) leftovers = [f for f in os.listdir(self.dir) if f.endswith(".tmp")] self.assertEqual(leftovers, []) def test_no_scratch_file_survives_a_failed_fetch(self): def fail(req, timeout=None): raise urllib.error.URLError("offline") largefiles.urllib.request.urlopen = fail self.assertIsNone(common.fetch_large_file("asset.bin", dest_dir=self.dir)) self.assertEqual(os.listdir(self.dir), []) def test_hash_mismatch_leaves_no_scratch_file(self): largefiles.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( PAYLOAD_A ) result = common.fetch_large_file( "asset.bin", dest_dir=self.dir, expected_sha1="00" * 20 ) self.assertIsNone(result) self.assertEqual(os.listdir(self.dir), []) def test_cached_file_is_returned_without_download(self): cached = Path(self.dir) / "asset.bin" cached.write_bytes(PAYLOAD_A) def fail(req, timeout=None): raise AssertionError("must not download when the cache is valid") largefiles.urllib.request.urlopen = fail self.assertEqual( common.fetch_large_file("asset.bin", dest_dir=self.dir), str(cached) ) def test_offline_cache_miss_never_opens_the_network(self): def fail(req, timeout=None): raise AssertionError("offline mode must not open the network") largefiles.urllib.request.urlopen = fail self.assertIsNone( common.fetch_large_file( "asset.bin", dest_dir=self.dir, offline=True ) ) self.assertEqual(os.listdir(self.dir), []) def test_offline_mode_still_uses_a_verified_cache_hit(self): cached = Path(self.dir) / "asset.bin" cached.write_bytes(PAYLOAD_A) def fail(req, timeout=None): raise AssertionError("offline cache hit must not open the network") largefiles.urllib.request.urlopen = fail self.assertEqual( common.fetch_large_file( "asset.bin", dest_dir=self.dir, expected_sha1=hashlib.sha1(PAYLOAD_A).hexdigest(), offline=True, ), str(cached), ) def _serve(self, payload: bytes, gets: list): class _Head(io.BytesIO): headers = {"Content-Length": str(len(payload))} def __enter__(self): return self def __exit__(self, *exc): return False def urlopen(req, timeout=None): if req.get_method() == "HEAD": return _Head() gets.append(req.full_url) return _SlowResponse(payload) largefiles.urllib.request.urlopen = urlopen def test_a_reuploaded_asset_replaces_the_cached_revision(self): """--clobber put new bytes under the name: the stale copy answered None on every run, without a single request.""" cached = Path(self.dir) / "asset.bin" cached.write_bytes(PAYLOAD_A) rebuilt = b"C" * 1000 gets: list = [] self._serve(rebuilt, gets) result = common.fetch_large_file( "asset.bin", dest_dir=self.dir, expected_sha1=hashlib.sha1(rebuilt).hexdigest(), ) self.assertEqual(result, str(cached)) self.assertEqual(cached.read_bytes(), rebuilt) self.assertEqual(len(gets), 1) def test_a_revision_the_release_still_serves_is_kept(self): cached = Path(self.dir) / "asset.bin" cached.write_bytes(PAYLOAD_A) gets: list = [] self._serve(PAYLOAD_A, gets) result = common.fetch_large_file( "asset.bin", dest_dir=self.dir, expected_sha1="00" * 20 ) self.assertIsNone(result) self.assertEqual(gets, []) self.assertEqual(cached.read_bytes(), PAYLOAD_A) def test_a_download_that_does_not_verify_keeps_the_cache(self): cached = Path(self.dir) / "asset.bin" cached.write_bytes(PAYLOAD_A) gets: list = [] self._serve(b"D" * 10, gets) result = common.fetch_large_file( "asset.bin", dest_dir=self.dir, expected_sha1="00" * 20 ) self.assertIsNone(result) self.assertEqual(cached.read_bytes(), PAYLOAD_A) self.assertEqual(os.listdir(self.dir), ["asset.bin"]) class HashCacheKeepsEveryDigest(unittest.TestCase): """A cache hit must serve the same five digests a fresh hash produces. The cache-hit path rebuilt the hash dict from a hand-written list that omitted adler32, then wrote the entry back without it, so one run without --force stripped the digest from all 7,850 entries for good. """ def setUp(self): import generate_db self.generate_db = generate_db self._tmp = tempfile.TemporaryDirectory() self.bios = Path(self._tmp.name) / "bios" (self.bios / "Sony" / "PS").mkdir(parents=True) (self.bios / "Sony" / "PS" / "boot.bin").write_bytes(b"CACHED PAYLOAD") def tearDown(self): self._tmp.cleanup() def test_cache_hit_serves_the_full_digest_set(self): files, _, cache = self.generate_db.scan_bios_dir(self.bios, {}, force=False) entry = next(iter(files.values())) self.assertTrue(set(self.generate_db.CACHED_HASHES).issubset(entry)) # Second pass, this time served entirely from the cache. again, _, cache2 = self.generate_db.scan_bios_dir(self.bios, cache, force=False) entry2 = next(iter(again.values())) self.assertTrue( set(self.generate_db.CACHED_HASHES).issubset(entry2), f"cache hit lost {set(self.generate_db.CACHED_HASHES) - set(entry2)}", ) self.assertEqual( {k: entry[k] for k in self.generate_db.CACHED_HASHES}, {k: entry2[k] for k in self.generate_db.CACHED_HASHES}, ) self.assertTrue( set(self.generate_db.CACHED_HASHES).issubset(next(iter(cache2.values()))) ) def test_a_warm_cache_serialises_exactly_like_a_fresh_hash(self): """Key order must not depend on whether the cache was warm. Rebuilding the dict by iterating a set made the order follow set hashing, so a run with a warm cache rewrote all 7,850 entries with their digests in a different order and no content change. """ fresh, _, cache = self.generate_db.scan_bios_dir(self.bios, {}, force=True) warm, _, _ = self.generate_db.scan_bios_dir(self.bios, cache, force=False) self.assertEqual( [list(entry) for entry in fresh.values()], [list(entry) for entry in warm.values()], ) self.assertEqual( json.dumps(fresh, indent=2), json.dumps(warm, indent=2) ) def test_a_partial_cache_entry_is_rehashed_instead_of_trusted(self): _, _, cache = self.generate_db.scan_bios_dir(self.bios, {}, force=False) key = next(iter(cache)) cache[key].pop("adler32") files, _, healed = self.generate_db.scan_bios_dir(self.bios, cache, force=False) entry = next(iter(files.values())) self.assertTrue(set(self.generate_db.CACHED_HASHES).issubset(entry)) self.assertTrue(set(self.generate_db.CACHED_HASHES).issubset(healed[key])) class PreservedLargeFileEntries(unittest.TestCase): """A preserved entry must never claim a path another entry already owns. A large file replaced on disk by a newer firmware revision left its old entry in the database forever, pointing at a path that now serves other bytes. """ def setUp(self): import generate_db self.generate_db = generate_db self._tmp = tempfile.TemporaryDirectory() self.tmp = Path(self._tmp.name) self._cwd = os.getcwd() os.chdir(self.tmp) (self.tmp / ".gitignore").write_text("bios/Sony/PS3/FW.PUP\n") self._real_fetch = common.fetch_large_file generate_db.__dict__.pop("fetch_large_file", None) def tearDown(self): os.chdir(self._cwd) common.fetch_large_file = self._real_fetch self._tmp.cleanup() def _write_db(self, entries: dict) -> str: path = str(self.tmp / "database.json") Path(path).write_text(json.dumps({"files": entries})) return path def test_stale_entry_for_a_rescanned_path_is_dropped(self): common.fetch_large_file = lambda *_a, **_k: None db_path = self._write_db( { "a" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}, "b" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}, } ) # The scan found the current revision at that path. files = {"a" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}} count = self.generate_db._preserve_large_file_entries(files, db_path) self.assertEqual(count, 0) self.assertEqual(list(files), ["a" * 40]) def test_absent_large_file_is_still_preserved(self): common.fetch_large_file = lambda *_a, **_k: None db_path = self._write_db( {"b" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}} ) files: dict = {} count = self.generate_db._preserve_large_file_entries(files, db_path) self.assertEqual(count, 1) self.assertIn("b" * 40, files) def test_a_fetched_asset_returns_to_its_registered_path(self): cache = self.tmp / "cache" / "FW.PUP" cache.parent.mkdir() cache.write_bytes(b"firmware") common.fetch_large_file = lambda *_a, **_k: str(cache) db_path = self._write_db( {"b" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}} ) files: dict = {} count = self.generate_db._preserve_large_file_entries(files, db_path) self.assertEqual(count, 1) self.assertEqual(files["b" * 40]["path"], "bios/Sony/PS3/FW.PUP") self.assertEqual((self.tmp / "bios/Sony/PS3/FW.PUP").read_bytes(), b"firmware") def test_an_interrupted_restore_leaves_no_file_behind(self): cache = self.tmp / "cache" / "FW.PUP" cache.parent.mkdir() cache.write_bytes(b"firmware") common.fetch_large_file = lambda *_a, **_k: str(cache) db_path = self._write_db( {"b" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}} ) def interrupted(_src, dst, *_a, **_k): Path(dst).write_bytes(b"firm") raise KeyboardInterrupt with mock.patch.object(self.generate_db.shutil, "copy2", interrupted), \ self.assertRaises(KeyboardInterrupt): self.generate_db._preserve_large_file_entries({}, db_path) directory = self.tmp / "bios/Sony/PS3" self.assertEqual(list(directory.iterdir()), []) def test_a_file_sharing_only_the_name_is_not_preserved(self): common.fetch_large_file = lambda *_a, **_k: "/cache/large/FW.PUP" db_path = self._write_db( {"c" * 40: {"name": "FW.PUP", "path": "bios/Sony/PSP/FW.PUP"}} ) files: dict = {} self.assertEqual( self.generate_db._preserve_large_file_entries(files, db_path), 0 ) self.assertEqual(files, {}) def test_a_second_revision_under_a_rescanned_path_is_dropped(self): common.fetch_large_file = lambda *_a, **_k: "/cache/large/FW.PUP" db_path = self._write_db( {"b" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}} ) files = {"a" * 40: {"name": "FW.PUP", "path": "bios/Sony/PS3/FW.PUP"}} self.assertEqual( self.generate_db._preserve_large_file_entries(files, db_path), 0 ) self.assertEqual(list(files), ["a" * 40]) class ReleaseAssetNames(unittest.TestCase): """One registered path, one asset name, the same for every consumer. Every consumer named an asset by the file's basename, so two collection files sharing one (`etmain/pak0.pk3` and `demomain/pak0.pk3`) could not both be published: the second upload replaced the first and every fetch received whichever was there. """ ET = "bios/Id Software/Wolfenstein Enemy Territory/etmain/pak0.pk3" RTCW = "bios/Id Software/Return to Castle Wolfenstein/demomain/pak0.pk3" PUP = "bios/Sony/PS3/PS3UPDAT.PUP" REGISTERED = [ET, RTCW, PUP] def test_a_shared_basename_gets_two_distinct_names(self): names = largefiles.asset_names(self.REGISTERED) self.assertNotEqual(names[self.ET], names[self.RTCW]) self.assertNotIn("pak0.pk3", (names[self.ET], names[self.RTCW])) for name in (names[self.ET], names[self.RTCW]): self.assertNotIn("/", name) self.assertNotIn(" ", name) self.assertTrue(name.endswith("pak0.pk3")) def test_the_name_depends_on_the_path_not_on_the_set_order(self): forward = largefiles.asset_names(self.REGISTERED) backward = largefiles.asset_names(list(reversed(self.REGISTERED))) self.assertEqual(forward, backward) def test_a_basename_used_once_keeps_its_name(self): self.assertEqual( largefiles.asset_name(self.PUP, self.REGISTERED), "PS3UPDAT.PUP" ) self.assertEqual( largefiles.asset_names(["bios/Arcade/MAME/MAME 0.174 Arcade XML.dat"]), {"bios/Arcade/MAME/MAME 0.174 Arcade XML.dat": "MAME 0.174 Arcade XML.dat"}, ) def test_the_published_register_has_no_two_paths_on_one_name(self): registered = largefiles.registered_paths( (REPO_ROOT / ".gitignore").read_text(encoding="utf-8") ) names = largefiles.asset_names(registered) self.assertEqual(len(set(names.values())), len(registered)) basenames = [os.path.basename(p) for p in registered] for registered_path, name in names.items(): if basenames.count(os.path.basename(registered_path)) == 1: self.assertEqual(name, os.path.basename(registered_path)) def test_manifest_checker_and_fetcher_agree_on_a_path(self): import generate_pack expected = largefiles.asset_names(self.REGISTERED) gitignore = "\n".join(["tmp/", *self.REGISTERED]) + "\n" with tempfile.TemporaryDirectory() as root: Path(root, ".gitignore").write_text(gitignore, encoding="utf-8") generate_pack._GITIGNORE_ENTRIES = None try: manifest = { p: generate_pack._release_asset_name(os.path.join(root, p), root) for p in self.REGISTERED } finally: generate_pack._GITIGNORE_ENTRIES = None self.assertEqual(manifest, expected) sizes = {self.ET: 228138631, self.RTCW: 122757192, self.PUP: 10} published = {expected[p]: size for p, size in sizes.items()} self.assertEqual( check_release_assets.compare( sizes, published, registered=self.REGISTERED ), [], ) # The old basename-only asset must not satisfy either pak. self.assertEqual( {f[1] for f in check_release_assets.compare( sizes, {"pak0.pk3": 228138631, "PS3UPDAT.PUP": 10}, registered=self.REGISTERED, )}, {self.ET, self.RTCW}, ) db = {"files": { "a" * 40: {"path": self.ET, "size": sizes[self.ET]}, "b" * 40: {"path": self.RTCW, "size": sizes[self.RTCW]}, "c" * 40: {"path": self.PUP, "size": sizes[self.PUP]}, }} body = check_release_assets.render_notes( db, gitignore, published, "", {}, {}, {} ) self.assertIn(f"[{expected[self.ET]}]", body) self.assertIn("a" * 40, body) self.assertIn("b" * 40, body) self.assertNotIn("## Not indexed", body) requested: list[str] = [] def fake_urlopen(req, timeout=None): requested.append(req.full_url.rsplit("/", 1)[1]) raise urllib.error.URLError("offline") self._urlopen = largefiles.urllib.request.urlopen largefiles.urllib.request.urlopen = fake_urlopen try: with tempfile.TemporaryDirectory() as cache: largefiles.fetch_large_file( self.RTCW, dest_dir=cache, registered=self.REGISTERED ) self.assertEqual( requested, [largefiles.urllib.parse.quote(expected[self.RTCW])], ) requested.clear() largefiles.fetch_large_file( "pak0.pk3", dest_dir=cache, registered=self.REGISTERED ) self.assertEqual( sorted(requested), sorted( largefiles.urllib.parse.quote(expected[p]) for p in (self.ET, self.RTCW) ), ) finally: largefiles.urllib.request.urlopen = self._urlopen def test_a_shared_basename_is_fetched_by_the_asset_its_hash_names(self): names = largefiles.asset_names(self.REGISTERED) with tempfile.TemporaryDirectory() as cache: Path(cache, names[self.ET]).write_bytes(PAYLOAD_A) Path(cache, names[self.RTCW]).write_bytes(PAYLOAD_B) got = largefiles.fetch_large_file( "pak0.pk3", dest_dir=cache, expected_sha1=hashlib.sha1(PAYLOAD_B).hexdigest(), offline=True, registered=self.REGISTERED, ) self.assertEqual(got, os.path.join(cache, names[self.RTCW])) # The cache entry of the other asset is a valid file, not stale. self.assertTrue(Path(cache, names[self.ET]).exists()) class _StalledResponse(io.BytesIO): """Serves one chunk, then the stream stalls.""" def __init__(self): super().__init__(PAYLOAD_A) self._served = False def read(self, size: int = -1) -> bytes: if self._served: raise TimeoutError("read timed out") self._served = True return super().read(4096) def __enter__(self): return self def __exit__(self, *exc): return False class DownloadFailuresAndRevisions(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp(dir=REPO_ROOT / "tmp") self._urlopen = largefiles.urllib.request.urlopen self.addCleanup(setattr, largefiles.urllib.request, "urlopen", self._urlopen) def tearDown(self): shutil.rmtree(self.dir, ignore_errors=True) def test_a_stalled_stream_leaves_no_scratch(self): largefiles.urllib.request.urlopen = lambda *_a, **_k: _StalledResponse() result = largefiles.fetch_large_file( "big.bin", self.dir, expected_md5=hashlib.md5(PAYLOAD_A).hexdigest(), registered=[], ) self.assertIsNone(result) self.assertEqual(os.listdir(self.dir), []) def test_another_revision_is_kept_not_evicted(self): cached = Path(self.dir, "PS3UPDAT.PUP") cached.write_bytes(PAYLOAD_A) largefiles.urllib.request.urlopen = lambda *_a, **_k: (_ for _ in ()).throw( urllib.error.URLError("offline") ) result = largefiles.fetch_large_file( "PS3UPDAT.PUP", self.dir, expected_md5=hashlib.md5(PAYLOAD_B).hexdigest(), registered=[], ) self.assertIsNone(result) self.assertTrue(cached.exists()) def test_a_variant_asset_is_a_candidate(self): registered = [ "bios/Sony/PlayStation 3/PS3UPDAT.PUP", "bios/Sony/PlayStation 3/.variants/PS3UPDAT.PUP.ed8ab192", ] self.assertEqual( largefiles.asset_candidates("PS3UPDAT.PUP", registered), ["PS3UPDAT.PUP", "PS3UPDAT.PUP.ed8ab192"], ) class GameDataHasOneDefinition(unittest.TestCase): """The release notes and the composition table sorted game data apart. The notes knew two engine trees, the table sixteen: fifteen assets of Doom, Quake, Half-Life and Descent were listed as "Other". The Ur-Quan Masters tree was in neither. """ def test_the_release_section_follows_the_composition_tier(self): for path in ( "bios/Id Software/Doom 3/demo/demo00.pk4", "bios/Valve/Half-Life/valve/pak0.pak", "bios/Toys for Bob/The Ur-Quan Masters/uqm-0.8.0-voice.uqm", ): with self.subTest(path=path): self.assertEqual(common.composition_tier(path), "game_data") self.assertEqual( check_release_assets._section_for(path), "Game engine data" ) def test_a_tree_of_game_data_alone_is_a_game_data_tree(self): db_path = REPO_ROOT / "database.json" if not db_path.exists(): self.skipTest("database.json not generated") db = json.loads(db_path.read_text(encoding="utf-8")) profiles = common.load_emulator_profiles(str(REPO_ROOT / "emulators")) names: dict[str, set[str]] = {} for profile in profiles.values(): for entry in profile.get("files") or []: category = entry.get("category", "bios") for name in [entry.get("name"), *(entry.get("aliases") or [])]: if name: names.setdefault(name, set()).add(category) trees: dict[str, set[str]] = {} for record in db["files"].values(): parts = record["path"].split("/") if parts[0] == "bios" and parts[2:]: trees.setdefault(parts[1], set()).update( names.get(record["name"], {"unreferenced"}) ) only_game_data = sorted( top for top, categories in trees.items() if categories == {"game_data"} ) self.assertEqual( [top for top in only_game_data if top not in common.GAME_DATA_TOPS], [] ) if __name__ == "__main__": unittest.main()