libretro/tests/test_scraper_contract.py

#!/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 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):
    """es_bios.xml names a libretro core as libretro/<name> and a standalone
    emulator by its bare name: the prefix is the build mode, and a platform
    that runs an emulator standalone must say so or the pack lays its files
    out for the libretro build."""

    def test_bare_names_are_the_standalone_emulators(self):
        from scraper.recalbox_scraper import split_cores

        cores, standalone = split_cores(
            ["libretro/uae", "amiberry", "", "dolphin", "libretro/dolphin", "libretro-hatari", "hatari"]
        )
        self.assertEqual(
            cores,
            ["amiberry", "dolphin", "hatari", "libretro-hatari", "libretro/dolphin", "libretro/uae"],
        )
        self.assertEqual(standalone, ["amiberry", "dolphin", "hatari"])



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")
            )


if __name__ == "__main__":
    unittest.main()