feat: copy BIOS files from a folder with --from

--from FOLDER (repeatable, a ZIP accepted) reads an old SD card, a
backup or a Downloads folder: files indexed by size, hashed only when
a size matches an entry, ZIP members included. Each file is copied,
checked again once written, and the source is never modified. Mac ._
files, system folders and links leading out of the folder are skipped.

An install also puts in place the files it finds misplaced in the BIOS
folder (bios/bios/, a pack extracted one level too deep, a name in
capitals, a ZIP never extracted, one folder up), leaving the user's
copy where it is. --check and --dry-run say what a run would copy.

Events gain one kind, found, sent before the copy, which a downloaded
event then reports. The end of the run says where each file came from.
This commit is contained in:
Abdessamad Derraz committed 2026-10-23 15:59:21 +02:00
1 parent d458cf653a
commit 44cdf2984b
4 files changed
+1059 -64

No files matched your search

+1 -1
View File
@@ -11,7 +11,7 @@
$ErrorActionPreference = "Stop"
$defaultInstallUrl = "https://raw.githubusercontent.com/Abdess/retrobios/main/install.py"
$defaultInstallSha256 = "8cfe5cb80b8df7f8c2736cfe5855a768c3eacad6ebf5bfd7336b3b4f5a4d5d76"
$defaultInstallSha256 = "9cc655f5e6e847ba30aa1058189e0ad8c1808945ce59dbac1a047d0241707d51"
$maximumInstallerBytes = 2MB
$installerExit = 0
$installer = if ($PSScriptRoot) { Join-Path $PSScriptRoot "install.py" } else { $null }
+587 -62
View File
@@ -16,6 +16,7 @@ from __future__ import annotations
import argparse
import concurrent.futures
import contextlib
import datetime
import errno
import hashlib
@@ -43,7 +44,7 @@ import zipfile
import zlib
from collections.abc import Callable, Collection, Iterator
from pathlib import Path, PurePosixPath
from typing import BinaryIO, NamedTuple, TextIO
from typing import BinaryIO, ContextManager, NamedTuple, TextIO
# Manifests are read from the same ref the bootstrap verified this installer
# against, so a file list and the code reading it always come from one commit.
@@ -487,7 +488,8 @@ EVENT_KEYS: dict[str, tuple[str, ...]] = {
# stage: manifest (reason: network, not_found, invalid, clock for a
# refused certificate under a clock set in the past, or no_ca for no
# trusted certificates on the machine) or detect (reason: conflict, two
# systems found on one machine).
# systems found on one machine) or from (reason: not_found or
# not_a_folder, a folder named with --from).
"error": ("stage", "reason", "platform", "message"),
# state: present (hash verified), wrong_hash or missing.
"file": ("platform", "dest", "systems", "size", "state"),
@@ -495,6 +497,11 @@ EVENT_KEYS: dict[str, tuple[str, ...]] = {
"wrong_hash", "bytes_needed"),
"check": ("platform", "name", "bios_dir", "files", "present", "missing",
"wrong_hash", "bytes_needed", "data_missing", "not_installable"),
# A file to install that this machine already holds: misplaced in the
# BIOS folder (folder is its bios_dir, reason says why in words) or in
# a folder named with --from (reason empty). member names the ZIP member
# when path is an archive, else "". Its copy is a downloaded event.
"found": ("platform", "dest", "folder", "path", "member", "reason"),
"download": ("platform", "files", "bytes"),
"progress": ("files_done", "files_total", "bytes_done", "bytes_total"),
"downloaded": ("dest", "ok", "reason", "size", "files_done", "files_total",
@@ -4172,9 +4179,10 @@ def restore_kept(run: Path, dry_run: bool = False) -> int:
def _download_one(
f: dict, bios_path: Path, verbose: bool = False,
progress: Progress | None = None,
progress: Progress | None = None, copy: LocalCopy | None = None,
) -> tuple[str, bool, str]:
"""Download a single file. Returns (dest, success, reason it failed)."""
"""Download a single file, or copy it from where copy found it on this
machine. Returns (dest, success, reason it failed)."""
if _stop.is_set():
return str(f.get("dest", "?")), False, "stopped"
try:
@@ -4188,6 +4196,18 @@ def _download_one(
return f["dest"], False, NOT_WRITABLE
return f["dest"], False, exc.strerror or str(exc)
if copy is not None:
result = _copy_local(f, bios_path, dest, copy, progress)
if result[1]:
copy.index.count_copy(copy.misplaced)
return result
if result[2] in ("stopped", NOT_WRITABLE, READ_ONLY) or result[2].startswith(NOT_KEPT):
return result
# The copy went or changed since it was read (a card pulled out, a
# file another run replaced): the file is fetched as if never found.
if verbose:
print(f" {f['dest']}: {result[2]}, fetching it instead", file=sys.stderr)
if LOCAL.pack is not None:
return _extract_from_pack(f, bios_path, dest, progress)
_network_allowed(f["dest"])
@@ -4592,7 +4612,7 @@ def _folder_writable(path: Path) -> bool:
# Options that take a value, as a hint gives them back.
_VALUED_OPTIONS = (
"--system", "--core", "--emulator", "--region", "--target", "--cores-dir", "--save-dir",
"--from-pack", "--manifest",
"--from-pack", "--from", "--manifest",
)
@@ -4781,6 +4801,7 @@ def _report_outcome(
omitted_entries: list[dict] | None = None,
platform_of: dict[Path, str] | None = None,
hints: list[str] | None = None,
sources: str = "",
) -> None:
"""Close the run on what happened, and leave with a status that says so.
@@ -4853,6 +4874,8 @@ def _report_outcome(
if errors:
print(f"\nFinished with errors. {_plural(downloaded, 'file')} installed, "
f"{_plural(errors, 'file')} failed.")
if sources:
print(f" {sources}")
print(f"Location: {_locations(paths)}")
for line in kept + advice:
print(line)
@@ -4861,6 +4884,8 @@ def _report_outcome(
print(f"\nDone. {_plural(downloaded, 'file')} installed, "
f"{_plural(up_to_date, 'file')} already up to date.")
if sources:
print(f" {sources}")
print(f"Location: {_locations(paths)}")
for line in kept:
print(line)
@@ -4959,6 +4984,7 @@ def _print_plan(
to_download: list[dict], bios_path: Path, term: Terminal,
trees: list[tuple[dict, dict[str, dict]]] = (),
kept: dict[Path, list[int]] | None = None,
copied: list[str] = (),
) -> None:
"""What a real run would fetch, by system, and nothing written.
@@ -4970,16 +4996,21 @@ def _print_plan(
trees = [(entry, needed) for entry, needed in trees if needed]
tree_files = sum(len(needed) for _entry, needed in trees)
tree_size = sum(meta.get("size", 0) for _entry, needed in trees for meta in needed.values())
for line in copied:
print(line)
if not to_download and not trees:
print("\n All files up to date. A real run would download nothing.")
return
if not copied:
print("\n All files up to date. A real run would download nothing.")
return
print("\n Nothing to download.")
parts = []
if to_download:
parts.append(f"{_plural(len(to_download), 'file')} ({format_size(total)})")
if trees:
parts.append(f"{_plural(tree_files, 'data file')} ({format_size(tree_size)}, "
f"extracted from {_plural(len(trees), 'archive')})")
print(f"\nWould download {' and '.join(parts)} to {bios_path}:")
if parts:
print(f"\nWould download {' and '.join(parts)} to {bios_path}:")
by_system: dict[str, list[dict]] = {}
for entry in to_download:
system = ", ".join(entry_systems(entry)) or "other"
@@ -5035,6 +5066,7 @@ def _check_free_space(bios_path: Path, needed: int, plat: str = "", code: int =
def download_files(
files: list[dict], bios_path: Path, jobs: int = 8, verbose: bool = False,
term: Terminal | None = None, progress: Progress | None = None,
copies: dict[int, LocalCopy] | None = None,
) -> tuple[list[tuple[str, str]], bool]:
"""Download files in parallel.
@@ -5054,7 +5086,8 @@ def download_files(
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
future_map = {
pool.submit(_download_one, f, bios_path, verbose, progress): f
pool.submit(_download_one, f, bios_path, verbose, progress,
(copies or {}).get(id(f))): f
for f in files
}
refused = False
@@ -6401,11 +6434,14 @@ class PackSource:
class LocalSources:
"""Where this run reads instead of the network: a file list, a pack."""
"""Where this run reads instead of the network: a file list, a pack, the
folders it copies BIOS files from."""
def __init__(self) -> None:
self.manifest: Path | None = None
self.pack: PackSource | None = None
# The folders named with --from, read once for every platform.
self.folders: list[FolderIndex] = []
@property
def offline(self) -> bool:
@@ -6416,6 +6452,7 @@ class LocalSources:
self.pack.close()
self.manifest = None
self.pack = None
self.folders = []
LOCAL = LocalSources()
@@ -6474,11 +6511,42 @@ def _extract_from_pack(
if held is None:
return f["dest"], False, NOT_IN_PACK
archive, info = held
return _write_verified(f, bios_path, dest, lambda: archive.open(info), progress,
"the pack")
@contextlib.contextmanager
def _open_copy(copy: LocalCopy) -> Iterator[BinaryIO]:
"""The bytes of a copy found on this machine, opened for reading only."""
if not copy.member:
with open(copy.path, "rb") as handle:
yield handle
return
with zipfile.ZipFile(copy.path) as archive, archive.open(copy.member) as handle:
yield handle
def _copy_local(
f: dict, bios_path: Path, dest: Path, copy: LocalCopy, progress: Progress | None,
) -> tuple[str, bool, str]:
"""Copy one file found on this machine to its place, checked like a
download. Its bytes count as already here, not in the download speed."""
return _write_verified(f, bios_path, dest, lambda: _open_copy(copy), progress,
f"the copy in {copy.index.label}", local=True)
def _write_verified(
f: dict, bios_path: Path, dest: Path,
source: Callable[[], ContextManager[BinaryIO]], progress: Progress | None,
held_in: str, local: bool = False,
) -> tuple[str, bool, str]:
"""Write the bytes source opens at dest once they match the list's size
and hashes, the file in place kept aside first when it differs."""
expected = {name: f[name].lower() for name in ("sha256", "sha1") if f.get(name)}
tmp_path: Path | None = None
counted = 0
try:
with archive.open(info) as src, tempfile.NamedTemporaryFile(
with source() as src, tempfile.NamedTemporaryFile(
mode="wb", dir=dest.parent, prefix=f".{dest.name}.", suffix=".part",
delete=False,
) as out:
@@ -6492,17 +6560,17 @@ def _extract_from_pack(
break
size += len(chunk)
if size > f["size"]:
raise ValueError("the pack holds more bytes than the list records")
raise ValueError(f"{held_in} holds more bytes than the list records")
out.write(chunk)
if progress is not None:
progress.add(len(chunk))
progress.add(len(chunk), resumed=local)
counted += len(chunk)
if size != f["size"]:
raise ValueError(f"the pack holds {size} bytes, the list records {f['size']}")
raise ValueError(f"{held_in} holds {size} bytes, the list records {f['size']}")
actual = _digest_file(tmp_path, tuple(expected))
for name, label in (("sha256", "SHA-256"), ("sha1", "SHA-1")):
if name in expected and actual[name] != expected[name]:
raise ValueError(f"{label} mismatch in the pack")
raise ValueError(f"{label} mismatch in {held_in}")
os.chmod(tmp_path, (0o777 if f.get("executable") else 0o666) & ~_UMASK)
_clear_read_only(dest)
KEPT.keep(*_kept_slot(*_entry_place(bios_path, f)), dest, expected, tmp_path)
@@ -6513,9 +6581,9 @@ def _extract_from_pack(
return f["dest"], False, "stopped"
except _NotKept as exc:
return f["dest"], False, f"{NOT_KEPT} ({exc})"
except (OSError, ValueError, zipfile.BadZipFile, zlib.error) as exc:
except (ValueError, *_MEMBER_ERRORS) as exc:
if progress is not None and counted:
progress.add(-counted)
progress.add(-counted, resumed=local)
if isinstance(exc, OSError) and _write_refused(exc):
return f["dest"], False, READ_ONLY if _read_only_file(dest) else NOT_WRITABLE
return f["dest"], False, str(exc)
@@ -6532,23 +6600,85 @@ _ARCHIVE_SUFFIXES = (".zip", ".7z", ".rar")
SHOWN_CORES = 4
class FolderIndex:
"""The files under a BIOS folder, by name and by size, hashed on demand."""
# Folders a system writes on a drive for itself. No BIOS file a person
# copied is in them, and a Mac's Spotlight index alone holds thousands.
SYSTEM_FOLDERS = frozenset(name.casefold() for name in (
".Spotlight-V100", ".Trashes", ".fseventsd", ".TemporaryItems",
".DocumentRevisions-V100", "System Volume Information", "$RECYCLE.BIN",
))
# A Mac copying to a FAT or exFAT card writes ._name beside every file: its
# Finder data, 4 KB, never the file.
APPLE_DOUBLE = "._"
# Errors a member of someone's ZIP can raise on reading: damaged, encrypted
# (RuntimeError), or packed with a method zipfile lacks.
_MEMBER_ERRORS = (OSError, EOFError, RuntimeError, NotImplementedError,
zipfile.BadZipFile, zlib.error)
def __init__(self, root: Path) -> None:
class LocalCopy(NamedTuple):
"""A file found on this machine with the contents an entry wants."""
index: FolderIndex
path: Path
# The member holding the bytes when path is a ZIP, else "".
member: str = ""
# In the BIOS folder at a place no entry of the list names: a mistake
# the copy fixes. A copy from another entry's place is no mistake.
misplaced: bool = False
def where(self) -> str:
"""The copy as the folder it was found in names it."""
rel = self.index.relative(self.path)
return f"{rel}, inside as {self.member}" if self.member else rel
class FolderIndex:
"""The files under a folder, by name and by size, hashed on demand.
The folder is only read. A link is followed only where it leads to a
file inside the folder, so an old card holding a link to the home
folder is not walked through it. A ZIP named as the root is read as a
folder holding that one archive.
"""
def __init__(self, root: Path, label: str = "") -> None:
self.label = label or str(root)
self.root = root = root.resolve()
self.by_name: dict[str, list[Path]] = {}
self.by_size: dict[int, list[Path]] = {}
self.size: dict[Path, int] = {}
self.archives: list[Path] = []
# Name a ._ file stands beside (casefolded) -> the ._ files.
self.apple_double: dict[str, list[Path]] = {}
# The list's destinations, casefolded (an archive among them is not
# unpacked for its members) and as the list writes them.
self.sealed: set[str] = set()
self.listed: set[str] = set()
self.truncated = False
self.copied = 0
self.fixed = 0
self._lock = threading.Lock()
self._sha1: dict[Path, str] = {}
self._digests: dict[tuple[Path, str, str], str] = {}
self._members: dict[Path, list[str]] = {}
self._member_sizes: dict[int, list[tuple[Path, str]]] | None = None
if root.is_file():
if root.name.casefold().endswith(".zip"):
self.archives.append(root)
return
seen = 0
for folder, dirs, names in os.walk(root):
dirs[:] = sorted(d for d in dirs if d != BACKUP_DIR)
dirs[:] = sorted(d for d in dirs
if d != BACKUP_DIR and d.casefold() not in SYSTEM_FOLDERS)
for name in sorted(names):
path = Path(folder) / name
if name.startswith(APPLE_DOUBLE):
self.apple_double.setdefault(name[2:].casefold(), []).append(path)
continue
if name.startswith(".") and name.endswith(".part"):
continue # a download of this installer, not finished
try:
if path.is_symlink() and not _inside(path, root):
continue
if not path.is_file():
continue
size = path.stat().st_size
@@ -6561,6 +6691,7 @@ class FolderIndex:
self.archives.append(path)
seen += 1
if seen >= MAX_SCANNED_FILES:
self.truncated = True
return
def seed(self, path: Path, sha1: str) -> None:
@@ -6575,6 +6706,23 @@ class FolderIndex:
self._sha1[path] = ""
return self._sha1[path]
def digest(self, path: Path, algorithm: str, member: str = "") -> str:
"""The file's (or the member's) digest, read once, "" when unreadable."""
if algorithm == "sha1" and not member:
return self.sha1(path)
key = (path, member, algorithm)
if key not in self._digests:
try:
if member:
with zipfile.ZipFile(path) as archive, archive.open(member) as handle:
value = _digest_handle(handle, (algorithm,))[algorithm]
else:
value = _digest_file(path, (algorithm,))[algorithm]
except _MEMBER_ERRORS:
value = ""
self._digests[key] = value
return self._digests[key]
def holding(self, sizes: set[int], sha1s: set[str]) -> list[Path]:
"""Files whose contents are one of sha1s, read only when the size fits."""
return [path for size in sorted(sizes) for path in self.by_size.get(size, [])
@@ -6590,11 +6738,70 @@ class FolderIndex:
self._members[path] = []
return self._members[path]
def _members_of_size(self, size: int) -> list[tuple[Path, str]]:
"""(archive, member) of every ZIP member of that size, the
archives' directories read once, the first time one is asked."""
if self._member_sizes is None:
self._member_sizes = {}
for path in self.archives:
if (not path.name.casefold().endswith(".zip")
or self.relative(path).casefold() in self.sealed):
continue
try:
with zipfile.ZipFile(path) as archive:
infos = archive.infolist()
except (OSError, zipfile.BadZipFile):
continue
for info in infos:
if not info.is_dir():
self._member_sizes.setdefault(info.file_size, []).append(
(path, info.filename))
return self._member_sizes.get(size, [])
def find(self, entry: dict, skip: Collection[Path] = ()) -> LocalCopy | None:
"""A copy of the entry's own contents: its size first, then its hash,
loose files before ZIP members, a file of the same name first."""
size = entry.get("size")
if not isinstance(size, int) or isinstance(size, bool):
return None
algorithm = "sha1" if entry.get("sha1") else "sha256"
wanted = str(entry.get(algorithm) or "").lower()
if not wanted:
return None
name = str(entry.get("dest", "")).rsplit("/", 1)[-1].casefold()
loose = sorted(self.by_size.get(size, []), key=lambda p: p.name.casefold() != name)
for path in loose:
if path not in skip and self.digest(path, algorithm) == wanted:
return LocalCopy(self, path)
members = sorted(self._members_of_size(size),
key=lambda item: item[1].rsplit("/", 1)[-1].casefold() != name)
for path, member in members:
if self.digest(path, algorithm, member) == wanted:
return LocalCopy(self, path, member)
return None
def count_copy(self, misplaced: bool = False) -> None:
with self._lock:
if misplaced:
self.fixed += 1
else:
self.copied += 1
def relative(self, path: Path) -> str:
try:
return path.relative_to(self.root).as_posix()
rel = path.relative_to(self.root).as_posix()
except ValueError:
return str(path)
return rel if rel != "." else path.name
def _inside(path: Path, root: Path) -> bool:
"""Whether a link at path leads to a place under root."""
try:
Path(os.path.realpath(path)).relative_to(root)
except ValueError:
return False
return True
def _copies(entry: dict) -> list[dict]:
@@ -6656,6 +6863,59 @@ def _content_owner(sha1: str, entry: dict, files: list[dict]) -> str:
return ""
def _place_reason(
entry: dict, path: Path, index: FolderIndex, expected: Path, known: set[str],
contents: str = "",
) -> tuple[str, str]:
"""(reason, folder too deep) for an entry's file found at path in the
BIOS folder, or ("", "") when that place says nothing. contents is
", with other contents" for a file of the right name and wrong bytes."""
dest = str(entry.get("dest", ""))
name = dest.rsplit("/", 1)[-1]
rel = index.relative(path)
# A dest below the save directory is not below this folder: a copy
# here under the same tail is in the wrong place, not too deep.
below = entry.get("root") != "save"
if below and rel.endswith("/" + dest) and path.name == name:
extra = rel[: -len(dest) - 1]
top = index.root.name
if extra.casefold() == top.casefold():
return (f"nested folder: it is at {top}/{rel}{contents}, a "
f"{top} folder inside {top}"), extra
depth = extra.count("/") + 1
levels = "one folder" if depth == 1 else f"{depth} folders"
return f"{levels} too deep: it is at {rel}{contents}", extra
if path.name.casefold() == name.casefold():
if path.parent == expected.parent and path.name != name:
return (f"case of the name: the folder holds {path.name}{contents}, "
f"the cores open {name}"), ""
return f"wrong folder: it is at {rel}{contents}", ""
if contents:
return "", ""
if rel.casefold() in known:
return f"its contents are at {rel}: copy that file to {dest}", ""
return f"a name no core opens: its contents are at {rel}, rename it {name}", ""
def _copy_reason(entry: dict, copy: LocalCopy, bios_path: Path, known: set[str]) -> str:
"""Why a copy found in the BIOS folder is not at the entry's place."""
if not copy.misplaced:
# np2kai/2608_bd.wav and np2/2608_bd.wav hold the same bytes. Told
# "wrong folder", a user moved np2's copy and broke that core.
return f"the same file is at {copy.where()}, where the list installs it too"
if copy.member:
return f"still zipped: it is inside {copy.index.relative(copy.path)}"
try:
copy.path.relative_to(copy.index.root)
except ValueError:
return f"one folder too high: it is at {copy.path}"
try:
expected = _entry_path(bios_path, entry)
except ValueError:
return ""
return _place_reason(entry, copy.path, copy.index, expected, known)[0]
def _found_elsewhere(
entry: dict, index: FolderIndex, bios_path: Path, files: list[dict],
known: set[str],
@@ -6678,32 +6938,15 @@ def _found_elsewhere(
same_name = [p for p in index.by_name.get(name.casefold(), []) if p != expected]
matching = [p for p in same_name if index.size[p] in sizes and index.sha1(p) in sha1s]
for path in matching or same_name:
rel = index.relative(path)
contents = "" if path in matching else ", with other contents"
# A dest below the save directory is not below this folder: a copy
# here under the same tail is in the wrong place, not too deep.
below = entry.get("root") != "save"
if below and rel.endswith("/" + dest) and path.name == name:
extra = rel[: -len(dest) - 1]
top = index.root.name
if extra.casefold() == top.casefold():
return (f"nested folder: it is at {top}/{rel}{contents}, a "
f"{top} folder inside {top}"), extra
depth = extra.count("/") + 1
levels = "one folder" if depth == 1 else f"{depth} folders"
return f"{levels} too deep: it is at {rel}{contents}", extra
if path.parent == expected.parent and path.name != name:
return f"case of the name: the folder holds {path.name}{contents}", ""
if path.name.casefold() == name.casefold():
return f"wrong folder: it is at {rel}{contents}", ""
found = _place_reason(entry, path, index, expected, known, contents)
if found[0]:
return found
above = index.root.parent / dest
if index.root.parent != index.root and above.is_file():
return f"one folder too high: it is at {above}", ""
for path in index.holding(sizes, sha1s):
rel = index.relative(path)
if rel.casefold() in known:
return f"its contents are at {rel}: copy that file to {dest}", ""
return f"a name no core opens: its contents are at {rel}, rename it {name}", ""
return _place_reason(entry, path, index, expected, known)
if name.casefold().endswith(_ARCHIVE_SUFFIXES):
return "", ""
stem = name.rsplit(".", 1)[0].casefold()
@@ -6716,6 +6959,10 @@ def _found_elsewhere(
return f"still zipped: it is inside {index.relative(path)}", ""
if lowered.rsplit(".", 1)[0] == stem:
return f"still zipped: {index.relative(path)} needs extracting", ""
for path in index.apple_double.get(name.casefold(), []):
if path.parent == expected.parent:
return (f"only a Mac's ._ file: {path.name} holds Finder data, not the "
f"file itself. Copy {name} again"), ""
return "", ""
@@ -6747,7 +6994,7 @@ class Diagnosis(NamedTuple):
def diagnose(
missing: list[dict], mismatched: list[dict], files: list[dict], bios_path: Path,
verified: list[dict] | None = None,
verified: list[dict] | None = None, index: FolderIndex | None = None,
) -> Diagnosis:
"""The reason --check gives for each file it flags, from what the folder
holds and what the list records: id(entry) -> reason, and the folders
@@ -6757,13 +7004,8 @@ def diagnose(
deep: dict[str, int] = {}
if not bios_path.is_dir() or not (missing or mismatched):
return Diagnosis(reasons, deep)
index = FolderIndex(bios_path)
for entry in verified or []:
if entry.get("sha1"):
try:
index.seed(_entry_path(bios_path, entry), entry["sha1"].lower())
except ValueError:
continue
if index is None:
index = bios_folder_index(bios_path, files, verified or [])
known = _known_destinations(files)
for entry in missing:
reason, extra = _found_elsewhere(entry, index, bios_path, files, known)
@@ -6785,21 +7027,26 @@ def diagnose(
def check_lines(
missing: list[dict], mismatched: list[dict], files: list[dict], manifest: dict,
bios_path: Path, every: bool, verified: list[dict] | None = None,
index: FolderIndex | None = None, copies: dict[int, LocalCopy] | None = None,
) -> list[str]:
"""The files --check flags, each with why and where its cores read it.
"missing: scph5501.bin" left a user with the file one folder too deep,
or still in its ZIP, to guess. A flagged file with a reason is always
listed. Plain missing ones stop at ten unless every is set.
listed, and so is one a folder named with --from holds. Plain missing
ones stop at ten unless every is set.
"""
found = diagnose(missing, mismatched, files, bios_path, verified)
found = diagnose(missing, mismatched, files, bios_path, verified, index)
copies = copies or {}
known = _known_destinations(files)
lines = []
for extra, count in sorted(found.deep.items(), key=lambda item: -item[1]):
if count >= 3:
lines.append(f" {_plural(count, 'file')} sit too deep, in "
f"{bios_path / extra}: move what it holds up into {bios_path}.")
explained = [e for e in missing if id(e) in found.reasons]
plain = [e for e in missing if id(e) not in found.reasons]
lines += _check_copy_lines(missing + mismatched, copies, index)
explained = [e for e in missing if id(e) in found.reasons or id(e) in copies]
plain = [e for e in missing if id(e) not in found.reasons and id(e) not in copies]
shown = plain if every else plain[:10]
def block(label: str, entry: dict) -> list[str]:
@@ -6807,6 +7054,16 @@ def check_lines(
rows = [f" {label}: {entry['dest']}{game}"]
if id(entry) in found.reasons:
rows.append(f" {found.reasons[id(entry)]}")
copy = copies.get(id(entry))
if copy is not None and copy.index is index and not copy.misplaced:
rows.append(f" a run without --check copies it from {copy.where()}")
elif copy is not None and copy.index is index:
where = _copy_reason(entry, copy, bios_path, known)
if where and where != found.reasons.get(id(entry)):
rows.append(f" the right file: {where}")
rows.append(" a run without --check copies it into place")
elif copy is not None:
rows.append(f" in {copy.index.label}: {copy.where()}")
rows.append(f" {opened_line(entry, manifest, bios_path)}")
return rows
@@ -6822,6 +7079,199 @@ def check_lines(
return lines
def bios_folder_index(
bios_path: Path, files: list[dict], verified: list[dict],
) -> FolderIndex | None:
"""The BIOS folder read for files misplaced in it. The files the check
verified are not read again: their SHA-1 is the one the list records.
An archive the list installs is not unpacked for its members: a MAME
set holding a ROM a core also reads loose is that core's to open."""
if not bios_path.is_dir():
return None
index = FolderIndex(bios_path)
for entry in verified:
if entry.get("sha1"):
try:
index.seed(_entry_path(bios_path, entry), entry["sha1"].lower())
except ValueError:
continue
index.sealed = _known_destinations(files)
index.listed = {str(f.get("dest", "")) for f in files}
return index
def _above(entry: dict, index: FolderIndex) -> LocalCopy | None:
"""The entry's file at its place one folder above the BIOS folder."""
if entry.get("root") == "save" or index.root.parent == index.root:
return None
try:
above = index.root.parent / _safe_relative_path(entry.get("dest"), "dest")
if above.stat().st_size != entry.get("size") or not above.is_file():
return None
except (OSError, ValueError):
return None
algorithm = "sha1" if entry.get("sha1") else "sha256"
if index.digest(above, algorithm) != str(entry.get(algorithm, "")).lower():
return None
return LocalCopy(index, above)
def plan_local_copies(
entries: list[dict], bios_path: Path, bios_index: FolderIndex | None,
folders: list[FolderIndex],
) -> dict[int, LocalCopy]:
"""id(entry) -> where this machine already holds the entry's bytes: the
BIOS folder itself first, a file misplaced in it, then each folder named
with --from in turn. Every copy is found by size, then by hash."""
if not entries or (bios_index is None and not folders):
return {}
# A file this run replaces is no source: its place is being written.
written = set()
for entry in entries:
try:
written.add(Path(os.path.realpath(_entry_path(bios_path, entry))))
except ValueError:
continue
copies: dict[int, LocalCopy] = {}
for entry in entries:
copy = None
if bios_index is not None:
copy = bios_index.find(entry, written) or _above(entry, bios_index)
if copy is not None:
copy = copy._replace(misplaced=_misplaced(entry, copy, bios_index))
for index in folders:
if copy is not None:
break
copy = index.find(entry, written)
if copy is not None:
copies[id(entry)] = copy
return copies
def _misplaced(entry: dict, copy: LocalCopy, index: FolderIndex) -> bool:
"""Whether a copy in the BIOS folder sits where no other entry of the
list goes: GBA_BIOS.BIN for gba_bios.bin is a mistake, np2/2608_bd.wav
for np2kai/2608_bd.wav is the other core's file."""
if copy.member:
return True
rel = index.relative(copy.path)
if rel in index.listed:
return False
own = str(entry.get("dest", "")).casefold()
return rel.casefold() not in index.sealed or rel.casefold() == own
def _check_copy_lines(
entries: list[dict], copies: dict[int, LocalCopy], bios_index: FolderIndex | None,
) -> list[str]:
"""What --check says a run would copy instead of downloading."""
fixed = sum(1 for e in entries if id(e) in copies and copies[id(e)].misplaced)
lines = []
if fixed:
verb = "is" if fixed == 1 else "are"
them = "it" if fixed == 1 else "them"
lines.append(f" {fixed} of these files {verb} in the BIOS folder at the wrong place: "
f"a run without --check copies {them} where the cores open {them}, and "
f"leaves your {'copy' if fixed == 1 else 'copies'} as "
f"{'it is' if fixed == 1 else 'they are'}.")
for label, count in _counts_by_folder(entries, copies, bios_index):
them = "it" if count == 1 else "them"
lines.append(f" {label} holds {count} of these files: a run without --check "
f"copies {them}.")
return lines
def _counts_by_folder(
entries: list[dict], copies: dict[int, LocalCopy], bios_index: FolderIndex | None,
) -> list[tuple[str, int]]:
"""(folder, files) for each --from folder that gives some of entries."""
counts: dict[str, int] = {}
for entry in entries:
copy = copies.get(id(entry))
if copy is not None and copy.index is not bios_index:
counts[copy.index.label] = counts.get(copy.index.label, 0) + 1
return list(counts.items())
def copy_plan_lines(
entries: list[dict], copies: dict[int, LocalCopy], bios_index: FolderIndex | None,
bios_path: Path, files: list[dict], every: bool, planned: bool,
) -> list[str]:
"""The files a run copies from this machine, by folder: the misplaced
ones with the reason each was not where its cores open it."""
known = _known_destinations(files)
# By --from folder, "" for the misplaced files and "=" for the ones the
# BIOS folder holds at another entry's place.
groups: dict[str, list[dict]] = {}
for entry in entries:
copy = copies.get(id(entry))
if copy is not None:
key = (copy.index.label if copy.index is not bios_index
else "" if copy.misplaced else "=")
groups.setdefault(key, []).append(entry)
lines = []
for key in sorted(groups, key=lambda k: (k != "", k != "=")):
chosen = groups[key]
sized = f"{_plural(len(chosen), 'file')} ({format_size(_bytes(chosen))})"
if key == "":
head = (f"Would fix {sized} misplaced in the BIOS folder, copied where the "
"cores open them:" if planned else
f"Fixing {sized} misplaced in the BIOS folder: copied where the cores "
"open them, your copies left as they are.")
elif key == "=":
head = (f"Would copy {sized} the BIOS folder already holds at another "
"place of the list:" if planned else
f"Copying {sized} the BIOS folder already holds at another place "
"of the list.")
else:
head = (f"Would copy {sized} from {key}:" if planned else
f"Copying {sized} from {key}, each one checked by hash.")
lines.append(f"\n {head}" if planned else f" {head}")
for entry in chosen if every else chosen[:10]:
copy = copies[id(entry)]
why = (_copy_reason(entry, copy, bios_path, known) if key == ""
else f"from {copy.where()}" if key == "=" else copy.where())
lines.append(f" {entry['dest']}: {why}" if why else f" {entry['dest']}")
if len(chosen) > 10 and not every:
lines.append(f" ... and {len(chosen) - 10} more (--verbose lists them all)")
return lines
def emit_copies(
plat: str, entries: list[dict], copies: dict[int, LocalCopy],
bios_index: FolderIndex | None, bios_path: Path, files: list[dict],
) -> None:
"""One found event per file this machine holds, before it is copied."""
known = _known_destinations(files)
for entry in entries:
copy = copies.get(id(entry))
if copy is None:
continue
reason = (_copy_reason(entry, copy, bios_path, known)
if copy.index is bios_index else "")
EVENTS.emit("found", platform=plat, dest=entry["dest"], folder=copy.index.label,
path=str(copy.path), member=copy.member, reason=reason)
def sources_line(
folders: list[tuple[str, int]], fixed: int, downloaded: int, missing: int,
fetched: str = "downloaded", shared: int = 0,
) -> str:
"""Where the files of a run that copied some came from, in words:
"38 files copied from E:\\, 4 downloaded, 2 still missing." """
parts = [(count, f"copied from {label}") for label, count in folders]
parts.append((fixed, "fixed from copies already in the BIOS folder"))
parts.append((shared, "copied from another place of the BIOS folder"))
if not any(count for count, _what in parts):
return ""
parts += [(downloaded, fetched), (missing, "still missing")]
words = []
for count, what in parts:
if count:
words.append(f"{_plural(count, 'file') if not words else count} {what}")
return ", ".join(words) + "."
# Below this a download is small enough that narrowing it is not worth a line.
BULK_HINT_FROM = 100 * 1024 * 1024
@@ -7082,6 +7532,15 @@ def main() -> None:
"(the others are found beside it) or the folder holding them. "
"Makes no network request, and checks every file's hash",
)
parser.add_argument(
"--from",
action="append",
type=Path,
metavar="FOLDER",
help="copy the BIOS files found in FOLDER (an old SD card, a backup, a "
"ZIP) instead of downloading them: every file recognised by its "
"hash, never moved. Repeat for several folders",
)
parser.add_argument(
"--manifest",
type=Path,
@@ -7126,7 +7585,7 @@ def main() -> None:
"cores_dir", "save_dir", "system", "core", "region", "no_game_data",
"list_systems",
"list_cores", "romm_firmware", "list_emulators", "events", "quiet",
"yes", "verbose", "from_pack", "manifest")
"yes", "verbose", "from_pack", "manifest", "from")
given = [name for name in alone if getattr(args, name)]
if args.jobs != 8:
given.append("jobs")
@@ -7139,6 +7598,12 @@ def main() -> None:
if args.from_pack and args.target:
parser.error("--target reads a core list from the network, which --from-pack "
"does not use")
if getattr(args, "from") and (
args.list_platforms or args.list_targets or args.list_systems
or args.list_cores or args.list_emulators
):
parser.error("--from copies files into the BIOS folder, which a listing "
"does not write")
if (args.from_pack or args.manifest) and (args.list_platforms or args.list_targets):
parser.error("--from-pack and --manifest give a file list, which --list-platforms "
"and --list-targets do not read")
@@ -7193,6 +7658,7 @@ def _open_local_sources(args: argparse.Namespace) -> None:
"""Take the file list and the pack this run was given, and the platform
they are for when none is named."""
LOCAL.reset()
_open_from_folders(getattr(args, "from", None) or [])
if args.manifest is not None:
LOCAL.manifest = args.manifest
named = local_manifest_platform(args.manifest)
@@ -7227,6 +7693,28 @@ def _open_local_sources(args: argparse.Namespace) -> None:
sys.exit(1)
def _open_from_folders(folders: list[Path]) -> None:
"""Read each folder named with --from, or stop on one that is not there."""
for folder in folders:
label = str(folder)
folder = Path(os.path.expanduser(label))
zipped = folder.is_file() and folder.name.casefold().endswith(".zip")
if not (folder.is_dir() or zipped):
reason = "not_a_folder" if folder.exists() else "not_found"
what = ("does not exist" if reason == "not_found"
else "is neither a folder nor a ZIP")
message = f"{label} {what}: --from names the folder to copy BIOS files from"
EVENTS.emit("error", stage="from", reason=reason, platform="", message=message)
print(f"Error: {message}.", file=sys.stderr)
sys.exit(1)
index = FolderIndex(folder, label)
LOCAL.folders.append(index)
print(f" Copying BIOS files from: {label}")
if index.truncated:
print(f" Read the first {MAX_SCANNED_FILES:,} files of {label} only. Name "
"the folder that holds the BIOS files to search it all.")
def _missing_dest(dest: Path, ask: bool, named: str = "named with --dest") -> None:
"""Say that a --dest folder does not exist before anything goes there.
@@ -7521,6 +8009,8 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
failures: list[tuple[str, str]] = []
installed_paths: list[Path] = []
bulk_lines: list[str] = []
# The BIOS folders read for misplaced files, one per platform.
bios_indexes: list[FolderIndex] = []
# --target is checked for every platform before anything is written, as
# --installed-cores is below: EmuDeck installed 3.2 GB and only then was
@@ -7820,6 +8310,14 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
# Mismatched files need re-download
to_download.extend(mismatched)
# The files this machine already holds: misplaced in the BIOS folder
# (bios/bios/, a pack extracted into a folder of its own, a name in
# capitals, still zipped), or in a folder named with --from.
bios_index = (bios_folder_index(bios_path, files, up_to_date)
if to_download else None)
if bios_index is not None:
bios_indexes.append(bios_index)
copies = plan_local_copies(to_download, bios_path, bios_index, LOCAL.folders)
hint = [] if LOCAL.pack is not None else bulk_hint(
to_download, manifest,
systems_chosen=bool(selection and selection[0]),
@@ -7843,6 +8341,7 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
sum(len(state[1]) for _entry, state in data_states), len(omitted_files),
chosen=selection[0] if selection else None,
)
emit_copies(plat_name, to_download, copies, bios_index, bios_path, files)
total_verified += len(up_to_date)
if args.check:
@@ -7859,7 +8358,7 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
print(line)
print(f"\n {_plural(len(to_download), 'file')} to download:")
for line in check_lines(missing, mismatched, files, manifest, bios_path,
args.verbose, up_to_date):
args.verbose, up_to_date, bios_index, copies):
print(line)
print(" Run the same command without --check to fetch them; "
"a file with a wrong hash is replaced, and the one in place "
@@ -7891,19 +8390,23 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
{"dest": f"{entry['dest']}/{rel}" if entry.get("dest") else rel}
for entry, needed in trees for rel in needed
], bios_path)
_print_plan(to_download, bios_path, term, trees, kept)
_print_plan([f for f in to_download if id(f) not in copies], bios_path, term,
trees, kept, copy_plan_lines(to_download, copies, bios_index,
bios_path, files, args.verbose, True))
for line in hint:
print(line)
continue
if to_download:
dl_size = sum(f.get("size", 0) for f in to_download)
fetched = [f for f in to_download if id(f) not in copies]
fetch_size = _bytes(fetched)
if (
interactive
and not args.yes
and dl_size >= CONFIRM_FROM_BYTES
and fetch_size >= CONFIRM_FROM_BYTES
and not _confirm(
f"\nDownload {format_size(dl_size)} to {bios_path}? [Y/n] "
f"\nDownload {format_size(fetch_size)} to {bios_path}? [Y/n] "
)
):
# A beginner who just passed the c entry was sent to a
@@ -7915,7 +8418,17 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
files_pending += len(to_download)
continue
EVENTS.emit("download", platform=plat_name, files=len(to_download), bytes=dl_size)
if LOCAL.pack is not None:
if copies:
print()
for line in copy_plan_lines(to_download, copies, bios_index, bios_path,
files, args.verbose, False):
print(line)
rest = (f" and {'copying' if LOCAL.pack is not None else 'downloading'} "
f"{len(fetched)}" if fetched else "")
where = f" from {LOCAL.pack.name}" if fetched and LOCAL.pack is not None else ""
print(f"\nCopying {_plural(len(copies), 'file')} found on this machine"
f"{rest}{where} ({format_size(dl_size)})...")
elif LOCAL.pack is not None:
print(f"\nCopying {len(to_download)} files ({format_size(dl_size)}) "
f"from {LOCAL.pack.name}...")
else:
@@ -7943,7 +8456,7 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
progress = Progress(len(to_download), dl_size, term)
failed, interrupted = download_files(
to_download, bios_path, jobs=args.jobs, verbose=args.verbose,
term=term, progress=progress,
term=term, progress=progress, copies=copies,
)
bulk_lines += hint
total_downloaded += progress.installed
@@ -8048,9 +8561,21 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
omitted_entries=omitted_seen,
platform_of={path: plat for plat, path in platforms},
hints=bulk_lines,
sources=_run_sources(bios_indexes, total_downloaded, total_errors),
)
def _run_sources(bios_indexes: list[FolderIndex], installed: int, failed: int) -> str:
"""The end line saying which folder gave how many of the files."""
folders = [(index.label, index.copied) for index in LOCAL.folders]
fixed = sum(index.fixed for index in bios_indexes)
shared = sum(index.copied for index in bios_indexes)
fetched = installed - fixed - shared - sum(count for _label, count in folders)
return sources_line(folders, fixed, max(fetched, 0), failed,
"copied from the pack" if LOCAL.pack is not None else "downloaded",
shared)
def _line_buffered_stdout() -> None:
"""Flush stdout line by line, as stderr is.
+1 -1
View File
@@ -18,7 +18,7 @@ esac
TEMP_INSTALLER=""
TEMP_DIRECTORY=""
DEFAULT_INSTALL_URL="https://raw.githubusercontent.com/Abdess/retrobios/main/install.py"
DEFAULT_INSTALL_SHA256="8cfe5cb80b8df7f8c2736cfe5855a768c3eacad6ebf5bfd7336b3b4f5a4d5d76"
DEFAULT_INSTALL_SHA256="9cc655f5e6e847ba30aa1058189e0ad8c1808945ce59dbac1a047d0241707d51"
MAX_INSTALLER_BYTES=2097152
cleanup() {
+470
View File
@@ -0,0 +1,470 @@
"""BIOS files the user already has, found by their contents.
A user who wiped the card a handheld came with asks where its BIOS files went,
and one who copied the folder from a Mac, unzipped a pack into bios/bios/ or
renamed scph1001.bin in capitals read "missing" for a file sitting in the
folder. --from FOLDER copies what an old card, a backup or a Downloads folder
holds, and an install puts the files it finds misplaced in the BIOS folder
where the cores open them. Every file is recognised by its hash, copied and
never moved, and checked again once written.
"""
from __future__ import annotations
import contextlib
import io
import json
import subprocess
import sys
import unittest
import zipfile
from pathlib import Path
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
sys.path.insert(0, str(REPO_ROOT / "tests"))
import install # noqa: E402
from test_install_offline import ( # noqa: E402
EXTRA, GBA, JP, PAYLOAD, US, Scratch, _entry, _offline_env,
)
from test_install_selection import FakeRepository # noqa: E402
def _zip(path: Path, members: dict[str, bytes]) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as archive:
for name, data in members.items():
archive.writestr(name, data)
return path
def _snapshot(folder: Path) -> dict[str, tuple[bytes, int]]:
"""Every file below folder, with its bytes and modification time."""
return {p.relative_to(folder).as_posix(): (p.read_bytes(), p.stat().st_mtime_ns)
for p in folder.rglob("*") if p.is_file() and not p.is_symlink()}
def _events(proc: subprocess.CompletedProcess) -> list[dict]:
return [json.loads(line) for line in proc.stdout.splitlines() if line.startswith("{")]
class FromFolder(Scratch):
def setUp(self):
super().setUp()
self.old = self.root / "old card"
self.old.mkdir()
def keep(self, relative: str, data: bytes) -> Path:
path = self.old / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
return path
def install(self, *options: str) -> subprocess.CompletedProcess:
return self.run_installer("--manifest", str(self.list), "--from", str(self.old),
"--no-input", *options)
def installed(self) -> dict[str, bytes]:
if not self.dest.exists():
return {}
return {p.relative_to(self.dest).as_posix(): p.read_bytes()
for p in self.dest.rglob("*") if p.is_file()}
class CopyFromAnOldCard(FromFolder):
def test_files_in_nested_folders_and_under_other_names_are_copied(self):
self.keep("BIOS/psx/SCPH5501.BIN", US)
self.keep("Roms/psx/japan.bin", JP)
self.keep("a/b/c/gba.rom", GBA)
self.keep("pcsx/extra.bin", EXTRA)
before = _snapshot(self.old)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), PAYLOAD)
self.assertIn(f"4 files copied from {self.old}.", proc.stdout)
self.assertEqual(_snapshot(self.old), before, "the old card is only read")
def test_members_of_a_zip_are_copied(self):
_zip(self.old / "Downloads" / "psx bios.zip",
{"PSX/scph5501.bin": US, "PSX/scph5500.bin": JP, "readme.txt": b"hi"})
self.keep("gba_bios.bin", GBA)
self.keep("pcsx/extra.bin", EXTRA)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), PAYLOAD)
def test_a_zip_given_alone_is_read_as_a_folder(self):
archive = _zip(self.root / "backup.zip", {f"system/{k}": v for k, v in PAYLOAD.items()})
proc = self.run_installer("--manifest", str(self.list), "--from", str(archive),
"--no-input")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), PAYLOAD)
self.assertIn(f"4 files copied from {archive}.", proc.stdout)
def test_an_archive_the_list_wants_whole_is_copied_whole(self):
inner = io.BytesIO()
with zipfile.ZipFile(inner, "w") as archive:
archive.writestr("sp-s2.sp1", b"neo geo bios" * 10)
neogeo = inner.getvalue()
self.list.write_text(json.dumps({
"manifest_version": 2, "platform": "retroarch",
"files": [_entry("neogeo.zip", neogeo, ["snk-neogeo"]),
_entry("fbneo/neogeo.zip", neogeo, ["snk-neogeo"])],
}))
self.keep("arcade/neogeo (1).zip", neogeo)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), {"neogeo.zip": neogeo, "fbneo/neogeo.zip": neogeo})
def test_an_archive_inside_an_archive_is_copied(self):
inner = io.BytesIO()
with zipfile.ZipFile(inner, "w") as archive:
archive.writestr("sp-s2.sp1", b"neo geo bios" * 10)
neogeo = inner.getvalue()
self.list.write_text(json.dumps({
"manifest_version": 2, "platform": "retroarch",
"files": [_entry("neogeo.zip", neogeo, ["snk-neogeo"])],
}))
_zip(self.old / "RetroArch_BIOS_Pack.zip", {"neogeo.zip": neogeo})
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), {"neogeo.zip": neogeo})
def test_an_empty_file_and_a_same_size_file_are_not_taken(self):
""""bios folder is 0kb" after flashing Garlic: a name is no proof."""
self.keep("scph5501.bin", b"")
self.keep("gba_bios.bin", b"x" * len(GBA))
self.keep("scph5500.bin", JP)
self.keep("pcsx/extra.bin", EXTRA)
proc = self.install()
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), {"scph5500.bin": JP, "pcsx/extra.bin": EXTRA})
self.assertIn(f"2 files copied from {self.old}, 2 still missing.", proc.stdout)
def test_a_link_out_of_the_folder_is_not_followed(self):
outside = self.root / "elsewhere"
outside.mkdir()
(outside / "scph5501.bin").write_bytes(US)
(outside / "gba").mkdir()
(outside / "gba" / "gba_bios.bin").write_bytes(GBA)
try:
(self.old / "scph5501.bin").symlink_to(outside / "scph5501.bin")
(self.old / "gba").symlink_to(outside / "gba", target_is_directory=True)
except (OSError, NotImplementedError):
self.skipTest("this system makes no symbolic link")
self.keep("scph5500.bin", JP)
proc = self.install()
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), {"scph5500.bin": JP})
def test_a_link_inside_the_folder_is_read(self):
self.keep("real/scph5501.bin", US)
try:
(self.old / "scph5501.bin").symlink_to(self.old / "real" / "scph5501.bin")
except (OSError, NotImplementedError):
self.skipTest("this system makes no symbolic link")
index = install.FolderIndex(self.old)
self.assertEqual(len(index.by_size[len(US)]), 2)
def test_mac_and_system_files_are_never_read(self):
"""A Mac writes ._name beside every file it copies to a FAT card,
and Spotlight keeps an index there: neither is a file a person put."""
self.keep("._scph5501.bin", US)
self.keep(".Spotlight-V100/Store-V2/scph5500.bin", JP)
self.keep("System Volume Information/gba_bios.bin", GBA)
self.keep("pcsx/extra.bin", EXTRA)
proc = self.install()
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), {"pcsx/extra.bin": EXTRA})
def test_the_rest_is_downloaded_and_the_summary_says_where_each_came_from(self):
with FakeRepository() as repo:
old = repo.root / "old"
payloads = FakeRepository.PAYLOADS
(old / "BIOS").mkdir(parents=True)
(old / "BIOS" / "SCPH5500.BIN").write_bytes(payloads["scph5500.bin"][0])
(old / "gba.bin").write_bytes(payloads["gba_bios.bin"][0])
(repo.root / "bios" / "mgba_extra.bin").unlink()
proc = repo.run("--from", str(old), "--no-input")
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertIn(f"2 files copied from {old}, 2 downloaded, 1 still missing.",
proc.stdout)
self.assertEqual(sorted(repo.installed()), sorted(
name for name in payloads if name != "mgba/extra.bin"))
def test_two_folders_are_read_in_turn(self):
second = self.root / "backup"
(second / "x").mkdir(parents=True)
(second / "x" / "gba_bios.bin").write_bytes(GBA)
(second / "scph5501.bin").write_bytes(US)
self.keep("scph5501.bin", US)
self.keep("scph5500.bin", JP)
self.keep("pcsx/extra.bin", EXTRA)
proc = self.install("--from", str(second))
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(self.installed(), PAYLOAD)
self.assertIn(f"3 files copied from {self.old}, 1 copied from {second}.", proc.stdout)
def test_check_names_what_the_folder_would_give_and_writes_nothing(self):
self.keep("BIOS/SCPH5501.BIN", US)
self.keep("gba_bios.bin", GBA)
proc = self.run_installer("--check", "--manifest", str(self.list), "--from",
str(self.old))
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertIn(f"{self.old} holds 2 of these files: a run without --check "
"copies them.", proc.stdout)
self.assertIn(f"in {self.old}: BIOS/SCPH5501.BIN", proc.stdout)
self.assertFalse(self.dest.exists())
def test_dry_run_counts_the_copies_and_writes_nothing(self):
self.keep("BIOS/SCPH5501.BIN", US)
self.keep("gba_bios.bin", GBA)
proc = self.install("--dry-run")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn(f"Would copy 2 files ({install.format_size(len(US) + len(GBA))}) "
f"from {self.old}", proc.stdout)
self.assertIn("Would download 2 files", proc.stdout)
self.assertFalse(self.dest.exists())
self.assertFalse((self.root / install.BACKUP_DIR).exists())
def test_a_folder_that_does_not_exist_is_refused(self):
proc = self.run_installer("--manifest", str(self.list), "--from",
str(self.root / "nowhere"))
self.assertEqual(proc.returncode, 1)
self.assertIn("does not exist", proc.stderr)
self.assertFalse(self.dest.exists())
def test_a_listing_refuses_the_option(self):
proc = self.run_installer("--manifest", str(self.list), "--from", str(self.old),
"--list-systems")
self.assertEqual(proc.returncode, 2)
self.assertIn("--from", proc.stderr)
def test_restore_refuses_the_option(self):
proc = subprocess.run(
[sys.executable, str(REPO_ROOT / "install.py"), "--restore", str(self.old),
"--from", str(self.old)],
env=_offline_env(), capture_output=True, text=True, timeout=60, check=False,
stdin=subprocess.DEVNULL)
self.assertEqual(proc.returncode, 2)
self.assertIn("takes no --from", proc.stderr)
class CaseTwinsOnAFolderThatFoldsCase(FromFolder):
"""An ext4 card holds font.bin and FONT.BIN, two files. On an exFAT
card or a Windows folder they are one name: the copy kept is the one
the list installs, found by its hash, never the other by its name."""
def test_the_hash_decides_which_twin_lands(self):
first, second = b"first font" * 9, b"second font" * 9
self.list.write_text(json.dumps({
"manifest_version": 2, "platform": "retroarch",
"files": [_entry("sys/font.bin", first, ["x"]),
_entry("SYS/FONT.BIN", second, ["x"])],
}))
self.keep("sys/FONT.BIN", first)
self.keep("sys/font.bin", second)
out = io.StringIO()
argv = ["install.py", "--platform", "retroarch", "--dest", str(self.dest),
"--manifest", str(self.list), "--from", str(self.old), "--no-input"]
with mock.patch.object(install, "keeps_case", return_value=False), \
mock.patch.object(install, "detect_os", return_value="linux"), \
mock.patch("sys.argv", argv), contextlib.redirect_stdout(out), \
contextlib.redirect_stderr(io.StringIO()):
try:
install.main()
except SystemExit as exc:
self.assertIn(exc.code, (0, None), out.getvalue())
self.addCleanup(install.LOCAL.reset)
self.assertEqual(self.installed(), {"sys/font.bin": first})
self.assertIn("1 file copied from", out.getvalue())
class EventsForAFolder(FromFolder):
def test_each_file_found_is_an_event_before_it_is_copied(self):
self.keep("BIOS/SCPH5501.BIN", US)
_zip(self.old / "gba.zip", {"gba_bios.bin": GBA})
proc = self.install("--events")
events = _events(proc)
self.assertEqual(proc.returncode, 1, proc.stderr)
for event in events:
self.assertEqual(list(event), ["event", *install.EVENT_KEYS[event["event"]]])
found = sorted((e for e in events if e["event"] == "found"), key=lambda e: e["dest"])
self.assertEqual(found, [
{"event": "found", "platform": "retroarch", "dest": "gba_bios.bin",
"folder": str(self.old), "path": str((self.old / "gba.zip").resolve()),
"member": "gba_bios.bin", "reason": ""},
{"event": "found", "platform": "retroarch", "dest": "scph5501.bin",
"folder": str(self.old),
"path": str((self.old / "BIOS" / "SCPH5501.BIN").resolve()),
"member": "", "reason": ""},
])
kinds = [e["event"] for e in events]
self.assertLess(kinds.index("found"), kinds.index("download"))
copied = {e["dest"] for e in events if e["event"] == "downloaded" and e["ok"]}
self.assertEqual(copied, {"gba_bios.bin", "scph5501.bin"})
summary = next(e for e in events if e["event"] == "summary")
self.assertEqual(summary["installed"], 2)
def test_a_check_writes_the_same_events_and_nothing_else(self):
self.keep("BIOS/SCPH5501.BIN", US)
proc = self.run_installer("--check", "--events", "--manifest", str(self.list),
"--from", str(self.old))
events = _events(proc)
self.assertEqual([e["dest"] for e in events if e["event"] == "found"],
["scph5501.bin"])
self.assertNotIn("download", [e["event"] for e in events])
self.assertFalse(self.dest.exists())
def test_a_folder_that_does_not_exist_is_an_error_event(self):
proc = self.run_installer("--events", "--manifest", str(self.list), "--from",
str(self.root / "nowhere"))
events = _events(proc)
error = next(e for e in events if e["event"] == "error")
self.assertEqual((error["stage"], error["reason"]), ("from", "not_found"))
self.assertEqual(events[-1], {"event": "end", "code": 1})
class MisplacedInTheBiosFolder(Scratch):
"""F12: the files are in the BIOS folder, at a place no core opens."""
def install(self, *options: str) -> subprocess.CompletedProcess:
return self.run_installer("--manifest", str(self.list), "--no-input", *options)
def test_a_nested_bios_folder_is_fixed_and_left_in_place(self):
for name, data in PAYLOAD.items():
self.place(f"system/{name}", data)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
for name, data in PAYLOAD.items():
self.assertEqual((self.dest / name).read_bytes(), data)
self.assertEqual((self.dest / "system" / name).read_bytes(), data)
self.assertIn("4 files fixed from copies already in the BIOS folder.", proc.stdout)
self.assertIn("nested folder: it is at system/system/scph5501.bin", proc.stdout)
def test_a_pack_extracted_into_a_folder_of_its_own_is_fixed(self):
for name, data in PAYLOAD.items():
self.place(f"RetroArch_BIOS_Pack/{name}", data)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("one folder too deep: it is at RetroArch_BIOS_Pack/scph5501.bin",
proc.stdout)
self.assertEqual((self.dest / "pcsx" / "extra.bin").read_bytes(), EXTRA)
def test_a_name_in_capitals_is_fixed_where_case_counts(self):
probe = self.root / "CaseProbe"
probe.write_text("x")
if (self.root / "caseprobe").exists():
self.skipTest("this filesystem does not keep case")
self.complete("gba_bios.bin")
self.place("GBA_BIOS.BIN", GBA)
check = self.check()
self.assertIn("case of the name: the folder holds GBA_BIOS.BIN, the cores open "
"gba_bios.bin", check)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual((self.dest / "gba_bios.bin").read_bytes(), GBA)
self.assertEqual((self.dest / "GBA_BIOS.BIN").read_bytes(), GBA)
self.assertIn("1 file fixed from copies already in the BIOS folder.", proc.stdout)
def test_a_file_left_in_its_zip_is_extracted(self):
self.complete("gba_bios.bin")
_zip(self.dest / "GBA BIOS.zip", {"bios/gba_bios.bin": GBA})
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual((self.dest / "gba_bios.bin").read_bytes(), GBA)
self.assertTrue((self.dest / "GBA BIOS.zip").is_file())
def test_a_file_above_the_folder_is_fixed(self):
self.complete("gba_bios.bin")
(self.root / "gba_bios.bin").write_bytes(GBA)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual((self.dest / "gba_bios.bin").read_bytes(), GBA)
self.assertTrue((self.root / "gba_bios.bin").is_file())
def test_a_wrong_file_in_place_is_kept_before_the_right_bytes_take_its_place(self):
self.complete("gba_bios.bin")
wrong = b"y" * len(GBA)
self.place("gba_bios.bin", wrong)
self.place("old/gba_bios.bin", GBA)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual((self.dest / "gba_bios.bin").read_bytes(), GBA)
kept = list((self.root / install.BACKUP_DIR).rglob("gba_bios.bin"))
self.assertEqual([p.read_bytes() for p in kept], [wrong])
def test_check_says_a_run_puts_it_in_place(self):
for name, data in PAYLOAD.items():
self.place(f"system/{name}", data)
out = self.check()
self.assertIn("4 of these files are in the BIOS folder at the wrong place: a run "
"without --check copies them where the cores open them, and leaves "
"your copies as they are.", out)
self.assertFalse((self.dest / "scph5501.bin").exists())
def test_dry_run_counts_them_and_writes_nothing(self):
for name, data in PAYLOAD.items():
self.place(f"system/{name}", data)
before = _snapshot(self.dest)
proc = self.install("--dry-run")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("Would fix 4 files", proc.stdout)
self.assertEqual(_snapshot(self.dest), before)
def test_a_mac_file_alone_is_named(self):
self.complete("gba_bios.bin")
self.place("._gba_bios.bin", b"\x00\x05\x16\x07" + b"\x00" * 60)
self.assertIn("only a Mac's ._ file: ._gba_bios.bin holds Finder data, not the "
"file itself", self.check())
def test_a_copy_at_another_place_of_the_list_is_no_mistake(self):
"""np2kai/2608_bd.wav and np2/2608_bd.wav hold the same bytes: the
second is copied from the first, and neither is called misplaced."""
probe = self.root / "CaseProbe"
probe.write_text("x")
if (self.root / "caseprobe").exists():
self.skipTest("this filesystem does not keep case")
data = json.loads(self.list.read_text())
twin = next(f for f in data["files"] if f["dest"] == "gba_bios.bin")
data["case_variants"] = [{**twin, "dest": "twin/gba_bios.bin"}]
self.list.write_text(json.dumps(data))
self.complete()
check = self.check()
self.assertIn("a run without --check copies it from gba_bios.bin", check)
self.assertNotIn("wrong place", check)
proc = self.install()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual((self.dest / "twin" / "gba_bios.bin").read_bytes(), GBA)
self.assertIn("1 file copied from another place of the BIOS folder.", proc.stdout)
self.assertNotIn("fixed", proc.stdout)
def test_a_folder_with_nothing_misplaced_says_nothing(self):
with FakeRepository() as repo:
proc = repo.run("--no-input")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertNotIn("fixed", proc.stdout)
self.assertNotIn("copied from", proc.stdout)
class TheSummaryLine(unittest.TestCase):
def test_the_parts_in_words(self):
line = install.sources_line([("E:\\", 38)], fixed=0, downloaded=4, missing=2)
self.assertEqual(line, "38 files copied from E:\\, 4 downloaded, 2 still missing.")
def test_one_file_and_fixes(self):
self.assertEqual(install.sources_line([("/old", 1)], fixed=3, downloaded=0, missing=0),
"1 file copied from /old, 3 fixed from copies already in the "
"BIOS folder.")
self.assertEqual(install.sources_line([], fixed=2, downloaded=1, missing=0),
"2 files fixed from copies already in the BIOS folder, "
"1 downloaded.")
def test_nothing_copied_is_no_line(self):
self.assertEqual(install.sources_line([("/old", 0)], 0, 5, 0), "")
if __name__ == "__main__":
unittest.main()