From b5fd643ccf38754fbf2e9b41fa98d988fb9a1f2f Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:49:54 +0200 Subject: [PATCH] refactor: give common.py's parts their own modules common.py had grown to 1833 lines by accumulation. Six coherent pieces move out - untrusted parsing, digests, archives, generated artefacts, release assets, dump catalogues - and common.py re-exports them, so the sixty existing import sites keep working and migrating them stays optional. The site build is now reproducible, which is what made the move checkable. It deleted its generated directories first, so every page was new and write_if_changed had no earlier version to compare against: a deploy republished six hundred pages for the clock alone. Directories are swept instead, a page is removed only once nothing produces it, and the body pass compares against the body of the file on disk rather than against the decorated page. Two consecutive builds on the same inputs now produce identical bytes; before, 1034 files differed. --- scripts/artifacts.py | 88 +++++ scripts/common.py | 610 +++------------------------------ scripts/dumpcatalog.py | 96 ++++++ scripts/generate_site.py | 85 ++++- scripts/hashing.py | 101 ++++++ scripts/largefiles.py | 106 ++++++ scripts/safeparse.py | 65 ++++ scripts/ziptools.py | 195 +++++++++++ tests/test_e2e.py | 7 +- tests/test_large_file_cache.py | 19 +- 10 files changed, 783 insertions(+), 589 deletions(-) create mode 100644 scripts/artifacts.py create mode 100644 scripts/dumpcatalog.py create mode 100644 scripts/hashing.py create mode 100644 scripts/largefiles.py create mode 100644 scripts/safeparse.py create mode 100644 scripts/ziptools.py diff --git a/scripts/artifacts.py b/scripts/artifacts.py new file mode 100644 index 00000000..0bc28eff --- /dev/null +++ b/scripts/artifacts.py @@ -0,0 +1,88 @@ +"""Writing generated files, and not writing them. + +A generated artefact carries a timestamp but must not be rewritten +when only the clock moved: the CI freshness guard is a git diff, and +it is only meaningful if the comparison ignores the hour.""" + +from __future__ import annotations + +import contextlib +import os +import re + + +_TIMESTAMP_PATTERNS = [ + re.compile(r'"generated_at":\s*"[^"]*"'), # database.json + re.compile(r'"imported_at":\s*"[^"]*"'), # provenance snapshots + re.compile(r"\*Auto-generated on [^*]*\*"), # README.md + re.compile(r"\*Generated on [^*]*\*"), # docs site pages + # The decorated pages carry the same stamp again as a rendered element, + # and missing it rewrote every page on every run for the clock alone. + re.compile(r'
[^<]*
'), +] + +def write_if_changed(path: str, content: str, normalize=None) -> bool: + """Write content to path only if the non-timestamp content differs. + + Compares new and existing content after stripping timestamp lines. + Returns True if the file was written, False if skipped (unchanged). + + A caller that writes a file in two passes -a body, then the same body + wrapped in front matter -passes ``normalize`` to reduce both sides to the + part it owns. Without it the second pass always sees a difference, the + file is rewritten, and the fresh timestamp defeats the comparison. + """ + if os.path.exists(path): + with open(path) as f: + existing = f.read() + before, after = ( + (normalize(existing), normalize(content)) + if normalize + else (existing, content) + ) + if _strip_timestamps(before) == _strip_timestamps(after): + return False + with open(path, "w") as f: + f.write(content) + return True + +def _strip_timestamps(text: str) -> str: + """Remove known timestamp patterns for content comparison.""" + result = text + for pattern in _TIMESTAMP_PATTERNS: + result = pattern.sub("", result) + return result + +class ArtifactLockBusy(RuntimeError): + """Raised when another process already holds the artifact directory.""" + +@contextlib.contextmanager +def artifact_lock(directory: str, exclusive: bool = True): + """Serialize access to a shared artifact directory across processes. + + Two pipeline runs building the same dist/ leave readers looking at + half-written ZIPs, which surfaces as BadZipFile far from its cause. + Writers take the lock exclusively, readers share it. On platforms + without flock the lock is a no-op. + """ + try: + import fcntl + except ImportError: + yield + return + + os.makedirs(directory, exist_ok=True) + lock_path = os.path.join(directory, ".lock") + mode = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH + with open(lock_path, "w") as handle: + try: + fcntl.flock(handle, mode | fcntl.LOCK_NB) + except OSError as exc: + raise ArtifactLockBusy( + f"{directory} is in use by another run " + f"(lock: {lock_path}). Wait for it to finish." + ) from exc + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) diff --git a/scripts/common.py b/scripts/common.py index 11fbdeb0..eb8cb8be 100644 --- a/scripts/common.py +++ b/scripts/common.py @@ -6,123 +6,24 @@ and file resolution - eliminates DRY violations across scripts. from __future__ import annotations -import contextlib import hashlib import json import os import re -import stat -import tempfile -import urllib.error -import urllib.parse -import urllib.request -import xml.etree.ElementTree as ET import zipfile -import zlib from pathlib import Path -def parse_untrusted_xml(content: str | bytes, label: str = "XML") -> ET.Element: - """Parse XML fetched from a third party. - - ElementTree expands internal entities, so a document that declares them - can make the parser build a payload far larger than the bytes downloaded. - Nothing this project reads (DAT packs, es_bios.xml, Emulators.xml) ever - declares one, so a declaration is grounds to refuse the document rather - than something to expand carefully. - - The check targets dict[str, str]: - """Compute file hashes. Pass *algorithms* to limit which are computed.""" - algos = algorithms or _ALL_ALGORITHMS - sha1 = hashlib.sha1() if "sha1" in algos else None - md5 = hashlib.md5() if "md5" in algos else None - sha256 = hashlib.sha256() if "sha256" in algos else None - do_crc = "crc32" in algos - do_adler = "adler32" in algos - crc = 0 - adler = 1 # zlib.adler32 initial value - with open(filepath, "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - if sha1: - sha1.update(chunk) - if md5: - md5.update(chunk) - if sha256: - sha256.update(chunk) - if do_crc: - crc = zlib.crc32(chunk, crc) - if do_adler: - adler = zlib.adler32(chunk, adler) - result: dict[str, str] = {} - if sha1: - result["sha1"] = sha1.hexdigest() - if md5: - result["md5"] = md5.hexdigest() - if sha256: - result["sha256"] = sha256.hexdigest() - if do_crc: - result["crc32"] = format(crc & 0xFFFFFFFF, "08x") - if do_adler: - result["adler32"] = format(adler & 0xFFFFFFFF, "08x") - return result def load_database(db_path: str) -> dict: @@ -131,20 +32,6 @@ def load_database(db_path: str) -> dict: return json.load(f) -def md5sum(source: str | Path | object) -> str: - """Compute MD5 of a file path or file-like object - matches Batocera's md5sum().""" - h = hashlib.md5() - if hasattr(source, "read"): - for chunk in iter(lambda: source.read(65536), b""): - h.update(chunk) - else: - with open(source, "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - h.update(chunk) - return h.hexdigest() - - -_md5_composite_cache: dict[str, str] = {} _casefold_index_cache: dict[int, dict[str, list[str]]] = {} @@ -186,40 +73,6 @@ def name_match_size_ok(file_entry: dict, candidate_size: int | None) -> bool: return not (high is not None and candidate_size > high) -def md5_composite(filepath: str | Path) -> str: - """Compute composite MD5 of a ZIP - matches Recalbox's Zip::Md5Composite(). - - Sorts filenames alphabetically, reads each file's contents in order, - feeds everything into a single MD5 hasher. The result is independent - of ZIP compression level or metadata. Results are cached per path. - """ - key = str(filepath) - cached = _md5_composite_cache.get(key) - if cached is not None: - return cached - with zipfile.ZipFile(filepath) as zf: - names = sorted(n for n in zf.namelist() if not n.endswith("/")) - h = hashlib.md5() - for name in names: - info = zf.getinfo(name) - if info.file_size > 512 * 1024 * 1024: - continue # skip oversized entries - h.update(zf.read(name)) - result = h.hexdigest() - _md5_composite_cache[key] = result - return result - - -def parse_md5_list(raw: str | list | None) -> list[str]: - """Normalize an md5 field into a lowercase list. - - Platform YAMLs carry Recalbox multi-hash as one comma-separated string, - emulator profiles carry a YAML list. Both reach here. - """ - if not raw: - return [] - values = raw if isinstance(raw, list) else str(raw).split(",") - return [str(m).strip().lower() for m in values if str(m).strip()] _shared_yml_cache: dict[str, dict] = {} @@ -923,67 +776,7 @@ def get_mame_clone_map() -> dict[str, str]: return _mame_clone_map_cache -def check_inside_zip(container: str, file_name: str, expected_md5: str) -> str: - """Check a ROM inside a ZIP -replicates Batocera checkInsideZip(). - Returns "ok", "untested", "not_in_zip", or "error". - """ - try: - with zipfile.ZipFile(container) as archive: - for fname in archive.namelist(): - if fname.casefold() == file_name.casefold(): - info = archive.getinfo(fname) - if info.file_size > 512 * 1024 * 1024: - return "error" - if expected_md5 == "": - return "ok" - with archive.open(fname) as entry: - actual = md5sum(entry) - return "ok" if actual == expected_md5 else "untested" - return "not_in_zip" - except (zipfile.BadZipFile, OSError, KeyError): - return "error" - - -_zip_contents_cache: tuple[frozenset[tuple[str, float]], dict] | None = None - - -def build_zip_contents_index(db: dict, max_entry_size: int = 512 * 1024 * 1024) -> dict: - """Build {inner_rom_md5: zip_file_sha1} for ROMs inside ZIP files. - - Results are cached in-process; repeated calls with unchanged ZIPs return - the cached index. - """ - global _zip_contents_cache - - # Build fingerprint from ZIP paths + mtimes for cache invalidation - zip_entries: list[tuple[str, str]] = [] - for sha1, entry in db.get("files", {}).items(): - path = entry["path"] - if path.endswith(".zip") and os.path.exists(path): - zip_entries.append((path, sha1)) - - fingerprint = frozenset((path, os.path.getmtime(path)) for path, _ in zip_entries) - if _zip_contents_cache is not None and _zip_contents_cache[0] == fingerprint: - return _zip_contents_cache[1] - - index: dict[str, str] = {} - for path, sha1 in zip_entries: - try: - with zipfile.ZipFile(path, "r") as zf: - for info in zf.infolist(): - if info.is_dir() or info.file_size > max_entry_size: - continue - h = hashlib.md5() - with zf.open(info.filename) as inner: - for chunk in iter(lambda: inner.read(65536), b""): - h.update(chunk) - index[h.hexdigest()] = sha1 - except (zipfile.BadZipFile, OSError): - continue - - _zip_contents_cache = (fingerprint, index) - return index _emulator_profiles_cache: dict[tuple[str, bool], dict[str, dict]] = {} @@ -1384,136 +1177,14 @@ def expand_platform_declared_names(config: dict, db: dict) -> set[str]: import re -_TIMESTAMP_PATTERNS = [ - re.compile(r'"generated_at":\s*"[^"]*"'), # database.json - re.compile(r'"imported_at":\s*"[^"]*"'), # provenance snapshots - re.compile(r"\*Auto-generated on [^*]*\*"), # README.md - re.compile(r"\*Generated on [^*]*\*"), # docs site pages - # The decorated pages carry the same stamp again as a rendered element, - # and missing it rewrote every page on every run for the clock alone. - re.compile(r'
[^<]*
'), -] -def write_if_changed(path: str, content: str) -> bool: - """Write content to path only if the non-timestamp content differs. - - Compares new and existing content after stripping timestamp lines. - Returns True if the file was written, False if skipped (unchanged). - """ - if os.path.exists(path): - with open(path) as f: - existing = f.read() - if _strip_timestamps(existing) == _strip_timestamps(content): - return False - with open(path, "w") as f: - f.write(content) - return True - - -def _strip_timestamps(text: str) -> str: - """Remove known timestamp patterns for content comparison.""" - result = text - for pattern in _TIMESTAMP_PATTERNS: - result = pattern.sub("", result) - return result - # Validation and mode filtering -extracted to validation.py for SoC. # Re-exported below for backward compatibility. -LARGE_FILES_RELEASE = "large-files" -LARGE_FILES_REPO = "Abdess/retrobios" -LARGE_FILES_CACHE = ".cache/large" - -def fetch_large_file( - name: str, - dest_dir: str = LARGE_FILES_CACHE, - expected_sha1: str = "", - expected_md5: str = "", - *, - offline: bool = False, -) -> str | None: - """Return a verified cached large file, downloading it only when allowed.""" - cached = os.path.join(dest_dir, name) - if os.path.exists(cached): - if expected_sha1 or expected_md5: - hashes = compute_hashes(cached) - if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): - os.unlink(cached) - elif expected_md5: - md5_list = [ - m.strip().lower() for m in expected_md5.split(",") if m.strip() - ] - if hashes["md5"].lower() not in md5_list: - os.unlink(cached) - else: - return cached - else: - return cached - else: - return cached - - if offline: - return None - - os.makedirs(dest_dir, exist_ok=True) - # A per-process scratch name: two runs fetching the same asset into one - # shared path interleave their writes into a full-size, corrupt file. - tmp_fd, tmp_path = tempfile.mkstemp( - dir=dest_dir, prefix=os.path.basename(cached) + ".", suffix=".tmp" - ) - os.close(tmp_fd) - # GitHub rewrites spaces to dots in release asset names, so a file whose - # name contains spaces is published under a dotted name. - candidates = [name] - if " " in name: - candidates.append(name.replace(" ", ".")) - - downloaded = False - for candidate in candidates: - encoded_name = urllib.parse.quote(candidate) - url = ( - f"https://github.com/{LARGE_FILES_REPO}/releases/download/" - f"{LARGE_FILES_RELEASE}/{encoded_name}" - ) - try: - req = urllib.request.Request(url, headers={"User-Agent": "retrobios/1.0"}) - with urllib.request.urlopen(req, timeout=300) as resp: - with open(tmp_path, "wb") as f: - while True: - chunk = resp.read(65536) - if not chunk: - break - f.write(chunk) - downloaded = True - break - except (urllib.error.URLError, urllib.error.HTTPError): - if os.path.exists(tmp_path): - os.unlink(tmp_path) - - if not downloaded: - if os.path.exists(tmp_path): - os.unlink(tmp_path) - return None - - if expected_sha1 or expected_md5: - hashes = compute_hashes(tmp_path) - if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): - os.unlink(tmp_path) - return None - if expected_md5: - md5_list = [m.strip().lower() for m in expected_md5.split(",") if m.strip()] - if hashes["md5"].lower() not in md5_list: - os.unlink(tmp_path) - return None - os.replace(tmp_path, cached) - return cached - - -MAX_ZIP_MEMBERS = 100_000 def sanitize_pack_path(raw: str) -> str: """Strip traversal components from a relative destination. @@ -1525,117 +1196,6 @@ def sanitize_pack_path(raw: str) -> str: return "/".join(p for p in raw.split("/") if p and p not in ("..", ".")) -MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024 -# The largest generated pack is already ~5 GB uncompressed and the collection -# only grows; this bounds a malicious archive without capping a real one. -MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024 -# DEFLATE cannot exceed roughly 1,032:1, so this rejects a declared ratio no -# real DEFLATE member can reach. Methods with a higher ceiling (bzip2, LZMA) -# are exempt and bounded by the per-member and per-archive size limits alone. -MAX_ZIP_COMPRESSION_RATIO = 1_100 -_BOUNDED_RATIO_METHODS = (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED) - - -def safe_extract_zip( - zip_path: str, - dest_dir: str, - *, - max_members: int = MAX_ZIP_MEMBERS, - max_member_size: int = MAX_ZIP_MEMBER_SIZE, - max_total_size: int = MAX_ZIP_TOTAL_SIZE, - max_compression_ratio: int = MAX_ZIP_COMPRESSION_RATIO, -) -> None: - """Extract a ZIP with traversal, link and resource-limit protection. - - Files are streamed to a temporary sibling and atomically installed only - after their declared length and CRC have been checked by ``zipfile``. - """ - dest = os.path.realpath(dest_dir) - os.makedirs(dest, exist_ok=True) - with zipfile.ZipFile(zip_path, "r") as zf: - members = zf.infolist() - if len(members) > max_members: - raise ValueError( - f"ZIP has {len(members)} members; limit is {max_members}" - ) - - declared_total = 0 - seen: set[str] = set() - for member in members: - # Archives written on Windows store a backslash separator. It is a - # separator, not a filename character, so it is normalized before - # the component checks rather than rejected. - name = member.filename.replace("\\", "/") - if not name or "\x00" in name: - raise ValueError(f"Unsafe ZIP member name: {member.filename!r}") - if name.startswith("/") or re.match(r"^[A-Za-z]:", name): - raise ValueError(f"Absolute ZIP member path: {name}") - parts = [part for part in name.split("/") if part] - if any(part in (".", "..") for part in parts): - raise ValueError(f"ZIP traversal detected: {name}") - normalized = "/".join(parts) - if normalized in seen: - raise ValueError(f"Duplicate ZIP member path: {name}") - seen.add(normalized) - - mode = (member.external_attr >> 16) & 0xFFFF - file_type = stat.S_IFMT(mode) - if file_type not in (0, stat.S_IFREG, stat.S_IFDIR): - raise ValueError(f"ZIP link or special file rejected: {name}") - if member.flag_bits & 0x1: - raise ValueError(f"Encrypted ZIP member rejected: {name}") - if member.file_size > max_member_size: - raise ValueError( - f"ZIP member {name} is {member.file_size} bytes; " - f"limit is {max_member_size}" - ) - declared_total += member.file_size - if declared_total > max_total_size: - raise ValueError( - f"ZIP expands to {declared_total} bytes; limit is {max_total_size}" - ) - if member.file_size and member.compress_type in _BOUNDED_RATIO_METHODS: - if member.compress_size == 0: - raise ValueError(f"Invalid compression size for ZIP member: {name}") - if member.file_size / member.compress_size > max_compression_ratio: - raise ValueError(f"Suspicious compression ratio for ZIP member: {name}") - - target = os.path.realpath(os.path.join(dest, *parts)) - if not target.startswith(dest + os.sep) and target != dest: - raise ValueError(f"ZIP traversal detected: {name}") - if member.is_dir() or name.endswith("/"): - os.makedirs(target, exist_ok=True) - continue - - os.makedirs(os.path.dirname(target), exist_ok=True) - tmp_path = "" - try: - with tempfile.NamedTemporaryFile( - mode="wb", dir=os.path.dirname(target), delete=False - ) as tmp_file: - tmp_path = tmp_file.name - actual_size = 0 - with zf.open(member, "r") as source: - while True: - chunk = source.read(1024 * 1024) - if not chunk: - break - actual_size += len(chunk) - if actual_size > member.file_size or actual_size > max_member_size: - raise ValueError( - f"ZIP member exceeded declared or configured size: {name}" - ) - tmp_file.write(chunk) - if actual_size != member.file_size: - raise ValueError( - f"ZIP member size mismatch for {name}: " - f"{actual_size} != {member.file_size}" - ) - os.replace(tmp_path, target) - tmp_path = "" - finally: - if tmp_path and os.path.exists(tmp_path): - os.unlink(tmp_path) def list_emulator_profiles(emulators_dir: str, skip_aliases: bool = True) -> None: @@ -1709,125 +1269,51 @@ def build_target_cores_cache( return cache, kept -DEFAULT_PROVENANCE_DIR = "provenance" - - -def load_provenance_snapshots(provenance_dir: str = DEFAULT_PROVENANCE_DIR) -> dict: - """Load dump-catalog snapshots from provenance/*.json. - - Returns {source_name: snapshot} where snapshot holds the normalized - entries written by the redump scraper or the pack importer. Missing - directory means no snapshots: returns an empty dict. - """ - snapshots = {} - prov_path = Path(provenance_dir) - if not prov_path.is_dir(): - return snapshots - for path in sorted(prov_path.glob("*.json")): - with open(path) as f: - snapshot = json.load(f) - source = snapshot.get("source") - if source and snapshot.get("entries"): - snapshots[source] = snapshot - return snapshots - - -def build_provenance_index(snapshots: dict) -> dict: - """Index snapshot entries by sha1 and by (md5, size) per source. - - First entry wins on hash collisions within a source; entries are - pre-sorted at snapshot write time so the outcome is deterministic. - """ - index = {} - for source, snapshot in snapshots.items(): - by_sha1 = {} - by_md5_size = {} - for entry in snapshot["entries"]: - sha1 = entry.get("sha1", "") - md5 = entry.get("md5", "") - if sha1 and sha1 not in by_sha1: - by_sha1[sha1] = entry - if md5 and entry.get("size"): - key = (md5, entry["size"]) - if key not in by_md5_size: - by_md5_size[key] = entry - index[source] = {"by_sha1": by_sha1, "by_md5_size": by_md5_size} - return index - - -def annotate_provenance(files: dict, snapshots: dict) -> dict[str, int]: - """Attach a provenance field to database file entries. - - Matches by SHA1 first, then MD5 + size. Returns per-source match - counts. Files without any catalog match keep no provenance field. - """ - index = build_provenance_index(snapshots) - counts = dict.fromkeys(index, 0) - for sha1, entry in files.items(): - matches = {} - for source in sorted(index): - src_index = index[source] - hit = src_index["by_sha1"].get(sha1) or src_index["by_md5_size"].get( - (entry.get("md5", ""), entry.get("size", 0)) - ) - if hit: - matches[source] = { - "dat": hit.get("dat", ""), - "name": hit.get("name", ""), - "description": hit.get("description", ""), - } - counts[source] += 1 - if matches: - entry["provenance"] = matches - else: - entry.pop("provenance", None) - return counts - - -def write_provenance_snapshot( - path: str, source: str, imported_at: str, dats: dict, entries: list[dict] -) -> bool: - """Write a normalized provenance snapshot, sorted for determinism.""" - snapshot = { - "source": source, - "imported_at": imported_at, - "dats": dict(sorted(dats.items())), - "entries": sorted(entries, key=lambda e: (e["dat"], e["name"])), - } - return write_if_changed(path, json.dumps(snapshot, indent=2) + "\n") - - -class ArtifactLockBusy(RuntimeError): - """Raised when another process already holds the artifact directory.""" - - -@contextlib.contextmanager -def artifact_lock(directory: str, exclusive: bool = True): - """Serialize access to a shared artifact directory across processes. - - Two pipeline runs building the same dist/ leave readers looking at - half-written ZIPs, which surfaces as BadZipFile far from its cause. - Writers take the lock exclusively, readers share it. On platforms - without flock the lock is a no-op. - """ - try: - import fcntl - except ImportError: - yield - return - - os.makedirs(directory, exist_ok=True) - lock_path = os.path.join(directory, ".lock") - mode = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH - with open(lock_path, "w") as handle: - try: - fcntl.flock(handle, mode | fcntl.LOCK_NB) - except OSError as exc: - raise ArtifactLockBusy( - f"{directory} is in use by another run " - f"(lock: {lock_path}). Wait for it to finish." - ) from exc - try: - yield - finally: - fcntl.flock(handle, fcntl.LOCK_UN) +# Re-exported so the existing call sites keep working while the +# modules above become the place to import from. +from safeparse import ( # noqa: E402,F401 + parse_untrusted_xml, + require_yaml, + yaml_load, + _pick_yaml_loader, + _YAML_LOADER, +) +from hashing import ( # noqa: E402,F401 + compute_hashes, + md5sum, + md5_composite, + parse_md5_list, + _ALL_ALGORITHMS, + _md5_composite_cache, +) +from ziptools import ( # noqa: E402,F401 + check_inside_zip, + build_zip_contents_index, + MAX_ZIP_MEMBERS, + MAX_ZIP_MEMBER_SIZE, + MAX_ZIP_TOTAL_SIZE, + MAX_ZIP_COMPRESSION_RATIO, + safe_extract_zip, + _zip_contents_cache, + _BOUNDED_RATIO_METHODS, +) +from artifacts import ( # noqa: E402,F401 + write_if_changed, + ArtifactLockBusy, + artifact_lock, + _TIMESTAMP_PATTERNS, + _strip_timestamps, +) +from largefiles import ( # noqa: E402,F401 + LARGE_FILES_RELEASE, + LARGE_FILES_REPO, + LARGE_FILES_CACHE, + fetch_large_file, +) +from dumpcatalog import ( # noqa: E402,F401 + DEFAULT_PROVENANCE_DIR, + load_provenance_snapshots, + build_provenance_index, + annotate_provenance, + write_provenance_snapshot, +) diff --git a/scripts/dumpcatalog.py b/scripts/dumpcatalog.py new file mode 100644 index 00000000..1211722e --- /dev/null +++ b/scripts/dumpcatalog.py @@ -0,0 +1,96 @@ +"""Dump catalogues joined onto the collection. + +Redump, No-Intro and TOSEC annotate what the repo holds. They are an +opinion on provenance, never an authority: the emulator source is.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from artifacts import write_if_changed + + +DEFAULT_PROVENANCE_DIR = "provenance" + +def load_provenance_snapshots(provenance_dir: str = DEFAULT_PROVENANCE_DIR) -> dict: + """Load dump-catalog snapshots from provenance/*.json. + + Returns {source_name: snapshot} where snapshot holds the normalized + entries written by the redump scraper or the pack importer. Missing + directory means no snapshots: returns an empty dict. + """ + snapshots = {} + prov_path = Path(provenance_dir) + if not prov_path.is_dir(): + return snapshots + for path in sorted(prov_path.glob("*.json")): + with open(path) as f: + snapshot = json.load(f) + source = snapshot.get("source") + if source and snapshot.get("entries"): + snapshots[source] = snapshot + return snapshots + +def build_provenance_index(snapshots: dict) -> dict: + """Index snapshot entries by sha1 and by (md5, size) per source. + + First entry wins on hash collisions within a source; entries are + pre-sorted at snapshot write time so the outcome is deterministic. + """ + index = {} + for source, snapshot in snapshots.items(): + by_sha1 = {} + by_md5_size = {} + for entry in snapshot["entries"]: + sha1 = entry.get("sha1", "") + md5 = entry.get("md5", "") + if sha1 and sha1 not in by_sha1: + by_sha1[sha1] = entry + if md5 and entry.get("size"): + key = (md5, entry["size"]) + if key not in by_md5_size: + by_md5_size[key] = entry + index[source] = {"by_sha1": by_sha1, "by_md5_size": by_md5_size} + return index + +def annotate_provenance(files: dict, snapshots: dict) -> dict[str, int]: + """Attach a provenance field to database file entries. + + Matches by SHA1 first, then MD5 + size. Returns per-source match + counts. Files without any catalog match keep no provenance field. + """ + index = build_provenance_index(snapshots) + counts = dict.fromkeys(index, 0) + for sha1, entry in files.items(): + matches = {} + for source in sorted(index): + src_index = index[source] + hit = src_index["by_sha1"].get(sha1) or src_index["by_md5_size"].get( + (entry.get("md5", ""), entry.get("size", 0)) + ) + if hit: + matches[source] = { + "dat": hit.get("dat", ""), + "name": hit.get("name", ""), + "description": hit.get("description", ""), + } + counts[source] += 1 + if matches: + entry["provenance"] = matches + else: + entry.pop("provenance", None) + return counts + +def write_provenance_snapshot( + path: str, source: str, imported_at: str, dats: dict, entries: list[dict] +) -> bool: + """Write a normalized provenance snapshot, sorted for determinism.""" + snapshot = { + "source": source, + "imported_at": imported_at, + "dats": dict(sorted(dats.items())), + "entries": sorted(entries, key=lambda e: (e["dat"], e["name"])), + } + return write_if_changed(path, json.dumps(snapshot, indent=2) + "\n") diff --git a/scripts/generate_site.py b/scripts/generate_site.py index e992349d..f0576227 100644 --- a/scripts/generate_site.py +++ b/scripts/generate_site.py @@ -39,11 +39,70 @@ from common import ( parse_md5_list, require_yaml, unique_emulator_profiles, - write_if_changed, + write_if_changed as _write_artifact, yaml_load, ) from nativemode import reads_file_contents +# Every path this run produced. The generated directories used to be deleted +# before the build, which made every page new: write_if_changed had no earlier +# version to compare against, so the whole site was rewritten for the clock +# alone and a deploy republished 600 unchanged pages. They are swept instead, +# and a page is only removed once nothing produced it. +_produced: set[str] = set() + + +def _record(path) -> None: + _produced.add(os.path.realpath(str(path))) + + +def _undecorated(markdown: str) -> str: + """Return a page body without the front matter the decoration pass adds.""" + if markdown.startswith("---\n") and "generated_by: retrobios-site" in markdown[:300]: + end = markdown.find("\n---\n", 4) + if end != -1: + markdown = markdown[end + 5:].lstrip("\n") + script_end = markdown.find("\n\n") + if markdown.startswith('\n\n"):] + return markdown + + +def write_if_changed(path: str, content: str) -> bool: + """Write a page body only when it moved, and remember it either way. + + A page is written twice: the body here, then the same body wrapped in + front matter by the decoration pass. This comparison therefore reduces + the file on disk to its body. The decoration pass compares in full - + normalizing there too would make its own write look like a no-op and + every page would lose its front matter. + """ + _record(path) + return _write_artifact(path, content, normalize=_undecorated) + + +def write_decorated(path, content: str) -> bool: + """Write the wrapped page, comparing front matter and body together.""" + _record(path) + return _write_artifact(str(path), content) + + +def _sweep_generated(docs: Path, directories: list[str]) -> int: + """Delete files in the generated tree that this run did not produce.""" + removed = 0 + for name in directories: + root = docs / name + if not root.is_dir(): + continue + for path in sorted(root.rglob("*")): + if path.is_file() and os.path.realpath(str(path)) not in _produced: + path.unlink() + removed += 1 + for path in sorted(root.rglob("*"), reverse=True): + if path.is_dir() and not any(path.iterdir()): + path.rmdir() + return removed + yaml = require_yaml() from generate_readme import compute_coverage, manifest_totals from profile_sync import source_ref_values, split_source_ref @@ -1134,6 +1193,7 @@ def generate_data_exports( ) for schema_name in schema_names: shutil.copy2(Path("schemas") / schema_name, schemas_dest / schema_name) + _record(schemas_dest / schema_name) file_rows = [] for sha1, entry in sorted(db.get("files", {}).items()): @@ -1197,6 +1257,7 @@ def generate_data_exports( downloads / "retrobios.sqlite", db, platform_items, platform_files, emulator_items, gap_rows, ) + _record(downloads / "retrobios.sqlite") assets = [ (api / "database.json", "Content database", "application/json", "schemas/database.schema.json"), @@ -1386,13 +1447,9 @@ def decorate_markdown_pages(docs: Path) -> None: if relative.parts and relative.parts[0] == "superpowers": continue markdown = path.read_text(encoding="utf-8") - if markdown.startswith("---\n") and "generated_by: retrobios-site" in markdown[:300]: - end = markdown.find("\n---\n", 4) - if end != -1: - markdown = markdown[end + 5:].lstrip("\n") - script_end = markdown.find("\n\n") - if markdown.startswith('\n\n"):] + stripped = _undecorated(markdown) + if stripped is not markdown: + markdown = stripped elif markdown.startswith("---\n"): continue @@ -1432,7 +1489,7 @@ def decorate_markdown_pages(docs: Path) -> None: f"{structured_json}\n" "\n\n" ) - write_if_changed(str(path), front_matter + markdown) + write_decorated(path, front_matter + markdown) # Platform pages @@ -3881,12 +3938,6 @@ def main(): db = load_database(args.db) docs = Path(args.docs_dir) - # Clean generated dirs (preserve docs/superpowers/) - for d in GENERATED_DIRS: - target = docs / d - if target.exists(): - shutil.rmtree(target) - # Ensure output dirs for d in GENERATED_DIRS: (docs / d).mkdir(parents=True, exist_ok=True) @@ -4059,6 +4110,7 @@ def main(): if wiki_src.is_dir(): for src_file in wiki_src.glob("*.md"): shutil.copy2(src_file, wiki_dest / src_file.name) + _record(wiki_dest / src_file.name) # data-model.md is generated (contains live DB stats) write_if_changed( str(wiki_dest / "data-model.md"), generate_wiki_data_model(db, profiles) @@ -4208,7 +4260,10 @@ validation: + 1 # contributing + 1 # data and API ) + stale = _sweep_generated(docs, GENERATED_DIRS) print(f"\nGenerated {total_pages} pages in {args.docs_dir}/") + if stale: + print(f"Removed {stale} page(s) nothing produced this run") if __name__ == "__main__": diff --git a/scripts/hashing.py b/scripts/hashing.py new file mode 100644 index 00000000..f1b95758 --- /dev/null +++ b/scripts/hashing.py @@ -0,0 +1,101 @@ +"""Content digests. + +SHA-1 is the collection's primary key; the rest are what the various +frontends check against.""" + +from __future__ import annotations + +import hashlib +import os +import zipfile +import zlib +from pathlib import Path + + +_ALL_ALGORITHMS = frozenset({"sha1", "md5", "sha256", "crc32", "adler32"}) + +def compute_hashes( + filepath: str | Path, + algorithms: frozenset[str] | None = None, +) -> dict[str, str]: + """Compute file hashes. Pass *algorithms* to limit which are computed.""" + algos = algorithms or _ALL_ALGORITHMS + sha1 = hashlib.sha1() if "sha1" in algos else None + md5 = hashlib.md5() if "md5" in algos else None + sha256 = hashlib.sha256() if "sha256" in algos else None + do_crc = "crc32" in algos + do_adler = "adler32" in algos + crc = 0 + adler = 1 # zlib.adler32 initial value + with open(filepath, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + if sha1: + sha1.update(chunk) + if md5: + md5.update(chunk) + if sha256: + sha256.update(chunk) + if do_crc: + crc = zlib.crc32(chunk, crc) + if do_adler: + adler = zlib.adler32(chunk, adler) + result: dict[str, str] = {} + if sha1: + result["sha1"] = sha1.hexdigest() + if md5: + result["md5"] = md5.hexdigest() + if sha256: + result["sha256"] = sha256.hexdigest() + if do_crc: + result["crc32"] = format(crc & 0xFFFFFFFF, "08x") + if do_adler: + result["adler32"] = format(adler & 0xFFFFFFFF, "08x") + return result + +def md5sum(source: str | Path | object) -> str: + """Compute MD5 of a file path or file-like object - matches Batocera's md5sum().""" + h = hashlib.md5() + if hasattr(source, "read"): + for chunk in iter(lambda: source.read(65536), b""): + h.update(chunk) + else: + with open(source, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + +_md5_composite_cache: dict[str, str] = {} + +def md5_composite(filepath: str | Path) -> str: + """Compute composite MD5 of a ZIP - matches Recalbox's Zip::Md5Composite(). + + Sorts filenames alphabetically, reads each file's contents in order, + feeds everything into a single MD5 hasher. The result is independent + of ZIP compression level or metadata. Results are cached per path. + """ + key = str(filepath) + cached = _md5_composite_cache.get(key) + if cached is not None: + return cached + with zipfile.ZipFile(filepath) as zf: + names = sorted(n for n in zf.namelist() if not n.endswith("/")) + h = hashlib.md5() + for name in names: + info = zf.getinfo(name) + if info.file_size > 512 * 1024 * 1024: + continue # skip oversized entries + h.update(zf.read(name)) + result = h.hexdigest() + _md5_composite_cache[key] = result + return result + +def parse_md5_list(raw: str | list | None) -> list[str]: + """Normalize an md5 field into a lowercase list. + + Platform YAMLs carry Recalbox multi-hash as one comma-separated string, + emulator profiles carry a YAML list. Both reach here. + """ + if not raw: + return [] + values = raw if isinstance(raw, list) else str(raw).split(",") + return [str(m).strip().lower() for m in values if str(m).strip()] diff --git a/scripts/largefiles.py b/scripts/largefiles.py new file mode 100644 index 00000000..348d65a4 --- /dev/null +++ b/scripts/largefiles.py @@ -0,0 +1,106 @@ +"""Files too large for the repository. + +They live as release assets and are fetched at build time, verified +against the hash the caller declares.""" + +from __future__ import annotations + +import os +import tempfile +import urllib.error +import urllib.request + +from hashing import compute_hashes + + +LARGE_FILES_RELEASE = "large-files" + +LARGE_FILES_RELEASE = "large-files" +LARGE_FILES_REPO = "Abdess/retrobios" + +LARGE_FILES_REPO = "Abdess/retrobios" +LARGE_FILES_CACHE = ".cache/large" + +def fetch_large_file( + name: str, + dest_dir: str = LARGE_FILES_CACHE, + expected_sha1: str = "", + expected_md5: str = "", + *, + offline: bool = False, +) -> str | None: + """Return a verified cached large file, downloading it only when allowed.""" + cached = os.path.join(dest_dir, name) + if os.path.exists(cached): + if expected_sha1 or expected_md5: + hashes = compute_hashes(cached) + if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): + os.unlink(cached) + elif expected_md5: + md5_list = [ + m.strip().lower() for m in expected_md5.split(",") if m.strip() + ] + if hashes["md5"].lower() not in md5_list: + os.unlink(cached) + else: + return cached + else: + return cached + else: + return cached + + if offline: + return None + + os.makedirs(dest_dir, exist_ok=True) + # A per-process scratch name: two runs fetching the same asset into one + # shared path interleave their writes into a full-size, corrupt file. + tmp_fd, tmp_path = tempfile.mkstemp( + dir=dest_dir, prefix=os.path.basename(cached) + ".", suffix=".tmp" + ) + os.close(tmp_fd) + # GitHub rewrites spaces to dots in release asset names, so a file whose + # name contains spaces is published under a dotted name. + candidates = [name] + if " " in name: + candidates.append(name.replace(" ", ".")) + + downloaded = False + for candidate in candidates: + encoded_name = urllib.parse.quote(candidate) + url = ( + f"https://github.com/{LARGE_FILES_REPO}/releases/download/" + f"{LARGE_FILES_RELEASE}/{encoded_name}" + ) + try: + req = urllib.request.Request(url, headers={"User-Agent": "retrobios/1.0"}) + with urllib.request.urlopen(req, timeout=300) as resp: + with open(tmp_path, "wb") as f: + while True: + chunk = resp.read(65536) + if not chunk: + break + f.write(chunk) + downloaded = True + break + except (urllib.error.URLError, urllib.error.HTTPError): + if os.path.exists(tmp_path): + os.unlink(tmp_path) + + if not downloaded: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + return None + + if expected_sha1 or expected_md5: + hashes = compute_hashes(tmp_path) + if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): + os.unlink(tmp_path) + return None + if expected_md5: + md5_list = [m.strip().lower() for m in expected_md5.split(",") if m.strip()] + if hashes["md5"].lower() not in md5_list: + os.unlink(tmp_path) + return None + os.replace(tmp_path, cached) + return cached diff --git a/scripts/safeparse.py b/scripts/safeparse.py new file mode 100644 index 00000000..c4aa9810 --- /dev/null +++ b/scripts/safeparse.py @@ -0,0 +1,65 @@ +"""Parsers for input the repo does not control. + +Upstream DATs and scraped documents are data, never instructions, and +the parsers here are the boundary that keeps it that way.""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET + +try: + import yaml +except ImportError: # optional at import time + yaml = None + + +def parse_untrusted_xml(content: str | bytes, label: str = "XML") -> ET.Element: + """Parse XML fetched from a third party. + + ElementTree expands internal entities, so a document that declares them + can make the parser build a payload far larger than the bytes downloaded. + Nothing this project reads (DAT packs, es_bios.xml, Emulators.xml) ever + declares one, so a declaration is grounds to refuse the document rather + than something to expand carefully. + + The check targets str: + """Check a ROM inside a ZIP -replicates Batocera checkInsideZip(). + + Returns "ok", "untested", "not_in_zip", or "error". + """ + try: + with zipfile.ZipFile(container) as archive: + for fname in archive.namelist(): + if fname.casefold() == file_name.casefold(): + info = archive.getinfo(fname) + if info.file_size > 512 * 1024 * 1024: + return "error" + if expected_md5 == "": + return "ok" + with archive.open(fname) as entry: + actual = md5sum(entry) + return "ok" if actual == expected_md5 else "untested" + return "not_in_zip" + except (zipfile.BadZipFile, OSError, KeyError): + return "error" + +_zip_contents_cache: tuple[frozenset[tuple[str, float]], dict] | None = None + +def build_zip_contents_index(db: dict, max_entry_size: int = 512 * 1024 * 1024) -> dict: + """Build {inner_rom_md5: zip_file_sha1} for ROMs inside ZIP files. + + Results are cached in-process; repeated calls with unchanged ZIPs return + the cached index. + """ + global _zip_contents_cache + + # Build fingerprint from ZIP paths + mtimes for cache invalidation + zip_entries: list[tuple[str, str]] = [] + for sha1, entry in db.get("files", {}).items(): + path = entry["path"] + if path.endswith(".zip") and os.path.exists(path): + zip_entries.append((path, sha1)) + + fingerprint = frozenset((path, os.path.getmtime(path)) for path, _ in zip_entries) + if _zip_contents_cache is not None and _zip_contents_cache[0] == fingerprint: + return _zip_contents_cache[1] + + index: dict[str, str] = {} + for path, sha1 in zip_entries: + try: + with zipfile.ZipFile(path, "r") as zf: + for info in zf.infolist(): + if info.is_dir() or info.file_size > max_entry_size: + continue + h = hashlib.md5() + with zf.open(info.filename) as inner: + for chunk in iter(lambda: inner.read(65536), b""): + h.update(chunk) + index[h.hexdigest()] = sha1 + except (zipfile.BadZipFile, OSError): + continue + + _zip_contents_cache = (fingerprint, index) + return index + +MAX_ZIP_MEMBERS = 100_000 + +MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024 + +MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024 +# The largest generated pack is already ~5 GB uncompressed and the collection +# only grows; this bounds a malicious archive without capping a real one. +MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024 + +MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024 +# DEFLATE cannot exceed roughly 1,032:1, so this rejects a declared ratio no +# real DEFLATE member can reach. Methods with a higher ceiling (bzip2, LZMA) +# are exempt and bounded by the per-member and per-archive size limits alone. +MAX_ZIP_COMPRESSION_RATIO = 1_100 + +MAX_ZIP_COMPRESSION_RATIO = 1_100 +_BOUNDED_RATIO_METHODS = (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED) + +def safe_extract_zip( + zip_path: str, + dest_dir: str, + *, + max_members: int = MAX_ZIP_MEMBERS, + max_member_size: int = MAX_ZIP_MEMBER_SIZE, + max_total_size: int = MAX_ZIP_TOTAL_SIZE, + max_compression_ratio: int = MAX_ZIP_COMPRESSION_RATIO, +) -> None: + """Extract a ZIP with traversal, link and resource-limit protection. + + Files are streamed to a temporary sibling and atomically installed only + after their declared length and CRC have been checked by ``zipfile``. + """ + dest = os.path.realpath(dest_dir) + os.makedirs(dest, exist_ok=True) + with zipfile.ZipFile(zip_path, "r") as zf: + members = zf.infolist() + if len(members) > max_members: + raise ValueError( + f"ZIP has {len(members)} members; limit is {max_members}" + ) + + declared_total = 0 + seen: set[str] = set() + for member in members: + # Archives written on Windows store a backslash separator. It is a + # separator, not a filename character, so it is normalized before + # the component checks rather than rejected. + name = member.filename.replace("\\", "/") + if not name or "\x00" in name: + raise ValueError(f"Unsafe ZIP member name: {member.filename!r}") + if name.startswith("/") or re.match(r"^[A-Za-z]:", name): + raise ValueError(f"Absolute ZIP member path: {name}") + parts = [part for part in name.split("/") if part] + if any(part in (".", "..") for part in parts): + raise ValueError(f"ZIP traversal detected: {name}") + normalized = "/".join(parts) + if normalized in seen: + raise ValueError(f"Duplicate ZIP member path: {name}") + seen.add(normalized) + + mode = (member.external_attr >> 16) & 0xFFFF + file_type = stat.S_IFMT(mode) + if file_type not in (0, stat.S_IFREG, stat.S_IFDIR): + raise ValueError(f"ZIP link or special file rejected: {name}") + if member.flag_bits & 0x1: + raise ValueError(f"Encrypted ZIP member rejected: {name}") + if member.file_size > max_member_size: + raise ValueError( + f"ZIP member {name} is {member.file_size} bytes; " + f"limit is {max_member_size}" + ) + declared_total += member.file_size + if declared_total > max_total_size: + raise ValueError( + f"ZIP expands to {declared_total} bytes; limit is {max_total_size}" + ) + if member.file_size and member.compress_type in _BOUNDED_RATIO_METHODS: + if member.compress_size == 0: + raise ValueError(f"Invalid compression size for ZIP member: {name}") + if member.file_size / member.compress_size > max_compression_ratio: + raise ValueError(f"Suspicious compression ratio for ZIP member: {name}") + + target = os.path.realpath(os.path.join(dest, *parts)) + if not target.startswith(dest + os.sep) and target != dest: + raise ValueError(f"ZIP traversal detected: {name}") + if member.is_dir() or name.endswith("/"): + os.makedirs(target, exist_ok=True) + continue + + os.makedirs(os.path.dirname(target), exist_ok=True) + tmp_path = "" + try: + with tempfile.NamedTemporaryFile( + mode="wb", dir=os.path.dirname(target), delete=False + ) as tmp_file: + tmp_path = tmp_file.name + actual_size = 0 + with zf.open(member, "r") as source: + while True: + chunk = source.read(1024 * 1024) + if not chunk: + break + actual_size += len(chunk) + if actual_size > member.file_size or actual_size > max_member_size: + raise ValueError( + f"ZIP member exceeded declared or configured size: {name}" + ) + tmp_file.write(chunk) + if actual_size != member.file_size: + raise ValueError( + f"ZIP member size mismatch for {name}: " + f"{actual_size} != {member.file_size}" + ) + os.replace(tmp_path, target) + tmp_path = "" + finally: + if tmp_path and os.path.exists(tmp_path): + os.unlink(tmp_path) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 6fdf22ad..b70bfb56 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -28,6 +28,7 @@ from pathlib import Path sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts")) +import largefiles import yaml from common import ( build_zip_contents_index, @@ -5120,8 +5121,8 @@ struct BurnDriver BurnDrvneogeo = { raise urllib.error.HTTPError(url, 404, "Not Found", {}, None) return FakeResponse(b"payload") - original = common.urllib.request.urlopen - common.urllib.request.urlopen = fake_urlopen + original = largefiles.urllib.request.urlopen + largefiles.urllib.request.urlopen = fake_urlopen try: with tempfile.TemporaryDirectory() as tmpdir: path = common.fetch_large_file( @@ -5131,7 +5132,7 @@ struct BurnDriver BurnDrvneogeo = { with open(path, "rb") as fh: self.assertEqual(fh.read(), b"payload") finally: - common.urllib.request.urlopen = original + largefiles.urllib.request.urlopen = original self.assertEqual(len(attempts), 2, attempts) self.assertIn("MAME%200.174", attempts[0]) diff --git a/tests/test_large_file_cache.py b/tests/test_large_file_cache.py index c0b1998c..e0f3b84e 100644 --- a/tests/test_large_file_cache.py +++ b/tests/test_large_file_cache.py @@ -23,6 +23,7 @@ REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "scripts")) import common # noqa: E402 +import largefiles PAYLOAD_A = b"A" * (256 * 1024) PAYLOAD_B = b"B" * (256 * 1024) @@ -54,10 +55,10 @@ class _SlowResponse(io.BytesIO): class LargeFileCacheTest(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp() - self._urlopen = common.urllib.request.urlopen + self._urlopen = largefiles.urllib.request.urlopen def tearDown(self): - common.urllib.request.urlopen = self._urlopen + largefiles.urllib.request.urlopen = self._urlopen def test_concurrent_fetches_do_not_mix(self): barrier = threading.Barrier(2, timeout=10) @@ -70,7 +71,7 @@ class LargeFileCacheTest(unittest.TestCase): i = next(index) return _SlowResponse(payloads[i], barrier) - common.urllib.request.urlopen = fake_urlopen + largefiles.urllib.request.urlopen = fake_urlopen results: list[str | None] = [None, None] def worker(slot: int): @@ -90,7 +91,7 @@ class LargeFileCacheTest(unittest.TestCase): self.assertIn(digest, accepted) def test_no_scratch_file_survives_a_successful_fetch(self): - common.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( + largefiles.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( PAYLOAD_A ) common.fetch_large_file("asset.bin", dest_dir=self.dir) @@ -101,12 +102,12 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise urllib.error.URLError("offline") - common.urllib.request.urlopen = fail + 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): - common.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( + largefiles.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( PAYLOAD_A ) result = common.fetch_large_file( @@ -122,7 +123,7 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise AssertionError("must not download when the cache is valid") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertEqual( common.fetch_large_file("asset.bin", dest_dir=self.dir), str(cached) ) @@ -131,7 +132,7 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise AssertionError("offline mode must not open the network") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertIsNone( common.fetch_large_file( "asset.bin", dest_dir=self.dir, offline=True @@ -146,7 +147,7 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise AssertionError("offline cache hit must not open the network") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertEqual( common.fetch_large_file( "asset.bin",