diff --git a/scripts/find_pinned_revision.py b/scripts/find_pinned_revision.py new file mode 100644 index 00000000..669a5f76 --- /dev/null +++ b/scripts/find_pinned_revision.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Find the upstream revision whose copy of a file a platform pins. + +A platform list pins the bytes a file had when the list was written: System.dat +names ScummVM's queen.tbl and FBNeo's hiscore.dat by an md5 that today's +upstream no longer carries. The bytes are still in the project's history. This +walks the commits that touched one path, newest first, and hashes the file at +each until one matches. + + python scripts/find_pinned_revision.py --repo https://github.com/scummvm/scummvm \\ + --path dists/engine-data/queen.tbl --md5 4f5b513b52c4ef6fb1c46e984760a3b7 + python scripts/find_pinned_revision.py ... --write bios/Other/ScummVM/.variants/queen.tbl.4f5b513b +""" + +from __future__ import annotations + +import argparse +import hashlib +import sys +import urllib.parse +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import upstream # noqa: E402 + +# Commits listed per page and the most pages walked: a data file older than +# 2 000 commits on its own path is not one a platform list still pins. +PER_PAGE = 100 +MAX_PAGES = 20 + + +def _history_url(repo: upstream.Repo, path: str, page: int) -> str: + quoted = urllib.parse.quote(path) + if repo.family == "gitlab": + project = urllib.parse.quote(repo.slug, safe="") + return (f"{repo.api_base}/projects/{project}/repository/commits" + f"?path={quoted}&per_page={PER_PAGE}&page={page}") + if repo.family == "github": + return (f"{repo.api_base}/repos/{repo.slug}/commits" + f"?path={quoted}&per_page={PER_PAGE}&page={page}") + return (f"{repo.api_base}/repos/{repo.slug}/commits" + f"?path={quoted}&limit={PER_PAGE}&page={page}") + + +def path_history(repo: upstream.Repo, path: str, cache_dir: str) -> list[str]: + """Commits that touched a path, newest first.""" + shas: list[str] = [] + for page in range(1, MAX_PAGES + 1): + payload = upstream._api(_history_url(repo, path, page), cache_dir, False, + max_age=upstream.MOVING_TTL) + if not isinstance(payload, list) or not payload: + break + shas += [str(entry.get("sha") or entry.get("id")) for entry in payload] + if len(payload) < PER_PAGE: + break + return shas + + +def find_revision( + repo: upstream.Repo, path: str, md5: str, cache_dir: str, +) -> tuple[str, bytes] | None: + """The newest revision whose copy of the path has this md5, and its bytes.""" + wanted = md5.lower() + for sha in path_history(repo, path, cache_dir): + data = upstream._fetch(upstream.raw_url(repo, sha, path)) + if data is not None and hashlib.md5(data).hexdigest() == wanted: + return sha, data + return None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--repo", required=True, help="forge URL of the project") + parser.add_argument("--path", required=True, help="path of the file in the project") + parser.add_argument("--md5", required=True, help="md5 the platform pins") + parser.add_argument("--write", type=Path, help="save the matching bytes here") + parser.add_argument("--cache-dir", default=upstream.CACHE_DIR) + args = parser.parse_args() + + repo = upstream.parse_repo(args.repo) + if repo is None: + parser.error(f"not a supported forge URL: {args.repo}") + found = find_revision(repo, args.path, args.md5, args.cache_dir) + if found is None: + print(f"no revision of {args.path} in {repo.slug} has md5 {args.md5}", + file=sys.stderr) + return 1 + sha, data = found + print(f"{sha} {args.path} {args.md5.lower()} {len(data)} bytes") + if args.write: + if args.write.exists() and args.write.read_bytes() != data: + print(f"refusing to overwrite {args.write}: it holds other bytes", + file=sys.stderr) + return 1 + args.write.parent.mkdir(parents=True, exist_ok=True) + args.write.write_bytes(data) + print(f"wrote {args.write}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_find_pinned_revision.py b/tests/test_find_pinned_revision.py new file mode 100644 index 00000000..4651b29f --- /dev/null +++ b/tests/test_find_pinned_revision.py @@ -0,0 +1,44 @@ +"""A platform's pinned revision found in its upstream history. + +RetroArch pins ScummVM's queen.tbl and FBNeo's hiscore.dat at revisions +today's upstream no longer carries; the bytes are in the project's history. +""" + +from __future__ import annotations + +import hashlib +import sys +import unittest +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "scripts")) + +import find_pinned_revision # noqa: E402 +import upstream # noqa: E402 + + +class ThePinnedRevisionIsInTheHistory(unittest.TestCase): + def test_the_newest_matching_revision_is_returned(self): + repo = upstream.parse_repo("https://github.com/scummvm/scummvm") + blobs = {"c3": b"today", "c2": b"pinned", "c1": b"older"} + + def api(url, *_a, **_k): + return [{"sha": s} for s in blobs] if "page=1" in url else [] + + def fetch(url, *_a, **_k): + return next(data for sha, data in blobs.items() if f"/{sha}/" in url) + + with mock.patch.object(upstream, "_api", api), mock.patch.object(upstream, "_fetch", fetch): + found = find_pinned_revision.find_revision( + repo, "dists/engine-data/queen.tbl", + hashlib.md5(b"pinned").hexdigest(), "unused") + missing = find_pinned_revision.find_revision( + repo, "dists/engine-data/queen.tbl", "0" * 32, "unused") + self.assertEqual(found, ("c2", b"pinned")) + self.assertIsNone(missing) + + +if __name__ == "__main__": + unittest.main()