diff --git a/scripts/profile_sync.py b/scripts/profile_sync.py new file mode 100644 index 00000000..336009a8 --- /dev/null +++ b/scripts/profile_sync.py @@ -0,0 +1,1236 @@ +#!/usr/bin/env python3 +"""Confront emulator profiles with their upstream source. + +Anchors every source_ref of a profile against the commit the profile was +written at and against upstream HEAD, separating what is mechanically +recalable from what needs the code read again. +""" + +from __future__ import annotations + +import argparse +import difflib +import json +import os +import posixpath +import re +import subprocess +import sys +from dataclasses import asdict, dataclass +from datetime import date +from pathlib import Path + +import yaml + +sys.path.insert(0, os.path.dirname(__file__)) + +import upstream +from common import load_emulator_profiles +from upstream import CompareResult, find_renamed + +DEFAULT_CACHE = ".cache/upstream" +ANON_QUOTA = 60 +TRIAGE_PATH_SAMPLE = 5 + +STATUS_ORDER = ("ANCHORED", "SHIFTED", "RENAMED", "AMBIGUOUS", "CHANGED", "GONE") +REVIEW_STATUSES = ("CHANGED", "GONE", "AMBIGUOUS") +REBASE_STATUSES = ("SHIFTED", "RENAMED") + +WIDEN_STEPS = (0, 3, 6, 12) +MAX_MATCH_LINES = 20000 + +PIN = "pin" +HEAD = "head" + +_REF_RE = re.compile(r"^(?P[^:]+?)(?::(?P\d+)(?:-(?P\d+))?)?$") +_BARE_RANGE_RE = re.compile(r"^\d+(?:-\d+)?$") + + +@dataclass(frozen=True) +class RefPart: + path: str + start: int | None + end: int | None + + +@dataclass(frozen=True) +class AnchorResult: + status: str + start: int | None + end: int | None + candidates: list[int] + reason: str | None = None + + +@dataclass(frozen=True) +class PartResult: + part: RefPart + status: str + new_path: str | None + start: int | None + end: int | None + candidates: list[int] + reason: str | None = None + repo: str | None = None + head_url: str | None = None + + +@dataclass(frozen=True) +class RepoView: + """One repository a profile cites, at the two revisions under comparison.""" + + repo: object + pin: str + head: str + origin: str + field: str + + +def parse_source_ref(ref: str) -> tuple[str, int | None, int | None]: + """Split 'path/file.cpp:123-129' into (path, start, end).""" + m = _REF_RE.match(ref.strip()) + if not m: + return ref.strip(), None, None + start = int(m.group("start")) if m.group("start") else None + end = int(m.group("end")) if m.group("end") else start + return m.group("path"), start, end + + +def split_source_ref(ref: str) -> list[RefPart]: + """A source_ref may carry several comma-separated references. + + A part reduced to a line or a line range continues the previous part's + file: `geo.c:234-243, 273-285` cites two ranges of the same file. 430 + parts across 69 profiles use that form. + """ + parts: list[RefPart] = [] + for chunk in (c.strip() for c in str(ref or "").split(",")): + if not chunk: + continue + if _BARE_RANGE_RE.match(chunk) and parts: + start, _, end = chunk.partition("-") + parts.append( + RefPart(parts[-1].path, int(start), int(end or start)) + ) + continue + path, start_line, end_line = parse_source_ref(chunk) + parts.append(RefPart(path, start_line, end_line)) + return parts + + +def collect_tokens(entry: dict) -> list[str]: + """Values declared by one file entry: hashes, then name as fallback.""" + tokens: list[str] = [] + for field in ("sha1", "md5", "crc32", "sha256", "known_hash_adler32"): + val = entry.get(field) + vals = val if isinstance(val, list) else [val] if val else [] + for v in vals: + v = str(v).lower().removeprefix("0x") + if v: + tokens.append(v) + if not tokens: + name = entry.get("name", "") + if name: + tokens.append(os.path.basename(name).lower()) + return tokens + + +def worst_status(statuses) -> str: + """Severity of an entry is the worst severity among its parts.""" + worst = "ANCHORED" + for status in statuses: + if STATUS_ORDER.index(status) > STATUS_ORDER.index(worst): + worst = status + return worst + + +def _normalize(lines: list[str]) -> list[str]: + return [line.strip() for line in lines] + + +def _find_all(haystack: list[str], needle: list[str]) -> list[int]: + """Zero-based offsets where the needle sequence occurs.""" + size = len(needle) + if not size or size > len(haystack): + return [] + first = needle[0] + hits = [] + for i in range(len(haystack) - size + 1): + if haystack[i] == first and haystack[i : i + size] == needle: + hits.append(i) + return hits + + +def _map_index(opcodes, index: int) -> int | None: + for tag, i1, i2, j1, j2 in opcodes: + if i1 <= index < i2: + return j1 + (index - i1) if tag == "equal" else j1 + return None + + +def _map_changed(pin: list[str], head: list[str], lo: int, hi: int) -> AnchorResult: + """Map a pinned range onto HEAD once exact anchoring has failed.""" + if len(pin) > MAX_MATCH_LINES or len(head) > MAX_MATCH_LINES: + return AnchorResult( + "CHANGED", None, None, [], f"file over {MAX_MATCH_LINES} lines" + ) + matcher = difflib.SequenceMatcher(None, pin, head, autojunk=False) + opcodes = matcher.get_opcodes() + if not any(tag == "equal" for tag, *_ in opcodes): + return AnchorResult("GONE", None, None, []) + new_lo = _map_index(opcodes, lo) + new_hi = _map_index(opcodes, max(lo, hi - 1)) + if new_lo is None or new_hi is None: + return AnchorResult("GONE", None, None, []) + return AnchorResult("CHANGED", new_lo + 1, max(new_lo, new_hi) + 1, []) + + +def anchor_block( + pin_lines: list[str], head_lines: list[str], start: int, end: int +) -> AnchorResult: + """Locate the pinned line range inside the HEAD revision of a file.""" + pin = _normalize(pin_lines) + head = _normalize(head_lines) + lo, hi = start - 1, end + if lo < 0 or lo >= len(pin): + return AnchorResult("GONE", None, None, []) + if not any(pin[lo:hi]): + return AnchorResult("CHANGED", None, None, [], "cited range is blank") + + candidates: list[int] = [] + for pad in WIDEN_STEPS: + a = max(0, lo - pad) + b = min(len(pin), hi + pad) + hits = _find_all(head, pin[a:b]) + offset = lo - a + if len(hits) == 1: + new_start = hits[0] + offset + 1 + new_end = new_start + (hi - lo) - 1 + status = "ANCHORED" if new_start == start else "SHIFTED" + return AnchorResult(status, new_start, new_end, []) + if not hits: + break + candidates = [h + offset + 1 for h in hits] + else: + return AnchorResult("AMBIGUOUS", None, None, candidates) + + if candidates: + return AnchorResult("AMBIGUOUS", None, None, candidates) + return _map_changed(pin, head, lo, hi) + + +def resolve_rename( + result: CompareResult, path: str, head_paths=() +) -> tuple[str | None, list[str]]: + """New path of a moved file, or the candidates when several match. + + The comparison is authoritative when it reports the rename. Otherwise the + HEAD tree is searched by basename, then by stem: a C++ file rewritten in C + keeps its stem and loses its basename. + """ + renamed = find_renamed(result, path) + if renamed: + return renamed, [] + pool = list(head_paths) or [change.path for change in result.files] + base = posixpath.basename(path) + matches = _narrow( + [p for p in pool if p != path and posixpath.basename(p) == base], path + ) + if len(matches) == 1: + return matches[0], [] + if not matches: + stem = base.rsplit(".", 1)[0] + matches = _narrow( + [ + p + for p in pool + if p != path and posixpath.basename(p).rsplit(".", 1)[0] == stem + ], + path, + ) + if len(matches) == 1: + return matches[0], [] + return None, matches + + +def _narrow(matches: list[str], path: str) -> list[str]: + """Prefer candidates sitting in the directory the file came from.""" + if len(matches) < 2: + return matches + directory = posixpath.dirname(path) + same = [p for p in matches if posixpath.dirname(p) == directory] + return same if len(same) == 1 else matches + + +def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResult: + """Locate one reference part at HEAD, following a rename when needed. + + `describe(path)` returns (repo slug, raw URL at HEAD) for the repository + that owns the path, or (None, None) when the caller does not track it. + """ + slug, url, actual = ( + describe(part.path) if describe else (None, None, part.path) + ) + head_lines = fetch(HEAD, part.path) + path = actual + renamed = actual != part.path + + if head_lines is None: + moved, candidates = rename_getter(part.path) + if moved is None: + if candidates: + return PartResult( + part, + "AMBIGUOUS", + None, + None, + None, + [], + f"{len(candidates)} candidates: {', '.join(candidates[:5])}", + slug, + url, + ) + return PartResult( + part, + "GONE", + None, + None, + None, + [], + "absent at HEAD, no rename found", + slug, + url, + ) + path = moved + head_lines = fetch(HEAD, path) + if head_lines is None: + return PartResult( + part, "GONE", None, None, None, [], None, slug, url + ) + renamed = True + if describe: + slug, url, _ = describe(path) + + if part.start is None: + status = "RENAMED" if renamed else "ANCHORED" + return PartResult( + part, status, path if renamed else None, None, None, [], None, slug, url + ) + + pin_lines = fetch(PIN, part.path) + if pin_lines is None: + return PartResult( + part, "GONE", None, None, None, [], "pin revision missing", slug, url + ) + + anchored = anchor_block(pin_lines, head_lines, part.start, part.end or part.start) + status = anchored.status + if renamed and status in ("ANCHORED", "SHIFTED"): + status = "RENAMED" + return PartResult( + part, + status, + path if renamed else None, + anchored.start, + anchored.end, + anchored.candidates, + anchored.reason, + slug, + url, + ) + + +@dataclass +class EntryReport: + name: str + source_ref: str + status: str + parts: list[PartResult] + + +@dataclass +class ProfileReport: + name: str + repo: str | None = None + repos: list[str] = None + host: str | None = None + pin: str | None = None + pin_origin: str | None = None + head: str | None = None + entries: list[EntryReport] = None + skipped: str | None = None + counts: dict[str, int] = None + + def needs_review(self) -> int: + counts = self.counts or {} + return sum(counts.get(status, 0) for status in REVIEW_STATUSES) + + +def select_repo(profile: dict) -> upstream.Repo | None: + """Repository the profile was read from: source first, then upstream.""" + for field in ("source", "upstream"): + repo = upstream.parse_repo(str(profile.get(field) or "")) + if repo is not None: + return repo + return None + + +def resolve_pin( + profile: dict, repo, cache_dir: str, offline: bool, field: str = "source" +) -> tuple[str | None, str | None]: + """Commit the profile was written at, and how it was obtained. + + `field` selects which declared pin applies: `source_commit` for the port + the profile was read from, `upstream_commit` for the original project. + """ + pinned = profile.get(f"{field}_commit") + if pinned: + return str(pinned), f"{field}_commit" + date = str(profile.get("profiled_date") or "") + if not date: + return None, None + sha = upstream.resolve_commit_at(repo, date, cache_dir, offline) + return sha, (f"profiled_date {date}" if sha else None) + + +def select_views( + profile: dict, cache_dir: str, offline: bool +) -> list[RepoView]: + """Every repository the profile cites, source before upstream. + + 241 profiles declare a `source` distinct from their `upstream`, and 168 of + those have both on a supported forge. Their refs mix paths from the two, so + a single repository cannot resolve them all. + """ + views: list[RepoView] = [] + seen: set[tuple[str, str]] = set() + for field in ("source", "upstream"): + repo = upstream.parse_repo(str(profile.get(field) or "")) + if repo is None or (repo.host, repo.slug) in seen: + continue + pin, origin = resolve_pin(profile, repo, cache_dir, offline, field) + if not pin: + continue + head = upstream.resolve_head(repo, cache_dir, offline) + if not head: + continue + seen.add((repo.host, repo.slug)) + views.append(RepoView(repo, pin, head, origin, field)) + return views + + +def build_report( + name: str, profile: dict, cache_dir: str, offline: bool = False +) -> ProfileReport: + """Confront one profile with its upstream.""" + report = ProfileReport(name=name, entries=[], counts={}) + + refs = [ + (entry.get("name", ""), str(entry.get("source_ref"))) + for entry in (profile.get("files") or []) + if isinstance(entry, dict) and entry.get("source_ref") + ] + + if select_repo(profile) is None: + declared = str(profile.get("source") or profile.get("upstream") or "") + report.skipped = f"unsupported host: {declared or 'none declared'}" + return report + + views = select_views(profile, cache_dir, offline) + if not views: + report.skipped = ( + "no source_commit and no resolvable profiled_date " + "on any declared repository" + ) + return report + primary = views[0] + report.repo, report.host = primary.repo.slug, primary.repo.host + report.repos = [v.repo.slug for v in views] + report.pin, report.pin_origin = primary.pin, primary.origin + report.head = primary.head + + # A profile carrying no source_ref still has a pin worth writing and a + # version worth checking, so the revisions above are resolved first. + if not refs: + report.skipped = "no source_ref" + return report + + owners: dict[str, RepoView] = {} + context: dict[str, object] = {} + + def _locate(path: str) -> tuple[RepoView, str]: + """Repository and real path carrying a cited path, HEAD before pin. + + A ref may prefix the path with the repository directory name, as it + appears in a parent folder holding both clones: 270 parts across 21 + profiles do. That prefix is stripped only as a last resort, once the + path as written has failed against every repository and revision. + """ + candidates = [path] + head, _, tail = path.partition("/") + if tail and any(head == v.repo.name for v in views): + candidates.append(tail) + for candidate in candidates: + for sha_of in (lambda v: v.head, lambda v: v.pin): + for view in views: + found = upstream.fetch_file( + view.repo, sha_of(view), candidate, cache_dir, offline + ) + if found is not None: + return view, candidate + return primary, path + + def resolve_path(path: str) -> tuple[RepoView, str]: + if path not in owners: + owners[path] = _locate(path) + return owners[path] + + def rename_getter(path: str) -> tuple[str | None, list[str]]: + """Resolved only when a cited file has vanished, then memoised.""" + view, actual = resolve_path(path) + key = view.repo.slug + if key not in context: + comparison = upstream.compare( + view.repo, view.pin, view.head, cache_dir, offline + ) + tree, truncated = upstream.list_tree( + view.repo, view.head, cache_dir, offline + ) + context[key] = (comparison, tree) + if truncated and tree: + print( + f"{name}: {key} HEAD tree truncated by the forge, " + "rename search is partial", + file=sys.stderr, + ) + comparison, tree = context[key] + return resolve_rename(comparison, actual, tree) + + def fetch(which: str, path: str): + view, actual = resolve_path(path) + sha = view.pin if which == PIN else view.head + return upstream.fetch_file(view.repo, sha, actual, cache_dir, offline) + + def describe(path: str) -> tuple[str | None, str | None, str]: + view, actual = resolve_path(path) + slug = view.repo.slug if view is not primary else None + return slug, upstream.raw_url(view.repo, view.head, actual), actual + + for entry_name, ref in refs: + parts = [ + anchor_part(part, fetch, rename_getter, describe) + for part in split_source_ref(ref) + ] + status = worst_status([p.status for p in parts]) + report.entries.append(EntryReport(entry_name, ref, status, parts)) + report.counts[status] = report.counts.get(status, 0) + 1 + + return report + + +def _render_span(path: str, start: int | None, end: int | None) -> str: + if start is None: + return path + if end and end != start: + return f"{path}:{start}-{end}" + return f"{path}:{start}" + + +def _original_part(part: PartResult) -> str: + return _render_span(part.part.path, part.part.start, part.part.end) + + +def _rendered_part(part: PartResult) -> str: + return _render_span(part.new_path or part.part.path, part.start, part.end) + + +def _part_line(part: PartResult) -> str: + detail = part.status + if part.new_path: + detail += f" -> {part.new_path}" + if part.start is not None and part.status != "ANCHORED": + span = f"{part.start}" + if part.end and part.end != part.start: + span += f"-{part.end}" + detail += f" -> {span}" + if part.candidates: + detail += f" candidats: {', '.join(str(c) for c in part.candidates)}" + if part.reason: + detail += f" ({part.reason})" + if part.repo: + detail += f" [{part.repo}]" + return f" {_original_part(part)} {detail}" + + +def format_report(report: ProfileReport, changed_only: bool = False) -> str: + """Human-readable report for one profile.""" + slugs = " + ".join(report.repos) if report.repos else (report.repo or "") + lines = [f"{report.name} {slugs}".rstrip()] + if report.pin and report.head: + lines.append( + f" {report.pin[:7]} -> {report.head[:7]} pin: {report.pin_origin}" + ) + if report.skipped: + lines.append(f" skipped: {report.skipped}") + return "\n".join(lines) + for entry in report.entries: + if changed_only and entry.status not in REVIEW_STATUSES: + continue + lines.append(f" {entry.name} {entry.source_ref}") + for part in entry.parts: + lines.append(_part_line(part)) + summary = ", ".join( + f"{count} {status.lower()}" for status, count in sorted(report.counts.items()) + ) + lines.append(f" {len(report.entries)} refs: {summary}") + lines.append(f" {report.needs_review()} demandent une relecture") + return "\n".join(lines) + + +def report_to_dict(report: ProfileReport) -> dict: + """Serialisable form of one report.""" + payload = asdict(report) + payload["needs_review"] = report.needs_review() + return payload + + +def format_markdown(reports: list[ProfileReport]) -> str: + """Report set as a markdown document.""" + lines = [ + "# profile-sync", + "", + "| profil | depot | refs | relecture |", + "|---|---|---|---|", + ] + for report in reports: + if report.skipped: + lines.append( + f"| {report.name} | {report.repo or ''} | skipped | {report.skipped} |" + ) + continue + lines.append( + f"| {report.name} | {report.repo} | {len(report.entries)} " + f"| {report.needs_review()} |" + ) + for report in reports: + lines.extend(["", "```", format_report(report), "```"]) + return "\n".join(lines) + + +def fetch_plan(report: ProfileReport) -> list[str]: + """Raw URLs at HEAD for the refs that need the code read again.""" + if report.skipped or not report.head or not report.host: + return [] + owner, _, name = (report.repo or "").partition("/") + fallback = upstream.make_repo(report.host, owner, name) + urls = [] + for entry in report.entries: + if entry.status not in REVIEW_STATUSES: + continue + for part in entry.parts: + url = part.head_url + if url is None: + if fallback is None: + continue + path = part.new_path or part.part.path + url = upstream.raw_url(fallback, report.head, path) + if part.part.start: + url += f"#L{part.part.start}" + urls.append(url) + return urls + + +@dataclass(frozen=True) +class VersionReport: + declared: str + latest_tag: str | None + latest_release: str | None + release_date: str | None + tag_matches_declared: bool + tag_commit: str | None + + +def check_version( + profile: dict, repo, cache_dir: str, offline: bool +) -> VersionReport | None: + """Declared core_version against the latest upstream tag and release.""" + declared = str(profile.get("core_version") or "") + if not declared: + return None + tags = upstream.list_tags(repo, cache_dir, offline) + release = upstream.latest_release(repo, cache_dir, offline) + matches = declared in tags + commit = ( + upstream.resolve_tag_commit(repo, declared, cache_dir, offline) + if matches + else None + ) + return VersionReport( + declared, + tags[0] if tags else None, + release.tag if release else None, + release.date if release else None, + matches, + commit, + ) + + +def version_warning(report: VersionReport | None, pin: str | None) -> str | None: + """Flag a declared version that names a tag away from the resolved pin.""" + if report is None or not report.tag_commit or not pin: + return None + if report.tag_commit == pin: + return None + return ( + f"core_version {report.declared} names tag commit {report.tag_commit[:7]}, " + f"the pin resolved to {pin[:7]}" + ) + + +FILE_EXTENSIONS = ( + "bin", "rom", "zip", "dat", "bios", "img", "chd", "nvram", "sav", + "ips", "pal", "fnt", +) + +_FILE_RE = re.compile( + r"[\w./+-]+\.(?:" + "|".join(FILE_EXTENSIONS) + r")\b", re.IGNORECASE +) +_SHA1_RE = re.compile(r"\b[0-9a-fA-F]{40}\b") +_MD5_RE = re.compile(r"\b[0-9a-fA-F]{32}\b") +_CRC_RE = re.compile(r"\b(?:0x)?([0-9a-fA-F]{8})\b") + + +def declared_names(profile: dict) -> set[str]: + """Every filename the profile covers, aliases included.""" + names = set() + for entry in profile.get("files") or []: + if not isinstance(entry, dict): + continue + for value in [entry.get("name", "")] + list(entry.get("aliases") or []): + if value: + names.add(os.path.basename(str(value)).casefold()) + return names + + +def declared_hashes(profile: dict) -> set[str]: + """Every hash the profile declares.""" + hashes = set() + for entry in profile.get("files") or []: + if isinstance(entry, dict): + hashes.update(collect_tokens(entry)) + return hashes + + +def detect_new_files( + head_lines: list[str], known: set[str], cap: int = 25 +) -> tuple[list[tuple[int, str]], int]: + """Filename literals at HEAD the profile does not declare.""" + found: list[tuple[int, str]] = [] + seen: set[str] = set() + elided = 0 + for number, line in enumerate(head_lines, start=1): + for match in _FILE_RE.findall(line): + name = os.path.basename(match).casefold() + if name in known or name in seen: + continue + seen.add(name) + if len(found) >= cap: + elided += 1 + continue + found.append((number, os.path.basename(match))) + return found, elided + + +def watch_hashes(added_lines: list[str], known: set[str]) -> list[tuple[str, str]]: + """Hash literals in added lines that no profile entry declares.""" + found: list[tuple[str, str]] = [] + for line in added_lines: + strong = bool(_SHA1_RE.search(line) or _MD5_RE.search(line)) + for pattern, kind in ((_SHA1_RE, "sha1"), (_MD5_RE, "md5")): + for value in pattern.findall(line): + if value.lower() not in known: + found.append((kind, value.lower())) + if strong or not _FILE_RE.search(line): + continue + for value in _CRC_RE.findall(line): + if value.lower() not in known: + found.append(("crc32", value.lower())) + return found + + +def unified_for_path( + pin_lines: list[str], head_lines: list[str], path: str, context: int +) -> str: + """Unified diff of one cited file between the two revisions.""" + diff = difflib.unified_diff( + pin_lines, + head_lines, + fromfile=f"a/{path}", + tofile=f"b/{path}", + n=context, + lineterm="", + ) + return "\n".join(diff) + + +def tree_diff(result: CompareResult, ref_dirs: set[str]) -> list[str]: + """Tree changes restricted to the directories the refs point at.""" + lines = [] + for change in result.files: + directory = posixpath.dirname(change.path) + if ref_dirs and not any( + directory == d or directory.startswith(d + "/") for d in ref_dirs + ): + continue + if change.status == "renamed" and change.previous_path: + lines.append(f"renamed {change.previous_path} -> {change.path}") + else: + lines.append(f"{change.status} {change.path}") + if result.truncated: + lines.append("comparison truncated by the forge, list is partial") + return lines + + +DRIFT_WEIGHTS = {"GONE": 40, "CHANGED": 30, "AMBIGUOUS": 15, "SHIFTED": 1} + + +def drift_score( + report: ProfileReport, version: VersionReport | None, commits: int +) -> int: + """Rank profiles by how far they have drifted from their pin.""" + counts = report.counts or {} + score = sum(DRIFT_WEIGHTS.get(k, 0) * v for k, v in counts.items()) + if version is not None and not version.tag_matches_declared: + if version.latest_release and version.latest_release != version.declared: + score += 10 + score += min(commits, 20) // 4 + return score + + +class YamlWriteError(Exception): + """A profile edit did not land exactly as intended.""" + + +def _top_level_indices(lines: list[str], key: str) -> list[int]: + """Indices of lines declaring a key at the top level of the document.""" + prefix = f"{key}:" + return [i for i, line in enumerate(lines) if line.startswith(prefix)] + + +def insert_after_line(text: str, key: str, new_line: str) -> str: + """Insert a line right after a top-level key.""" + lines = text.splitlines() + indices = _top_level_indices(lines, key) + if len(indices) != 1: + raise YamlWriteError( + f"{key}: expected one top-level line, found {len(indices)}" + ) + lines.insert(indices[0] + 1, new_line) + return "\n".join(lines) + ("\n" if text.endswith("\n") else "") + + +def replace_field_line( + text: str, key: str, occurrence: int, expected: str, value: str +) -> str: + """Rewrite the Nth key line, only if it currently holds the expected value.""" + lines = text.splitlines() + marker = f"{key}:" + found = [i for i, line in enumerate(lines) if line.strip().startswith(marker)] + if occurrence >= len(found): + raise YamlWriteError(f"{key}: occurrence {occurrence} not found") + index = found[occurrence] + current = lines[index].split(":", 1)[1].strip().strip('"').strip("'") + if current != expected: + raise YamlWriteError( + f"{key}: line {index + 1} holds {current!r}, expected {expected!r}" + ) + indent = lines[index][: len(lines[index]) - len(lines[index].lstrip())] + lines[index] = f'{indent}{key}: "{value}"' + return "\n".join(lines) + ("\n" if text.endswith("\n") else "") + + +def apply_edit(path: Path, new_text: str, expected: dict) -> None: + """Write, then verify the parsed document matches what was intended.""" + original = path.read_text(encoding="utf-8") + path.write_text(new_text, encoding="utf-8") + try: + written = yaml.safe_load(new_text) + except yaml.YAMLError as exc: + path.write_text(original, encoding="utf-8") + raise YamlWriteError(f"{path}: parse failed after write: {exc}") from exc + if written != expected: + path.write_text(original, encoding="utf-8") + raise YamlWriteError(f"{path}: structure changed beyond the intended field") + + +def backfill_commit(path: Path, sha: str) -> bool: + """Insert source_commit after profiled_date. False when already present.""" + text = path.read_text(encoding="utf-8") + document = yaml.safe_load(text) + if document.get("source_commit"): + return False + new_text = insert_after_line(text, "profiled_date", f'source_commit: "{sha}"') + expected = dict(document) + expected["source_commit"] = sha + apply_edit(path, new_text, expected) + return True + + +def rebase_refs(path: Path, report: ProfileReport) -> list[str]: + """Recale the line ranges of parts whose content is unchanged.""" + text = path.read_text(encoding="utf-8") + document = yaml.safe_load(text) + carriers = [ + entry + for entry in (document.get("files") or []) + if isinstance(entry, dict) and entry.get("source_ref") + ] + applied: list[str] = [] + + for occurrence, entry in enumerate(report.entries): + rendered = [] + touched = False + for part in entry.parts: + if part.status in REBASE_STATUSES and part.start is not None: + recaled = _rendered_part(part) + if recaled != _original_part(part): + touched = True + rendered.append(recaled) + else: + rendered.append(_original_part(part)) + if not touched: + continue + new_ref = ", ".join(rendered) + text = replace_field_line( + text, "source_ref", occurrence, entry.source_ref, new_ref + ) + carriers[occurrence]["source_ref"] = new_ref + applied.append(f"{entry.source_ref} -> {new_ref}") + + if applied: + apply_edit(path, text, document) + return applied + + +def bump_commit(path: Path, report: ProfileReport) -> bool: + """Advance source_commit to HEAD when nothing needs a read again.""" + if report.skipped or report.needs_review() or not report.head: + return False + text = path.read_text(encoding="utf-8") + document = yaml.safe_load(text) + expected = dict(document) + expected["source_commit"] = report.head + if document.get("source_commit"): + new_text = replace_field_line( + text, "source_commit", 0, str(document["source_commit"]), report.head + ) + else: + new_text = insert_after_line( + text, "profiled_date", f'source_commit: "{report.head}"' + ) + apply_edit(path, new_text, expected) + return True + + +def emulators_dir_is_dirty(emulators_dir: str) -> bool: + """True when the profile directory carries uncommitted changes.""" + result = subprocess.run( + ["git", "status", "--porcelain", "--", emulators_dir], + capture_output=True, + text=True, + check=False, + ) + return bool(result.stdout.strip()) + + +def build_parser() -> argparse.ArgumentParser: + """Command line: selection, output, detection, writes, network.""" + parser = argparse.ArgumentParser( + description="Confront emulator profiles with their upstream source" + ) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--emulator", help="a single profile") + group.add_argument("--all", action="store_true", help="every profile") + parser.add_argument("--emulators-dir", default="emulators") + parser.add_argument("--system", help="keep profiles covering this system") + parser.add_argument("--type", dest="type_", help="keep profiles of this type") + parser.add_argument("--stale-before", help="keep profiles profiled before a date") + parser.add_argument("--limit", type=int, help="stop after N profiles") + parser.add_argument("--changed-only", action="store_true") + parser.add_argument("--json", action="store_true", dest="as_json") + parser.add_argument("--markdown", action="store_true") + parser.add_argument("--fetch-plan", action="store_true") + parser.add_argument("--full-diff", action="store_true") + parser.add_argument("--ref", help="restrict --full-diff to one cited path") + parser.add_argument("--context", type=int, default=3) + parser.add_argument("--check-version", action="store_true") + parser.add_argument("--detect-new-files", action="store_true") + parser.add_argument("--watch-hashes", action="store_true") + parser.add_argument("--tree-diff", action="store_true") + parser.add_argument("--triage", action="store_true") + parser.add_argument("--backfill-commits", action="store_true") + parser.add_argument("--rebase-refs", action="store_true") + parser.add_argument("--bump-commit", action="store_true") + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--force", action="store_true") + parser.add_argument("--offline", action="store_true") + parser.add_argument("--cache-dir", default=DEFAULT_CACHE) + return parser + + +def select_profiles(profiles: dict, args) -> dict: + """Profiles matching the selection flags.""" + if args.emulator: + if args.emulator not in profiles: + print(f"unknown profile: {args.emulator}", file=sys.stderr) + raise SystemExit(1) + return {args.emulator: profiles[args.emulator]} + + selected = {} + for name, profile in sorted(profiles.items()): + if profile.get("type") in ("alias", "test"): + continue + if args.system and args.system not in (profile.get("systems") or []): + continue + if args.type_ and profile.get("type") != args.type_: + continue + if args.stale_before: + profiled = str(profile.get("profiled_date") or "") + if not profiled or profiled >= args.stale_before: + continue + selected[name] = profile + if args.limit and len(selected) >= args.limit: + break + return selected + + +def _check_quota(count: int, offline: bool) -> None: + """Refuse a run the anonymous quota cannot carry to completion.""" + if offline or os.environ.get("GITHUB_TOKEN"): + return + if count * 4 > ANON_QUOTA: + print( + f"{count} profiles need roughly {count * 4} API calls, the anonymous " + f"quota is {ANON_QUOTA} per hour. Set GITHUB_TOKEN or use --offline.", + file=sys.stderr, + ) + raise SystemExit(1) + + +def _apply_writes(args, name: str, profile: dict, report: ProfileReport) -> None: + path = Path(args.emulators_dir) / f"{name}.yml" + if not path.is_file(): + return + if args.backfill_commits and report.pin and not profile.get("source_commit"): + if args.dry_run: + print(f"{name}: would write source_commit {report.pin[:7]}") + elif backfill_commit(path, report.pin): + print(f"{name}: source_commit {report.pin[:7]}") + if args.rebase_refs and not args.dry_run and not report.skipped: + for line in rebase_refs(path, report): + print(f"{name}: {line}") + if args.bump_commit and not args.dry_run and bump_commit(path, report): + print(f"{name}: source_commit -> {report.head[:7]}") + + +def _cited_paths(report: ProfileReport) -> list[str]: + return sorted( + {part.new_path or part.part.path for e in report.entries for part in e.parts} + ) + + +def _print_version(args, profile: dict, report: ProfileReport, repo) -> None: + version = check_version(profile, repo, args.cache_dir, args.offline) + if version is None: + return + print( + f" version: declared {version.declared}, " + f"latest tag {version.latest_tag}, " + f"latest release {version.latest_release} ({version.release_date})" + ) + warning = version_warning(version, report.pin) + if warning: + print(f" {warning}") + + +def _print_detection(args, profile: dict, report: ProfileReport, repo) -> None: + names, hashes = declared_names(profile), declared_hashes(profile) + for path in _cited_paths(report): + head_lines = upstream.fetch_file( + repo, report.head, path, args.cache_dir, args.offline + ) + if head_lines is None: + continue + if args.detect_new_files: + found, elided = detect_new_files(head_lines, names) + for number, name in found: + print(f" new file: {path}:{number} {name}") + if elided: + print(f" new file: {elided} further matches not shown") + if args.watch_hashes: + pin_lines = ( + upstream.fetch_file( + repo, report.pin, path, args.cache_dir, args.offline + ) + or [] + ) + added = [ + line + for line in difflib.unified_diff(pin_lines, head_lines, n=0) + if line.startswith("+") and not line.startswith("+++") + ] + for kind, value in watch_hashes(added, hashes): + print(f" new {kind}: {path} {value}") + + +def _print_extras(args, profile: dict, report: ProfileReport) -> None: + """Optional per-profile sections beyond the ref report.""" + wants = ( + args.check_version + or args.detect_new_files + or args.watch_hashes + or args.full_diff + or args.tree_diff + ) + if not wants or not report.head: + return + repo = select_repo(profile) + if repo is None: + return + + if args.check_version: + _print_version(args, profile, report, repo) + if args.detect_new_files or args.watch_hashes: + _print_detection(args, profile, report, repo) + if args.full_diff: + for path in _cited_paths(report): + if args.ref and args.ref not in path: + continue + pin_lines = ( + upstream.fetch_file( + repo, report.pin, path, args.cache_dir, args.offline + ) + or [] + ) + head_lines = ( + upstream.fetch_file( + repo, report.head, path, args.cache_dir, args.offline + ) + or [] + ) + text = unified_for_path(pin_lines, head_lines, path, args.context) + if text: + print(text) + if args.tree_diff: + comparison = upstream.compare( + repo, report.pin, report.head, args.cache_dir, args.offline + ) + # The repository root is a directory like any other: dropping the + # empty dirname would turn the filter off and list the whole tree. + ref_dirs = {posixpath.dirname(p) for p in _cited_paths(report)} + for line in tree_diff(comparison, ref_dirs): + print(f" {line}") + + +def _print_triage(args, selected: dict, reports: list[ProfileReport]) -> None: + """Profiles ranked by drift, worst first.""" + ranked = [] + for report in reports: + profile = selected[report.name] + version = None + commits = 0 + repo = select_repo(profile) + if report.head and repo is not None: + version = check_version(profile, repo, args.cache_dir, args.offline) + paths = _cited_paths(report) + for path in paths[:TRIAGE_PATH_SAMPLE]: + commits += upstream.commits_touching( + repo, path, report.pin, args.cache_dir, args.offline + ) + if len(paths) > TRIAGE_PATH_SAMPLE: + print( + f"{report.name}: commit count sampled on " + f"{TRIAGE_PATH_SAMPLE} of {len(paths)} paths", + file=sys.stderr, + ) + ranked.append((drift_score(report, version, commits), report)) + for score, report in sorted(ranked, key=lambda item: -item[0]): + state = report.skipped or f"{report.needs_review()} to review" + print(f"{score:5d} {report.name:30s} {state}") + + +def main() -> None: + """Build one report per selected profile, then apply writes and output.""" + args = build_parser().parse_args() + profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False) + selected = select_profiles(profiles, args) + _check_quota(len(selected), args.offline) + + writes = args.backfill_commits or args.rebase_refs or args.bump_commit + if ( + writes + and not args.dry_run + and not args.force + and emulators_dir_is_dirty(args.emulators_dir) + ): + print( + f"{args.emulators_dir} carries uncommitted changes. " + "Commit them first or pass --force.", + file=sys.stderr, + ) + raise SystemExit(1) + + reports = [] + for name, profile in selected.items(): + report = build_report(name, profile, args.cache_dir, args.offline) + reports.append(report) + if writes: + _apply_writes(args, name, profile, report) + + if args.as_json: + print(json.dumps([report_to_dict(r) for r in reports], indent=2)) + return + if args.markdown: + target = Path("claudedocs") / f"profile-sync-{date.today().isoformat()}.md" + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(format_markdown(reports), encoding="utf-8") + print(f"written: {target}") + return + if args.fetch_plan: + for report in reports: + for url in fetch_plan(report): + print(url) + return + if args.triage: + _print_triage(args, selected, reports) + return + printed = 0 + for report in reports: + if args.changed_only and not report.needs_review(): + continue + printed += 1 + print(format_report(report, args.changed_only)) + _print_extras(args, selected[report.name], report) + print() + if args.changed_only: + _print_elided(reports, printed) + + +def _print_elided(reports: list[ProfileReport], printed: int) -> None: + """Account for what --changed-only left out, grouped by reason.""" + reasons: dict[str, int] = {} + for report in reports: + if report.needs_review(): + continue + key = report.skipped or "nothing to review" + reasons[key] = reasons.get(key, 0) + 1 + if not reasons: + return + total = sum(reasons.values()) + print(f"{printed} profils listes, {total} non listes:") + for reason, count in sorted(reasons.items(), key=lambda item: -item[1]): + print(f" {count:4d} {reason}") + + +if __name__ == "__main__": + main() diff --git a/scripts/upstream.py b/scripts/upstream.py new file mode 100644 index 00000000..f7429f06 --- /dev/null +++ b/scripts/upstream.py @@ -0,0 +1,420 @@ +"""Access to upstream source repositories. + +Resolves revisions, fetches files by sha, and compares trees across the +forge families the emulator profiles point at. Knows nothing about profile +structure. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from pathlib import Path + +USER_AGENT = "retrobios-profile-sync/1.0" +ABSENT = "\0absent\0" +GITHUB_COMPARE_CAP = 300 + +_HOSTS: dict[str, tuple[str, str, str]] = { + "github.com": ( + "github", + "https://api.github.com", + "https://raw.githubusercontent.com", + ), + "gitlab.com": ("gitlab", "https://gitlab.com/api/v4", "https://gitlab.com"), + "codeberg.org": ( + "forgejo", + "https://codeberg.org/api/v1", + "https://codeberg.org", + ), + "git.citron-emu.org": ( + "forgejo", + "https://git.citron-emu.org/api/v1", + "https://git.citron-emu.org", + ), + "git.eden-emu.dev": ( + "forgejo", + "https://git.eden-emu.dev/api/v1", + "https://git.eden-emu.dev", + ), +} + + +class UpstreamError(Exception): + """Any failure while talking to a forge.""" + + +class RateLimitError(UpstreamError): + """The forge refused the request for quota reasons.""" + + +@dataclass(frozen=True) +class Repo: + host: str + family: str + api_base: str + raw_base: str + owner: str + name: str + + @property + def slug(self) -> str: + return f"{self.owner}/{self.name}" + + +@dataclass(frozen=True) +class Release: + tag: str + date: str + is_prerelease: bool + + +@dataclass(frozen=True) +class FileChange: + status: str + path: str + previous_path: str | None + + +@dataclass(frozen=True) +class CompareResult: + files: list[FileChange] + truncated: bool + + +def parse_repo(url: str) -> Repo | None: + """Build a Repo from a forge URL, or None when the host is unknown.""" + if not url: + return None + parts = urllib.parse.urlsplit(url.strip()) + entry = _HOSTS.get(parts.netloc) + if entry is None: + return None + segments = [s for s in parts.path.split("/") if s] + if len(segments) < 2: + return None + owner, name = segments[0], segments[1] + if name.endswith(".git"): + name = name[:-4] + family, api_base, raw_base = entry + return Repo(parts.netloc, family, api_base, raw_base, owner, name) + + +def make_repo(host: str, owner: str, name: str) -> Repo | None: + """Rebuild a Repo from a host and slug already known to be supported.""" + entry = _HOSTS.get(host) + if entry is None: + return None + family, api_base, raw_base = entry + return Repo(host, family, api_base, raw_base, owner, name) + + +def raw_url(repo: Repo, sha: str, path: str) -> str: + """URL serving the raw bytes of one path at one revision.""" + quoted = urllib.parse.quote(path) + if repo.family == "github": + return f"{repo.raw_base}/{repo.owner}/{repo.name}/{sha}/{quoted}" + if repo.family == "gitlab": + return f"{repo.raw_base}/{repo.owner}/{repo.name}/-/raw/{sha}/{quoted}" + return f"{repo.raw_base}/{repo.owner}/{repo.name}/raw/commit/{sha}/{quoted}" + + +def _headers(accept_json: bool = False) -> dict[str, str]: + headers = {"User-Agent": USER_AGENT} + if accept_json: + headers["Accept"] = "application/json" + token = os.environ.get("GITHUB_TOKEN", "") + if token: + headers["Authorization"] = f"token {token}" + return headers + + +def _http_text(url: str) -> str | None: + """Body of a GET, or None on 404. Replaced in tests.""" + req = urllib.request.Request(url, headers=_headers()) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + return resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + if exc.code == 404: + return None + if exc.code in (403, 429): + raise RateLimitError(f"{url}: HTTP {exc.code}") from exc + raise UpstreamError(f"{url}: HTTP {exc.code}") from exc + except urllib.error.URLError as exc: + raise UpstreamError(f"{url}: {exc.reason}") from exc + + +def _http_json(url: str) -> object | None: + """Parsed JSON body of a GET, or None on 404. Replaced in tests.""" + req = urllib.request.Request(url, headers=_headers(accept_json=True)) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + return json.loads(resp.read().decode()) + except urllib.error.HTTPError as exc: + if exc.code == 404: + return None + if exc.code in (403, 429): + raise RateLimitError(f"{url}: HTTP {exc.code}") from exc + raise UpstreamError(f"{url}: HTTP {exc.code}") from exc + except urllib.error.URLError as exc: + raise UpstreamError(f"{url}: {exc.reason}") from exc + + +def cache_path(cache_dir: str, repo: Repo, sha: str, path: str) -> Path: + """Content-addressed location for one path at one revision.""" + root = Path(cache_dir) / repo.host / repo.owner / repo.name / sha + safe = path.replace("\\", "/").strip("/") + target = (root / safe).resolve() + base = root.resolve() + if not str(target).startswith(str(base) + os.sep) and target != base: + target = base / safe.replace("/", "_").replace("..", "_") + return target + + +def write_cache(target: Path, text: str) -> None: + """Atomic write: unique scratch in the target directory, then replace.""" + target.parent.mkdir(parents=True, exist_ok=True) + fd, scratch = tempfile.mkstemp(dir=str(target.parent), suffix=".part") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + os.replace(scratch, target) + except OSError: + Path(scratch).unlink(missing_ok=True) + raise + + +def fetch_file( + repo: Repo, + sha: str, + path: str, + cache_dir: str, + offline: bool = False, +) -> list[str] | None: + """Lines of one path at one revision, or None when absent.""" + target = cache_path(cache_dir, repo, sha, path) + if target.is_file(): + text = target.read_text(encoding="utf-8") + return None if text == ABSENT else text.splitlines() + if offline: + return None + text = _http_text(raw_url(repo, sha, path)) + write_cache(target, ABSENT if text is None else text) + return None if text is None else text.splitlines() + + +def _api(url: str, cache_dir: str, offline: bool) -> object | None: + """Cached API call. The cache stores absence as well as payloads.""" + key = hashlib.sha256(url.encode()).hexdigest() + target = Path(cache_dir) / "_api" / f"{key}.json" + if target.is_file(): + raw = target.read_text(encoding="utf-8") + return None if raw == ABSENT else json.loads(raw) + if offline: + return None + payload = _http_json(url) + write_cache(target, ABSENT if payload is None else json.dumps(payload)) + return payload + + +def _project(repo: Repo) -> str: + return urllib.parse.quote(f"{repo.owner}/{repo.name}", safe="") + + +def _commits_url(repo: Repo, date: str | None) -> str: + if repo.family == "github": + base = f"{repo.api_base}/repos/{repo.slug}/commits?per_page=1" + elif repo.family == "gitlab": + base = ( + f"{repo.api_base}/projects/{_project(repo)}" + f"/repository/commits?per_page=1" + ) + else: + base = f"{repo.api_base}/repos/{repo.slug}/commits?limit=1" + return f"{base}&until={date}T23:59:59Z" if date else base + + +def _first_sha(payload: object) -> str | None: + if isinstance(payload, list) and payload: + head = payload[0] + if isinstance(head, dict): + return head.get("sha") or head.get("id") + return None + + +def resolve_head(repo: Repo, cache_dir: str, offline: bool = False) -> str | None: + """Sha of the default branch tip.""" + return _first_sha(_api(_commits_url(repo, None), cache_dir, offline)) + + +def resolve_commit_at( + repo: Repo, date: str, cache_dir: str, offline: bool = False +) -> str | None: + """Last default-branch commit on or before a date.""" + return _first_sha(_api(_commits_url(repo, date), cache_dir, offline)) + + +def _tags_url(repo: Repo) -> str: + if repo.family == "gitlab": + return f"{repo.api_base}/projects/{_project(repo)}/repository/tags" + return f"{repo.api_base}/repos/{repo.slug}/tags" + + +def list_tags(repo: Repo, cache_dir: str, offline: bool = False) -> list[str]: + """Tag names, newest first as the forge orders them.""" + payload = _api(_tags_url(repo), cache_dir, offline) + if not isinstance(payload, list): + return [] + return [t["name"] for t in payload if isinstance(t, dict) and t.get("name")] + + +def resolve_tag_commit( + repo: Repo, tag: str, cache_dir: str, offline: bool = False +) -> str | None: + """Commit a tag points at.""" + payload = _api(_tags_url(repo), cache_dir, offline) + if not isinstance(payload, list): + return None + for entry in payload: + if not isinstance(entry, dict) or entry.get("name") != tag: + continue + commit = entry.get("commit") + if isinstance(commit, dict): + return commit.get("sha") or commit.get("id") + return None + + +def _releases_url(repo: Repo) -> str: + if repo.family == "gitlab": + return f"{repo.api_base}/projects/{_project(repo)}/releases" + return f"{repo.api_base}/repos/{repo.slug}/releases/latest" + + +def latest_release( + repo: Repo, cache_dir: str, offline: bool = False +) -> Release | None: + """Most recent release the forge exposes.""" + payload = _api(_releases_url(repo), cache_dir, offline) + if isinstance(payload, list): + payload = payload[0] if payload else None + if not isinstance(payload, dict): + return None + tag = payload.get("tag_name") or payload.get("tag") or "" + stamp = payload.get("published_at") or payload.get("released_at") or "" + if not tag: + return None + return Release(str(tag), str(stamp)[:10], bool(payload.get("prerelease"))) + + +def _compare_url(repo: Repo, base: str, head: str) -> str: + if repo.family == "gitlab": + return ( + f"{repo.api_base}/projects/{_project(repo)}" + f"/repository/compare?from={base}&to={head}" + ) + return f"{repo.api_base}/repos/{repo.slug}/compare/{base}...{head}" + + +def _changes_from_github(payload: dict) -> list[FileChange]: + return [ + FileChange( + entry.get("status", "modified"), + entry.get("filename", ""), + entry.get("previous_filename"), + ) + for entry in (payload.get("files") or []) + if isinstance(entry, dict) + ] + + +def _changes_from_gitlab(payload: dict) -> list[FileChange]: + changes = [] + for entry in payload.get("diffs") or []: + if not isinstance(entry, dict): + continue + if entry.get("renamed_file"): + status = "renamed" + elif entry.get("new_file"): + status = "added" + elif entry.get("deleted_file"): + status = "removed" + else: + status = "modified" + previous = entry.get("old_path") if status == "renamed" else None + changes.append(FileChange(status, entry.get("new_path", ""), previous)) + return changes + + +def compare( + repo: Repo, base: str, head: str, cache_dir: str, offline: bool = False +) -> CompareResult: + """Tree difference between two revisions.""" + payload = _api(_compare_url(repo, base, head), cache_dir, offline) + if not isinstance(payload, dict): + return CompareResult([], True) + if repo.family == "gitlab": + files = _changes_from_gitlab(payload) + else: + files = _changes_from_github(payload) + truncated = bool(payload.get("truncated")) or len(files) >= GITHUB_COMPARE_CAP + return CompareResult(files, truncated) + + +def find_renamed(result: CompareResult, path: str) -> str | None: + """New path of a file the comparison reports as renamed.""" + for change in result.files: + if change.status == "renamed" and change.previous_path == path: + return change.path + return None + + +def _tree_url(repo: Repo, sha: str) -> str | None: + if repo.family == "gitlab": + return None + return f"{repo.api_base}/repos/{repo.slug}/git/trees/{sha}?recursive=1" + + +def list_tree( + repo: Repo, sha: str, cache_dir: str, offline: bool = False +) -> tuple[list[str], bool]: + """Every blob path at one revision, and whether the forge truncated it.""" + url = _tree_url(repo, sha) + if url is None: + return [], True + payload = _api(url, cache_dir, offline) + if not isinstance(payload, dict): + return [], True + paths = [ + entry["path"] + for entry in payload.get("tree") or [] + if isinstance(entry, dict) and entry.get("type") == "blob" and entry.get("path") + ] + return paths, bool(payload.get("truncated")) + + +def commits_touching( + repo: Repo, path: str, base: str, cache_dir: str, offline: bool = False +) -> int: + """Commits touching one path since a revision.""" + quoted = urllib.parse.quote(path) + if repo.family == "gitlab": + url = ( + f"{repo.api_base}/projects/{_project(repo)}" + f"/repository/commits?path={quoted}&per_page=100" + ) + elif repo.family == "github": + url = ( + f"{repo.api_base}/repos/{repo.slug}/commits" + f"?path={quoted}&sha={base}&per_page=100" + ) + else: + url = f"{repo.api_base}/repos/{repo.slug}/commits?path={quoted}&limit=100" + payload = _api(url, cache_dir, offline) + return len(payload) if isinstance(payload, list) else 0 diff --git a/tests/test_profile_sync.py b/tests/test_profile_sync.py new file mode 100644 index 00000000..2cd7b054 --- /dev/null +++ b/tests/test_profile_sync.py @@ -0,0 +1,1161 @@ +"""Tests for the profile synchronisation tool (no network).""" + +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +import yaml + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts")) + +import profile_sync +from profile_sync import ( + EntryReport, + PartResult, + ProfileReport, + RefPart, + VersionReport, + YamlWriteError, + anchor_block, + anchor_part, + apply_edit, + backfill_commit, + build_parser, + build_report, + bump_commit, + check_version, + collect_tokens, + declared_hashes, + declared_names, + detect_new_files, + drift_score, + fetch_plan, + format_markdown, + format_report, + insert_after_line, + parse_source_ref, + rebase_refs, + replace_field_line, + report_to_dict, + resolve_rename, + select_profiles, + select_repo, + split_source_ref, + tree_diff, + unified_for_path, + version_warning, + watch_hashes, + worst_status, +) +from upstream import CompareResult, FileChange + + +class TestParseSourceRef(unittest.TestCase): + def test_path_with_single_line(self): + self.assertEqual( + parse_source_ref("src/core/bios.cpp:258"), + ("src/core/bios.cpp", 258, 258), + ) + + def test_path_with_line_range(self): + self.assertEqual( + parse_source_ref("Machines/Utility/ROMCatalogue.cpp:123-129"), + ("Machines/Utility/ROMCatalogue.cpp", 123, 129), + ) + + def test_path_without_line(self): + self.assertEqual( + parse_source_ref("FirmwareDatabase.cs"), + ("FirmwareDatabase.cs", None, None), + ) + + +class TestSplitSourceRef(unittest.TestCase): + def test_single_part(self): + parts = split_source_ref("a.c:1-3") + self.assertEqual(len(parts), 1) + self.assertEqual( + (parts[0].path, parts[0].start, parts[0].end), ("a.c", 1, 3) + ) + + def test_multiple_parts(self): + parts = split_source_ref("a.c:1-3, b.c:9, c.h") + self.assertEqual([p.path for p in parts], ["a.c", "b.c", "c.h"]) + self.assertIsNone(parts[2].start) + + def test_empty_ref(self): + self.assertEqual(split_source_ref(""), []) + + def test_bare_range_continues_the_previous_file(self): + parts = split_source_ref("src/geo.c:234-243, 273-285") + self.assertEqual( + [(p.path, p.start, p.end) for p in parts], + [("src/geo.c", 234, 243), ("src/geo.c", 273, 285)], + ) + + def test_bare_single_line_continues_the_previous_file(self): + parts = split_source_ref("HW_.cpp:2813,2820") + self.assertEqual( + [(p.path, p.start, p.end) for p in parts], + [("HW_.cpp", 2813, 2813), ("HW_.cpp", 2820, 2820)], + ) + + def test_bare_range_carries_across_several_continuations(self): + parts = split_source_ref("a.c:1-2, 5-6, 9") + self.assertEqual([p.path for p in parts], ["a.c", "a.c", "a.c"]) + self.assertEqual(parts[2].start, 9) + + def test_continuation_stops_at_a_new_path(self): + parts = split_source_ref("a.c:1-2, 5-6, b.c:3, 8") + self.assertEqual( + [(p.path, p.start) for p in parts], + [("a.c", 1), ("a.c", 5), ("b.c", 3), ("b.c", 8)], + ) + + def test_leading_bare_range_stays_a_path(self): + parts = split_source_ref("273-285") + self.assertEqual(parts[0].path, "273-285") + + +class TestCollectTokens(unittest.TestCase): + def test_hashes_preferred_over_name(self): + entry = { + "name": "kernal.bin", + "sha1": "6C4FA9465F6091B174DF27DFE679499DF447503C", + "crc32": "789c8cc5", + } + tokens = collect_tokens(entry) + self.assertIn("6c4fa9465f6091b174df27dfe679499df447503c", tokens) + self.assertIn("789c8cc5", tokens) + self.assertNotIn("kernal.bin", tokens) + + def test_name_fallback_without_hashes(self): + self.assertEqual(collect_tokens({"name": "GC/USA/IPL.bin"}), ["ipl.bin"]) + + def test_adler_prefix_stripped(self): + self.assertEqual( + collect_tokens({"known_hash_adler32": "0x4f1f6f5c"}), ["4f1f6f5c"] + ) + + def test_hash_lists(self): + self.assertEqual( + collect_tokens({"md5": ["AABB", "ccdd"]}), ["aabb", "ccdd"] + ) + + +class TestWorstStatus(unittest.TestCase): + def test_gone_beats_everything(self): + self.assertEqual(worst_status(["ANCHORED", "GONE", "SHIFTED"]), "GONE") + + def test_changed_beats_ambiguous(self): + self.assertEqual(worst_status(["AMBIGUOUS", "CHANGED"]), "CHANGED") + + def test_all_anchored(self): + self.assertEqual(worst_status(["ANCHORED", "ANCHORED"]), "ANCHORED") + + def test_empty(self): + self.assertEqual(worst_status([]), "ANCHORED") + + +class TestAnchorBlock(unittest.TestCase): + def test_anchored_same_position(self): + lines = ["a", "target", "b"] + result = anchor_block(lines, list(lines), 2, 2) + self.assertEqual(result.status, "ANCHORED") + self.assertEqual((result.start, result.end), (2, 2)) + + def test_shifted_when_block_moved(self): + pin = ["a", "target line", "b"] + head = ["x", "y", "z", "a", "target line", "b"] + result = anchor_block(pin, head, 2, 2) + self.assertEqual(result.status, "SHIFTED") + self.assertEqual((result.start, result.end), (5, 5)) + + def test_shifted_on_whitespace_only_change(self): + self.assertEqual( + anchor_block([" value = 1"], ["\t\tvalue = 1"], 1, 1).status, "ANCHORED" + ) + + def test_range_preserved_when_shifted(self): + pin = ["p", "one", "two", "three", "q"] + head = ["x", "x", "p", "one", "two", "three", "q"] + result = anchor_block(pin, head, 2, 4) + self.assertEqual((result.status, result.start, result.end), ("SHIFTED", 4, 6)) + + def test_widening_disambiguates_duplicate_line(self): + pin = ["header A", "dup", "tail A", "header B", "dup", "tail B"] + head = ["pad", "header A", "dup", "tail A", "header B", "dup", "tail B"] + result = anchor_block(pin, head, 5, 5) + self.assertEqual((result.status, result.start), ("SHIFTED", 6)) + + def test_ambiguous_when_widening_never_resolves(self): + result = anchor_block(["dup"] * 40, ["dup"] * 40, 5, 5) + self.assertEqual(result.status, "AMBIGUOUS") + self.assertGreater(len(result.candidates), 1) + + def test_changed_maps_to_new_range(self): + pin = ["a", "b", "value = 1", "c", "d"] + head = ["a", "b", "value = 2", "c", "d"] + result = anchor_block(pin, head, 3, 3) + self.assertEqual(result.status, "CHANGED") + self.assertEqual(result.start, 3) + + def test_gone_when_content_vanished(self): + result = anchor_block(["a", "unique marker", "b"], ["x"] * 200, 2, 2) + self.assertEqual(result.status, "GONE") + + def test_gone_when_start_beyond_pin_file(self): + self.assertEqual(anchor_block(["a"], ["a"], 50, 50).status, "GONE") + + def test_blank_cited_range_is_changed(self): + result = anchor_block(["a", "", "b"], ["a", "", "b"], 2, 2) + self.assertEqual(result.status, "CHANGED") + + def test_large_file_skips_mapping_with_reason(self): + pin = [f"line {i}" for i in range(25000)] + head = [f"other {i}" for i in range(25000)] + result = anchor_block(pin, head, 10, 10) + self.assertEqual(result.status, "CHANGED") + self.assertIsNotNone(result.reason) + self.assertIsNone(result.start) + + +PIN = profile_sync.PIN +HEAD = profile_sync.HEAD + + +def make_fetch(files: dict[tuple[str, str], list[str]]): + return lambda sha, path: files.get((sha, path)) + + +def renamer(result: CompareResult, head_paths=()): + return lambda path: resolve_rename(result, path, head_paths) + + +class TestResolveRename(unittest.TestCase): + def test_uses_comparison_when_available(self): + result = CompareResult([FileChange("renamed", "new.c", "old.c")], False) + self.assertEqual(resolve_rename(result, "old.c")[0], "new.c") + + def test_basename_fallback_when_truncated(self): + result = CompareResult( + [ + FileChange("added", "a/b/driver.cpp", None), + FileChange("added", "z/other.cpp", None), + ], + True, + ) + found, candidates = resolve_rename(result, "src/driver.cpp") + self.assertEqual(found, "a/b/driver.cpp") + self.assertEqual(candidates, []) + + def test_basename_fallback_ambiguous(self): + result = CompareResult( + [ + FileChange("added", "a/driver.cpp", None), + FileChange("added", "b/driver.cpp", None), + ], + True, + ) + found, candidates = resolve_rename(result, "src/driver.cpp") + self.assertIsNone(found) + self.assertEqual(len(candidates), 2) + + def test_nothing_found(self): + self.assertEqual(resolve_rename(CompareResult([], False), "x.c"), (None, [])) + + def test_head_tree_wins_over_comparison_files(self): + result = CompareResult([FileChange("added", "wrong/a.c", None)], True) + found, _ = resolve_rename(result, "old/a.c", ["src/a.c", "docs/readme.md"]) + self.assertEqual(found, "src/a.c") + + def test_stem_fallback_when_extension_changed(self): + result = CompareResult([], True) + found, _ = resolve_rename( + result, "libretro.cpp", ["libretro.c", "libretro_cbs.h", "other.c"] + ) + self.assertEqual(found, "libretro.c") + + def test_same_directory_breaks_a_stem_tie(self): + found, _ = resolve_rename( + CompareResult([], True), + "libretro.cpp", + ["libretro-common/include/libretro.h", "libretro.c"], + ) + self.assertEqual(found, "libretro.c") + + def test_same_directory_tie_break_needs_a_single_winner(self): + found, candidates = resolve_rename( + CompareResult([], True), "src/a.cpp", ["src/a.c", "src/a.h"] + ) + self.assertIsNone(found) + self.assertEqual(len(candidates), 2) + + def test_stem_fallback_ambiguous_across_directories(self): + found, candidates = resolve_rename( + CompareResult([], True), "src/driver.cpp", ["a/driver.c", "b/driver.cc"] + ) + self.assertIsNone(found) + self.assertEqual(len(candidates), 2) + + +class TestAnchorPart(unittest.TestCase): + def test_anchored(self): + fetch = make_fetch({(PIN, "a.c"): ["x", "hit"], (HEAD, "a.c"): ["x", "hit"]}) + result = anchor_part( + RefPart("a.c", 2, 2), fetch, renamer(CompareResult([], False)) + ) + self.assertEqual(result.status, "ANCHORED") + + def test_path_only_ref_present_at_head(self): + fetch = make_fetch({(PIN, "a.c"): ["x"], (HEAD, "a.c"): ["y"]}) + result = anchor_part( + RefPart("a.c", None, None), fetch, renamer(CompareResult([], False)) + ) + self.assertEqual(result.status, "ANCHORED") + + def test_path_only_ref_absent_at_head(self): + fetch = make_fetch({(PIN, "a.c"): ["x"]}) + result = anchor_part( + RefPart("a.c", None, None), fetch, renamer(CompareResult([], False)) + ) + self.assertEqual(result.status, "GONE") + + def test_renamed_then_anchored_at_new_path(self): + fetch = make_fetch( + {(PIN, "old.c"): ["x", "hit"], (HEAD, "new.c"): ["x", "hit"]} + ) + comparison = CompareResult([FileChange("renamed", "new.c", "old.c")], False) + result = anchor_part(RefPart("old.c", 2, 2), fetch, renamer(comparison)) + self.assertEqual(result.status, "RENAMED") + self.assertEqual(result.new_path, "new.c") + self.assertEqual(result.start, 2) + + def test_gone_when_pin_file_missing(self): + result = anchor_part( + RefPart("a.c", 1, 1), make_fetch({}), renamer(CompareResult([], False)) + ) + self.assertEqual(result.status, "GONE") + + def test_ambiguous_rename_candidates_reported(self): + fetch = make_fetch({(PIN, "src/d.cpp"): ["x"]}) + comparison = CompareResult( + [ + FileChange("added", "a/d.cpp", None), + FileChange("added", "b/d.cpp", None), + ], + True, + ) + result = anchor_part(RefPart("src/d.cpp", 1, 1), fetch, renamer(comparison)) + self.assertEqual(result.status, "AMBIGUOUS") + self.assertIsNotNone(result.reason) + + def test_rename_resolver_is_not_called_when_file_is_present(self): + calls = [] + + def getter(path): + calls.append(path) + return None, [] + + fetch = make_fetch({(PIN, "a.c"): ["x"], (HEAD, "a.c"): ["x"]}) + anchor_part(RefPart("a.c", 1, 1), fetch, getter) + self.assertEqual(calls, []) + + def test_gone_carries_a_reason(self): + result = anchor_part( + RefPart("a.c", 1, 1), make_fetch({}), renamer(CompareResult([], False)) + ) + self.assertIn("no rename found", result.reason) + + +class TestSelectRepo(unittest.TestCase): + def test_source_wins_over_upstream(self): + repo = select_repo( + { + "source": "https://github.com/libretro/beetle-psx-libretro", + "upstream": "https://mednafen.github.io/", + } + ) + self.assertEqual(repo.name, "beetle-psx-libretro") + + def test_falls_back_to_upstream(self): + repo = select_repo( + { + "source": "https://www.mamedev.org/", + "upstream": "https://github.com/mamedev/mame", + } + ) + self.assertEqual(repo.name, "mame") + + def test_none_when_no_supported_host(self): + self.assertIsNone(select_repo({"source": "https://www.6809.org.uk/"})) + + +class TestBuildReport(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.files: dict[tuple[str, str], list[str]] = {} + self._orig = ( + profile_sync.upstream.fetch_file, + profile_sync.upstream.resolve_head, + profile_sync.upstream.resolve_commit_at, + profile_sync.upstream.compare, + profile_sync.upstream.list_tree, + profile_sync.upstream._http_json, + profile_sync.upstream._http_text, + ) + # Any code path reaching the HTTP layer is a leak, not a slow test. + def _no_network(url): + raise AssertionError(f"test reached the network: {url}") + + profile_sync.upstream._http_json = _no_network + profile_sync.upstream._http_text = _no_network + profile_sync.upstream.list_tree = ( + lambda repo, sha, cache_dir, offline=False: ( + sorted({path for _, path in self.files}), False + ) + ) + profile_sync.upstream.fetch_file = ( + lambda repo, sha, path, cache_dir, offline=False: self.files.get( + (sha, path) + ) + ) + profile_sync.upstream.resolve_head = ( + lambda repo, cache_dir, offline=False: "headsha" + ) + profile_sync.upstream.resolve_commit_at = ( + lambda repo, date, cache_dir, offline=False: "pinsha" + ) + profile_sync.upstream.compare = ( + lambda repo, base, head, cache_dir, offline=False: CompareResult([], False) + ) + + def tearDown(self): + ( + profile_sync.upstream.fetch_file, + profile_sync.upstream.resolve_head, + profile_sync.upstream.resolve_commit_at, + profile_sync.upstream.compare, + profile_sync.upstream.list_tree, + profile_sync.upstream._http_json, + profile_sync.upstream._http_text, + ) = self._orig + self.tmp.cleanup() + + def _profile(self, refs): + return { + "emulator": "Test", + "source": "https://github.com/o/n", + "profiled_date": "2026-03-29", + "files": [ + {"name": f"f{i}.bin", "source_ref": ref} + for i, ref in enumerate(refs) + ], + } + + def test_counts_by_status(self): + self.files[("pinsha", "a.c")] = ["x", "hit", "y"] + self.files[("headsha", "a.c")] = ["pad", "x", "hit", "y"] + report = build_report("test", self._profile(["a.c:2"]), self.dir) + self.assertEqual(report.pin, "pinsha") + self.assertEqual(report.head, "headsha") + self.assertEqual(report.host, "github.com") + self.assertEqual(report.counts["SHIFTED"], 1) + + def test_entry_takes_worst_part_status(self): + self.files[("pinsha", "a.c")] = ["x", "hit"] + self.files[("headsha", "a.c")] = ["x", "hit"] + self.files[("pinsha", "b.c")] = ["gone marker"] + self.files[("headsha", "b.c")] = ["z"] * 100 + report = build_report("test", self._profile(["a.c:2, b.c:1"]), self.dir) + self.assertEqual(report.entries[0].status, "GONE") + + def test_existing_source_commit_is_used(self): + profile = self._profile(["a.c:1"]) + profile["source_commit"] = "pinned" + self.files[("pinned", "a.c")] = ["x"] + self.files[("headsha", "a.c")] = ["x"] + report = build_report("test", profile, self.dir) + self.assertEqual(report.pin, "pinned") + self.assertEqual(report.pin_origin, "source_commit") + + def test_unsupported_host_is_skipped_with_reason(self): + report = build_report( + "test", {"emulator": "T", "source": "https://www.6809.org.uk/"}, self.dir + ) + self.assertIn("unsupported host", report.skipped) + + def test_profile_without_refs_is_reported_not_empty(self): + profile = { + "emulator": "T", + "source": "https://github.com/o/n", + "profiled_date": "2026-03-29", + "files": [{"name": "a.bin"}], + } + report = build_report("test", profile, self.dir) + self.assertEqual(report.entries, []) + self.assertEqual(report.skipped, "no source_ref") + + def test_profile_without_refs_still_resolves_its_pin(self): + profile = { + "emulator": "T", + "source": "https://github.com/o/n", + "profiled_date": "2026-03-29", + "files": [{"name": "a.bin"}], + } + report = build_report("test", profile, self.dir) + self.assertEqual(report.pin, "pinsha") + self.assertEqual(report.head, "headsha") + self.assertEqual(report.needs_review(), 0) + + def _two_repo_profile(self, refs): + profile = self._profile(refs) + profile["upstream"] = "https://gitlab.com/g/p" + return profile + + def test_path_absent_from_source_resolves_against_upstream(self): + self.files[("pinsha", "src/a.c")] = ["x", "hit"] + self.files[("headsha", "src/a.c")] = ["x", "hit"] + report = build_report( + "test", self._two_repo_profile(["src/a.c:2"]), self.dir + ) + self.assertEqual(report.entries[0].status, "ANCHORED") + self.assertEqual(report.repos, ["o/n", "g/p"]) + + def test_part_records_the_repository_that_carries_it(self): + self.files[("pinsha", "only_here.c")] = ["x", "hit"] + self.files[("headsha", "only_here.c")] = ["x", "hit"] + report = build_report( + "test", self._two_repo_profile(["only_here.c:2"]), self.dir + ) + part = report.entries[0].parts[0] + self.assertIsNotNone(part.head_url) + + def test_single_repo_profile_lists_one_slug(self): + self.files[("pinsha", "a.c")] = ["x"] + self.files[("headsha", "a.c")] = ["x"] + report = build_report("test", self._profile(["a.c:1"]), self.dir) + self.assertEqual(report.repos, ["o/n"]) + + def test_upstream_commit_pins_the_second_repository(self): + profile = self._two_repo_profile(["a.c:1"]) + profile["upstream_commit"] = "upinned" + self.files[("pinsha", "a.c")] = ["x"] + self.files[("headsha", "a.c")] = ["x"] + report = build_report("test", profile, self.dir) + self.assertEqual(report.pin, "pinsha") + self.assertEqual(report.repos, ["o/n", "g/p"]) + + def test_repo_name_prefix_is_stripped_as_a_last_resort(self): + self.files[("pinsha", "Source/HW_.cpp")] = ["x", "hit"] + self.files[("headsha", "Source/HW_.cpp")] = ["x", "hit"] + report = build_report( + "test", self._profile(["n/Source/HW_.cpp:2"]), self.dir + ) + part = report.entries[0].parts[0] + self.assertEqual(part.status, "RENAMED") + self.assertEqual(part.new_path, "Source/HW_.cpp") + + def test_declared_path_wins_over_the_stripped_one(self): + self.files[("pinsha", "n/a.c")] = ["x", "declared"] + self.files[("headsha", "n/a.c")] = ["x", "declared"] + self.files[("pinsha", "a.c")] = ["x", "stripped"] + self.files[("headsha", "a.c")] = ["x", "stripped"] + report = build_report("test", self._profile(["n/a.c:2"]), self.dir) + part = report.entries[0].parts[0] + self.assertEqual(part.status, "ANCHORED") + self.assertIsNone(part.new_path) + + def test_prefix_matching_no_repository_is_not_stripped(self): + self.files[("pinsha", "a.c")] = ["x", "hit"] + self.files[("headsha", "a.c")] = ["x", "hit"] + report = build_report("test", self._profile(["other/a.c:2"]), self.dir) + self.assertEqual(report.entries[0].parts[0].status, "GONE") + + def test_missing_date_and_commit_is_skipped(self): + report = build_report( + "test", + { + "emulator": "T", + "source": "https://github.com/o/n", + "files": [{"name": "a", "source_ref": "a.c:1"}], + }, + self.dir, + ) + self.assertIn("no source_commit", report.skipped) + + +def _sample_report(): + shifted = PartResult(RefPart("a.c", 2, 4), "SHIFTED", None, 10, 12, []) + changed = PartResult(RefPart("b.c", 5, 5), "CHANGED", None, 7, 7, []) + return ProfileReport( + name="demo", + repo="o/n", + host="github.com", + pin="pinsha0000", + pin_origin="profiled_date 2026-03-29", + head="headsha000", + entries=[ + EntryReport("f0.bin", "a.c:2-4", "SHIFTED", [shifted]), + EntryReport("f1.bin", "b.c:5", "CHANGED", [changed]), + ], + counts={"SHIFTED": 1, "CHANGED": 1}, + ) + + +class TestFormatReport(unittest.TestCase): + def test_header_shows_both_revisions(self): + text = format_report(_sample_report()) + self.assertIn("pinsha0", text) + self.assertIn("headsha", text) + + def test_shifted_shows_new_range(self): + self.assertIn("10-12", format_report(_sample_report())) + + def test_anchored_does_not_repeat_its_range(self): + part = PartResult(RefPart("a.c", 2, 4), "ANCHORED", None, 2, 4, []) + report = ProfileReport( + name="d", repo="o/n", host="github.com", pin="p" * 10, head="h" * 10, + pin_origin="source_commit", + entries=[EntryReport("f.bin", "a.c:2-4", "ANCHORED", [part])], + counts={"ANCHORED": 1}, + ) + self.assertEqual(format_report(report).count("2-4"), 2) + + def test_review_count_in_summary(self): + self.assertIn("1 demandent une relecture", format_report(_sample_report())) + + def test_skipped_profile_states_reason(self): + text = format_report(ProfileReport(name="x", skipped="no source_ref")) + self.assertIn("no source_ref", text) + + def test_skipped_profile_still_shows_a_known_pin(self): + text = format_report( + ProfileReport( + name="x", + pin="pinsha0000", + head="headsha000", + pin_origin="profiled_date 2026-03-29", + skipped="no source_ref", + ) + ) + self.assertIn("pinsha0", text) + self.assertIn("no source_ref", text) + + def test_changed_only_hides_clean_entries(self): + text = format_report(_sample_report(), changed_only=True) + self.assertIn("b.c:5", text) + self.assertNotIn("a.c:2-4", text) + + +class TestElidedSummary(unittest.TestCase): + def test_groups_untouched_profiles_by_reason(self): + reports = [ + ProfileReport(name="a", entries=[], counts={"CHANGED": 1}), + ProfileReport(name="b", entries=[], counts={}, skipped="no source_ref"), + ProfileReport(name="c", entries=[], counts={}, skipped="no source_ref"), + ProfileReport(name="d", entries=[], counts={"ANCHORED": 4}), + ] + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + profile_sync._print_elided(reports, printed=1) + text = buffer.getvalue() + self.assertIn("3 non listes", text) + self.assertIn("2 no source_ref", text) + self.assertIn("1 nothing to review", text) + + def test_silent_when_everything_was_listed(self): + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + profile_sync._print_elided( + [ProfileReport(name="a", entries=[], counts={"GONE": 1})], printed=1 + ) + self.assertEqual(buffer.getvalue(), "") + + +class TestReportToDict(unittest.TestCase): + def test_round_trips_through_json(self): + payload = report_to_dict(_sample_report()) + self.assertEqual(json.loads(json.dumps(payload))["counts"]["CHANGED"], 1) + + def test_parts_are_serialisable(self): + payload = report_to_dict(_sample_report()) + self.assertEqual(payload["entries"][0]["parts"][0]["status"], "SHIFTED") + + def test_markdown_lists_every_profile(self): + text = format_markdown([_sample_report()]) + self.assertIn("demo", text) + + +class TestFetchPlan(unittest.TestCase): + def test_only_entries_needing_review(self): + urls = fetch_plan(_sample_report()) + self.assertEqual(len(urls), 1) + self.assertIn("b.c", urls[0]) + + def test_url_carries_head_revision(self): + self.assertIn("headsha000", fetch_plan(_sample_report())[0]) + + def test_url_uses_the_recorded_host(self): + report = _sample_report() + report.host = "codeberg.org" + self.assertIn("codeberg.org", fetch_plan(report)[0]) + + def test_empty_when_nothing_to_review(self): + report = _sample_report() + report.entries = [report.entries[0]] + report.counts = {"SHIFTED": 1} + self.assertEqual(fetch_plan(report), []) + + +PROFILES = { + "alpha": { + "emulator": "A", + "type": "libretro", + "systems": ["sony-playstation"], + "profiled_date": "2026-03-01", + }, + "beta": { + "emulator": "B", + "type": "standalone", + "systems": ["nintendo-64"], + "profiled_date": "2026-08-01", + }, + "gamma": { + "emulator": "C", + "type": "alias", + "systems": ["sony-playstation"], + "profiled_date": "2026-03-01", + }, +} + + +def _args(**kwargs): + argv = [] + for key, value in kwargs.items(): + flag = "--" + key.replace("_", "-") + if value is True: + argv.append(flag) + else: + argv.extend([flag, str(value)]) + return build_parser().parse_args(argv) + + +class TestSelectProfiles(unittest.TestCase): + def test_single_emulator(self): + self.assertEqual( + list(select_profiles(PROFILES, _args(emulator="alpha"))), ["alpha"] + ) + + def test_all_excludes_aliases(self): + self.assertEqual( + sorted(select_profiles(PROFILES, _args(all=True))), ["alpha", "beta"] + ) + + def test_filter_by_system(self): + self.assertEqual( + list( + select_profiles( + PROFILES, _args(all=True, system="sony-playstation") + ) + ), + ["alpha"], + ) + + def test_filter_by_type(self): + self.assertEqual( + list(select_profiles(PROFILES, _args(all=True, type="standalone"))), + ["beta"], + ) + + def test_stale_before(self): + self.assertEqual( + list( + select_profiles(PROFILES, _args(all=True, stale_before="2026-06-01")) + ), + ["alpha"], + ) + + def test_limit(self): + self.assertEqual(len(select_profiles(PROFILES, _args(all=True, limit=1))), 1) + + def test_unknown_emulator_raises(self): + with self.assertRaises(SystemExit): + select_profiles(PROFILES, _args(emulator="nope")) + + +SAMPLE = '''emulator: Test +source: "https://github.com/o/n" +profiled_date: "2026-03-29" + +notes: > + A note mentioning profiled_date: "1999-01-01" inside prose. + +files: + - name: "a.bin" + source_ref: "a.c:10-12" + - name: "b.bin" + source_ref: "b.c:5" +''' + + +class TestLineEdits(unittest.TestCase): + def test_insert_after_profiled_date(self): + out = insert_after_line(SAMPLE, "profiled_date", 'source_commit: "abc"') + lines = out.splitlines() + index = lines.index('profiled_date: "2026-03-29"') + self.assertEqual(lines[index + 1], 'source_commit: "abc"') + + def test_insert_ignores_indented_occurrence_in_prose(self): + out = insert_after_line(SAMPLE, "profiled_date", 'source_commit: "abc"') + self.assertEqual(out.count("source_commit"), 1) + + def test_replace_nth_source_ref(self): + out = replace_field_line(SAMPLE, "source_ref", 1, "b.c:5", "b.c:9") + self.assertIn('source_ref: "b.c:9"', out) + self.assertIn('source_ref: "a.c:10-12"', out) + + def test_replace_refuses_on_value_mismatch(self): + with self.assertRaises(YamlWriteError): + replace_field_line(SAMPLE, "source_ref", 1, "not-there", "x") + + def test_replace_refuses_missing_occurrence(self): + with self.assertRaises(YamlWriteError): + replace_field_line(SAMPLE, "source_ref", 9, "b.c:5", "x") + + +class TestApplyEdit(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) / "p.yml" + self.path.write_text(SAMPLE, encoding="utf-8") + + def tearDown(self): + self.tmp.cleanup() + + def test_backfill_writes_commit_and_preserves_comments(self): + self.assertTrue(backfill_commit(self.path, "abc123")) + text = self.path.read_text(encoding="utf-8") + self.assertIn('source_commit: "abc123"', text) + self.assertIn("A note mentioning", text) + self.assertEqual(yaml.safe_load(text)["source_commit"], "abc123") + + def test_backfill_skips_when_already_present(self): + backfill_commit(self.path, "abc123") + self.assertFalse(backfill_commit(self.path, "def456")) + + def test_guard_restores_file_on_structural_drift(self): + expected = yaml.safe_load(SAMPLE) + broken = SAMPLE.replace("emulator: Test", "emulator: Other") + with self.assertRaises(YamlWriteError): + apply_edit(self.path, broken, expected) + self.assertEqual(self.path.read_text(encoding="utf-8"), SAMPLE) + + def test_guard_accepts_single_field_change(self): + expected = yaml.safe_load(SAMPLE) + expected["source_commit"] = "abc" + new_text = insert_after_line(SAMPLE, "profiled_date", 'source_commit: "abc"') + apply_edit(self.path, new_text, expected) + self.assertEqual( + yaml.safe_load(self.path.read_text())["source_commit"], "abc" + ) + + +class TestRebaseRefs(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) / "p.yml" + self.path.write_text(SAMPLE, encoding="utf-8") + + def tearDown(self): + self.tmp.cleanup() + + def _report(self, entries): + return ProfileReport( + name="p", repo="o/n", pin="pin", head="head", entries=entries, counts={} + ) + + def test_shifted_range_is_rewritten(self): + part = PartResult(RefPart("a.c", 10, 12), "SHIFTED", None, 20, 22, []) + applied = rebase_refs( + self.path, + self._report([EntryReport("a.bin", "a.c:10-12", "SHIFTED", [part])]), + ) + self.assertEqual(applied, ["a.c:10-12 -> a.c:20-22"]) + self.assertIn('source_ref: "a.c:20-22"', self.path.read_text()) + + def test_renamed_path_is_rewritten(self): + part = PartResult(RefPart("a.c", 10, 12), "RENAMED", "src/a.c", 10, 12, []) + rebase_refs( + self.path, + self._report([EntryReport("a.bin", "a.c:10-12", "RENAMED", [part])]), + ) + self.assertIn('source_ref: "src/a.c:10-12"', self.path.read_text()) + + def test_changed_is_never_touched(self): + part = PartResult(RefPart("b.c", 5, 5), "CHANGED", None, 9, 9, []) + self.assertEqual( + rebase_refs( + self.path, + self._report([EntryReport("b.bin", "b.c:5", "CHANGED", [part])]), + ), + [], + ) + self.assertIn('source_ref: "b.c:5"', self.path.read_text()) + + def test_ambiguous_is_never_touched(self): + part = PartResult(RefPart("b.c", 5, 5), "AMBIGUOUS", None, None, None, [3, 9]) + self.assertEqual( + rebase_refs( + self.path, + self._report([EntryReport("b.bin", "b.c:5", "AMBIGUOUS", [part])]), + ), + [], + ) + + def test_shifted_part_beside_changed_part_is_rebased(self): + shifted = PartResult(RefPart("a.c", 10, 12), "SHIFTED", None, 20, 22, []) + changed = PartResult(RefPart("b.c", 5, 5), "CHANGED", None, 9, 9, []) + rebase_refs( + self.path, + self._report( + [ + EntryReport("a.bin", "a.c:10-12", "SHIFTED", [shifted]), + EntryReport("b.bin", "b.c:5", "CHANGED", [changed]), + ] + ), + ) + text = self.path.read_text() + self.assertIn('source_ref: "a.c:20-22"', text) + self.assertIn('source_ref: "b.c:5"', text) + + +class TestBumpCommit(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) / "p.yml" + self.path.write_text(SAMPLE, encoding="utf-8") + + def tearDown(self): + self.tmp.cleanup() + + def test_refused_while_a_changed_remains(self): + report = ProfileReport( + name="p", repo="o/n", pin="pin", head="newhead", + entries=[], counts={"CHANGED": 1}, + ) + self.assertFalse(bump_commit(self.path, report)) + + def test_accepted_when_everything_anchors(self): + report = ProfileReport( + name="p", repo="o/n", pin="pin", head="newhead", + entries=[], counts={"ANCHORED": 3, "SHIFTED": 1}, + ) + self.assertTrue(bump_commit(self.path, report)) + self.assertEqual( + yaml.safe_load(self.path.read_text())["source_commit"], "newhead" + ) + + +class TestCheckVersion(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.repo = profile_sync.upstream.parse_repo("https://github.com/o/n") + self._orig = ( + profile_sync.upstream.latest_release, + profile_sync.upstream.list_tags, + profile_sync.upstream.resolve_tag_commit, + ) + profile_sync.upstream.latest_release = ( + lambda repo, cache, offline=False: profile_sync.upstream.Release( + "v3.1.2", "2026-06-14", False + ) + ) + profile_sync.upstream.list_tags = ( + lambda repo, cache, offline=False: ["v3.1.2", "v3.0.0"] + ) + profile_sync.upstream.resolve_tag_commit = ( + lambda repo, tag, cache, offline=False: ( + "tagsha" if tag == "v3.0.0" else None + ) + ) + + def tearDown(self): + ( + profile_sync.upstream.latest_release, + profile_sync.upstream.list_tags, + profile_sync.upstream.resolve_tag_commit, + ) = self._orig + self.tmp.cleanup() + + def test_reports_declared_and_latest(self): + result = check_version({"core_version": "v3.0.0"}, self.repo, self.dir, False) + self.assertEqual(result.declared, "v3.0.0") + self.assertEqual(result.latest_release, "v3.1.2") + self.assertEqual(result.release_date, "2026-06-14") + + def test_detects_declared_version_is_a_tag(self): + result = check_version({"core_version": "v3.0.0"}, self.repo, self.dir, False) + self.assertTrue(result.tag_matches_declared) + self.assertEqual(result.tag_commit, "tagsha") + + def test_declared_version_not_a_tag(self): + result = check_version({"core_version": "5.2"}, self.repo, self.dir, False) + self.assertFalse(result.tag_matches_declared) + self.assertIsNone(result.tag_commit) + + def test_none_without_core_version(self): + self.assertIsNone(check_version({}, self.repo, self.dir, False)) + + def test_warning_when_tag_commit_differs_from_pin(self): + result = check_version({"core_version": "v3.0.0"}, self.repo, self.dir, False) + self.assertIsNotNone(version_warning(result, "otherpin")) + + def test_no_warning_when_tag_commit_matches_pin(self): + result = check_version({"core_version": "v3.0.0"}, self.repo, self.dir, False) + self.assertIsNone(version_warning(result, "tagsha")) + + +HASH_PROFILE = { + "files": [ + { + "name": "scph5500.bin", + "sha1": "B05DEF971D8EC59F346F2D9AC21FB742E3EB6917", + "aliases": ["SCPH-5500.bin"], + }, + {"name": "gba_bios.bin", "md5": "a860e8c0b6d573d191e4ec7db1b1e4f6"}, + ] +} + + +class TestDeclared(unittest.TestCase): + def test_names_include_aliases_casefolded(self): + names = declared_names(HASH_PROFILE) + self.assertIn("scph5500.bin", names) + self.assertIn("scph-5500.bin", names) + + def test_hashes_lowercased(self): + self.assertIn( + "b05def971d8ec59f346f2d9ac21fb742e3eb6917", declared_hashes(HASH_PROFILE) + ) + + +class TestDetectNewFiles(unittest.TestCase): + def test_finds_undeclared_filename(self): + lines = ['load("scph5500.bin");', 'load("scph7001.bin");'] + found, elided = detect_new_files(lines, declared_names(HASH_PROFILE)) + self.assertEqual(found, [(2, "scph7001.bin")]) + self.assertEqual(elided, 0) + + def test_ignores_declared_alias(self): + found, _ = detect_new_files( + ['open("SCPH-5500.bin")'], declared_names(HASH_PROFILE) + ) + self.assertEqual(found, []) + + def test_cap_reports_elided_count(self): + lines = [f'load("rom{i}.bin");' for i in range(40)] + found, elided = detect_new_files(lines, set(), cap=25) + self.assertEqual(len(found), 25) + self.assertEqual(elided, 15) + + def test_source_file_extensions_are_not_matched(self): + found, _ = detect_new_files(['#include "driver.cpp"'], set()) + self.assertEqual(found, []) + + +class TestWatchHashes(unittest.TestCase): + def test_finds_new_sha1(self): + added = ['+ { "x.bin", "0123456789abcdef0123456789abcdef01234567" },'] + found = watch_hashes(added, declared_hashes(HASH_PROFILE)) + self.assertEqual(found[0][1], "0123456789abcdef0123456789abcdef01234567") + + def test_ignores_declared_hash(self): + added = ["+ b05def971d8ec59f346f2d9ac21fb742e3eb6917"] + self.assertEqual(watch_hashes(added, declared_hashes(HASH_PROFILE)), []) + + def test_crc32_needs_a_filename_on_the_line(self): + self.assertEqual(watch_hashes(["+ mask = 0xdeadbeef;"], set()), []) + found = watch_hashes(['+ { "a.rom", 0xdeadbeef }'], set()) + self.assertEqual(found[0][1], "deadbeef") + + +class TestUnifiedForPath(unittest.TestCase): + def test_produces_a_hunk(self): + out = unified_for_path(["a", "b"], ["a", "c"], "f.c", 1) + self.assertIn("-b", out) + self.assertIn("+c", out) + self.assertIn("f.c", out) + + def test_identical_files_produce_nothing(self): + self.assertEqual(unified_for_path(["a"], ["a"], "f.c", 3), "") + + +class TestTreeDiff(unittest.TestCase): + def test_filters_to_ref_directories(self): + result = CompareResult( + [ + FileChange("added", "src/new.cpp", None), + FileChange("added", "docs/readme.md", None), + ], + False, + ) + self.assertEqual(tree_diff(result, {"src"}), ["added src/new.cpp"]) + + def test_rename_shows_both_paths(self): + result = CompareResult([FileChange("renamed", "src/b.c", "src/a.c")], False) + self.assertEqual(tree_diff(result, {"src"}), ["renamed src/a.c -> src/b.c"]) + + def test_repository_root_is_a_directory(self): + result = CompareResult( + [ + FileChange("modified", "libretro.c", None), + FileChange("modified", "deps/lightning/lib/jit_arm.c", None), + ], + False, + ) + self.assertEqual(tree_diff(result, {""}), ["modified libretro.c"]) + + def test_empty_ref_dirs_show_everything(self): + result = CompareResult([FileChange("modified", "a/b.c", None)], False) + self.assertEqual(tree_diff(result, set()), ["modified a/b.c"]) + + def test_truncation_is_announced(self): + result = CompareResult([FileChange("added", "src/a.c", None)], True) + self.assertIn("truncated", "\n".join(tree_diff(result, {"src"}))) + + +class TestDriftScore(unittest.TestCase): + def test_review_statuses_dominate(self): + heavy = ProfileReport(name="a", entries=[], counts={"GONE": 1}) + light = ProfileReport(name="b", entries=[], counts={"SHIFTED": 20}) + self.assertGreater(drift_score(heavy, None, 0), drift_score(light, None, 0)) + + def test_version_mismatch_adds_weight(self): + report = ProfileReport(name="a", entries=[], counts={}) + stale = VersionReport("v1.0", "v2.0", "v2.0", "2026-06-01", False, None) + current = VersionReport("v2.0", "v2.0", "v2.0", "2026-06-01", True, None) + self.assertGreater( + drift_score(report, stale, 0), drift_score(report, current, 0) + ) + + def test_commit_count_is_a_light_signal(self): + report = ProfileReport(name="a", entries=[], counts={}) + self.assertGreater(drift_score(report, None, 100), drift_score(report, None, 0)) + + def test_clean_profile_scores_zero(self): + report = ProfileReport(name="a", entries=[], counts={"ANCHORED": 5}) + self.assertEqual(drift_score(report, None, 0), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_upstream.py b/tests/test_upstream.py new file mode 100644 index 00000000..7e43b650 --- /dev/null +++ b/tests/test_upstream.py @@ -0,0 +1,318 @@ +"""Tests for the upstream repository access module (no network).""" + +from __future__ import annotations + +import os +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts")) + +import upstream +from upstream import make_repo, parse_repo, raw_url + + +def _no_network(url: str): + """Any call reaching here is a leak: the suite must stay offline.""" + raise AssertionError(f"test reached the network: {url}") + + +class TestParseRepo(unittest.TestCase): + def test_github(self): + repo = parse_repo("https://github.com/libretro/beetle-psx-libretro") + self.assertEqual(repo.family, "github") + self.assertEqual(repo.owner, "libretro") + self.assertEqual(repo.name, "beetle-psx-libretro") + + def test_github_trailing_git_and_slash(self): + self.assertEqual( + parse_repo("https://github.com/mamedev/mame.git/"), + parse_repo("https://github.com/mamedev/mame"), + ) + + def test_gitlab(self): + repo = parse_repo("https://gitlab.com/recalbox/recalbox") + self.assertEqual(repo.family, "gitlab") + + def test_codeberg_is_forgejo(self): + self.assertEqual(parse_repo("https://codeberg.org/a/b").family, "forgejo") + + def test_known_forgejo_instances(self): + for url in ( + "https://git.citron-emu.org/citron/emu", + "https://git.eden-emu.dev/eden-emu/eden", + ): + self.assertEqual(parse_repo(url).family, "forgejo") + + def test_unsupported_host(self): + self.assertIsNone(parse_repo("https://sourceforge.net/projects/vice")) + + def test_non_repository_url(self): + self.assertIsNone(parse_repo("https://mednafen.github.io/")) + + def test_empty(self): + self.assertIsNone(parse_repo("")) + + +class TestRawUrl(unittest.TestCase): + def test_github_raw(self): + repo = parse_repo("https://github.com/libretro/x") + self.assertEqual( + raw_url(repo, "abc123", "src/main.cpp"), + "https://raw.githubusercontent.com/libretro/x/abc123/src/main.cpp", + ) + + def test_gitlab_raw(self): + repo = parse_repo("https://gitlab.com/g/p") + self.assertEqual( + raw_url(repo, "abc123", "a.c"), + "https://gitlab.com/g/p/-/raw/abc123/a.c", + ) + + def test_forgejo_raw(self): + repo = parse_repo("https://codeberg.org/g/p") + self.assertEqual( + raw_url(repo, "abc123", "a.c"), + "https://codeberg.org/g/p/raw/commit/abc123/a.c", + ) + + def test_path_is_quoted(self): + repo = parse_repo("https://github.com/o/n") + self.assertIn("src/a%20b.cpp", raw_url(repo, "s", "src/a b.cpp")) + + +class TestMakeRepo(unittest.TestCase): + def test_rebuilds_a_known_host(self): + self.assertEqual( + make_repo("github.com", "o", "n"), parse_repo("https://github.com/o/n") + ) + + def test_unknown_host_returns_none(self): + self.assertIsNone(make_repo("example.invalid", "o", "n")) + + +class TestCache(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.repo = parse_repo("https://github.com/o/n") + self.calls: list[str] = [] + self._orig = (upstream._http_text, upstream._http_json) + upstream._http_text = self._fake + upstream._http_json = _no_network + + def tearDown(self): + upstream._http_text, upstream._http_json = self._orig + self.tmp.cleanup() + + def _fake(self, url: str) -> str | None: + self.calls.append(url) + return "line one\nline two\n" + + def test_path_includes_host_slug_and_sha(self): + p = upstream.cache_path(self.dir, self.repo, "abc123", "src/a.cpp") + self.assertIn("github.com", str(p)) + self.assertIn("o", str(p)) + self.assertIn("abc123", str(p)) + + def test_path_is_contained_in_cache_dir(self): + p = upstream.cache_path(self.dir, self.repo, "abc", "../../escape.c") + self.assertTrue(str(p.resolve()).startswith(str(Path(self.dir).resolve()))) + + def test_fetch_then_cache_hit(self): + first = upstream.fetch_file(self.repo, "abc", "a.c", self.dir) + second = upstream.fetch_file(self.repo, "abc", "a.c", self.dir) + self.assertEqual(first, ["line one", "line two"]) + self.assertEqual(second, first) + self.assertEqual(len(self.calls), 1) + + def test_offline_miss_returns_none_without_request(self): + self.assertIsNone( + upstream.fetch_file(self.repo, "abc", "a.c", self.dir, offline=True) + ) + self.assertEqual(self.calls, []) + + def test_offline_hit_serves_cache(self): + upstream.fetch_file(self.repo, "abc", "a.c", self.dir) + self.calls.clear() + self.assertEqual( + upstream.fetch_file(self.repo, "abc", "a.c", self.dir, offline=True), + ["line one", "line two"], + ) + self.assertEqual(self.calls, []) + + def test_missing_file_is_cached_as_absent(self): + upstream._http_text = lambda url: None + self.assertIsNone(upstream.fetch_file(self.repo, "abc", "gone.c", self.dir)) + self.assertIsNone( + upstream.fetch_file(self.repo, "abc", "gone.c", self.dir, offline=True) + ) + + def test_write_leaves_no_temporary_behind(self): + target = Path(self.dir) / "sub" / "f.txt" + upstream.write_cache(target, "payload") + self.assertEqual(target.read_text(encoding="utf-8"), "payload") + siblings = list(target.parent.iterdir()) + self.assertEqual([p.name for p in siblings], ["f.txt"]) + + +class TestRevisions(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.repo = parse_repo("https://github.com/o/n") + self.responses: dict[str, object] = {} + self.calls: list[str] = [] + self._orig = (upstream._http_json, upstream._http_text) + upstream._http_json = self._fake + upstream._http_text = _no_network + + def tearDown(self): + upstream._http_json, upstream._http_text = self._orig + self.tmp.cleanup() + + def _fake(self, url: str): + self.calls.append(url) + for fragment, payload in self.responses.items(): + if fragment in url: + return payload + return None + + def test_resolve_head_github(self): + self.responses["/commits"] = [{"sha": "deadbeef"}] + self.assertEqual(upstream.resolve_head(self.repo, self.dir), "deadbeef") + + def test_resolve_head_gitlab_uses_id(self): + repo = parse_repo("https://gitlab.com/g/p") + self.responses["/repository/commits"] = [{"id": "cafe"}] + self.assertEqual(upstream.resolve_head(repo, self.dir), "cafe") + + def test_resolve_commit_at_passes_date(self): + self.responses["/commits"] = [{"sha": "abc"}] + self.assertEqual( + upstream.resolve_commit_at(self.repo, "2026-03-29", self.dir), "abc" + ) + self.assertIn("2026-03-29", self.calls[0]) + + def test_api_response_is_cached(self): + self.responses["/commits"] = [{"sha": "abc"}] + upstream.resolve_commit_at(self.repo, "2026-03-29", self.dir) + upstream.resolve_commit_at(self.repo, "2026-03-29", self.dir) + self.assertEqual(len(self.calls), 1) + + def test_offline_without_cache_returns_none(self): + self.assertIsNone(upstream.resolve_head(self.repo, self.dir, offline=True)) + self.assertEqual(self.calls, []) + + def test_empty_history_returns_none(self): + self.responses["/commits"] = [] + self.assertIsNone(upstream.resolve_head(self.repo, self.dir)) + + def test_list_tags(self): + self.responses["/tags"] = [{"name": "v1.2"}, {"name": "v1.1"}] + self.assertEqual(upstream.list_tags(self.repo, self.dir), ["v1.2", "v1.1"]) + + def test_resolve_tag_commit(self): + self.responses["/tags"] = [{"name": "v1.2", "commit": {"sha": "tagsha"}}] + self.assertEqual( + upstream.resolve_tag_commit(self.repo, "v1.2", self.dir), "tagsha" + ) + + def test_latest_release(self): + self.responses["/releases/latest"] = { + "tag_name": "v3.0.0", + "published_at": "2026-06-14T10:00:00Z", + "prerelease": False, + } + rel = upstream.latest_release(self.repo, self.dir) + self.assertEqual(rel.tag, "v3.0.0") + self.assertEqual(rel.date, "2026-06-14") + self.assertFalse(rel.is_prerelease) + + def test_missing_release_returns_none(self): + self.assertIsNone(upstream.latest_release(self.repo, self.dir)) + + def test_rate_limit_propagates(self): + def boom(url: str): + raise upstream.RateLimitError("quota") + + upstream._http_json = boom + with self.assertRaises(upstream.RateLimitError): + upstream.resolve_head(self.repo, self.dir) + + +class TestCompare(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.repo = parse_repo("https://github.com/o/n") + self.payload: object = None + self._orig = (upstream._http_json, upstream._http_text) + upstream._http_json = lambda url: self.payload + upstream._http_text = _no_network + + def tearDown(self): + upstream._http_json, upstream._http_text = self._orig + self.tmp.cleanup() + + def test_renamed_file_carries_previous_path(self): + self.payload = { + "files": [ + { + "status": "renamed", + "filename": "src/new.cpp", + "previous_filename": "old.cpp", + }, + {"status": "modified", "filename": "a.c"}, + ] + } + result = upstream.compare(self.repo, "a", "b", self.dir) + self.assertFalse(result.truncated) + self.assertEqual(upstream.find_renamed(result, "old.cpp"), "src/new.cpp") + + def test_no_rename_returns_none(self): + self.payload = {"files": [{"status": "modified", "filename": "a.c"}]} + result = upstream.compare(self.repo, "a", "b", self.dir) + self.assertIsNone(upstream.find_renamed(result, "old.cpp")) + + def test_truncated_at_github_cap(self): + self.payload = { + "files": [ + {"status": "modified", "filename": f"f{i}.c"} for i in range(300) + ] + } + self.assertTrue(upstream.compare(self.repo, "a", "b", self.dir).truncated) + + def test_missing_comparison_is_empty_and_truncated(self): + self.payload = None + result = upstream.compare(self.repo, "a", "b", self.dir) + self.assertEqual(result.files, []) + self.assertTrue(result.truncated) + + def test_gitlab_diffs_shape(self): + repo = parse_repo("https://gitlab.com/g/p") + self.payload = { + "diffs": [ + { + "new_path": "new.c", + "old_path": "old.c", + "renamed_file": True, + "new_file": False, + "deleted_file": False, + } + ] + } + result = upstream.compare(repo, "a", "b", self.dir) + self.assertEqual(upstream.find_renamed(result, "old.c"), "new.c") + + def test_commits_touching_counts_entries(self): + self.payload = [{"sha": "1"}, {"sha": "2"}, {"sha": "3"}] + self.assertEqual( + upstream.commits_touching(self.repo, "a.c", "base", self.dir), 3 + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/wiki/architecture.md b/wiki/architecture.md index c063d9e1..6607e13e 100644 --- a/wiki/architecture.md +++ b/wiki/architecture.md @@ -275,19 +275,23 @@ user's platform, filter files by hardware target, and download with SHA1 verific ## Tests -10 test files, 400 tests total: +14 test files, 586 tests total: | File | Tests | Coverage | |------|-------|----------| -| `test_e2e.py` | 217 | file resolution, verification, severity, cross-reference, aliases, inheritance, shared groups, data dirs, storage tiers, HLE, launchers, platform grouping, core resolution, target filtering, truth/diff, exporters | +| `test_e2e.py` | 218 | file resolution, verification, severity, cross-reference, aliases, inheritance, shared groups, data dirs, storage tiers, HLE, launchers, platform grouping, core resolution, target filtering, truth/diff, exporters | +| `test_profile_sync.py` | 129 | ref anchoring, guarded profile writes, detection, triage | | `test_install.py` | 70 | `install.py` platform detection, config-file parsing, manifest handling | +| `test_upstream.py` | 38 | forge URL parsing, cache, revision resolution, tree comparison | | `test_provenance.py` | 29 | Logiqx/Redump parsers, DAT pack import, provenance join, coverage report | -| `test_mame_parser.py` | 22 | BIOS root set detection, ROM block parsing, macro expansion | +| `test_mame_parser.py` | 25 | BIOS root set detection, ROM block parsing, macro expansion | | `test_hash_merge.py` | 17 | MAME/FBNeo YAML merge, diff detection, formatting preservation | | `test_fbneo_parser.py` | 16 | BIOS set detection, ROM info parsing | -| `test_profile_refs.py` | 12 | `check_profile_refs` pure functions (no network) | +| `test_deterministic_zip.py` | 12 | streaming rebuild, metadata normalisation, entry ordering, source CRC | +| `test_artifact_lock.py` | 10 | writer/writer and writer/reader exclusion, reader sharing, release on error | | `test_pack_integrity.py` | 8 | extract ZIP packs to disk, verify paths + hashes per platform's native mode | | `test_torrentzip.py` | 8 | TorrentZip builder against real MAME romsets | +| `test_large_file_cache.py` | 5 | concurrent downloads, temporary file residue, hash rejection | | `test_no_case_collisions.py` | 1 | guard against case-colliding paths in `bios/` | ```bash @@ -295,9 +299,9 @@ python -m unittest discover tests -v # full suite python -m unittest tests.test_e2e -v # single module ``` -`test_e2e.py`, `test_install.py`, `test_provenance.py`, the parser tests and -`test_profile_refs.py` run on synthetic fixtures with no network and no real -BIOS files. `test_pack_integrity.py`, `test_torrentzip.py` and +`test_e2e.py`, `test_install.py`, `test_provenance.py`, the parser tests, +`test_profile_sync.py` and `test_upstream.py` run on synthetic fixtures with no +network and no real BIOS files. `test_pack_integrity.py`, `test_torrentzip.py` and `test_no_case_collisions.py` read the working tree and skip when the data they need is absent. See the [testing guide](testing-guide.md) for the fixture pattern and how to add a test. diff --git a/wiki/profiling.md b/wiki/profiling.md index 9871f8e0..30cce568 100644 --- a/wiki/profiling.md +++ b/wiki/profiling.md @@ -215,7 +215,7 @@ A few field conventions that protect the toolchain: python scripts/cross_reference.py --emulator dolphin --json python scripts/verify.py --emulator dolphin python scripts/verify.py --emulator dolphin --verbose # per-core checks + source refs -python scripts/check_profile_refs.py --emulator dolphin # do the source_ref lines still hold +python scripts/profile_sync.py --emulator dolphin # do the source_ref lines still hold ``` The profile also has to satisfy `schemas/emulator.schema.json`, which CI checks diff --git a/wiki/testing-guide.md b/wiki/testing-guide.md index dfd25401..232d87da 100644 --- a/wiki/testing-guide.md +++ b/wiki/testing-guide.md @@ -23,7 +23,11 @@ python -m unittest tests.test_provenance -v python -m unittest tests.test_mame_parser -v python -m unittest tests.test_hash_merge -v python -m unittest tests.test_fbneo_parser -v -python -m unittest tests.test_profile_refs -v +python -m unittest tests.test_profile_sync -v +python -m unittest tests.test_upstream -v +python -m unittest tests.test_deterministic_zip -v +python -m unittest tests.test_artifact_lock -v +python -m unittest tests.test_large_file_cache -v python -m unittest tests.test_pack_integrity -v python -m unittest tests.test_torrentzip -v python -m unittest tests.test_no_case_collisions -v @@ -36,15 +40,19 @@ library `unittest` module. | Module | Tests | Fixtures | What it covers | |--------|-------|----------|----------------| -| `test_e2e.py` | 217 | synthetic | resolution, verification, packs, cross-reference, targets, truth | +| `test_e2e.py` | 218 | synthetic | resolution, verification, packs, cross-reference, targets, truth | +| `test_profile_sync.py` | 129 | synthetic | ref anchoring, guarded profile writes, detection, triage | | `test_install.py` | 70 | synthetic | `install.py` detection, config parsing, manifest handling | +| `test_upstream.py` | 38 | synthetic | forge URL parsing, cache, revision resolution, tree comparison | | `test_provenance.py` | 29 | synthetic | Logiqx/Redump parsing, DAT import, provenance join, coverage report | -| `test_mame_parser.py` | 22 | inline C | BIOS root sets, ROM blocks, macro expansion | +| `test_mame_parser.py` | 25 | inline C | BIOS root sets, ROM blocks, macro expansion | | `test_hash_merge.py` | 17 | synthetic | YAML hash merge, diff, formatting preservation | | `test_fbneo_parser.py` | 16 | inline C | `BDF_BOARDROM` sets, ROM info parsing | -| `test_profile_refs.py` | 12 | synthetic | `check_profile_refs` pure functions, no network | +| `test_deterministic_zip.py` | 12 | synthetic | streaming rebuild, metadata normalisation, entry ordering, source CRC | +| `test_artifact_lock.py` | 10 | synthetic | writer/writer and writer/reader exclusion, reader sharing, release on error | | `test_pack_integrity.py` | 8 | real packs | extract each ZIP, verify paths and hashes | | `test_torrentzip.py` | 8 | real romsets | TorrentZip builder byte-for-byte | +| `test_large_file_cache.py` | 5 | synthetic | concurrent downloads, temporary file residue, hash rejection | | `test_no_case_collisions.py` | 1 | real `bios/` | no case-colliding paths on Windows/macOS clones | ## Test architecture @@ -123,9 +131,15 @@ detection, each registry detection method (`config_file`, `path_exists`, `file_exists`), config-file key parsing, manifest loading, target filtering, and destination resolution. -**test_profile_refs.** Covers the pure functions of `check_profile_refs` -(anchor matching, line-window search, hash extraction). The GitHub fetching -path is not exercised, so the module runs offline. +**test_profile_sync.** Covers ref anchoring end to end: the six statuses, +widening a one-line anchor until it is unique, refusing to rebase an ambiguous +one, following a rename, and the guarded YAML writes. Network access is +replaced by an injected fetch function. + +**test_upstream.** Covers forge URL parsing for GitHub, GitLab and Forgejo, +the content-addressed cache and its atomic write, revision and tag resolution, +and tree comparison. The HTTP layer is replaced at module level, so nothing +leaves the machine. ### Tests that read the working tree diff --git a/wiki/tools.md b/wiki/tools.md index d42bc675..77c54501 100644 --- a/wiki/tools.md +++ b/wiki/tools.md @@ -208,25 +208,55 @@ to the files inside the archive, so the exported DAT lists those entries without a container sha1. Anyone submitting the DAT upstream should mention this. -### check_profile_refs.py +### profile_sync.py -Audit `source_ref` line references against the profiled upstream. The -commit under audit is the profile's `source_commit` when present, else the -last upstream commit at `profiled_date`. Each referenced file is fetched -at that commit and at HEAD, and the entry's declared hashes (or filename) -are searched around the cited lines. +Confront a profile with its upstream. The pinned commit is the profile's +`source_commit` when present, else the last upstream commit at +`profiled_date`. Each cited line range is extracted at the pin and located +in the HEAD revision of the same file. ```bash -python scripts/check_profile_refs.py --emulator vice -python scripts/check_profile_refs.py --all --json +python scripts/profile_sync.py --emulator vice +python scripts/profile_sync.py --emulator vice --full-diff +python scripts/profile_sync.py --all --triage +python scripts/profile_sync.py --all --changed-only --json +python scripts/profile_sync.py --emulator vice --fetch-plan ``` -Per ref and revision: `anchored` (found at the cited lines), `moved` -(found elsewhere in the file), `gone` (absent from the file). `moved` at -HEAD means upstream shifted since profiling; `gone` at pin means the -declared value does not come from the referenced file and the entry -deserves a re-read. Uses `GITHUB_TOKEN` when set; GitHub-hosted upstreams -only. +Per part of a ref: `ANCHORED` (same content, same lines), `SHIFTED` (same +content, moved), `RENAMED` (the source file moved), `CHANGED` (content +edited), `AMBIGUOUS` (several equally good candidates), `GONE` (nothing +left to anchor to). An entry carries the worst status of its parts. + +A single cited line is often not distinctive, so the anchor widens by +steps of ±3, ±6 and ±12 lines until it is unique. Anything still ambiguous +is reported, never guessed. + +Three shorthand forms appear in the corpus and are resolved rather than +reported missing. A part reduced to a line range continues the previous +part's file (`geo.c:234-243, 273-285`). A profile whose `source` differs +from its `upstream` may cite paths from both, and each path is attributed +to the repository that carries it. A path prefixed with a repository +directory name (`EightyOne/Source/HW_.cpp`) is stripped as a last resort, +only after the path as written has failed everywhere, and the result is +reported as `RENAMED` so `--rebase-refs` cleans the profile. + +`--check-version` compares `core_version` with the latest upstream tag and +release. `--detect-new-files` lists filename literals at HEAD the profile +does not declare. `--watch-hashes` lists hash literals added upstream that +match no entry. `--tree-diff` shows added, removed and renamed files in the +directories the refs point at. + +Writes are explicit and mechanical only. `--backfill-commits` fills a +missing `source_commit`, `--rebase-refs` recales `SHIFTED` and `RENAMED` +line ranges, `--bump-commit` advances `source_commit` to HEAD only when +nothing needs a re-read. All three refuse to run on a dirty `emulators/` +without `--force`, and every write is verified by reparsing the document. + +Uses `GITHUB_TOKEN` when set, which `--all` requires. Responses are cached +under `.cache/upstream/`, addressed by commit sha, so `--offline` replays a +previous run. GitHub, GitLab and Forgejo upstreams; other hosts are +reported as skipped. ### validation.py