diff --git a/scripts/download.py b/scripts/download.py
index cfaf3692..1a978c00 100644
--- a/scripts/download.py
+++ b/scripts/download.py
@@ -3,11 +3,13 @@
Cross-platform tool (Linux/macOS/Windows) using only Python stdlib.
+A pack over 2 GB is published as numbered volumes (`.zip.001`, `.zip.002`),
+so a platform is a group of assets rather than a single one.
+
Usage:
python scripts/download.py --list # List platforms
python scripts/download.py retroarch ~/path/ # Download pack
- python scripts/download.py --verify retroarch ~/path # Verify local files
- python scripts/download.py --info retroarch # Show coverage info
+ python scripts/download.py --info retroarch # Show pack info
"""
from __future__ import annotations
@@ -15,22 +17,63 @@ from __future__ import annotations
import argparse
import json
import os
+import re
+import shutil
import sys
import urllib.error
+import urllib.parse
import urllib.request
-import zipfile
+from dataclasses import dataclass
from pathlib import Path
sys.path.insert(0, os.path.dirname(__file__))
from common import compute_hashes, safe_extract_zip
-GITHUB_API = "https://api.github.com"
+DEFAULT_API = "https://api.github.com"
REPO = "Abdess/retrobios"
+PACK_SUFFIX = "_BIOS_Pack.zip"
+CHECKSUMS_ASSET = "SHA256SUMS.txt"
+STAGING_DIR = ".retrobios-download"
+MAX_CHECKSUMS_BYTES = 1 << 20
+CHUNK = 1 << 20
+
+_VOLUME_RE = re.compile(rf"^(?P.+{re.escape(PACK_SUFFIX)})\.(?P\d+)$")
+_LOOPBACK_HOSTS = {"127.0.0.1", "::1", "localhost"}
+
+
+def _checked_url(value: str, label: str) -> str:
+ """Refuse a URL that is neither HTTPS nor loopback.
+
+ This endpoint names the assets and, through them, the bytes that land in
+ the BIOS directory, so it is the trust anchor of the whole download.
+ Plain HTTP to loopback stays allowed: it is how this is tested end to end.
+ """
+ parsed = urllib.parse.urlparse(value)
+ if parsed.scheme == "https":
+ return value.rstrip("/")
+ host = (parsed.hostname or "").lower()
+ if parsed.scheme == "http" and host in _LOOPBACK_HOSTS:
+ return value.rstrip("/")
+ print(f"Error: {label} must use HTTPS, got {value!r}", file=sys.stderr)
+ sys.exit(1)
+
+
+API = _checked_url(os.environ.get("RETROBIOS_API", DEFAULT_API), "RETROBIOS_API")
+
+
+@dataclass(frozen=True)
+class Pack:
+ """A platform pack: one asset, or the volumes it was split into."""
+
+ name: str
+ platform: str
+ parts: tuple[dict, ...]
+ size: int
def get_latest_release() -> dict:
"""Fetch latest release info from GitHub API."""
- url = f"{GITHUB_API}/repos/{REPO}/releases/latest"
+ url = f"{API}/repos/{REPO}/releases/latest"
req = urllib.request.Request(
url,
headers={
@@ -49,31 +92,75 @@ def get_latest_release() -> dict:
raise
-def list_platforms(release: dict) -> list[str]:
- """List available platform packs from release assets."""
- platforms = []
+def group_packs(release: dict) -> list[Pack]:
+ """Group release assets into packs, volumes folded into their archive."""
+ groups: dict[str, list[tuple[int, dict]]] = {}
for asset in release.get("assets", []):
name = asset["name"]
- if name.endswith("_BIOS_Pack.zip"):
- platform = name.replace("_BIOS_Pack.zip", "").replace("_", " ")
- platforms.append(platform)
- return sorted(platforms)
+ volume = _VOLUME_RE.match(name)
+ if volume:
+ groups.setdefault(volume["base"], []).append((int(volume["index"]), asset))
+ elif name.endswith(PACK_SUFFIX):
+ groups.setdefault(name, []).append((0, asset))
+
+ packs = []
+ for base, volumes in sorted(groups.items()):
+ parts = tuple(asset for _, asset in sorted(volumes, key=lambda v: v[0]))
+ packs.append(
+ Pack(
+ name=base,
+ platform=base[: -len(PACK_SUFFIX)].replace("_", " "),
+ parts=parts,
+ size=sum(part.get("size", 0) for part in parts),
+ )
+ )
+ return packs
-def find_asset(release: dict, platform: str) -> dict | None:
- """Find the release asset for a specific platform."""
- normalized = platform.lower().replace(" ", "_").replace("-", "_")
+def list_platforms(release: dict) -> list[str]:
+ """List available platform packs from release assets."""
+ return sorted(pack.platform for pack in group_packs(release))
- for asset in release.get("assets", []):
- asset_name = asset["name"].lower().replace(" ", "_").replace("-", "_")
- if normalized in asset_name and asset_name.endswith("_bios_pack.zip"):
- return asset
+def _match_key(value: str) -> str:
+ """Letters and digits only: the platform is `misterfpga`, the asset MiSTer_FPGA."""
+ return re.sub(r"[^a-z0-9]", "", value.lower())
+
+
+def find_pack(release: dict, platform: str) -> Pack | None:
+ """Find the pack for a platform name."""
+ needle = _match_key(platform)
+ if not needle:
+ return None
+ for pack in group_packs(release):
+ if needle in _match_key(pack.name):
+ return pack
return None
-def download_file(url: str, dest: str, expected_size: int = 0):
+def fetch_checksums(release: dict) -> dict[str, str]:
+ """Read the published SHA-256 of each pack, keyed by archive name."""
+ for asset in release.get("assets", []):
+ if asset["name"] != CHECKSUMS_ASSET:
+ continue
+ url = _checked_url(asset["browser_download_url"], "asset URL")
+ req = urllib.request.Request(
+ url, headers={"User-Agent": "retrobios-downloader/1.0"}
+ )
+ with urllib.request.urlopen(req, timeout=60) as resp:
+ body = resp.read(MAX_CHECKSUMS_BYTES).decode("utf-8", "replace")
+ sums = {}
+ for line in body.splitlines():
+ digest, _, name = line.strip().partition(" ")
+ if digest and name:
+ sums[name] = digest.lower()
+ return sums
+ return {}
+
+
+def download_file(url: str, dest: Path, expected_size: int = 0, label: str = ""):
"""Download a file with progress indication."""
+ url = _checked_url(url, "asset URL")
req = urllib.request.Request(
url, headers={"User-Agent": "retrobios-downloader/1.0"}
)
@@ -94,95 +181,76 @@ def download_file(url: str, dest: str, expected_size: int = 0):
pct = downloaded * 100 // total
bar = "=" * (pct // 2) + " " * (50 - pct // 2)
print(
- f"\r [{bar}] {pct}% ({downloaded:,}/{total:,})",
+ f"\r {label}[{bar}] {pct}% ({downloaded:,}/{total:,})",
end="",
flush=True,
)
print()
+ if expected_size and downloaded != expected_size:
+ raise ValueError(f"downloaded {downloaded} bytes; expected {expected_size}")
-def extract_pack(zip_path: str, dest_dir: str):
- """Extract a BIOS pack ZIP to destination."""
- with zipfile.ZipFile(zip_path, "r") as zf:
- members = zf.namelist()
- print(f" Extracting {len(members)} files to {dest_dir}/")
- safe_extract_zip(zip_path, dest_dir)
+def join_volumes(parts: list[Path], dest: Path) -> None:
+ """Concatenate split volumes into one archive, streaming."""
+ with open(dest, "wb") as out:
+ for part in parts:
+ with open(part, "rb") as src:
+ shutil.copyfileobj(src, out, CHUNK)
-def verify_files(platform: str, dest_dir: str, release: dict):
- """Verify local files against database.json from release."""
- db_asset = None
- for asset in release.get("assets", []):
- if asset["name"] == "database.json":
- db_asset = asset
- break
+def fetch_pack(pack: Pack, staging: Path, checksums: dict[str, str]) -> Path:
+ """Download every volume of a pack and return the assembled archive."""
+ staging.mkdir(parents=True, exist_ok=True)
+ volumes = []
+ for index, part in enumerate(pack.parts, start=1):
+ label = f"part {index}/{len(pack.parts)} " if len(pack.parts) > 1 else ""
+ target = staging / part["name"]
+ print(f"Downloading {part['name']} ({part.get('size', 0):,} bytes)...")
+ download_file(part["browser_download_url"], target, part.get("size", 0), label)
+ volumes.append(target)
- if not db_asset:
- print("No database.json found in release assets. Cannot verify.")
- return
+ archive = staging / pack.name
+ if volumes != [archive]:
+ if len(volumes) > 1:
+ print(f"Joining {len(volumes)} parts into {pack.name}...")
+ join_volumes(volumes, archive)
+ for volume in volumes:
+ volume.unlink()
+ else:
+ volumes[0].replace(archive)
- import tempfile
-
- tmp = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
- tmp.close()
-
- try:
- download_file(
- db_asset["browser_download_url"], tmp.name, db_asset.get("size", 0)
- )
- with open(tmp.name) as f:
- db = json.load(f)
- finally:
- os.unlink(tmp.name)
-
- dest = Path(dest_dir)
- verified = 0
- missing = 0
- mismatched = 0
-
- for sha1, entry in db.get("files", {}).items():
- name = entry["name"]
- found = False
- for local_file in dest.rglob(name):
- if local_file.is_file():
- local_sha1 = compute_hashes(local_file)["sha1"]
-
- if local_sha1 == sha1:
- verified += 1
- found = True
- break
- else:
- mismatched += 1
- print(
- f" MISMATCH: {name} (expected {sha1[:12]}..., got {local_sha1[:12]}...)"
- )
- found = True
- break
-
- if not found:
- missing += 1
-
- total = verified + missing + mismatched
- print(f"\n Verified: {verified}/{total}")
- if missing:
- print(f" Missing: {missing}")
- if mismatched:
- print(f" Mismatched: {mismatched}")
+ expected = checksums.get(pack.name)
+ if expected:
+ print("Checking the archive...")
+ actual = compute_hashes(str(archive))["sha256"].lower()
+ if actual != expected:
+ archive.unlink()
+ print(
+ f"Error: checksum mismatch for {pack.name}\n"
+ f" expected {expected}\n got {actual}\n"
+ "Download the parts again; a truncated part gives this.",
+ file=sys.stderr,
+ )
+ sys.exit(1)
+ else:
+ print(f"No {CHECKSUMS_ASSET} in the release, skipping the checksum.")
+ return archive
def show_info(platform: str, release: dict):
- """Show coverage information for a platform."""
- asset = find_asset(release, platform)
- if not asset:
+ """Show pack information for a platform."""
+ pack = find_pack(release, platform)
+ if not pack:
print(f"Platform '{platform}' not found in release")
return
- print(f" Platform: {platform}")
- print(f" File: {asset['name']}")
- print(f" Size: {asset['size']:,} bytes ({asset['size'] / (1024 * 1024):.1f} MB)")
- print(f" Downloads: {asset.get('download_count', 'N/A')}")
- print(f" Updated: {asset.get('updated_at', 'N/A')}")
+ print(f" Platform: {pack.platform}")
+ print(f" File: {pack.name}")
+ print(f" Parts: {len(pack.parts)} part{'s' if len(pack.parts) > 1 else ''}")
+ print(f" Size: {pack.size:,} bytes ({pack.size / (1024 * 1024):.1f} MB)")
+ for part in pack.parts:
+ print(f" {part['name']} ({part.get('size', 0):,} bytes)")
def main():
@@ -193,14 +261,14 @@ def main():
Examples:
%(prog)s --list List available platforms
%(prog)s retroarch ~/RetroArch/system Download RetroArch pack
- %(prog)s --verify retroarch ~/path Verify local files
%(prog)s --info retroarch Show pack info
+
+To check files already in place, use: python install.py --check
""",
)
parser.add_argument("platform", nargs="?", help="Platform name")
parser.add_argument("dest", nargs="?", help="Destination directory")
parser.add_argument("--list", action="store_true", help="List available platforms")
- parser.add_argument("--verify", action="store_true", help="Verify existing files")
parser.add_argument("--info", action="store_true", help="Show platform info")
args = parser.parse_args()
@@ -220,7 +288,8 @@ Examples:
OSError,
json.JSONDecodeError,
) as e:
- print(f"Error: {e}")
+ print(f"Error: {e}", file=sys.stderr)
+ sys.exit(1)
return
if not args.platform:
@@ -229,43 +298,35 @@ Examples:
try:
release = get_latest_release()
except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e:
- print(f"Error fetching release info: {e}")
+ print(f"Error fetching release info: {e}", file=sys.stderr)
sys.exit(1)
if args.info:
show_info(args.platform, release)
return
- if args.verify:
- if not args.dest:
- parser.error("Destination directory required for --verify")
- verify_files(args.platform, args.dest, release)
- return
-
if not args.dest:
parser.error("Destination directory required")
- asset = find_asset(release, args.platform)
- if not asset:
+ pack = find_pack(release, args.platform)
+ if not pack:
print(f"Platform '{args.platform}' not found in release.")
print("Available:", ", ".join(list_platforms(release)))
sys.exit(1)
- import tempfile
+ dest = Path(os.path.expanduser(args.dest))
+ dest.mkdir(parents=True, exist_ok=True)
- fd, zip_path = tempfile.mkstemp(suffix=".zip")
- os.close(fd)
-
- print(f"Downloading {asset['name']} ({asset['size']:,} bytes)...")
- download_file(asset["browser_download_url"], zip_path, asset["size"])
-
- dest = os.path.expanduser(args.dest)
- os.makedirs(dest, exist_ok=True)
-
- print(f"Extracting to {dest}/...")
- extract_pack(zip_path, dest)
-
- os.unlink(zip_path)
+ # Staged inside the destination: a pack is gigabytes, and that is the
+ # filesystem the user picked for them. The system temp directory is a RAM
+ # disk on the appliances these packs target.
+ staging = dest / STAGING_DIR
+ try:
+ archive = fetch_pack(pack, staging, fetch_checksums(release))
+ print(f"Extracting to {dest}/...")
+ safe_extract_zip(str(archive), str(dest))
+ finally:
+ shutil.rmtree(staging, ignore_errors=True)
print("Done!")
diff --git a/scripts/download.sh b/scripts/download.sh
index b69b239a..70996665 100755
--- a/scripts/download.sh
+++ b/scripts/download.sh
@@ -1,16 +1,32 @@
#!/usr/bin/env bash
# Download BIOS pack from GitHub Releases (Linux/macOS one-liner compatible)
#
+# A pack over 2 GB is published as numbered volumes (.zip.001, .zip.002),
+# which are downloaded, joined and checked here.
+#
# Usage:
# bash scripts/download.sh retroarch ~/RetroArch/system/
# bash scripts/download.sh --list
#
-# Requires: curl, unzip, jq (optional, for --list)
+# Requires: curl, unzip
set -euo pipefail
REPO="Abdess/retrobios"
-API="https://api.github.com/repos/${REPO}/releases/latest"
+API_BASE="${RETROBIOS_API:-https://api.github.com}"
+
+# This endpoint names the assets and so the bytes that land in the BIOS
+# directory. Plain HTTP stays allowed to loopback, which is how the script
+# is exercised end to end.
+case "$API_BASE" in
+ https://*) ;;
+ http://127.0.0.1[:/]*|http://localhost[:/]*|"http://[::1]"[:/]*) ;;
+ *)
+ echo "Error: RETROBIOS_API must use HTTPS, got '$API_BASE'" >&2
+ exit 1
+ ;;
+esac
+API="${API_BASE%/}/repos/${REPO}/releases/latest"
usage() {
echo "Usage: $0 "
@@ -25,53 +41,159 @@ usage() {
exit 1
}
+# One asset URL per line, in the order the release lists them.
+asset_urls() {
+ printf '%s' "$1" | tr ',' '\n' |
+ sed -n 's/.*"browser_download_url"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p'
+}
+
+# The archive name behind each asset, volumes folded into their archive.
+pack_names() {
+ asset_urls "$1" | sed 's#.*/##' |
+ sed -n 's/\(.*_BIOS_Pack\.zip\)\(\.[0-9][0-9]*\)\{0,1\}$/\1/p' |
+ LC_ALL=C sort -u
+}
+
+platform_names() {
+ pack_names "$1" | sed 's/_BIOS_Pack\.zip$//' | tr '_' ' '
+}
+
+# Letters and digits only: the platform is `misterfpga`, the asset MiSTer_FPGA.
+normalize() {
+ printf '%s' "$1" | tr '[:upper:]' '[:lower:]' | tr -cd 'a-z0-9'
+}
+
+fetch_release() {
+ curl -fsSL "$API"
+}
+
list_platforms() {
echo "Fetching available platforms..."
- if command -v jq &>/dev/null; then
- curl -sL "$API" | jq -r '.assets[].name' | grep '_BIOS_Pack.zip' | sed 's/_BIOS_Pack.zip//' | tr '_' ' '
- else
- curl -sL "$API" | grep -oP '"name":\s*"\K[^"]*_BIOS_Pack\.zip' | sed 's/_BIOS_Pack.zip//' | tr '_' ' '
- fi
+ platform_names "$(fetch_release)"
}
download_pack() {
local platform="$1"
local dest="$2"
local normalized
- normalized=$(echo "$platform" | tr ' ' '_' | tr '[:upper:]' '[:lower:]')
+ normalized=$(normalize "$platform")
echo "Fetching release info..."
local release_json
- release_json=$(curl -sL "$API")
+ release_json=$(fetch_release)
- # Find matching asset URL
- local download_url
- download_url=$(echo "$release_json" | grep -oP "\"browser_download_url\":\s*\"[^\"]*${normalized}[^\"]*_BIOS_Pack\.zip\"" | head -1 | grep -oP 'https://[^"]+')
+ local archive=""
+ local candidate
+ while read -r candidate; do
+ [ -n "$candidate" ] || continue
+ case "$(normalize "$candidate")" in
+ *"$normalized"*)
+ archive="$candidate"
+ break
+ ;;
+ esac
+ done <&2
+ exit 1
+ fi
- local tmpfile
- tmpfile=$(mktemp "/tmp/${filename}.XXXXXX")
+ # Staged inside the destination: a pack is gigabytes, and /tmp is a RAM
+ # disk on the appliances these packs target.
+ mkdir -p "$dest"
+ local staging="${dest%/}/.retrobios-download"
+ rm -rf "$staging"
+ mkdir -p "$staging"
+ trap 'rm -rf "$staging"' EXIT
- echo "Downloading ${filename}..."
- curl -L --progress-bar -o "$tmpfile" "$download_url"
+ local count
+ count=$(printf '%s\n' "$volumes" | wc -l | tr -d ' ')
+
+ local index=0
+ local url name
+ while read -r url; do
+ [ -n "$url" ] || continue
+ index=$((index + 1))
+ name=$(basename "$url")
+ if [ "$count" -gt 1 ]; then
+ echo "Downloading ${name} (part ${index}/${count})..."
+ else
+ echo "Downloading ${name}..."
+ fi
+ curl -fL --progress-bar -o "${staging}/${name}" "$url"
+ done < "${staging}/${archive}"
+ rm -f "${staging}/${archive}".[0-9][0-9][0-9]
+ fi
+
+ verify_checksum "$release_json" "$staging" "$archive"
echo "Extracting to ${dest}/..."
- mkdir -p "$dest"
- unzip -o -q "$tmpfile" -d "$dest"
+ unzip -o -q "${staging}/${archive}" -d "$dest"
- rm -f "$tmpfile"
+ rm -rf "$staging"
+ trap - EXIT
echo "Done! BIOS files extracted to ${dest}/"
}
+verify_checksum() {
+ local release_json="$1" staging="$2" archive="$3"
+
+ local sums_url
+ sums_url=$(asset_urls "$release_json" | grep -E '/SHA256SUMS\.txt$' | head -1)
+ if [ -z "$sums_url" ]; then
+ echo "No SHA256SUMS.txt in the release, skipping the checksum."
+ return 0
+ fi
+
+ local hasher
+ if command -v sha256sum >/dev/null 2>&1; then
+ hasher="sha256sum"
+ elif command -v shasum >/dev/null 2>&1; then
+ hasher="shasum -a 256"
+ else
+ echo "No sha256sum available, skipping the checksum."
+ return 0
+ fi
+
+ local expected
+ expected=$(curl -fsSL "$sums_url" | awk -v name="$archive" '$2 == name {print $1}')
+ if [ -z "$expected" ]; then
+ echo "No checksum published for ${archive}, skipping."
+ return 0
+ fi
+
+ echo "Checking the archive..."
+ local actual
+ actual=$($hasher "${staging}/${archive}" | cut -d' ' -f1)
+ if [ "$actual" != "$expected" ]; then
+ echo "Error: checksum mismatch for ${archive}" >&2
+ echo " expected ${expected}" >&2
+ echo " got ${actual}" >&2
+ echo "Download the parts again; a truncated part gives this." >&2
+ exit 1
+ fi
+}
+
# Main
case "${1:-}" in
--list|-l)
diff --git a/tests/test_download.py b/tests/test_download.py
new file mode 100644
index 00000000..d5137666
--- /dev/null
+++ b/tests/test_download.py
@@ -0,0 +1,425 @@
+"""Tests for the release pack downloaders.
+
+Packs over 2 GB are published as numbered volumes (`.zip.001`, `.zip.002`),
+so a downloader that expects one asset per platform finds nothing for most
+of them. These tests pin the grouping, the join and the staging location.
+"""
+from __future__ import annotations
+
+import functools
+import hashlib
+import http.server
+import importlib.util
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+import zipfile
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+_spec = importlib.util.spec_from_file_location(
+ "download", REPO_ROOT / "scripts" / "download.py"
+)
+download = importlib.util.module_from_spec(_spec)
+# Registered before execution: a dataclass resolves its annotations through
+# sys.modules and cannot be built from a module that is not there yet.
+sys.modules["download"] = download
+_spec.loader.exec_module(download)
+
+SHELL = REPO_ROOT / "scripts" / "download.sh"
+
+# A slice of a real release: two whole packs, three split, and the checksums.
+RELEASE_ASSETS = [
+ "Batocera_43.1_BIOS_Pack.zip.001",
+ "Batocera_43.1_BIOS_Pack.zip.002",
+ "BizHawk_2.11.1_BIOS_Pack.zip",
+ "EmuDeck_2.3.8_BIOS_Pack.zip",
+ "RetroArch_Lakka_v1.22.2_BIOS_Pack.zip.001",
+ "RetroArch_Lakka_v1.22.2_BIOS_Pack.zip.002",
+ "RetroDECK_0.10.9b_BIOS_Pack.zip.001",
+ "RetroDECK_0.10.9b_BIOS_Pack.zip.002",
+ "RetroDECK_0.10.9b_BIOS_Pack.zip.003",
+ "SHA256SUMS.txt",
+]
+
+
+def _closed_port() -> int:
+ """A loopback port nothing listens on, so connecting is refused at once."""
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ return sock.getsockname()[1]
+
+
+class _QuietHandler(http.server.SimpleHTTPRequestHandler):
+ def log_message(self, *args): # keep the test output pristine
+ pass
+
+
+def _release(names, base="https://example.invalid", size=10):
+ return {
+ "assets": [
+ {
+ "name": name,
+ "size": size,
+ "browser_download_url": f"{base}/assets/{name}",
+ }
+ for name in names
+ ]
+ }
+
+
+class TestPackGrouping(unittest.TestCase):
+ def test_whole_pack_has_one_part(self):
+ packs = download.group_packs(_release(["EmuDeck_2.3.8_BIOS_Pack.zip"]))
+ self.assertEqual([p.name for p in packs], ["EmuDeck_2.3.8_BIOS_Pack.zip"])
+ self.assertEqual(len(packs[0].parts), 1)
+
+ def test_volumes_group_into_one_pack(self):
+ packs = download.group_packs(
+ _release(
+ [
+ "Batocera_43.1_BIOS_Pack.zip.001",
+ "Batocera_43.1_BIOS_Pack.zip.002",
+ ]
+ )
+ )
+ self.assertEqual([p.name for p in packs], ["Batocera_43.1_BIOS_Pack.zip"])
+ self.assertEqual(
+ [part["name"] for part in packs[0].parts],
+ [
+ "Batocera_43.1_BIOS_Pack.zip.001",
+ "Batocera_43.1_BIOS_Pack.zip.002",
+ ],
+ )
+
+ def test_pack_size_is_the_sum_of_its_volumes(self):
+ packs = download.group_packs(
+ _release(
+ [
+ "Batocera_43.1_BIOS_Pack.zip.001",
+ "Batocera_43.1_BIOS_Pack.zip.002",
+ ],
+ size=7,
+ )
+ )
+ self.assertEqual(packs[0].size, 14)
+
+ def test_volumes_order_numerically_not_lexically(self):
+ packs = download.group_packs(
+ _release(
+ [
+ "Big_BIOS_Pack.zip.010",
+ "Big_BIOS_Pack.zip.009",
+ "Big_BIOS_Pack.zip.001",
+ ]
+ )
+ )
+ self.assertEqual(
+ [part["name"] for part in packs[0].parts],
+ ["Big_BIOS_Pack.zip.001", "Big_BIOS_Pack.zip.009", "Big_BIOS_Pack.zip.010"],
+ )
+
+ def test_other_assets_are_not_packs(self):
+ packs = download.group_packs(_release(["SHA256SUMS.txt", "database.json"]))
+ self.assertEqual(packs, [])
+
+ def test_list_platforms_names_split_packs(self):
+ names = download.list_platforms(_release(RELEASE_ASSETS))
+ self.assertIn("Batocera 43.1", names)
+ self.assertIn("RetroDECK 0.10.9b", names)
+ self.assertEqual(len(names), 5)
+
+ def test_find_pack_resolves_a_split_platform(self):
+ pack = download.find_pack(_release(RELEASE_ASSETS), "batocera")
+ self.assertIsNotNone(pack)
+ self.assertEqual(pack.name, "Batocera_43.1_BIOS_Pack.zip")
+ self.assertEqual(len(pack.parts), 2)
+
+ def test_find_pack_maps_retroarch_to_the_lakka_asset(self):
+ pack = download.find_pack(_release(RELEASE_ASSETS), "retroarch")
+ self.assertEqual(pack.name, "RetroArch_Lakka_v1.22.2_BIOS_Pack.zip")
+
+ def test_find_pack_returns_none_for_an_unknown_platform(self):
+ self.assertIsNone(download.find_pack(_release(RELEASE_ASSETS), "nintendo"))
+
+ def test_find_pack_resolves_a_platform_id_that_drops_separators(self):
+ # The platform is `misterfpga` everywhere else; the asset is MiSTer_FPGA.
+ pack = download.find_pack(
+ _release(RELEASE_ASSETS + ["MiSTer_FPGA_2026-08-29_BIOS_Pack.zip"]),
+ "misterfpga",
+ )
+ self.assertIsNotNone(pack)
+ self.assertEqual(pack.name, "MiSTer_FPGA_2026-08-29_BIOS_Pack.zip")
+
+
+class TestJoinVolumes(unittest.TestCase):
+ def test_joined_volumes_reproduce_the_archive(self):
+ tmp = Path(tempfile.mkdtemp())
+ archive = tmp / "pack.zip"
+ with zipfile.ZipFile(archive, "w") as zf:
+ zf.writestr("bios/scph5501.bin", b"\x01\x02" * 5000)
+ zf.writestr("bios/dc_boot.bin", b"\x03\x04" * 5000)
+ raw = archive.read_bytes()
+ cut = len(raw) // 3
+ parts = []
+ for index, start in enumerate(range(0, len(raw), cut), start=1):
+ part = tmp / f"pack.zip.{index:03d}"
+ part.write_bytes(raw[start : start + cut])
+ parts.append(part)
+ self.assertGreater(len(parts), 1)
+
+ joined = tmp / "joined.zip"
+ download.join_volumes(parts, joined)
+
+ self.assertEqual(joined.read_bytes(), raw)
+ with zipfile.ZipFile(joined) as zf:
+ self.assertEqual(zf.testzip(), None)
+
+
+class ReleaseServer:
+ """Serves a release index and its assets over loopback."""
+
+ def __init__(self, pack_name: str, payload: dict[str, bytes], volumes: int):
+ self.root = Path(tempfile.mkdtemp())
+ (self.root / "assets").mkdir()
+ archive = self.root / "assets" / pack_name
+ with zipfile.ZipFile(archive, "w") as zf:
+ for name, data in payload.items():
+ zf.writestr(name, data)
+ raw = archive.read_bytes()
+ self.digest = hashlib.sha256(raw).hexdigest()
+ archive.unlink()
+
+ self.names: list[str] = []
+ if volumes == 1:
+ (self.root / "assets" / pack_name).write_bytes(raw)
+ self.names.append(pack_name)
+ else:
+ cut = len(raw) // volumes + 1
+ for index, start in enumerate(range(0, len(raw), cut), start=1):
+ name = f"{pack_name}.{index:03d}"
+ (self.root / "assets" / name).write_bytes(raw[start : start + cut])
+ self.names.append(name)
+
+ (self.root / "assets" / "SHA256SUMS.txt").write_text(
+ f"{self.digest} {pack_name}\n"
+ )
+
+ handler = functools.partial(_QuietHandler, directory=str(self.root))
+ self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
+ self.base = f"http://127.0.0.1:{self.httpd.server_address[1]}"
+
+ index = self.root / "repos" / "Abdess" / "retrobios" / "releases"
+ index.mkdir(parents=True)
+ assets = [
+ {
+ "name": name,
+ "size": (self.root / "assets" / name).stat().st_size,
+ "browser_download_url": f"{self.base}/assets/{name}",
+ }
+ for name in self.names + ["SHA256SUMS.txt"]
+ ]
+ (index / "latest").write_text(json.dumps({"assets": assets}))
+ threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
+
+ def corrupt_last_volume(self) -> None:
+ target = self.root / "assets" / self.names[-1]
+ target.write_bytes(target.read_bytes()[:-16] + b"0" * 16)
+
+ def close(self) -> None:
+ self.httpd.shutdown()
+ self.httpd.server_close()
+
+
+class DownloaderCase(unittest.TestCase):
+ """Common fixture: a two-volume pack served over loopback."""
+
+ volumes = 2
+ payload = {
+ "bios/scph5501.bin": b"\x10\x20" * 4096,
+ "bios/dc/dc_boot.bin": b"\x30\x40" * 4096,
+ }
+
+ pack_name = "Batocera_43.1_BIOS_Pack.zip"
+ platform = "batocera"
+ platform_label = "Batocera 43.1"
+
+ def setUp(self):
+ self.server = ReleaseServer(self.pack_name, self.payload, self.volumes)
+ self.addCleanup(self.server.close)
+ self.dest = Path(tempfile.mkdtemp()) / "bios"
+ # A pack is gigabytes: staging it in the system temp directory fills
+ # the RAM disk that /tmp is on the appliances these packs target.
+ self.tmpdir = Path(tempfile.mkdtemp())
+
+ def env(self) -> dict[str, str]:
+ env = dict(os.environ)
+ env.update(
+ RETROBIOS_API=self.base_api(),
+ TMPDIR=str(self.tmpdir),
+ TMP=str(self.tmpdir),
+ TEMP=str(self.tmpdir),
+ )
+ return env
+
+ def base_api(self) -> str:
+ return self.server.base
+
+ def assert_extracted(self):
+ for name, data in self.payload.items():
+ extracted = self.dest / name
+ self.assertTrue(extracted.is_file(), f"{name} not extracted")
+ self.assertEqual(extracted.read_bytes(), data)
+
+ def assert_listed(self, out: str):
+ names = [
+ line.strip().lstrip("- ").strip()
+ for line in out.splitlines()
+ if "BIOS" not in line and line.strip()
+ ]
+ self.assertIn(self.platform_label, names)
+ self.assertNotIn(".001", out)
+
+ def assert_refused(self, proc):
+ self.assertNotEqual(proc.returncode, 0)
+ self.assertIn("checksum", (proc.stdout + proc.stderr).lower())
+ self.assertFalse((self.dest / "bios/scph5501.bin").exists())
+
+ def assert_no_leftovers(self):
+ self.assertEqual(
+ sorted(p.name for p in self.tmpdir.iterdir()),
+ [],
+ "pack staged in the system temp directory",
+ )
+ searched = list(self.dest.parent.iterdir()) + list(self.dest.rglob("*"))
+ leftovers = sorted(
+ p.name for p in searched if "_BIOS_Pack" in p.name or p.name.startswith(".")
+ )
+ self.assertEqual(leftovers, [], "download staging left behind")
+
+
+class WholePackCase(DownloaderCase):
+ """A pack published whole, under a name the platform id does not spell."""
+
+ volumes = 1
+ pack_name = "MiSTer_FPGA_2026-08-29_BIOS_Pack.zip"
+ platform = "misterfpga"
+ platform_label = "MiSTer FPGA 2026-08-29"
+
+
+class TestDownloadPython(DownloaderCase):
+ def run_cli(self, *args, expect_success=True):
+ proc = subprocess.run(
+ ["python3", str(REPO_ROOT / "scripts" / "download.py"), *args],
+ env=self.env(),
+ capture_output=True,
+ text=True,
+ timeout=120,
+ )
+ if expect_success:
+ self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+ return proc
+
+ def test_list_names_the_pack(self):
+ proc = self.run_cli("--list")
+ self.assert_listed(proc.stdout)
+
+ def test_pack_downloads_and_extracts(self):
+ self.run_cli(self.platform, str(self.dest))
+ self.assert_extracted()
+
+ def test_volumes_are_not_staged_in_the_system_temp_directory(self):
+ self.run_cli(self.platform, str(self.dest))
+ self.assert_no_leftovers()
+
+ def test_a_corrupt_volume_is_refused(self):
+ self.server.corrupt_last_volume()
+ proc = self.run_cli(self.platform, str(self.dest), expect_success=False)
+ self.assert_refused(proc)
+
+ def test_info_reports_every_volume(self):
+ proc = self.run_cli("--info", self.platform)
+ self.assertIn("2 parts", proc.stdout)
+
+ def test_an_unreachable_release_endpoint_fails(self):
+ env = self.env()
+ env["RETROBIOS_API"] = f"http://127.0.0.1:{_closed_port()}"
+ proc = subprocess.run(
+ ["python3", str(REPO_ROOT / "scripts" / "download.py"), "--list"],
+ env=env,
+ capture_output=True,
+ text=True,
+ timeout=60,
+ )
+ self.assertNotEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+
+ def test_a_non_loopback_http_api_is_refused(self):
+ env = self.env()
+ env["RETROBIOS_API"] = "http://api.example.com"
+ proc = subprocess.run(
+ ["python3", str(REPO_ROOT / "scripts" / "download.py"), "--list"],
+ env=env,
+ capture_output=True,
+ text=True,
+ timeout=60,
+ )
+ self.assertNotEqual(proc.returncode, 0)
+ self.assertIn("RETROBIOS_API", proc.stderr)
+
+
+@unittest.skipUnless(
+ shutil.which("curl") and shutil.which("unzip"), "curl and unzip required"
+)
+class TestDownloadShell(DownloaderCase):
+ def run_cli(self, *args, expect_success=True):
+ proc = subprocess.run(
+ ["bash", str(SHELL), *args],
+ env=self.env(),
+ capture_output=True,
+ text=True,
+ timeout=120,
+ )
+ if expect_success:
+ self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+ return proc
+
+ def test_list_names_the_pack(self):
+ proc = self.run_cli("--list")
+ self.assert_listed(proc.stdout)
+
+ def test_pack_downloads_and_extracts(self):
+ self.run_cli(self.platform, str(self.dest))
+ self.assert_extracted()
+
+ def test_volumes_are_not_staged_in_the_system_temp_directory(self):
+ self.run_cli(self.platform, str(self.dest))
+ self.assert_no_leftovers()
+
+ def test_a_corrupt_volume_is_refused(self):
+ self.server.corrupt_last_volume()
+ proc = self.run_cli(self.platform, str(self.dest), expect_success=False)
+ self.assert_refused(proc)
+
+ def test_an_unknown_platform_lists_what_exists(self):
+ proc = self.run_cli("nintendo", str(self.dest), expect_success=False)
+ self.assertIn(self.platform_label, proc.stdout + proc.stderr)
+
+
+class TestWholePackPython(WholePackCase, TestDownloadPython):
+ def test_info_reports_every_volume(self):
+ proc = self.run_cli("--info", self.platform)
+ self.assertIn("1 part", proc.stdout)
+
+
+class TestWholePackShell(WholePackCase, TestDownloadShell):
+ pass
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/wiki/troubleshooting.md b/wiki/troubleshooting.md
index 129e6161..25c06166 100644
--- a/wiki/troubleshooting.md
+++ b/wiki/troubleshooting.md
@@ -261,3 +261,26 @@ bash scripts/download.sh --list
Some platforms share packs (Lakka uses the RetroArch pack). The installer handles
this mapping automatically, but if you're downloading manually, check which pack
name corresponds to your platform.
+
+**A split pack will not extract:**
+
+A pack over 2 GB is published as `.zip.001`, `.zip.002`. Put every part in one
+folder; 7-Zip and PeaZip open the `.001` directly. A frontend's own extractor
+may refuse it, Batocera among them:
+
+```
+Archive type: '001' is not yet supported
+```
+
+The volumes are plain byte ranges, so joining them from a shell rebuilds the
+ZIP:
+
+```bash
+cat Pack.zip.0* > Pack.zip
+unzip Pack.zip -d /userdata/bios/
+rm Pack.zip
+```
+
+`scripts/download.sh` and `scripts/download.py` do this on their own, checksum
+included. Both stage inside the destination directory, never in `/tmp`, which
+is a RAM disk on most of these systems.