#!/usr/bin/env python3 """The BaseScraper contract the wiki asks contributors to implement. wiki/adding-a-scraper.md and wiki/adding-a-platform.md document compare_with_config, ChangeSet.has_changes and test_connection as the interface a new scraper plugs into. No scraper in the repository calls them, so nothing proved they still worked; a documented contract that is never exercised is a promise to contributors backed by nothing. """ from __future__ import annotations import importlib import subprocess import sys import tempfile import unittest import urllib.error from pathlib import Path from unittest import mock REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "scripts")) from scraper import emudeck_scraper, libretro_scraper # noqa: E402 from scraper.base_scraper import ( # noqa: E402 BaseScraper, BiosRequirement, ChangeSet, fetch_github_latest_version, github_headers, ) class _StubScraper(BaseScraper): """Minimal scraper: returns what it was handed, or raises.""" def __init__(self, requirements=None, error: Exception | None = None): super().__init__(url="https://example.test/bios.xml") self._requirements = requirements or [] self._error = error def fetch_requirements(self) -> list[BiosRequirement]: if self._error is not None: raise self._error return self._requirements def validate_format(self, raw_data: str) -> bool: return True def _config(*files: dict) -> dict: return {"systems": {"test-sys": {"files": list(files)}}} class ChangeSetSummary(unittest.TestCase): def test_empty_changeset_has_no_changes(self): changes = ChangeSet() self.assertFalse(changes.has_changes) self.assertEqual(changes.summary(), "no changes") def test_summary_counts_each_kind(self): req = BiosRequirement(name="a.bin", system="test-sys") changes = ChangeSet(added=[req], removed=[req], modified=[(req, req)]) self.assertTrue(changes.has_changes) self.assertEqual(changes.summary(), "+1 added, -1 removed, ~1 modified") class CompareWithConfig(unittest.TestCase): def test_new_upstream_file_is_reported_as_added(self): scraper = _StubScraper([BiosRequirement(name="new.bin", system="test-sys")]) changes = scraper.compare_with_config(_config()) self.assertEqual([r.name for r in changes.added], ["new.bin"]) self.assertTrue(changes.has_changes) def test_file_dropped_upstream_is_reported_as_removed(self): scraper = _StubScraper([]) changes = scraper.compare_with_config(_config({"name": "old.bin"})) self.assertEqual([r.name for r in changes.removed], ["old.bin"]) def test_changed_sha1_is_reported_as_modified(self): scraper = _StubScraper( [BiosRequirement(name="a.bin", system="test-sys", sha1="b" * 40)] ) changes = scraper.compare_with_config(_config({"name": "a.bin", "sha1": "a" * 40})) self.assertEqual(len(changes.modified), 1) before, after = changes.modified[0] self.assertEqual((before.sha1, after.sha1), ("a" * 40, "b" * 40)) def test_changed_md5_is_reported_when_no_sha1_is_declared(self): scraper = _StubScraper( [BiosRequirement(name="a.bin", system="test-sys", md5="b" * 32)] ) changes = scraper.compare_with_config(_config({"name": "a.bin", "md5": "a" * 32})) self.assertEqual(len(changes.modified), 1) def test_identical_hashes_are_not_a_change(self): scraper = _StubScraper( [BiosRequirement(name="a.bin", system="test-sys", sha1="a" * 40)] ) changes = scraper.compare_with_config(_config({"name": "a.bin", "sha1": "a" * 40})) self.assertFalse(changes.has_changes) def test_a_file_in_another_system_is_not_the_same_file(self): scraper = _StubScraper([BiosRequirement(name="a.bin", system="other-sys")]) changes = scraper.compare_with_config(_config({"name": "a.bin"})) self.assertEqual([r.name for r in changes.added], ["a.bin"]) self.assertEqual([r.name for r in changes.removed], ["a.bin"]) class TestConnection(unittest.TestCase): def test_reachable_source_reports_true(self): self.assertTrue(_StubScraper([]).test_connection()) def test_network_failure_reports_false(self): self.assertFalse(_StubScraper(error=OSError("unreachable")).test_connection()) def test_bad_payload_reports_false(self): self.assertFalse(_StubScraper(error=ValueError("bad format")).test_connection()) def test_an_unexpected_error_is_not_swallowed(self): """Only reachability failures are absorbed; a bug must surface.""" with self.assertRaises(KeyError): _StubScraper(error=KeyError("bug")).test_connection() class RateLimitedForge(unittest.TestCase): """GitLab answered 429 midway through Recalbox's system.ini files and the whole scrape stopped.""" @staticmethod def _limited(retry_after: str | None): import email.message # noqa: PLC0415 import io # noqa: PLC0415 headers = email.message.Message() if retry_after is not None: headers["Retry-After"] = retry_after return urllib.error.HTTPError("u", 429, "Too Many Requests", headers, io.BytesIO()) def test_a_rate_limit_is_waited_out(self): import io # noqa: PLC0415 from scraper.base_scraper import fetch_bytes # noqa: PLC0415 waits: list[float] = [] answers = [self._limited("7"), self._limited(None), io.BytesIO(b"ok")] def urlopen(_request, timeout): answer = answers.pop(0) if isinstance(answer, Exception): raise answer return answer with mock.patch("urllib.request.urlopen", urlopen): self.assertEqual(fetch_bytes("https://x", sleep=waits.append), b"ok") self.assertEqual(waits, [7.0, 4.0]) def test_a_lasting_limit_or_a_missing_file_fails(self): from scraper.base_scraper import fetch_bytes # noqa: PLC0415 def limited(_request, timeout): raise self._limited(None) with mock.patch("urllib.request.urlopen", limited): with self.assertRaises(ConnectionError): fetch_bytes("https://x", attempts=3, sleep=lambda _s: None) def gone(_request, timeout): raise urllib.error.HTTPError("u", 404, "Not Found", None, None) waits: list[float] = [] with mock.patch("urllib.request.urlopen", gone): with self.assertRaises(ConnectionError): fetch_bytes("https://x", sleep=waits.append) self.assertEqual(waits, []) class RecalboxReadsItsSystemsInBatches(unittest.TestCase): """One raw read per system.ini, about 150, kept GitLab throttling the scrape past its half-hour timeout.""" def test_one_query_per_fifty_systems(self): import json as _json # noqa: PLC0415 from scraper import recalbox_scraper # noqa: PLC0415 systems = [f"sys{i:03}" for i in range(120)] queries: list[list[str]] = [] def fetch(url, data=None, headers=None): if "repository/tree" in url: page = int(url.rsplit("page=", 1)[1]) names = systems[(page - 1) * 100:page * 100] return _json.dumps([{"name": n, "type": "tree"} for n in names]).encode() paths = _json.loads(data)["variables"]["paths"] queries.append(paths) nodes = [{"path": p, "rawBlob": "[core.0]\nemulator = libretro\ncore = " + p.split("/")[-2] + "\n"} for p in paths] return _json.dumps({"data": {"project": {"repository": { "blobs": {"nodes": nodes}}}}}).encode() scraper = recalbox_scraper.Scraper.__new__(recalbox_scraper.Scraper) scraper.tag = "10.1.1" with mock.patch.object(recalbox_scraper, "fetch_bytes", fetch): cores, _standalone, _dual = scraper._fetch_cores() self.assertEqual([len(q) for q in queries], [50, 50, 20]) self.assertEqual(len(cores), 120) class EveryNamedScraperExists(unittest.TestCase): """The registry named lakka_targets, a scraper no file defines: a refresh of stale targets would have failed on its import.""" def test_registry_scrapers_are_modules(self): import yaml # noqa: PLC0415 registry = yaml.safe_load((REPO_ROOT / "platforms" / "_registry.yml").read_text()) scraper_dir = REPO_ROOT / "scripts" / "scraper" for name, meta in registry["platforms"].items(): with self.subTest(platform=name): if meta.get("scraper"): self.assertTrue((scraper_dir / f"{meta['scraper']}_scraper.py").is_file()) if meta.get("target_scraper"): self.assertTrue((scraper_dir / "targets" / f"{meta['target_scraper']}_scraper.py").is_file()) class RetroPiePackageList(unittest.TestCase): """RetroPie states its packages, and nothing else states them for it.""" @staticmethod def _archive(modules: dict[str, str]) -> bytes: import io import tarfile buffer = io.BytesIO() with tarfile.open(fileobj=buffer, mode="w:gz") as archive: for path, body in modules.items(): payload = body.encode() info = tarfile.TarInfo(f"RetroPie-Setup-master/{path}") info.size = len(payload) archive.addfile(info, io.BytesIO(payload)) return buffer.getvalue() def _scraper(self, modules: dict[str, str]): from scraper.retropie_scraper import Scraper scraper = Scraper() scraper.module_ids(self._archive(modules)) return scraper def test_a_package_is_read_from_the_id_its_script_declares(self): scraper = self._scraper( { "scriptmodules/libretrocores/lr-mgba.sh": 'rp_module_id="lr-mgba"\n', "scriptmodules/emulators/openmsx.sh": 'rp_module_id="openmsx"\n', } ) self.assertEqual(scraper.module_ids(), ["lr-mgba", "openmsx"]) def test_a_script_outside_the_package_directories_is_not_a_package(self): """Setup helpers, themes and drivers carry no core.""" scraper = self._scraper( { "scriptmodules/libretrocores/lr-mgba.sh": 'rp_module_id="lr-mgba"\n', "scriptmodules/supplementary/autostart.sh": ( 'rp_module_id="autostart"\n' ), "scriptmodules/admin/setup.sh": 'rp_module_id="setup"\n', } ) self.assertEqual(scraper.module_ids(), ["lr-mgba"]) def test_the_core_list_drops_the_libretro_prefix(self): scraper = self._scraper( { "retropie_packages.sh": '__version="4.8.12"\n', "scriptmodules/libretrocores/lr-mgba.sh": 'rp_module_id="lr-mgba"\n', "scriptmodules/emulators/openmsx.sh": 'rp_module_id="openmsx"\n', } ) config = scraper.generate_platform_yaml() self.assertEqual(config["cores"], ["mgba", "openmsx"]) self.assertEqual(config["inherits"], "retroarch") self.assertEqual(config["base_destination"], "BIOS") def test_the_version_is_the_one_the_setup_script_declares(self): scraper = self._scraper( { "retropie_packages.sh": '#!/bin/bash\n__version="4.8.12"\n', "scriptmodules/libretrocores/lr-mgba.sh": 'rp_module_id="lr-mgba"\n', } ) self.assertEqual(scraper.generate_platform_yaml()["version"], "4.8.12") def test_a_version_assigned_elsewhere_is_not_the_setup_version(self): """Only the top-level retropie_packages.sh speaks for the release.""" scraper = self._scraper( { "scriptmodules/emulators/openmsx.sh": ( 'rp_module_id="openmsx"\n__version="1.0"\n' ), } ) with self.assertRaises(ValueError): scraper.generate_platform_yaml() def test_no_requirement_is_invented_from_prose(self): """RetroPie names BIOS in prose, without a hash to transcribe.""" scraper = self._scraper( { "scriptmodules/libretrocores/lr-mgba.sh": ( 'rp_module_id="lr-mgba"\n' 'rp_module_help="Copy the required BIOS file gba_bios.bin' ' to $biosdir"\n' ), } ) self.assertEqual(scraper.fetch_requirements(), []) class RecalboxBuildMode(unittest.TestCase): """Recalbox's cores come from each system.ini of recalbox-romfs2: a `[core.N]` section names `emulator = "libretro"` with the core, or a standalone emulator. es_bios.xml only names the cores that need a BIOS, so read alone it left most of Recalbox's cores, mame0258 and mame0278 among them, outside the pack.""" SYSTEM_INI = """\ [system] name = "mame" fullname = "MAME" extra line the parser must survive. [core.0] package = BR2_PACKAGE_LIBRETRO_MAME0278 emulator = "libretro" core = "mame0278" [core.1] emulator = "advancemame" core = "advancemame" [core.2.rpi4] emulator = "libretro" core = "mame2003_plus" """ def test_libretro_cores_and_standalone_emulators(self): from scraper.recalbox_scraper import system_ini_cores cores, standalone = system_ini_cores(self.SYSTEM_INI) self.assertEqual(cores, {"mame0278", "advancemame", "mame2003_plus"}) self.assertEqual(standalone, {"advancemame"}) def test_the_registry_leaves_the_list_to_the_scraper(self): import yaml root = Path(__file__).resolve().parents[1] registry = yaml.safe_load((root / "platforms" / "_registry.yml").read_text()) self.assertNotIn("cores", registry["platforms"]["recalbox"]) class RetroBatCoresFromItsES(unittest.TestCase): """RetroBat's YAML carried no core list: batocera-systems.json names BIOS files only, and a registry list copied from Batocera stood in, with abuse, clk or etlegacy that RetroBat never runs.""" ES_SYSTEMS = """\ snes snes9xbsnes samcoupe simcoupe """ def test_libretro_cores_and_standalone_emulators(self): from scraper.retrobat_scraper import es_systems_cores cores, standalone = es_systems_cores(self.ES_SYSTEMS) self.assertEqual(cores, {"snes9x", "bsnes", "simcoupe"}) self.assertEqual(standalone, {"bsnes", "simcoupe"}) def test_the_registry_leaves_the_list_to_the_scraper(self): import yaml root = Path(__file__).resolve().parents[1] registry = yaml.safe_load((root / "platforms" / "_registry.yml").read_text()) self.assertNotIn("cores", registry["platforms"]["retrobat"]) platform = yaml.safe_load((root / "platforms" / "retrobat.yml").read_text()) self.assertIn("snes9x", platform["cores"]) class RocknixRunsItsStandaloneEmulators(unittest.TestCase): """ROCKNIX declared all_libretro while it packages duckstation, dolphin, ppsspp and 33 other emulators standalone, and read rocknix-systems on the next branch under the release's version.""" def test_package_names_reach_the_profiles(self): from scraper.rocknix_scraper import standalone_emulators self.assertEqual( standalone_emulators(["duckstation-sa", "hatarisa", "openbor"]), ["duckstation", "duckstation-sa", "hatari", "hatarisa", "openbor"], ) def test_the_platform_file_is_read_at_its_release(self): import yaml root = Path(__file__).resolve().parents[1] platform = yaml.safe_load((root / "platforms" / "rocknix.yml").read_text()) self.assertIn(f"/distribution/{platform['version']}/", platform["source"]) self.assertIn("retroarch", platform["cores"]) registry = yaml.safe_load((root / "platforms" / "_registry.yml").read_text()) self.assertNotIn("cores", registry["platforms"]["rocknix"]) class RetroDeckCoresAreItsComponents(unittest.TestCase): """The registry added clk and gsplus to RetroDECK, which ships neither as a component, current or archived.""" def test_the_registry_leaves_the_list_to_the_scraper(self): import yaml root = Path(__file__).resolve().parents[1] registry = yaml.safe_load((root / "platforms" / "_registry.yml").read_text()) self.assertNotIn("cores", registry["platforms"]["retrodeck"]) class BizHawkRunsAresForN64Only(unittest.TestCase): """BizHawk embeds ares as Ares64, its N64 core; mapping it to the ares profile claimed thirty systems and accused FirmwareDatabase.cs of a wrong MSX.rom it never declared for ares.""" def test_ares_is_not_a_bizhawk_core(self): import yaml root = Path(__file__).resolve().parents[1] platform = yaml.safe_load((root / "platforms" / "bizhawk.yml").read_text()) registry = yaml.safe_load((root / "platforms" / "_registry.yml").read_text()) self.assertNotIn("ares", platform["cores"]) self.assertNotIn("ares", registry["platforms"]["bizhawk"]["cores"]) class EveryTargetCoreReachesAProfile(unittest.TestCase): """A target lists cores by their buildbot or package name; a name no profile's cores: claims makes --target drop that core and its systems without a word. Only check_freshness compared them, on a fresh scrape.""" # RetroPie's packages that no profile describes yet, counted down, never up. RETROPIE_UNPROFILED = 53 def test_the_committed_targets_resolve(self): import glob import os import yaml sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import check_freshness from common import load_emulator_profiles root = Path(__file__).resolve().parents[1] index = check_freshness.profile_name_index( load_emulator_profiles(str(root / "emulators")) ) overrides = yaml.safe_load((root / "platforms/targets/_overrides.yml").read_text()) or {} for path in sorted(glob.glob(str(root / "platforms/targets/*.yml"))): platform = os.path.basename(path)[:-4] if platform.startswith("_"): continue unresolved = check_freshness.unresolved_target_cores( yaml.safe_load(Path(path).read_text()), index, check_freshness._platform_overrides(overrides, platform), ) with self.subTest(platform=platform): if platform == "retropie": self.assertLessEqual(len(unresolved), self.RETROPIE_UNPROFILED) else: self.assertEqual(unresolved, {}) class UnreadableReleaseStopsTheScrape(unittest.TestCase): """A failed tag lookup fell back to master, or to an empty or invented version. BizHawk, RomM, Batocera and Recalbox then wrote a YAML read from master under the previous release's version (Recalbox under "10.0"), and the exporter patched a revision the data never described.""" @staticmethod def _refuse(*_args, **_kwargs): raise urllib.error.HTTPError("https://api.github.com/x", 403, "rate limited", {}, None) def test_every_pinned_scraper_raises(self): for name in ( "scraper.bizhawk_scraper", "scraper.romm_scraper", "scraper.batocera_scraper", "scraper.recalbox_scraper", ): module = importlib.import_module(name) with self.subTest(scraper=name), mock.patch( "urllib.request.urlopen", self._refuse ), self.assertRaises(RuntimeError): module.Scraper() def test_a_release_lookup_raises(self): with mock.patch("urllib.request.urlopen", self._refuse), self.assertRaises( RuntimeError ): fetch_github_latest_version("libretro/RetroArch") def test_the_token_is_sent(self): with mock.patch.dict("os.environ", {"GITHUB_TOKEN": "t0k"}): self.assertEqual(github_headers()["Authorization"], "Bearer t0k") def test_emudeck_cores_raise(self): """EmuDeck wrote cores: [] when the GitHub listing was refused.""" with mock.patch("urllib.request.urlopen", self._refuse), self.assertRaises( RuntimeError ): emudeck_scraper.Scraper()._fetch_installed_emulators() def test_libretro_core_metadata_raises(self): """libretro dropped every system name and manufacturer when the core-info listing was refused.""" with mock.patch("urllib.request.urlopen", self._refuse), self.assertRaises( RuntimeError ): libretro_scraper.Scraper()._fetch_core_metadata() def test_a_core_info_file_the_forge_refuses_raises(self): """Read as empty, the core left the result and --compare-db counted none of its firmware missing.""" from scraper import coreinfo_scraper # noqa: PLC0415 scraper = coreinfo_scraper.Scraper() with mock.patch.object( scraper, "_fetch_info_list", return_value=["mednafen_psx_libretro.info"] ), mock.patch("urllib.request.urlopen", self._refuse), self.assertRaises( ConnectionError ): scraper.fetch_requirements() class EveryGitHubApiCallerSendsTheToken(unittest.TestCase): """romset_dat_importer and coreinfo_scraper called api.github.com with no token: past the anonymous quota, --fetch failed with 403 while refresh_stale had put a token in their environment.""" def test_each_module_naming_the_api_reads_the_token(self): scripts = Path(__file__).resolve().parent.parent / "scripts" silent = sorted( str(path.relative_to(scripts)) for path in scripts.rglob("*.py") if "api.github.com" in (source := path.read_text(encoding="utf-8")) and "GITHUB_TOKEN" not in source and "github_headers(" not in source ) self.assertEqual(silent, []) def test_each_function_building_an_api_request_reads_the_token(self): """Per module, one call of github_headers anywhere covered them all: fetch_github_latest_tag sent no token beside a helper that did.""" import ast # noqa: PLC0415 scripts = Path(__file__).resolve().parent.parent / "scripts" silent = [] for path in sorted(scripts.rglob("*.py")): source = path.read_text(encoding="utf-8") for node in ast.walk(ast.parse(source)): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue body = ast.get_source_segment(source, node) or "" requests = any( isinstance(call, ast.Call) and getattr(call.func, "attr", getattr(call.func, "id", "")) in ("Request", "urlopen") for call in ast.walk(node) ) if ( requests and "api.github.com" in body and "github_headers(" not in body and "GITHUB_TOKEN" not in body ): silent.append(f"{path.relative_to(scripts)}:{node.name}") self.assertEqual(silent, []) def test_the_tag_helper_sends_the_token_and_raises(self): from scraper import base_scraper # noqa: PLC0415 seen = {} def answer(request, timeout=None): seen.update(request.headers) raise urllib.error.HTTPError(request.full_url, 403, "rate limited", {}, None) with mock.patch.dict("os.environ", {"GITHUB_TOKEN": "t0k"}), mock.patch( "urllib.request.urlopen", answer ), self.assertRaises(RuntimeError): base_scraper.fetch_github_latest_tag("o/r") self.assertEqual(seen.get("Authorization"), "Bearer t0k") _IMPORT_WITHOUT_NETWORK = """ import importlib, socket, sys from pathlib import Path root, package = Path(sys.argv[1]), sys.argv[2] sys.path[:0] = [str(root), str(root / "scripts")] attempts = [] def refuse(*_args, **_kwargs): attempts.append(1) raise OSError("network refused at import") socket.socket.connect = refuse socket.create_connection = refuse socket.getaddrinfo = refuse for path in sorted((root / package.replace(".", "/")).rglob("*.py")): if path.name == "__init__.py": continue name = ".".join(path.relative_to(root).with_suffix("").parts) before = len(attempts) try: importlib.import_module(name) except Exception as exc: print(f"{name}: {type(exc).__name__}: {exc}") continue # A fetch whose error the module swallows ("or 'master'") still tried. if len(attempts) > before: print(f"{name}: reached for the network") """ class ImportingAScraperFetchesNothing(unittest.TestCase): """A tag resolved at module load made a refused API call fall back to master before any scraper ran; the rule since is no network at import. Each module is imported in a fresh interpreter with sockets refused, so nothing already cached hides an import-time fetch.""" @staticmethod def _offenders(root: Path, package: str) -> str: return subprocess.run( [sys.executable, "-I", "-c", _IMPORT_WITHOUT_NETWORK, str(root), package], capture_output=True, text=True, check=True, timeout=300, ).stdout.strip() def test_every_scraper_imports_offline(self): self.assertEqual(self._offenders(REPO_ROOT, "scripts.scraper"), "") def test_a_swallowed_import_time_fetch_is_caught(self): with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory: package = Path(directory) / "planted" package.mkdir() (package / "__init__.py").write_text("") (package / "quiet_scraper.py").write_text( "import urllib.request\n" "def _tag():\n" " try:\n" " return urllib.request.urlopen('https://api.github.com/x').read()\n" " except OSError:\n" " return None\n" "TAG = _tag() or 'master'\n" ) self.assertIn( "planted.quiet_scraper: reached for the network", self._offenders(Path(directory), "planted"), ) def test_an_import_time_fetch_is_caught(self): with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory: package = Path(directory) / "planted" package.mkdir() (package / "__init__.py").write_text("") (package / "eager_scraper.py").write_text( "import urllib.request\n" "urllib.request.urlopen('https://api.github.com/repos/o/r/tags')\n" ) self.assertIn( "planted.eager_scraper", self._offenders(Path(directory), "planted") ) class OneNativeIdOneSystem(unittest.TestCase): """RetroDECK filed coco as tandy-coco and Batocera as coco: xroar's 28 CoCo and Dragon ROMs reached no RetroDECK system, and --system left them out of the pack and the installer alike.""" # A folder each platform fills with something else, not a system name: # EmuDeck checks the Neo Geo BIOS there, RetroDECK keeps arcade sets. # Several platforms file the PC Engine CD under the HuCard system, the # native_system of each entry keeping the platform's own grouping. FOLDS = {"fbneo", "pcenginecd"} def test_scrapers_agree_on_every_shared_native_id(self): from common import _norm_system_id # noqa: PLC0415 seen: dict[str, dict[str, str]] = {} for path in sorted((REPO_ROOT / "scripts" / "scraper").glob("*_scraper.py")): if "SYSTEM_SLUG_MAP" not in path.read_text(): continue module = importlib.import_module(f"scraper.{path.stem}") for native, slug in getattr(module, "SYSTEM_SLUG_MAP", {}).items(): slug = slug[0] if isinstance(slug, tuple) else slug if isinstance(slug, str): seen.setdefault(native.lower(), {})[path.stem] = slug split = { native: by for native, by in seen.items() if native not in self.FOLDS and len({_norm_system_id(s) for s in by.values()}) > 1 } self.assertEqual(split, {}) def test_retrobat_files_its_systems_as_batocera_does(self): """RetroBat kept psx, nds and atari800: no profile system met them, so no core extra joined them and --system found nothing.""" import yaml # noqa: PLC0415 from scraper import batocera_scraper, retrobat_scraper # noqa: PLC0415 for native, slug in batocera_scraper.SYSTEM_SLUG_MAP.items(): self.assertEqual(retrobat_scraper.SYSTEM_SLUG_MAP[native], slug) data = yaml.safe_load((REPO_ROOT / "platforms" / "retrobat.yml").read_text()) self.assertTrue({"sony-playstation", "nintendo-ds"} <= set(data["systems"])) self.assertFalse({"psx", "nds", "atari800"} & set(data["systems"])) # A platform system no profile names gets no core extra and answers no # --system. Named here: Batocera's family folds Naomi and Atomiswave into # one system the profiles keep apart, and MiSTer cores nothing profiles. UNMATCHED = { "*": {"sega-dreamcast-arcade"}, "misterfpga": {"gameking", "interact", "marty", "studio-ii", "superacan", "videobrain"}, } def test_every_platform_system_meets_a_profile_system(self): import yaml # noqa: PLC0415 from common import ( # noqa: PLC0415 _norm_system_id, load_emulator_profiles, load_platform_config, ) known = set() for profile in load_emulator_profiles(str(REPO_ROOT / "emulators")).values(): known |= {_norm_system_id(s) for s in profile.get("systems") or []} known |= {_norm_system_id(f["system"]) for f in profile.get("files") or [] if isinstance(f, dict) and f.get("system")} registry = yaml.safe_load((REPO_ROOT / "platforms" / "_registry.yml").read_text()) for name in registry["platforms"]: with self.subTest(platform=name): config = load_platform_config(name, str(REPO_ROOT / "platforms")) allowed = self.UNMATCHED["*"] | self.UNMATCHED.get(name, set()) stray = {s for s in config.get("systems", {}) if _norm_system_id(s) not in known} - allowed self.assertEqual(stray, set()) def test_xroar_machines_reach_the_platforms_that_run_it(self): import yaml # noqa: PLC0415 from common import _norm_system_id # noqa: PLC0415 xroar = yaml.safe_load((REPO_ROOT / "emulators" / "xroar.yml").read_text()) machines = {_norm_system_id(s) for s in xroar["systems"]} for platform, wanted in (("retrodeck", {"coco", "dragon64"}), ("recalbox", {"coco", "dragon64"}), ("misterfpga", {"coco3"})): with self.subTest(platform=platform): data = yaml.safe_load((REPO_ROOT / "platforms" / f"{platform}.yml").read_text()) ids = {_norm_system_id(s) for s in data["systems"]} self.assertEqual(wanted & ids & machines, wanted) if __name__ == "__main__": unittest.main()