libretro/tests/test_game_headers.py

"""Header readers: schema, references between readers and cores, pinned citations."""

from __future__ import annotations

import json
import re
import sys
import unittest
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "scripts"))

from common import load_emulator_profiles  # noqa: E402
from region import REGIONS  # noqa: E402

try:
    from jsonschema import Draft202012Validator
except ImportError:  # pragma: no cover - CI installs it
    Draft202012Validator = None

HEADERS = ROOT / "games" / "headers.yml"
_CITATION = re.compile(r"^([\w.-]+/[\w.-]+)@([0-9a-f]{8,40}) ")
# Tools the containers cite that are not emulators: no profile pins them.
TOOL_REPOS = {"python/cpython", "ip7z/7zip"}


def load() -> dict:
    return yaml.safe_load(HEADERS.read_text(encoding="utf-8"))


def pins(profile: dict) -> set[str]:
    out: set[str] = set()
    for key in ("source_commit", "upstream_commit"):
        value = profile.get(key)
        if isinstance(value, dict):
            out.update(str(v) for v in value.values() if v)
        elif value:
            out.add(str(value))
    return out


def outcomes(node: object) -> list[list[str]]:
    """Every region list anywhere below a node."""
    found: list[list[str]] = []
    if isinstance(node, dict):
        for key, value in node.items():
            if key == "regions" and isinstance(value, list):
                found.append(value)
            else:
                found += outcomes(value)
    elif isinstance(node, list):
        if node and all(isinstance(v, str) and v in REGIONS for v in node):
            found.append(node)
        for value in node:
            found += outcomes(value)
    return found


def decode(reader: dict, raw: object) -> object:
    """A reference decoder for the kinds whose rule is a plain lookup."""
    rule = reader["decode"]
    kind = rule["kind"]
    for entry in rule["values"]:
        value = entry["value"]
        hit = (
            (kind == "exact-prefix" and str(raw).startswith(value))
            or (kind == "prefix-nocase" and str(raw).lower().startswith(value))
            or (kind == "integer" and raw == value)
            or (kind == "char-at" and len(str(raw)) > rule["index"]
                and str(raw)[rule["index"]] == value)
        )
        if hit:
            return entry.get("regions") or entry.get("standard") or entry.get("label")
    return rule["otherwise"]


def check(cond: dict, data: bytes) -> bool:
    """A reference evaluator for the byte, text and size tests of a predicate."""
    if "all" in cond:
        return all(check(c, data) for c in cond["all"])
    if "any" in cond:
        return any(check(c, data) for c in cond["any"])
    if "file_size" in cond:
        bounds = cond["file_size"]
        return bounds.get("min", 0) <= len(data) <= bounds.get("max", len(data))
    offset = cond["offset"]
    if "text" in cond or "not_text" in cond:
        want = cond.get("text", cond.get("not_text"))
        hit = data[offset:offset + len(want)] == want.encode("latin-1")
        return hit if "text" in cond else not hit
    value = data[offset] & cond.get("mask", 0xFF)
    if "equals" in cond:
        return value == cond["equals"]
    if "not_equals" in cond:
        return value != cond["not_equals"]
    return value in cond["in"]


class HeaderReaders(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.doc = load()
        cls.profiles = load_emulator_profiles(str(ROOT / "emulators"))

    @unittest.skipIf(Draft202012Validator is None, "jsonschema not installed")
    def test_schema(self) -> None:
        schema = json.loads((ROOT / "schemas" / "game-headers.schema.json").read_text())
        errors = [f"{list(e.path)}: {e.message}" for e in Draft202012Validator(schema).iter_errors(self.doc)]
        self.assertEqual(errors, [])

    def test_core_order_names_readers_of_its_system(self) -> None:
        for system, block in self.doc["systems"].items():
            for core, spec in block["cores"].items():
                missing = [r for r in spec["order"] if r not in block["readers"]]
                self.assertEqual(missing, [], f"{system}/{core}")

    def test_every_core_is_a_profile(self) -> None:
        for system, block in self.doc["systems"].items():
            for core in block["cores"]:
                self.assertIn(core, self.profiles, system)

    def test_every_system_is_a_repository_system(self) -> None:
        from romfolders import SystemResolver

        ids = SystemResolver.from_repo().ids
        self.assertEqual([s for s in self.doc["systems"] if s not in ids], [])

    def test_regions_are_vocabulary(self) -> None:
        for regions in outcomes(self.doc["systems"]):
            self.assertLessEqual(set(regions), set(REGIONS), regions)

    def test_core_citations_name_the_profile_pin(self) -> None:
        for system, block in self.doc["systems"].items():
            for core, spec in block["cores"].items():
                match = _CITATION.match(spec["source_ref"])
                self.assertIsNotNone(match, f"{system}/{core}")
                sha = match.group(2)
                self.assertTrue(any(p.startswith(sha) for p in pins(self.profiles[core])),
                                f"{system}/{core}: {sha} is not a pin of the profile")

    def test_reader_citations_name_a_profile_pin(self) -> None:
        every_pin = {p for profile in self.profiles.values() for p in pins(profile)}
        refs = [(f"{s}/{r}", spec["source_ref"]) for s, block in self.doc["systems"].items()
                for r, spec in block["readers"].items()]
        refs += [(f"containers/{c}", spec["source_ref"]) for c, spec in self.doc["containers"].items()]
        for label, ref in refs:
            match = _CITATION.match(ref)
            self.assertIsNotNone(match, label)
            if match.group(1) in TOOL_REPOS:
                continue
            self.assertTrue(any(p.startswith(match.group(2)) for p in every_pin),
                            f"{label}: {match.group(2)} pins no profile")

    def test_reference_decodes(self) -> None:
        psx = self.doc["systems"]["sony-playstation"]["readers"]
        self.assertEqual(decode(psx["psx-serial-prefix"], "SLUS_012.34"), ["north-america"])
        self.assertEqual(decode(psx["psx-serial-prefix"], "SCES_000.01"), ["europe"])
        self.assertEqual(decode(psx["psx-serial-prefix"], "XXXX_000.01"), "unknown")
        self.assertEqual(decode(psx["psx-serial-letter"], "SLPS_012.04"), ["japan"])
        self.assertEqual(decode(psx["psx-licence-offset60"], "Euro pe   "), ["europe"])
        mcd = self.doc["systems"]["sega-megacd"]["readers"]["mcd-security-code"]
        self.assertEqual(decode(mcd, 0x64), ["europe"])
        self.assertEqual(decode(mcd, 0x7A), ["north-america"])
        gc = self.doc["systems"]["nintendo-gamecube"]["readers"]["gc-region"]
        self.assertEqual(decode(gc, 3), "unknown")
        n64 = self.doc["systems"]["nintendo-64"]["readers"]["n64-country"]
        self.assertEqual(decode(n64, 0x45), {"standard": "ntsc"})
        self.assertEqual(decode(n64, 0x50), "pal")

    def test_reader_ids_are_unique(self) -> None:
        ids = [r for block in self.doc["systems"].values() for r in block["readers"]]
        self.assertEqual(len(ids), len(set(ids)))

    def test_every_reader_extras_names_exists(self) -> None:
        extras = yaml.safe_load((ROOT / "games" / "extras.yml").read_text(encoding="utf-8"))
        defined = {r for block in self.doc["systems"].values() for r in block["readers"]}
        named = {
            entry["trigger"]["header"]["reader"]
            for entry in extras["extras"]
            if "header" in entry.get("trigger", {})
        }
        self.assertTrue(named)
        self.assertEqual(sorted(named - defined), [])

    def test_predicate_readers_decode_reference_bytes(self) -> None:
        systems = self.doc["systems"]
        sgb = systems["nintendo-sgb"]["readers"]["gb-sgb-flag"]["decode"]
        header = bytearray(0x150)
        header[0x146], header[0x14B] = 0x03, 0x33
        self.assertTrue(all(check(c, bytes(header)) for c in sgb["all"]))
        header[0x143] = 0x80
        self.assertFalse(all(check(c, bytes(header)) for c in sgb["all"]))
        sufami = systems["nintendo-sufami-turbo"]["readers"]["sufami-magic"]["decode"]
        cart = bytearray(0x80000)
        cart[0:14] = b"BANDAI SFC-ADX"
        self.assertTrue(all(check(c, bytes(cart)) for c in sufami["all"]))
        cart[0x10:0x1E] = b"SFC-ADX BACKUP"
        self.assertFalse(all(check(c, bytes(cart)) for c in sufami["all"]))

    def test_cores_that_pick_no_bios_read_nothing_for_it(self) -> None:
        for system, block in self.doc["systems"].items():
            for core, spec in block["cores"].items():
                if spec.get("selects_bios") is False:
                    self.assertNotIn("preference", spec, f"{system}/{core}")


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