refactor: read the asset registry with one parser

generate_pack kept a private .gitignore parser whose cache ignored the
root it was asked about, and generate_db a copy of the shared one. Both
now read largefiles.load_registered_paths, cached per root. Test modules
whose main block preceded their last class end with it again.
This commit is contained in:
Abdessamad Derraz committed 2026-10-17 19:35:42 +02:00
1 parent 0754d3b64f
commit 18e61fabcb
6 files changed
+57 -51

No files matched your search

+3 -8
View File
@@ -271,14 +271,9 @@ def save_cache(cache_path: str, cache: dict) -> None:
def _load_gitignored_large_files() -> set[str]:
"""The bios/ paths .gitignore registers as release assets."""
gitignore = Path(".gitignore")
if not gitignore.exists():
return set()
return {
line.strip()
for line in gitignore.read_text().splitlines()
if line.strip().startswith("bios/")
}
from largefiles import load_registered_paths
return set(load_registered_paths(".gitignore"))
def _restore(cached: str, path: str) -> None:
+9 -25
View File
@@ -15,6 +15,7 @@ from __future__ import annotations
from artifacts import _build_timestamp
import argparse
import contextlib
import functools
import hashlib
import json
import os
@@ -71,7 +72,7 @@ import slot as slot_mod
import slots
import split_pack
from deterministic_zip import _FIXED_DATE_TIME, rebuild_zip_deterministic
from largefiles import asset_name
from largefiles import asset_name, load_registered_paths
from nativemode import (
digest_algorithm,
hash_mismatch_excludes_file,
@@ -3103,24 +3104,11 @@ def _main(held: contextlib.ExitStack) -> None:
# Manifest generation (JSON inventory for install.py)
_GITIGNORE_ENTRIES: set[str] | None = None
def _load_gitignore_entries(repo_root: str) -> set[str]:
"""Load gitignored paths (large files) from .gitignore at repo root."""
global _GITIGNORE_ENTRIES
if _GITIGNORE_ENTRIES is not None:
return _GITIGNORE_ENTRIES
gitignore = os.path.join(repo_root, ".gitignore")
entries: set[str] = set()
if os.path.exists(gitignore):
with open(gitignore) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
entries.add(line)
_GITIGNORE_ENTRIES = entries
return entries
@functools.lru_cache(maxsize=None)
def _registered(repo_root: str) -> frozenset[str]:
"""The paths a repository's .gitignore registers as release assets, read
by the parser the fetcher and the release check use, once per root."""
return frozenset(load_registered_paths(os.path.join(repo_root, ".gitignore")))
def _is_release_asset(local_path: str, repo_root: str) -> bool:
@@ -3132,7 +3120,7 @@ def _is_release_asset(local_path: str, repo_root: str) -> bool:
is still served by the repository, and announcing it as a release asset
sends the installer to an asset nobody uploaded.
"""
return _repo_relative(local_path, repo_root) in _load_gitignore_entries(repo_root)
return _repo_relative(local_path, repo_root) in _registered(repo_root)
def _repo_relative(local_path: str, repo_root: str) -> str:
@@ -3144,11 +3132,7 @@ def _repo_relative(local_path: str, repo_root: str) -> str:
def _release_asset_name(local_path: str, repo_root: str) -> str:
"""The large-files asset install.py downloads for a gitignored file."""
registered = [
entry for entry in _load_gitignore_entries(repo_root)
if entry.startswith("bios/")
]
return asset_name(_repo_relative(local_path, repo_root), registered)
return asset_name(_repo_relative(local_path, repo_root), sorted(_registered(repo_root)))
def _get_repo_path(sha1: str, db: dict) -> str:
+31 -7
View File
@@ -1618,11 +1618,11 @@ class ReleaseAssetCriterion(unittest.TestCase):
big.parent.mkdir()
with big.open("wb") as handle:
handle.truncate(60_000_000)
generate_pack._GITIGNORE_ENTRIES = None
generate_pack._registered.cache_clear()
try:
self.assertFalse(generate_pack._is_release_asset(str(big), root))
finally:
generate_pack._GITIGNORE_ENTRIES = None
generate_pack._registered.cache_clear()
def test_a_gitignored_file_is_served_by_the_release_whatever_its_size(self):
from scripts import generate_pack
@@ -1632,11 +1632,11 @@ class ReleaseAssetCriterion(unittest.TestCase):
small = Path(root, "bios", "small.zip")
small.parent.mkdir()
small.write_bytes(b"PK")
generate_pack._GITIGNORE_ENTRIES = None
generate_pack._registered.cache_clear()
try:
self.assertTrue(generate_pack._is_release_asset(str(small), root))
finally:
generate_pack._GITIGNORE_ENTRIES = None
generate_pack._registered.cache_clear()
class ReleaseAssetsMatchTheCollection(unittest.TestCase):
@@ -2860,9 +2860,6 @@ class ADuplicateYamlKeyIsRefused(unittest.TestCase):
self.assertIn("duplicate key 'target_source'", str(caught.exception))
self.assertEqual(yaml_load("a: &x {b: 1}\nc: *x\n"), {"a": {"b": 1}, "c": {"b": 1}})
if __name__ == "__main__":
unittest.main()
class AnAcceptedVersionIsNoCoreDiscrepancy(unittest.TestCase):
"""eduke32 lists the retail and demo NWINTER.GRP under one name: the extra
@@ -3244,3 +3241,30 @@ class ADataDirectoryIsHeldWhileRead(unittest.TestCase):
self.assertIn("hold_data_dirs(held,", verify_src)
refresh = pack.index("results = refresh_all(")
self.assertIn("hold_data_dirs(held, data_registry)", pack[refresh:])
class EachRootReadsItsOwnRegistry(unittest.TestCase):
"""The release-asset registry was cached once per process whatever root
was asked: a second tree got the first tree's .gitignore."""
def test_two_roots_in_one_process(self):
from scripts import generate_pack
with tempfile.TemporaryDirectory(dir=TMP_ROOT) as a, \
tempfile.TemporaryDirectory(dir=TMP_ROOT) as b:
Path(a, ".gitignore").write_text("bios/one.bin\n")
Path(b, ".gitignore").write_text("bios/two.bin\n")
self.assertTrue(generate_pack._is_release_asset(str(Path(a, "bios/one.bin")), a))
self.assertFalse(generate_pack._is_release_asset(str(Path(b, "bios/one.bin")), b))
self.assertTrue(generate_pack._is_release_asset(str(Path(b, "bios/two.bin")), b))
def test_one_parser_reads_the_registry(self):
for module in ("generate_pack.py", "generate_db.py"):
source = (ROOT / "scripts" / module).read_text()
with self.subTest(module=module):
self.assertIn("load_registered_paths", source)
self.assertNotIn('startswith("bios/")\n }', source)
if __name__ == "__main__":
unittest.main()
+4 -3
View File
@@ -93,9 +93,6 @@ class ACoreFileFillsAFreeDeclaration(unittest.TestCase):
)
if __name__ == "__main__":
unittest.main()
class AFileDeclaredUnderAnotherName(unittest.TestCase):
"""The platform declares ROM1 with the md5 of the core's ROM. The
@@ -142,3 +139,7 @@ class AFileDeclaredUnderAnotherName(unittest.TestCase):
self.assertIn("gs/ROM", zf.namelist())
reported = [(u["name"], u["in_repo"]) for u in result["undeclared_files"]]
self.assertIn(("ROM", True), reported)
if __name__ == "__main__":
unittest.main()
+4 -3
View File
@@ -1359,9 +1359,6 @@ class UpstreamFidelity(unittest.TestCase):
)
if __name__ == "__main__":
unittest.main()
class ARecordDescribesOneDump(unittest.TestCase):
"""syscard2.pce: System.dat states v2.1 (crc 283b74e0, md5, sha1),
@@ -1504,3 +1501,7 @@ class RocknixNamesRunnersItsOwnWay(unittest.TestCase):
self.assertIn('"file": "bios/a.bin", "emulator": "retroarch", "core": "beetle_psx" }', text)
self.assertIn('"file": "bios/b.bin", "emulator": "duckstation", "core": "duckstation-sa" }', text)
self.assertNotIn("c.bin", text)
if __name__ == "__main__":
unittest.main()
+6 -5
View File
@@ -480,14 +480,14 @@ class ReleaseAssetNames(unittest.TestCase):
with tempfile.TemporaryDirectory() as root:
Path(root, ".gitignore").write_text(gitignore, encoding="utf-8")
generate_pack._GITIGNORE_ENTRIES = None
generate_pack._registered.cache_clear()
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
generate_pack._registered.cache_clear()
self.assertEqual(manifest, expected)
sizes = {self.ET: 228138631, self.RTCW: 122757192, self.PUP: 10}
@@ -677,9 +677,6 @@ class GameDataHasOneDefinition(unittest.TestCase):
)
if __name__ == "__main__":
unittest.main()
class ASameSizeReuploadIsSeen(unittest.TestCase):
"""A DSi NAND rebuilt at the same 251658304 bytes and re-uploaded with
@@ -730,3 +727,7 @@ class ASameSizeReuploadIsSeen(unittest.TestCase):
),
{"x": "abc"},
)
if __name__ == "__main__":
unittest.main()