From dc4b23a8838842ef46dedb0d0de7e5b2bb334ed7 Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Sat, 8 Aug 2026 12:24:31 +0200 Subject: [PATCH] feat: anchor on declared values when refs miss --- scripts/profile_sync.py | 71 ++++++++++++++++++++++++++++++++++---- tests/test_profile_sync.py | 63 +++++++++++++++++++++++++++++++-- 2 files changed, 126 insertions(+), 8 deletions(-) diff --git a/scripts/profile_sync.py b/scripts/profile_sync.py index 33a04a07..f1e3bc22 100644 --- a/scripts/profile_sync.py +++ b/scripts/profile_sync.py @@ -171,6 +171,20 @@ def collect_tokens(entry: dict) -> list[str]: return tokens +def _anchor_tokens(entry: dict) -> list[str]: + """Values worth searching for when a cited line misses its subject. + + Archive entries are named `stvbios.zip` while the driver source writes the + set name alone, so the stem is searched as well as the full name. + """ + tokens = collect_tokens(entry) + name = os.path.basename(str(entry.get("name", ""))).lower() + stem = name.rsplit(".", 1)[0] + if stem and stem not in tokens and len(stem) > 2: + tokens.append(stem) + return tokens + + def worst_status(statuses) -> str: """Severity of an entry is the worst severity among its parts.""" worst = "ANCHORED" @@ -255,6 +269,36 @@ def anchor_block( return _map_changed(pin, head, lo, hi) +NUDGE_WINDOW = 60 + + +def nudge_to_declared( + pin_lines: list[str], start: int, end: int, tokens +) -> tuple[int, int] | None: + """Move a cited range onto the nearest line carrying a declared value. + + Some refs sit a few lines off their subject, and a few land on a blank + line, which anchors nothing. When the entry declares a hash or a name and + that value appears close by, the ref plainly meant that line. + """ + if not tokens: + return None + lowered = [line.lower() for line in pin_lines] + hits = [ + i + 1 + for i, line in enumerate(lowered) + if any(token in line for token in tokens) + ] + if not hits: + return None + nearest = min(hits, key=lambda n: abs(n - start)) + if abs(nearest - start) > NUDGE_WINDOW or nearest == start: + return None + if sum(1 for h in hits if abs(h - nearest) <= 1) > 1: + return None + return nearest, nearest + (end - start) + + def resolve_rename( result: CompareResult, path: str, head_paths=() ) -> tuple[str | None, list[str]]: @@ -298,7 +342,9 @@ def _narrow(matches: list[str], path: str) -> list[str]: return same if len(same) == 1 else matches -def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResult: +def anchor_part( + part: RefPart, fetch, rename_getter, describe=None, tokens=() +) -> PartResult: """Locate one reference part at HEAD, following a rename when needed. `describe(path)` returns (repo slug, raw URL at HEAD) for the repository @@ -364,7 +410,16 @@ def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResul part, "GONE", None, None, None, [], "pin revision missing", slug, url ) - anchored = anchor_block(pin_lines, head_lines, part.start, part.end or part.start) + end = part.end or part.start + anchored = anchor_block(pin_lines, head_lines, part.start, end) + note = anchored.reason + if anchored.status in REVIEW_STATUSES: + nudged = nudge_to_declared(pin_lines, part.start, end, tokens) + if nudged is not None: + retry = anchor_block(pin_lines, head_lines, *nudged) + if retry.status not in REVIEW_STATUSES: + anchored = retry + note = f"cited line was {nudged[0] - part.start:+d} off its subject" status = anchored.status if renamed and status in ("ANCHORED", "SHIFTED"): status = "RENAMED" @@ -375,7 +430,7 @@ def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResul anchored.start, anchored.end, anchored.candidates, - anchored.reason, + note, slug, url, ) @@ -518,7 +573,11 @@ def build_report( report = ProfileReport(name=name, entries=[], counts={}) refs = [ - (entry.get("name", "") + (f" [{label}]" if label else ""), value) + ( + entry.get("name", "") + (f" [{label}]" if label else ""), + value, + _anchor_tokens(entry), + ) for entry in (profile.get("files") or []) if isinstance(entry, dict) and entry.get("source_ref") for label, value in source_ref_values(entry.get("source_ref")) @@ -629,10 +688,10 @@ def build_report( staged = [ (entry_name, ref, [ - anchor_part(part, fetch, rename_getter, describe) + anchor_part(part, fetch, rename_getter, describe, tokens) for part in split_source_ref(ref) ]) - for entry_name, ref in refs + for entry_name, ref, tokens in refs ] shifts = dominant_shifts([p for _, _, parts in staged for p in parts]) diff --git a/tests/test_profile_sync.py b/tests/test_profile_sync.py index eb05af8d..6ebfe240 100644 --- a/tests/test_profile_sync.py +++ b/tests/test_profile_sync.py @@ -280,7 +280,64 @@ def _part(path, old, new, status): def _ambiguous(path, old, candidates): - return PartResult(RefPart(path, old, old), "AMBIGUOUS", None, None, None, candidates) + return PartResult( + RefPart(path, old, old), "AMBIGUOUS", None, None, None, candidates + ) + + +class TestAnchorTokens(unittest.TestCase): + def test_archive_stem_is_searched_too(self): + tokens = profile_sync._anchor_tokens({"name": "stvbios.zip"}) + self.assertIn("stvbios.zip", tokens) + self.assertIn("stvbios", tokens) + + def test_hashes_still_win(self): + tokens = profile_sync._anchor_tokens( + {"name": "a.zip", "sha1": "AABB" * 10} + ) + self.assertIn("aabb" * 10, tokens) + + def test_very_short_stem_is_skipped(self): + self.assertNotIn("ab", profile_sync._anchor_tokens({"name": "ab.zip"})) + + +class TestNudgeToDeclared(unittest.TestCase): + LINES = ["pad"] * 20 + ["ROM_START( stvbios )"] + ["pad"] * 20 + + def test_moves_onto_the_declared_value(self): + self.assertEqual( + profile_sync.nudge_to_declared(self.LINES, 19, 19, ["stvbios"]), + (21, 21), + ) + + def test_range_length_is_kept(self): + self.assertEqual( + profile_sync.nudge_to_declared(self.LINES, 19, 22, ["stvbios"]), + (21, 24), + ) + + def test_nothing_when_already_on_target(self): + self.assertIsNone( + profile_sync.nudge_to_declared(self.LINES, 21, 21, ["stvbios"]) + ) + + def test_nothing_beyond_the_window(self): + lines = ["pad"] * 500 + ["ROM_START( stvbios )"] + self.assertIsNone(profile_sync.nudge_to_declared(lines, 1, 1, ["stvbios"])) + + def test_nothing_without_tokens(self): + self.assertIsNone(profile_sync.nudge_to_declared(self.LINES, 19, 19, [])) + + def test_nothing_when_the_value_is_absent(self): + self.assertIsNone( + profile_sync.nudge_to_declared(self.LINES, 19, 19, ["absent"]) + ) + + def test_adjacent_duplicates_are_refused(self): + lines = ["pad"] * 20 + ["stvbios a", "stvbios b"] + ["pad"] * 20 + self.assertIsNone( + profile_sync.nudge_to_declared(lines, 19, 19, ["stvbios"]) + ) class TestDominantShift(unittest.TestCase): @@ -315,7 +372,9 @@ class TestDominantShift(unittest.TestCase): "AMBIGUOUS") def test_range_length_is_preserved(self): - part = PartResult(RefPart("a.c", 100, 104), "AMBIGUOUS", None, None, None, [120]) + part = PartResult( + RefPart("a.c", 100, 104), "AMBIGUOUS", None, None, None, [120] + ) settled = profile_sync.resolve_by_shift(part, {"a.c": 20}) self.assertEqual((settled.start, settled.end), (120, 124))