From a9e6fff22771027eda562ccbe8fadde8f44fd8a0 Mon Sep 17 00:00:00 2001 From: plexerPK <160128888+plexerPK@users.noreply.github.com> Date: Sun, 25 Oct 2026 11:16:16 +0100 Subject: [PATCH] fix(tools): preserve debt entry identities during conflict resolution --- tools/debt_split.py | 65 +++++++++++++++++++++++----------- tools/tests/test_debt_split.py | 63 ++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 21 deletions(-) diff --git a/tools/debt_split.py b/tools/debt_split.py index 1460f46bca..ab66d0de76 100644 --- a/tools/debt_split.py +++ b/tools/debt_split.py @@ -149,6 +149,10 @@ def git(root, *args, check=True): def apply_change(root, base, new, index_text): + class Entry: + def __init__(self, text): + self.text = text + order = sections_of(index_text) or sections_of(new) before = [(kind, section, item, key(kind, item)) for kind, section, item in parse(base) if kind == "entry" and key(kind, item)] after = [(kind, section, item, key(kind, item)) for kind, section, item in parse(new) if kind == "entry" and key(kind, item)] @@ -156,51 +160,70 @@ def apply_change(root, base, new, index_text): def load(name): if name not in files: - files[name] = read_debt(root, name) + files[name] = {section: [Entry(text) for text in entries] + for section, entries in read_debt(root, name).items()} return files[name] - def locate(entry): - _, section, item, name = entry - entries = load(name).get(section, []) - moved = relink(item, INDEX, debt_file(name)) - if moved not in entries: - raise LookupError(f"{debt_file(name)}: entry not found under '{section}': {item.splitlines()[0][:100]}") - return entries, entries.index(moved) + matcher = difflib.SequenceMatcher(None, [entry[1:3] for entry in before], [entry[1:3] for entry in after], autojunk=False) + operations = matcher.get_opcodes() + names = {entry[3] for tag, i1, i2, j1, j2 in operations if tag != "equal" + for entry in before[i1:i2] + after[j1:j2]} + available = {} + for name in names: + for section, entries in load(name).items(): + for entry in entries: + available.setdefault((name, section, entry.text), []).append(entry) + available = {key: iter(entries) for key, entries in available.items()} + originals = [next(available.get((name, section, relink(item, INDEX, debt_file(name))), iter(())), None) + for _, section, item, name in before] + resolved = [None] * len(after) - def insert(entry, anchor): - _, section, item, name = entry + def locate(index): + _, section, item, name = before[index] + entries = load(name).get(section, []) + original = originals[index] + if original is None or original not in entries: + raise LookupError(f"{debt_file(name)}: entry not found under '{section}': {item.splitlines()[0][:100]}") + return entries, entries.index(original) + + def insert(index, anchor): + _, section, item, name = after[index] entries = load(name).setdefault(section, []) - moved = relink(item, INDEX, debt_file(name)) + moved = Entry(relink(item, INDEX, debt_file(name))) position = 0 if anchor is None else len(entries) - if anchor is not None and anchor[1] == section and anchor[3] == name: - anchored = relink(anchor[2], INDEX, debt_file(name)) + if anchor is not None: + anchored = resolved[anchor] if anchored in entries: position = entries.index(anchored) + 1 entries.insert(position, moved) + resolved[index] = moved - matcher = difflib.SequenceMatcher(None, [entry[1:3] for entry in before], [entry[1:3] for entry in after], autojunk=False) - for tag, i1, i2, j1, j2 in matcher.get_opcodes(): + for tag, i1, i2, j1, j2 in operations: if tag == "equal": + resolved[j1:j2] = originals[i1:i2] continue removed, added = before[i1:i2], after[j1:j2] kept_old, kept_new = set(), set() for offset, (old, fresh) in enumerate(zip(removed, added)): if old[1] == fresh[1] and old[3] == fresh[3]: - entries, position = locate(old) - entries[position] = relink(fresh[2], INDEX, debt_file(fresh[3])) + entries, position = locate(i1 + offset) + entries[position].text = relink(fresh[2], INDEX, debt_file(fresh[3])) + resolved[j1 + offset] = entries[position] kept_old.add(offset) kept_new.add(offset) for offset, old in enumerate(removed): if offset not in kept_old: - entries, position = locate(old) + entries, position = locate(i1 + offset) del entries[position] for offset, fresh in enumerate(added): if offset in kept_new: continue - anchor = next((entry for entry in reversed(after[:j1 + offset]) if entry[1] == fresh[1] and entry[3] == fresh[3]), None) - insert(fresh, anchor) + anchor = next((index for index in range(j1 + offset - 1, -1, -1) + if after[index][1] == fresh[1] and after[index][3] == fresh[3]), None) + insert(j1 + offset, anchor) for name, sections in files.items(): - write_debt(root, name, sections, order) + write_debt(root, name, {section: [entry.text for entry in entries] + for section, entries in sections.items()}, order) return sorted(debt_file(name) for name in files) diff --git a/tools/tests/test_debt_split.py b/tools/tests/test_debt_split.py index ae68754774..d26cdd035e 100644 --- a/tools/tests/test_debt_split.py +++ b/tools/tests/test_debt_split.py @@ -224,6 +224,69 @@ class DebtSplitTests(unittest.TestCase): "\n\n### Functional\n\n" + "\n".join(entry.replace("../../", "../../../") for entry in (added, A_FUNCTIONAL)) + "\n") + def test_resolve_preserves_an_entry_moved_before_another_in_its_file(self): + first = "- [libSceA](../../core/libs/prx/libSceA/First.cpp) - first limitation" + second = "- [libSceA](../../core/libs/prx/libSceA/Second.cpp) - second limitation" + self.write(INDEX, index([first, second], [])) + self.git("add", "--", INDEX) + self.git("commit", "-qm", "docs: record ordered debt") + self.git("checkout", "-qb", "reorder") + self.write(INDEX, index([second, first], [])) + self.git("commit", "-qam", "docs: reorder debt") + self.git("checkout", "-q", "main") + result = self.tool("split") + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.git("add", "-A") + self.git("commit", "-qm", "docs: split technical debt") + self.git("checkout", "-q", "reorder") + result = self.git("rebase", "main", check=False) + self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) + result = self.tool("resolve") + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.git("rebase", "--continue") + self.assertEqual(self.read("docs/dev/debt/libSceA.md"), + "# libSceA technical debt\n\n### Silent stubs\n\n" + + "\n".join(entry.replace("../../", "../../../") for entry in (second, first)) + "\n") + + def test_apply_changes_preserve_original_occurrences_and_upstream_entries(self): + a = "- [libSceA](../../core/libs/prx/libSceA/A.cpp) - a" + b = "- [libSceA](../../core/libs/prx/libSceA/B.cpp) - b" + c = "- [libSceA](../../core/libs/prx/libSceA/C.cpp) - c" + u, v = "- upstream first", "- upstream last" + cases = ( + ([a, b], [a, b], [b, a], [b, a]), + ([a, b, c], [a, u, b, c, v], [c, a, b], [c, a, u, b, v]), + ([a, b, a], [a, u, b, a, v], [a, a, b], [a, a, u, b, v]), + ([a, b, c, a], [a, u, b, c, a, v], [a, c, a, b], [a, c, a, u, b, v]), + ([a, b, a], [a, u, b, a, v], [b, a], [u, b, a, v]), + ([a, b, a], [a, u, b, a, v], [a, b], [a, u, b, v]), + ([a, b, a], [a, u, b, a, v], [a, b, a + " revised"], [a, u, b, a + " revised", v]), + ([a, b, a], [a, u, b, a, v], [a, b, a, c], [a, u, b, a, c, v]), + ([a, b], [a, u, b, v], [a, c, c, b], [a, c, c, u, b, v]), + ([a, b], [a, u, b, v], [], [u, v]), + ([a, b, a], [a, u, b, a, v], [a + " revised", a, c], [a + " revised", a, u, c, v]), + ([a, b], [a, u, b + " upstream", v], [a + " revised", b], [a + " revised", u, b + " upstream", v]), + ) + path = "docs/dev/debt/libSceA.md" + for before, upstream, after, expected in cases: + with self.subTest(before=before, after=after): + self.write(path, "# libSceA technical debt\n\n### Functional\n\n" + + "\n".join(entry.replace("../../", "../../../") for entry in upstream) + "\n") + debt_split.apply_change(self.root, index([], before), index([], after), index([], [])) + self.assertEqual(self.read(path), "# libSceA technical debt\n\n### Functional\n\n" + + "\n".join(entry.replace("../../", "../../../") for entry in expected) + "\n") + + def test_apply_changes_missing_original_does_not_write_files(self): + original = "# libSceA technical debt\n\n### Silent stubs\n\n" + A_STUB.replace("../../", "../../../") + "\n" + a_path, b_path = "docs/dev/debt/libSceA.md", "docs/dev/debt/libSceB.md" + self.write(a_path, original) + self.write(b_path, "# libSceB technical debt\n\n### Silent stubs\n\n- changed upstream\n") + before = {path: self.read(path) for path in (a_path, b_path)} + with self.assertRaises(LookupError): + debt_split.apply_change(self.root, index([A_STUB, B_STUB], []), + index([A_STUB + " revised", B_STUB + " revised"], []), index([], [])) + self.assertEqual({path: self.read(path) for path in (a_path, b_path)}, before) + def test_indented_continuation_after_a_blank_line_moves_with_its_entry(self): note = " - sub note about sceAOne" self.write(INDEX, "# Project technical debt\n\n### Silent stubs\n\n" + A_STUB + "\n\n" + note + "\n")