diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 743bb00..f3eaf68 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,6 +52,12 @@ jobs: find config packaging/flatpak -type f -name '*.json' -print0 \ | xargs -0 -n1 python3 -m json.tool >/dev/null + - name: Run script unit tests + shell: bash + run: | + python3 -m pip install --disable-pip-version-check requests + python3 -m unittest discover -s tests -p '*_test.py' -v + - name: Verify build and packaging contracts shell: bash run: | diff --git a/scripts/check_payload_source_drift.py b/scripts/check_payload_source_drift.py new file mode 100755 index 0000000..6037799 --- /dev/null +++ b/scripts/check_payload_source_drift.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +# Copyright 2026 Mocktail Project Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 + +"""Report when the pinned Roblox payload metadata has fallen behind. + +Two kinds of drift make a first launch fail, and neither one shows up until a +user hits it: + +* the compatibility catalog no longer lists the Roblox version the provider + publishes, so every new install lands on a version Roblox has since retired; +* the preferred supported profile has no pinned bootstrap source, so the + pinned bootstrap path cannot produce the version the updater falls back to. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import subprocess +import sys + +MAX_MANIFEST_BYTES = 4 * 1024 * 1024 +PACKAGE = "com.roblox.client" +ARCHITECTURE = "x86_64" + + +class DriftError(RuntimeError): + """Raised when the pinned metadata cannot be inspected at all.""" + + +def _load(path: Path) -> dict: + try: + encoded = path.read_bytes() + except OSError as error: + raise DriftError(f"cannot read {path}") from error + if len(encoded) > MAX_MANIFEST_BYTES: + raise DriftError(f"{path} exceeds its size limit") + try: + document = json.loads(encoded.decode("utf-8")) + except (UnicodeError, json.JSONDecodeError) as error: + raise DriftError(f"cannot parse {path}") from error + if not isinstance(document, dict) or document.get("schema_version") != 1: + raise DriftError(f"{path} has an unsupported schema") + return document + + +def supported_profiles(catalog: dict) -> list[dict]: + profiles = catalog.get("profiles") + if not isinstance(profiles, list): + raise DriftError("compatibility catalog has no profiles") + return [ + profile + for profile in profiles + if isinstance(profile, dict) + and profile.get("status") == "supported" + and profile.get("default_allowed") is True + and isinstance(profile.get("version_name"), str) + and isinstance(profile.get("version_code"), int) + ] + + +def preferred_profile(catalog: dict) -> dict: + """The profile the updater falls back to, matching the C++ coordinator.""" + profiles = supported_profiles(catalog) + if not profiles: + raise DriftError("compatibility catalog has no supported profile") + return max(profiles, key=lambda profile: profile["version_code"]) + + +def pinned_versions(sources: dict) -> set[str]: + entries = sources.get("sources") + if not isinstance(entries, list): + raise DriftError("bootstrap source manifest has no sources") + return { + entry["version_name"] + for entry in entries + if isinstance(entry, dict) + and entry.get("package") == PACKAGE + and entry.get("abi") == ARCHITECTURE + and isinstance(entry.get("version_name"), str) + } + + +def latest_published_version(provider: Path) -> tuple[str, int]: + completed = subprocess.run( + [sys.executable, str(provider), "--check", "--package", PACKAGE, + "--arch", ARCHITECTURE], + capture_output=True, + text=True, + timeout=180, + check=False, + ) + if completed.returncode != 0: + raise DriftError( + "provider metadata is unavailable: " + + (completed.stderr.strip() or "no diagnostic") + ) + try: + reported = json.loads(completed.stdout) + return str(reported["latest_version_name"]), int( + reported["latest_version_code"] + ) + except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error: + raise DriftError("provider metadata is not understood") from error + + +def findings(catalog: dict, sources: dict, latest: tuple[str, int] | None + ) -> list[str]: + reported: list[str] = [] + preferred = preferred_profile(catalog) + pinned = pinned_versions(sources) + if preferred["version_name"] not in pinned: + reported.append( + f"The preferred supported profile {preferred['version_name']} " + f"({preferred['version_code']}) has no pinned bootstrap source. " + "The pinned bootstrap path cannot produce the version the updater " + "falls back to." + ) + if latest is not None: + latest_name, latest_code = latest + if latest_code > preferred["version_code"]: + reported.append( + f"Roblox {latest_name} ({latest_code}) is published, but the " + f"newest supported profile is {preferred['version_name']} " + f"({preferred['version_code']}). New installs land on a " + "version Roblox may already have retired." + ) + return reported + + +def main(argv: list[str] | None = None) -> int: + root = Path(__file__).resolve().parents[1] + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--catalog", type=Path, + default=root / "config/roblox_compatibility.json") + parser.add_argument( + "--sources", type=Path, + default=root / "config/roblox_bootstrap_sources.json") + parser.add_argument( + "--provider", type=Path, + default=root / "scripts/apk_providers/direct_apkpure.py") + parser.add_argument( + "--check-latest", action="store_true", + help="also ask the provider which version is published") + arguments = parser.parse_args(argv) + + try: + catalog = _load(arguments.catalog) + sources = _load(arguments.sources) + latest = ( + latest_published_version(arguments.provider) + if arguments.check_latest + else None + ) + reported = findings(catalog, sources, latest) + except DriftError as error: + print(f"payload source drift: {error}", file=sys.stderr) + return 2 + + if not reported: + print("Pinned Roblox payload metadata is current.") + return 0 + for finding in reported: + print(f"- {finding}") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/check_payload_source_drift_test.py b/tests/check_payload_source_drift_test.py new file mode 100644 index 0000000..9295ed4 --- /dev/null +++ b/tests/check_payload_source_drift_test.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +# Copyright 2026 Mocktail Project Authors +# Licensed under the Apache License, Version 2.0. + +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest + + +SCRIPT = Path(__file__).parents[1] / "scripts/check_payload_source_drift.py" +SPEC = importlib.util.spec_from_file_location("check_payload_source_drift", + SCRIPT) +assert SPEC is not None and SPEC.loader is not None +drift = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(drift) + + +def catalog(*profiles: tuple[str, int, bool]) -> dict: + return { + "schema_version": 1, + "profiles": [ + { + "version_name": name, + "version_code": code, + "status": "supported" if allowed else "legacy-researched", + "default_allowed": allowed, + } + for name, code, allowed in profiles + ], + } + + +def sources(*versions: str) -> dict: + return { + "schema_version": 1, + "sources": [ + { + "package": "com.roblox.client", + "version_name": version, + "abi": "x86_64", + "provider": "uptodown", + } + for version in versions + ], + } + + +class PreferredProfileTest(unittest.TestCase): + def test_picks_the_highest_allowed_version_code(self) -> None: + preferred = drift.preferred_profile( + catalog(("2.725.1142", 2546, True), ("2.727.1199", 2628, True)) + ) + self.assertEqual(preferred["version_name"], "2.727.1199") + + def test_ignores_profiles_that_are_not_default_allowed(self) -> None: + preferred = drift.preferred_profile( + catalog(("2.725.1142", 2546, True), ("2.740.1", 2900, False)) + ) + self.assertEqual(preferred["version_name"], "2.725.1142") + + def test_rejects_a_catalog_without_a_supported_profile(self) -> None: + with self.assertRaises(drift.DriftError): + drift.preferred_profile(catalog(("2.721.1108", 2350, False))) + + +class FindingsTest(unittest.TestCase): + def test_reports_nothing_when_metadata_is_current(self) -> None: + reported = drift.findings( + catalog(("2.727.1199", 2628, True)), + sources("2.727.1199"), + ("2.727.1199", 2628), + ) + self.assertEqual(reported, []) + + def test_reports_an_unpinned_fallback(self) -> None: + reported = drift.findings( + catalog(("2.725.1142", 2546, True), ("2.727.1199", 2628, True)), + sources("2.725.1142"), + None, + ) + self.assertEqual(len(reported), 1) + self.assertIn("no pinned bootstrap source", reported[0]) + self.assertIn("2.727.1199", reported[0]) + + def test_reports_a_catalog_behind_the_published_version(self) -> None: + reported = drift.findings( + catalog(("2.727.1199", 2628, True)), + sources("2.727.1199"), + ("2.734.917", 2908), + ) + self.assertEqual(len(reported), 1) + self.assertIn("2.734.917", reported[0]) + + def test_accepts_a_catalog_ahead_of_the_published_version(self) -> None: + reported = drift.findings( + catalog(("2.727.1199", 2628, True)), + sources("2.727.1199"), + ("2.725.1142", 2546), + ) + self.assertEqual(reported, []) + + def test_ignores_pinned_sources_for_another_architecture(self) -> None: + manifest = sources("2.727.1199") + manifest["sources"][0]["abi"] = "arm64-v8a" + reported = drift.findings( + catalog(("2.727.1199", 2628, True)), manifest, None + ) + self.assertEqual(len(reported), 1) + + +class MainTest(unittest.TestCase): + def test_exits_non_zero_on_drift(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "catalog.json").write_text( + json.dumps(catalog(("2.727.1199", 2628, True))) + ) + (root / "sources.json").write_text(json.dumps(sources())) + self.assertEqual( + drift.main( + [ + "--catalog", str(root / "catalog.json"), + "--sources", str(root / "sources.json"), + ] + ), + 1, + ) + + def test_reports_a_broken_manifest_separately(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "catalog.json").write_text("{") + (root / "sources.json").write_text(json.dumps(sources())) + self.assertEqual( + drift.main( + [ + "--catalog", str(root / "catalog.json"), + "--sources", str(root / "sources.json"), + ] + ), + 2, + ) + + +if __name__ == "__main__": + unittest.main()