test_core_cohen_transform.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Tests for muse.core.cohen_transform — the Cohen Transform three-way merge. |
| 2 | |
| 3 | Named in honour of Bram Cohen (creator of BitTorrent, Manyana CRDT weave), |
| 4 | whose conflict-presentation insight is the direct inspiration for this module. |
| 5 | |
| 6 | Test categories |
| 7 | --------------- |
| 8 | TestClassifyAction — unit: classify_action() |
| 9 | TestAnnotateHunkAction — unit: annotate_hunk_action() |
| 10 | TestComputeRegions — unit: compute_regions() / _find_sync_regions() |
| 11 | TestThreeWayMergeLines — unit: three_way_merge_lines() clean + conflict |
| 12 | TestConflictMarkerFormat — unit: marker syntax, diff3 style, label content |
| 13 | TestFormatConflictDiff — unit: format_conflict_diff() rendering |
| 14 | TestEdgeCases — unit: empty sequences, binary-safe text, unicode |
| 15 | TestIntegration — integration: realistic multi-hunk scenarios |
| 16 | TestStress — stress: large files, many conflicts, many hunks |
| 17 | TestDataIntegrity — data integrity: determinism, idempotency |
| 18 | TestSecurity — security: ANSI injection, path traversal, null bytes |
| 19 | """ |
| 20 | |
| 21 | from __future__ import annotations |
| 22 | |
| 23 | import difflib |
| 24 | import pathlib |
| 25 | import threading |
| 26 | import time |
| 27 | from unittest.mock import MagicMock |
| 28 | |
| 29 | import pytest |
| 30 | |
| 31 | from muse.core.cohen_transform import ( |
| 32 | CONFLICT_SEPARATOR, |
| 33 | MergeRegion, |
| 34 | annotate_hunk_action, |
| 35 | classify_action, |
| 36 | compute_regions, |
| 37 | format_conflict_diff, |
| 38 | three_way_merge_lines, |
| 39 | ) |
| 40 | |
| 41 | |
| 42 | # ────────────────────────────────────────────────────────────────────────────── |
| 43 | # Helpers |
| 44 | # ────────────────────────────────────────────────────────────────────────────── |
| 45 | |
| 46 | def _lines(*text: str) -> list[str]: |
| 47 | """Split text into lines, each ending with \\n.""" |
| 48 | return [ln + "\n" for ln in "\n".join(text).split("\n") if ln + "\n"] |
| 49 | |
| 50 | |
| 51 | def _merge(base, ours, theirs, **kw): |
| 52 | return three_way_merge_lines( |
| 53 | _lines(*base) if isinstance(base, (list, tuple)) else base, |
| 54 | _lines(*ours) if isinstance(ours, (list, tuple)) else ours, |
| 55 | _lines(*theirs) if isinstance(theirs, (list, tuple)) else theirs, |
| 56 | **kw, |
| 57 | ) |
| 58 | |
| 59 | |
| 60 | def _has_marker(lines: list[str], marker: str) -> bool: |
| 61 | return any(marker in ln for ln in lines) |
| 62 | |
| 63 | |
| 64 | # ────────────────────────────────────────────────────────────────────────────── |
| 65 | # TestClassifyAction |
| 66 | # ────────────────────────────────────────────────────────────────────────────── |
| 67 | |
| 68 | class TestClassifyAction: |
| 69 | def test_inserted_when_base_empty(self) -> None: |
| 70 | assert classify_action([], ["new\n"]) == "inserted" |
| 71 | |
| 72 | def test_deleted_when_other_empty(self) -> None: |
| 73 | assert classify_action(["old\n"], []) == "deleted" |
| 74 | |
| 75 | def test_modified_when_both_non_empty(self) -> None: |
| 76 | assert classify_action(["old\n"], ["new\n"]) == "modified" |
| 77 | |
| 78 | def test_modified_when_multi_line_both(self) -> None: |
| 79 | assert classify_action(["a\n", "b\n"], ["c\n", "d\n"]) == "modified" |
| 80 | |
| 81 | def test_inserted_single_line(self) -> None: |
| 82 | assert classify_action([], ["x\n"]) == "inserted" |
| 83 | |
| 84 | def test_deleted_single_line(self) -> None: |
| 85 | assert classify_action(["x\n"], []) == "deleted" |
| 86 | |
| 87 | |
| 88 | # ────────────────────────────────────────────────────────────────────────────── |
| 89 | # TestAnnotateHunkAction |
| 90 | # ────────────────────────────────────────────────────────────────────────────── |
| 91 | |
| 92 | class TestAnnotateHunkAction: |
| 93 | def _make_hunk(self, adds: int = 0, dels: int = 0) -> list[str]: |
| 94 | lines = ["--- a/f.py", "+++ b/f.py", "@@ -1,3 +1,3 @@", " context\n"] |
| 95 | lines.extend([f"+added{i}\n" for i in range(adds)]) |
| 96 | lines.extend([f"-deleted{i}\n" for i in range(dels)]) |
| 97 | return lines |
| 98 | |
| 99 | def test_pure_insert_hunk_labelled_inserted(self) -> None: |
| 100 | hunk = self._make_hunk(adds=2, dels=0) |
| 101 | result = annotate_hunk_action(hunk, "ours") |
| 102 | at_line = next(ln for ln in result if ln.startswith("@@")) |
| 103 | assert "[ours: inserted]" in at_line |
| 104 | |
| 105 | def test_pure_delete_hunk_labelled_deleted(self) -> None: |
| 106 | hunk = self._make_hunk(adds=0, dels=2) |
| 107 | result = annotate_hunk_action(hunk, "theirs") |
| 108 | at_line = next(ln for ln in result if ln.startswith("@@")) |
| 109 | assert "[theirs: deleted]" in at_line |
| 110 | |
| 111 | def test_mixed_hunk_labelled_modified(self) -> None: |
| 112 | hunk = self._make_hunk(adds=1, dels=1) |
| 113 | result = annotate_hunk_action(hunk, "ours") |
| 114 | at_line = next(ln for ln in result if ln.startswith("@@")) |
| 115 | assert "[ours: modified]" in at_line |
| 116 | |
| 117 | def test_header_lines_preserved(self) -> None: |
| 118 | hunk = self._make_hunk(adds=1) |
| 119 | result = annotate_hunk_action(hunk, "ours") |
| 120 | assert any(ln.startswith("---") for ln in result) |
| 121 | assert any(ln.startswith("+++") for ln in result) |
| 122 | |
| 123 | def test_multiple_hunks_each_annotated(self) -> None: |
| 124 | hunk = [ |
| 125 | "--- a/f.py", "+++ b/f.py", |
| 126 | "@@ -1,2 +1,2 @@", " ctx\n", "+add1\n", |
| 127 | "@@ -10,2 +10,2 @@", " ctx2\n", "-del1\n", |
| 128 | ] |
| 129 | result = annotate_hunk_action(hunk, "ours") |
| 130 | at_lines = [ln for ln in result if ln.startswith("@@")] |
| 131 | assert len(at_lines) == 2 |
| 132 | assert "[ours: inserted]" in at_lines[0] |
| 133 | assert "[ours: deleted]" in at_lines[1] |
| 134 | |
| 135 | def test_empty_hunk_list_returns_empty(self) -> None: |
| 136 | assert annotate_hunk_action([], "ours") == [] |
| 137 | |
| 138 | def test_no_at_markers_unchanged(self) -> None: |
| 139 | lines = ["--- a/f.py", "+++ b/f.py", " context\n"] |
| 140 | result = annotate_hunk_action(lines, "ours") |
| 141 | assert result == lines |
| 142 | |
| 143 | def test_side_label_appears_in_annotation(self) -> None: |
| 144 | hunk = self._make_hunk(adds=1) |
| 145 | for label in ("main", "feature/auth", "theirs"): |
| 146 | result = annotate_hunk_action(hunk, label) |
| 147 | at_line = next(ln for ln in result if ln.startswith("@@")) |
| 148 | assert label in at_line |
| 149 | |
| 150 | |
| 151 | # ────────────────────────────────────────────────────────────────────────────── |
| 152 | # TestComputeRegions |
| 153 | # ────────────────────────────────────────────────────────────────────────────── |
| 154 | |
| 155 | class TestComputeRegions: |
| 156 | def test_all_stable_no_changes(self) -> None: |
| 157 | lines = ["a\n", "b\n", "c\n"] |
| 158 | regions = compute_regions(lines, lines, lines) |
| 159 | kinds = [r.kind for r in regions] |
| 160 | assert all(k == "stable" for k in kinds) |
| 161 | |
| 162 | def test_ours_only(self) -> None: |
| 163 | base = ["a\n", "b\n", "c\n"] |
| 164 | ours = ["a\n", "B\n", "c\n"] |
| 165 | regions = compute_regions(base, ours, base) |
| 166 | conflict_kinds = [r.kind for r in regions] |
| 167 | assert "ours_only" in conflict_kinds |
| 168 | assert "conflict" not in conflict_kinds |
| 169 | |
| 170 | def test_theirs_only(self) -> None: |
| 171 | base = ["a\n", "b\n", "c\n"] |
| 172 | theirs = ["a\n", "T\n", "c\n"] |
| 173 | regions = compute_regions(base, base, theirs) |
| 174 | kinds = [r.kind for r in regions] |
| 175 | assert "theirs_only" in kinds |
| 176 | assert "conflict" not in kinds |
| 177 | |
| 178 | def test_both_same(self) -> None: |
| 179 | base = ["a\n", "b\n"] |
| 180 | changed = ["a\n", "X\n"] |
| 181 | regions = compute_regions(base, changed, changed) |
| 182 | kinds = [r.kind for r in regions] |
| 183 | assert "both_same" in kinds |
| 184 | assert "conflict" not in kinds |
| 185 | |
| 186 | def test_conflict(self) -> None: |
| 187 | base = ["a\n", "b\n"] |
| 188 | ours = ["a\n", "O\n"] |
| 189 | theirs = ["a\n", "T\n"] |
| 190 | regions = compute_regions(base, ours, theirs) |
| 191 | kinds = [r.kind for r in regions] |
| 192 | assert "conflict" in kinds |
| 193 | |
| 194 | def test_empty_sequences(self) -> None: |
| 195 | regions = compute_regions([], [], []) |
| 196 | assert regions == [] |
| 197 | |
| 198 | def test_region_lines_are_lists(self) -> None: |
| 199 | base = ["a\n", "b\n"] |
| 200 | ours = ["a\n", "X\n"] |
| 201 | for r in compute_regions(base, ours, base): |
| 202 | assert isinstance(r.base_lines, list) |
| 203 | assert isinstance(r.ours_lines, list) |
| 204 | assert isinstance(r.theirs_lines, list) |
| 205 | |
| 206 | def test_stable_region_all_three_equal(self) -> None: |
| 207 | lines = ["x\n", "y\n"] |
| 208 | regions = compute_regions(lines, lines, lines) |
| 209 | for r in regions: |
| 210 | assert r.base_lines == r.ours_lines == r.theirs_lines |
| 211 | |
| 212 | def test_pure_insertion_ours(self) -> None: |
| 213 | base = ["a\n", "c\n"] |
| 214 | ours = ["a\n", "b\n", "c\n"] # inserted b |
| 215 | regions = compute_regions(base, ours, base) |
| 216 | kinds = [r.kind for r in regions] |
| 217 | assert "ours_only" in kinds or "stable" in kinds |
| 218 | assert "conflict" not in kinds |
| 219 | |
| 220 | def test_pure_insertion_conflict(self) -> None: |
| 221 | base = ["a\n", "c\n"] |
| 222 | ours = ["a\n", "B\n", "c\n"] |
| 223 | theirs = ["a\n", "T\n", "c\n"] |
| 224 | regions = compute_regions(base, ours, theirs) |
| 225 | kinds = [r.kind for r in regions] |
| 226 | assert "conflict" in kinds |
| 227 | |
| 228 | |
| 229 | # ────────────────────────────────────────────────────────────────────────────── |
| 230 | # TestThreeWayMergeLines — clean merges |
| 231 | # ────────────────────────────────────────────────────────────────────────────── |
| 232 | |
| 233 | class TestThreeWayMergeLines: |
| 234 | |
| 235 | # ── Clean cases ────────────────────────────────────────────────────────── |
| 236 | |
| 237 | def test_no_changes_returns_base(self) -> None: |
| 238 | base = ["a\n", "b\n", "c\n"] |
| 239 | merged, conflict = three_way_merge_lines(base, base, base) |
| 240 | assert merged == base |
| 241 | assert not conflict |
| 242 | |
| 243 | def test_ours_only_change_applied(self) -> None: |
| 244 | base = ["a\n", "b\n", "c\n"] |
| 245 | ours = ["a\n", "B\n", "c\n"] |
| 246 | merged, conflict = three_way_merge_lines(base, ours, base) |
| 247 | assert merged == ours |
| 248 | assert not conflict |
| 249 | |
| 250 | def test_theirs_only_change_applied(self) -> None: |
| 251 | base = ["a\n", "b\n", "c\n"] |
| 252 | theirs = ["a\n", "T\n", "c\n"] |
| 253 | merged, conflict = three_way_merge_lines(base, base, theirs) |
| 254 | assert merged == theirs |
| 255 | assert not conflict |
| 256 | |
| 257 | def test_both_same_change_applied_once(self) -> None: |
| 258 | base = ["a\n", "b\n"] |
| 259 | changed = ["a\n", "X\n"] |
| 260 | merged, conflict = three_way_merge_lines(base, changed, changed) |
| 261 | assert merged == changed |
| 262 | assert not conflict |
| 263 | |
| 264 | def test_non_overlapping_changes_both_applied(self) -> None: |
| 265 | base = ["a\n", "b\n", "c\n", "d\n"] |
| 266 | ours = ["a\n", "B\n", "c\n", "d\n"] # b→B |
| 267 | theirs = ["a\n", "b\n", "c\n", "D\n"] # d→D |
| 268 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 269 | assert "B\n" in merged |
| 270 | assert "D\n" in merged |
| 271 | assert not conflict |
| 272 | |
| 273 | def test_ours_inserts_line(self) -> None: |
| 274 | base = ["a\n", "c\n"] |
| 275 | ours = ["a\n", "b\n", "c\n"] |
| 276 | merged, conflict = three_way_merge_lines(base, ours, base) |
| 277 | assert merged == ours |
| 278 | assert not conflict |
| 279 | |
| 280 | def test_theirs_deletes_line(self) -> None: |
| 281 | base = ["a\n", "b\n", "c\n"] |
| 282 | theirs = ["a\n", "c\n"] |
| 283 | merged, conflict = three_way_merge_lines(base, base, theirs) |
| 284 | assert merged == theirs |
| 285 | assert not conflict |
| 286 | |
| 287 | def test_empty_base_both_add(self) -> None: |
| 288 | # Both add same content to empty base → clean |
| 289 | added = ["line1\n", "line2\n"] |
| 290 | merged, conflict = three_way_merge_lines([], added, added) |
| 291 | assert merged == added |
| 292 | assert not conflict |
| 293 | |
| 294 | # ── Conflict cases ─────────────────────────────────────────────────────── |
| 295 | |
| 296 | def test_conflict_detected(self) -> None: |
| 297 | base = ["a\n", "b\n"] |
| 298 | ours = ["a\n", "O\n"] |
| 299 | theirs = ["a\n", "T\n"] |
| 300 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 301 | assert conflict |
| 302 | |
| 303 | def test_conflict_has_ours_marker(self) -> None: |
| 304 | base = ["a\n"] |
| 305 | ours = ["O\n"] |
| 306 | theirs = ["T\n"] |
| 307 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 308 | assert _has_marker(merged, "<<<<<<<") |
| 309 | |
| 310 | def test_conflict_has_base_marker(self) -> None: |
| 311 | base = ["a\n"] |
| 312 | ours = ["O\n"] |
| 313 | theirs = ["T\n"] |
| 314 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 315 | assert _has_marker(merged, "|||||||") |
| 316 | |
| 317 | def test_conflict_has_sep_marker(self) -> None: |
| 318 | base = ["a\n"] |
| 319 | ours = ["O\n"] |
| 320 | theirs = ["T\n"] |
| 321 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 322 | assert _has_marker(merged, "=======") |
| 323 | |
| 324 | def test_conflict_has_end_marker(self) -> None: |
| 325 | base = ["a\n"] |
| 326 | ours = ["O\n"] |
| 327 | theirs = ["T\n"] |
| 328 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 329 | assert _has_marker(merged, ">>>>>>> end conflict") |
| 330 | |
| 331 | def test_conflict_ours_content_present(self) -> None: |
| 332 | base = ["x\n"] |
| 333 | ours = ["ours_line\n"] |
| 334 | theirs = ["theirs_line\n"] |
| 335 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 336 | assert any("ours_line" in ln for ln in merged) |
| 337 | |
| 338 | def test_conflict_theirs_content_present(self) -> None: |
| 339 | base = ["x\n"] |
| 340 | ours = ["ours_line\n"] |
| 341 | theirs = ["theirs_line\n"] |
| 342 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 343 | assert any("theirs_line" in ln for ln in merged) |
| 344 | |
| 345 | def test_conflict_base_content_in_diff3_section(self) -> None: |
| 346 | base = ["base_line\n"] |
| 347 | ours = ["ours_line\n"] |
| 348 | theirs = ["theirs_line\n"] |
| 349 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 350 | assert any("base_line" in ln for ln in merged) |
| 351 | |
| 352 | def test_stable_lines_outside_conflict_preserved(self) -> None: |
| 353 | base = ["preamble\n", "conflict_zone\n", "epilogue\n"] |
| 354 | ours = ["preamble\n", "ours_content\n", "epilogue\n"] |
| 355 | theirs = ["preamble\n", "theirs_content\n", "epilogue\n"] |
| 356 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 357 | assert conflict |
| 358 | assert any("preamble" in ln for ln in merged) |
| 359 | assert any("epilogue" in ln for ln in merged) |
| 360 | |
| 361 | def test_multiple_independent_conflicts(self) -> None: |
| 362 | base = ["a\n", "b\n", "c\n", "d\n", "e\n"] |
| 363 | ours = ["a\n", "O1\n", "c\n", "O2\n", "e\n"] |
| 364 | theirs = ["a\n", "T1\n", "c\n", "T2\n", "e\n"] |
| 365 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 366 | assert conflict |
| 367 | assert sum(1 for ln in merged if ln.startswith("<<<<<<< ")) == 2 |
| 368 | |
| 369 | def test_both_empty_insertion_at_same_point_conflicts(self) -> None: |
| 370 | base = ["a\n", "c\n"] |
| 371 | ours = ["a\n", "B\n", "c\n"] |
| 372 | theirs = ["a\n", "T\n", "c\n"] |
| 373 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 374 | assert conflict |
| 375 | |
| 376 | def test_custom_labels_appear_in_markers(self) -> None: |
| 377 | base = ["x\n"] |
| 378 | ours = ["O\n"] |
| 379 | theirs = ["T\n"] |
| 380 | merged, _ = three_way_merge_lines( |
| 381 | base, ours, theirs, |
| 382 | label_ours="feature/login", |
| 383 | label_base="merge-base", |
| 384 | label_theirs="main", |
| 385 | ) |
| 386 | assert any("feature/login" in ln for ln in merged) |
| 387 | assert any("main" in ln for ln in merged) |
| 388 | assert any("merge-base" in ln for ln in merged) |
| 389 | |
| 390 | |
| 391 | # ────────────────────────────────────────────────────────────────────────────── |
| 392 | # TestConflictMarkerFormat — exact marker syntax / Cohen action labels |
| 393 | # ────────────────────────────────────────────────────────────────────────────── |
| 394 | |
| 395 | class TestConflictMarkerFormat: |
| 396 | |
| 397 | def _conflict_merged(self, base_text: str, ours_text: str, theirs_text: str) -> list[str]: |
| 398 | base = base_text.splitlines(keepends=True) |
| 399 | ours = ours_text.splitlines(keepends=True) |
| 400 | theirs = theirs_text.splitlines(keepends=True) |
| 401 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 402 | return merged |
| 403 | |
| 404 | def test_deleted_action_on_ours_marker(self) -> None: |
| 405 | merged = self._conflict_merged("func()\n", "", "other()\n") |
| 406 | assert any("<<<<<<< ours [deleted]" in ln for ln in merged) |
| 407 | |
| 408 | def test_inserted_action_on_theirs_marker(self) -> None: |
| 409 | merged = self._conflict_merged("", "ours_line\n", "theirs_line\n") |
| 410 | assert any("[inserted]" in ln for ln in merged) |
| 411 | |
| 412 | def test_modified_action_on_modified_conflict(self) -> None: |
| 413 | merged = self._conflict_merged("original\n", "ours_ver\n", "theirs_ver\n") |
| 414 | assert any("[modified]" in ln for ln in merged) |
| 415 | |
| 416 | def test_base_section_marker_present(self) -> None: |
| 417 | merged = self._conflict_merged("base\n", "ours\n", "theirs\n") |
| 418 | assert any("|||||||" in ln for ln in merged) |
| 419 | |
| 420 | def test_end_conflict_marker_present(self) -> None: |
| 421 | merged = self._conflict_merged("base\n", "ours\n", "theirs\n") |
| 422 | assert any(">>>>>>> end conflict" in ln for ln in merged) |
| 423 | |
| 424 | def test_marker_ordering(self) -> None: |
| 425 | """<<<<<<< must precede ||||||| must precede ======= must precede >>>>>>>.""" |
| 426 | merged = self._conflict_merged("base\n", "ours\n", "theirs\n") |
| 427 | positions = {} |
| 428 | for i, ln in enumerate(merged): |
| 429 | if "<<<<<<<" in ln: |
| 430 | positions["start"] = i |
| 431 | elif "|||||||" in ln: |
| 432 | positions["base"] = i |
| 433 | elif "=======" in ln: |
| 434 | positions["sep"] = i |
| 435 | elif ">>>>>>>" in ln: |
| 436 | positions["end"] = i |
| 437 | assert positions["start"] < positions["base"] < positions["sep"] < positions["end"] |
| 438 | |
| 439 | |
| 440 | # ────────────────────────────────────────────────────────────────────────────── |
| 441 | # TestFormatConflictDiff |
| 442 | # ────────────────────────────────────────────────────────────────────────────── |
| 443 | |
| 444 | class TestFormatConflictDiff: |
| 445 | """Tests for format_conflict_diff() — the muse diff --conflict renderer.""" |
| 446 | |
| 447 | def _make_manifests(self) -> tuple[dict, dict, dict, MagicMock]: |
| 448 | base_content = b"base line\n" |
| 449 | ours_content = b"ours line\n" |
| 450 | theirs_content = b"theirs line\n" |
| 451 | path = "src/util.py" |
| 452 | |
| 453 | base_oid = "aaaa" * 16 |
| 454 | ours_oid = "bbbb" * 16 |
| 455 | theirs_oid = "cccc" * 16 |
| 456 | |
| 457 | base_manifest = {path: base_oid} |
| 458 | ours_manifest = {path: ours_oid} |
| 459 | theirs_manifest = {path: theirs_oid} |
| 460 | |
| 461 | def _read(root, oid): |
| 462 | return {base_oid: base_content, ours_oid: ours_content, theirs_oid: theirs_content}.get(oid) |
| 463 | |
| 464 | return base_manifest, ours_manifest, theirs_manifest, _read |
| 465 | |
| 466 | def test_output_contains_conflict_header(self, tmp_path: pathlib.Path) -> None: |
| 467 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 468 | lines = format_conflict_diff( |
| 469 | "src/util.py", tmp_path, base_m, ours_m, theirs_m, read_fn, |
| 470 | ) |
| 471 | assert any("CONFLICT" in ln for ln in lines) |
| 472 | assert any("src/util.py" in ln for ln in lines) |
| 473 | |
| 474 | def test_output_contains_ours_section(self, tmp_path: pathlib.Path) -> None: |
| 475 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 476 | lines = format_conflict_diff( |
| 477 | "src/util.py", tmp_path, base_m, ours_m, theirs_m, read_fn, |
| 478 | ) |
| 479 | assert any("[ours]" in ln for ln in lines) |
| 480 | |
| 481 | def test_output_contains_theirs_section(self, tmp_path: pathlib.Path) -> None: |
| 482 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 483 | lines = format_conflict_diff( |
| 484 | "src/util.py", tmp_path, base_m, ours_m, theirs_m, read_fn, |
| 485 | ) |
| 486 | assert any("[theirs]" in ln for ln in lines) |
| 487 | |
| 488 | def test_custom_labels_appear(self, tmp_path: pathlib.Path) -> None: |
| 489 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 490 | lines = format_conflict_diff( |
| 491 | "src/util.py", tmp_path, base_m, ours_m, theirs_m, read_fn, |
| 492 | ours_label="feature/auth", |
| 493 | theirs_label="main", |
| 494 | ) |
| 495 | assert any("feature/auth" in ln for ln in lines) |
| 496 | assert any("main" in ln for ln in lines) |
| 497 | |
| 498 | def test_returns_list_of_strings(self, tmp_path: pathlib.Path) -> None: |
| 499 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 500 | result = format_conflict_diff( |
| 501 | "src/util.py", tmp_path, base_m, ours_m, theirs_m, read_fn, |
| 502 | ) |
| 503 | assert isinstance(result, list) |
| 504 | assert all(isinstance(ln, str) for ln in result) |
| 505 | |
| 506 | def test_ansi_sanitized_in_path(self, tmp_path: pathlib.Path) -> None: |
| 507 | """ANSI escape sequences in the path must not appear in output.""" |
| 508 | evil_path = "\x1b[31mevil\x1b[0m/file.py" |
| 509 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 510 | # Use the evil path (will have no manifest entry → empty diffs) |
| 511 | lines = format_conflict_diff( |
| 512 | evil_path, tmp_path, {}, {}, {}, read_fn, |
| 513 | ) |
| 514 | output = "\n".join(lines) |
| 515 | assert "\x1b[31m" not in output |
| 516 | |
| 517 | def test_missing_file_shows_no_changes_message(self, tmp_path: pathlib.Path) -> None: |
| 518 | def _read(root, oid): |
| 519 | return None |
| 520 | |
| 521 | lines = format_conflict_diff( |
| 522 | "missing.py", tmp_path, {}, {}, {}, _read, |
| 523 | ) |
| 524 | assert any("no changes" in ln.lower() for ln in lines) |
| 525 | |
| 526 | def test_no_color_mode(self, tmp_path: pathlib.Path) -> None: |
| 527 | base_m, ours_m, theirs_m, read_fn = self._make_manifests() |
| 528 | lines = format_conflict_diff( |
| 529 | "src/util.py", tmp_path, base_m, ours_m, theirs_m, read_fn, |
| 530 | use_color=False, |
| 531 | ) |
| 532 | output = "\n".join(lines) |
| 533 | assert "\x1b[" not in output |
| 534 | |
| 535 | |
| 536 | # ────────────────────────────────────────────────────────────────────────────── |
| 537 | # TestEdgeCases |
| 538 | # ────────────────────────────────────────────────────────────────────────────── |
| 539 | |
| 540 | class TestEdgeCases: |
| 541 | def test_all_empty(self) -> None: |
| 542 | merged, conflict = three_way_merge_lines([], [], []) |
| 543 | assert merged == [] |
| 544 | assert not conflict |
| 545 | |
| 546 | def test_base_empty_ours_adds(self) -> None: |
| 547 | merged, conflict = three_way_merge_lines([], ["new\n"], []) |
| 548 | assert merged == ["new\n"] |
| 549 | assert not conflict |
| 550 | |
| 551 | def test_base_empty_theirs_adds(self) -> None: |
| 552 | merged, conflict = three_way_merge_lines([], [], ["new\n"]) |
| 553 | assert merged == ["new\n"] |
| 554 | assert not conflict |
| 555 | |
| 556 | def test_base_empty_both_add_same(self) -> None: |
| 557 | merged, conflict = three_way_merge_lines([], ["same\n"], ["same\n"]) |
| 558 | assert merged == ["same\n"] |
| 559 | assert not conflict |
| 560 | |
| 561 | def test_base_empty_both_add_different_conflicts(self) -> None: |
| 562 | _, conflict = three_way_merge_lines([], ["A\n"], ["B\n"]) |
| 563 | assert conflict |
| 564 | |
| 565 | def test_ours_equals_base_theirs_deletes(self) -> None: |
| 566 | base = ["x\n"] |
| 567 | merged, conflict = three_way_merge_lines(base, base, []) |
| 568 | assert merged == [] |
| 569 | assert not conflict |
| 570 | |
| 571 | def test_both_delete_everything(self) -> None: |
| 572 | base = ["x\n", "y\n"] |
| 573 | merged, conflict = three_way_merge_lines(base, [], []) |
| 574 | assert merged == [] |
| 575 | assert not conflict |
| 576 | |
| 577 | def test_unicode_content(self) -> None: |
| 578 | base = ["héllo\n", "wörld\n"] |
| 579 | ours = ["héllo\n", "WÖRLD\n"] |
| 580 | theirs = ["héllo\n", "wörld\n"] |
| 581 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 582 | assert not conflict |
| 583 | assert "WÖRLD\n" in merged |
| 584 | |
| 585 | def test_long_lines_no_crash(self) -> None: |
| 586 | long_line = "x" * 10_000 + "\n" |
| 587 | base = [long_line] |
| 588 | ours = ["y" * 10_000 + "\n"] |
| 589 | theirs = [long_line] |
| 590 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 591 | assert not conflict |
| 592 | |
| 593 | def test_many_identical_lines(self) -> None: |
| 594 | base = ["same\n"] * 1000 |
| 595 | merged, conflict = three_way_merge_lines(base, base, base) |
| 596 | assert merged == base |
| 597 | assert not conflict |
| 598 | |
| 599 | def test_lines_without_trailing_newline(self) -> None: |
| 600 | # Not all editors guarantee trailing newlines. |
| 601 | base = ["no newline"] |
| 602 | ours = ["changed"] |
| 603 | merged, conflict = three_way_merge_lines(base, ours, base) |
| 604 | assert not conflict |
| 605 | assert "changed" in merged[0] |
| 606 | |
| 607 | |
| 608 | # ────────────────────────────────────────────────────────────────────────────── |
| 609 | # TestIntegration — realistic multi-hunk merges |
| 610 | # ────────────────────────────────────────────────────────────────────────────── |
| 611 | |
| 612 | class TestIntegration: |
| 613 | """Realistic multi-hunk merge scenarios matching real developer workflows.""" |
| 614 | |
| 615 | def test_ours_adds_docstring_theirs_renames_param(self) -> None: |
| 616 | base = [ |
| 617 | "def calculate(x):\n", |
| 618 | " return x * 2\n", |
| 619 | ] |
| 620 | ours = [ |
| 621 | "def calculate(x):\n", |
| 622 | ' """Double the input."""\n', |
| 623 | " return x * 2\n", |
| 624 | ] |
| 625 | theirs = [ |
| 626 | "def calculate(value):\n", |
| 627 | " return value * 2\n", |
| 628 | ] |
| 629 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 630 | # ours inserted a docstring; theirs renamed x→value. |
| 631 | # They touch different parts of the function — should be conflict-free |
| 632 | # if ours and theirs don't overlap on line 1. |
| 633 | # (They DO overlap on line 1 if difflib groups them — conflict is acceptable) |
| 634 | # Just assert no crash and well-formed output. |
| 635 | assert isinstance(merged, list) |
| 636 | assert isinstance(conflict, bool) |
| 637 | |
| 638 | def test_both_fix_same_typo(self) -> None: |
| 639 | base = ["# Calcualtion module\n"] |
| 640 | ours = ["# Calculation module\n"] |
| 641 | theirs = ["# Calculation module\n"] |
| 642 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 643 | assert not conflict |
| 644 | assert merged == ours |
| 645 | |
| 646 | def test_ours_deletes_block_theirs_modifies_different_block(self) -> None: |
| 647 | base = [ |
| 648 | "block_a_line_1\n", |
| 649 | "block_a_line_2\n", |
| 650 | "separator\n", |
| 651 | "block_b_line_1\n", |
| 652 | "block_b_line_2\n", |
| 653 | ] |
| 654 | ours = [ |
| 655 | "separator\n", |
| 656 | "block_b_line_1\n", |
| 657 | "block_b_line_2\n", |
| 658 | ] |
| 659 | theirs = [ |
| 660 | "block_a_line_1\n", |
| 661 | "block_a_line_2\n", |
| 662 | "separator\n", |
| 663 | "block_b_LINE_1\n", # modified |
| 664 | "block_b_line_2\n", |
| 665 | ] |
| 666 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 667 | # No overlap between ours' deletion (top) and theirs' modification (bottom) |
| 668 | assert not conflict |
| 669 | assert "block_b_LINE_1\n" in merged |
| 670 | assert "block_a_line_1\n" not in merged |
| 671 | |
| 672 | def test_conflict_preserves_stable_context_above_and_below(self) -> None: |
| 673 | base = ["header\n", "conflict_zone\n", "footer\n"] |
| 674 | ours = ["header\n", "ours_version\n", "footer\n"] |
| 675 | theirs = ["header\n", "theirs_version\n", "footer\n"] |
| 676 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 677 | assert conflict |
| 678 | text = "".join(merged) |
| 679 | assert "header\n" in text |
| 680 | assert "footer\n" in text |
| 681 | |
| 682 | def test_append_only_change_ours(self) -> None: |
| 683 | base = ["line1\n", "line2\n"] |
| 684 | ours = ["line1\n", "line2\n", "appended\n"] |
| 685 | merged, conflict = three_way_merge_lines(base, ours, base) |
| 686 | assert not conflict |
| 687 | assert merged == ours |
| 688 | |
| 689 | def test_prepend_only_change_theirs(self) -> None: |
| 690 | base = ["line1\n"] |
| 691 | theirs = ["prepended\n", "line1\n"] |
| 692 | merged, conflict = three_way_merge_lines(base, base, theirs) |
| 693 | assert not conflict |
| 694 | assert merged == theirs |
| 695 | |
| 696 | def test_roundtrip_clean_merge_deterministic(self) -> None: |
| 697 | base = ["a\n", "b\n", "c\n"] |
| 698 | ours = ["a\n", "B\n", "c\n"] |
| 699 | theirs = ["a\n", "b\n", "C\n"] |
| 700 | m1, c1 = three_way_merge_lines(base, ours, theirs) |
| 701 | m2, c2 = three_way_merge_lines(base, ours, theirs) |
| 702 | assert m1 == m2 |
| 703 | assert c1 == c2 |
| 704 | |
| 705 | |
| 706 | # ────────────────────────────────────────────────────────────────────────────── |
| 707 | # TestStress |
| 708 | # ────────────────────────────────────────────────────────────────────────────── |
| 709 | |
| 710 | class TestStress: |
| 711 | def test_large_file_clean_merge_fast(self) -> None: |
| 712 | """500-line file with non-overlapping changes in each half: < 2 s.""" |
| 713 | base = [f"line_{i:04d}\n" for i in range(500)] |
| 714 | ours = list(base) |
| 715 | ours[50] = "OURS_CHANGE\n" |
| 716 | theirs = list(base) |
| 717 | theirs[450] = "THEIRS_CHANGE\n" |
| 718 | |
| 719 | start = time.monotonic() |
| 720 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 721 | elapsed = time.monotonic() - start |
| 722 | |
| 723 | assert not conflict |
| 724 | assert elapsed < 2.0, f"merge took {elapsed:.2f}s" |
| 725 | assert "OURS_CHANGE\n" in merged |
| 726 | assert "THEIRS_CHANGE\n" in merged |
| 727 | |
| 728 | def test_many_conflicts_no_crash(self) -> None: |
| 729 | """Alternating conflict zones throughout a 200-line file.""" |
| 730 | base = [f"line_{i}\n" for i in range(200)] |
| 731 | ours = list(base) |
| 732 | theirs = list(base) |
| 733 | # Every 10th line conflicts. |
| 734 | for i in range(0, 200, 10): |
| 735 | ours[i] = f"ours_{i}\n" |
| 736 | theirs[i] = f"theirs_{i}\n" |
| 737 | |
| 738 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 739 | assert conflict |
| 740 | assert isinstance(merged, list) |
| 741 | |
| 742 | def test_concurrent_merges_consistent(self) -> None: |
| 743 | """Concurrent invocations must return identical results (no shared mutable state).""" |
| 744 | base = ["a\n", "conflict\n", "b\n"] |
| 745 | ours = ["a\n", "ours\n", "b\n"] |
| 746 | theirs = ["a\n", "theirs\n", "b\n"] |
| 747 | results: list[tuple[list[str], bool]] = [] |
| 748 | errors: list[Exception] = [] |
| 749 | |
| 750 | def _run(): |
| 751 | try: |
| 752 | results.append(three_way_merge_lines(base, ours, theirs)) |
| 753 | except Exception as exc: |
| 754 | errors.append(exc) |
| 755 | |
| 756 | threads = [threading.Thread(target=_run) for _ in range(10)] |
| 757 | for t in threads: |
| 758 | t.start() |
| 759 | for t in threads: |
| 760 | t.join() |
| 761 | |
| 762 | assert not errors |
| 763 | assert len(results) == 10 |
| 764 | assert all(r == results[0] for r in results) |
| 765 | |
| 766 | |
| 767 | # ────────────────────────────────────────────────────────────────────────────── |
| 768 | # TestDataIntegrity |
| 769 | # ────────────────────────────────────────────────────────────────────────────── |
| 770 | |
| 771 | class TestDataIntegrity: |
| 772 | def test_clean_merge_is_deterministic(self) -> None: |
| 773 | base = ["a\n", "b\n", "c\n"] |
| 774 | ours = ["a\n", "B\n", "c\n"] |
| 775 | theirs = ["a\n", "b\n", "C\n"] |
| 776 | r1, c1 = three_way_merge_lines(base, ours, theirs) |
| 777 | r2, c2 = three_way_merge_lines(base, ours, theirs) |
| 778 | assert r1 == r2 |
| 779 | assert c1 == c2 |
| 780 | |
| 781 | def test_conflict_output_is_deterministic(self) -> None: |
| 782 | base = ["x\n"] |
| 783 | ours = ["O\n"] |
| 784 | theirs = ["T\n"] |
| 785 | r1, _ = three_way_merge_lines(base, ours, theirs) |
| 786 | r2, _ = three_way_merge_lines(base, ours, theirs) |
| 787 | assert r1 == r2 |
| 788 | |
| 789 | def test_input_sequences_not_mutated(self) -> None: |
| 790 | base = ["a\n", "b\n"] |
| 791 | ours = ["a\n", "O\n"] |
| 792 | theirs = ["a\n", "T\n"] |
| 793 | base_copy = list(base) |
| 794 | ours_copy = list(ours) |
| 795 | theirs_copy = list(theirs) |
| 796 | three_way_merge_lines(base, ours, theirs) |
| 797 | assert base == base_copy |
| 798 | assert ours == ours_copy |
| 799 | assert theirs == theirs_copy |
| 800 | |
| 801 | def test_merged_content_contains_all_clean_changes(self) -> None: |
| 802 | # ours changes b→B (pos 1); theirs changes d→D (pos 3). |
| 803 | # 'a' (pos 0) and 'c' (pos 2) are stable in both LCS runs, |
| 804 | # so the two changes land in separate non-overlapping regions → clean. |
| 805 | base = ["a\n", "b\n", "c\n", "d\n"] |
| 806 | ours = ["a\n", "B\n", "c\n", "d\n"] |
| 807 | theirs = ["a\n", "b\n", "c\n", "D\n"] |
| 808 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 809 | assert not conflict |
| 810 | merged_str = "".join(merged) |
| 811 | assert "B\n" in merged_str |
| 812 | assert "D\n" in merged_str |
| 813 | assert "a\n" in merged_str |
| 814 | assert "c\n" in merged_str |
| 815 | |
| 816 | def test_conflict_markers_are_balanced(self) -> None: |
| 817 | """Every <<<<<<< must have a matching >>>>>>>.""" |
| 818 | base = ["x\n", "y\n"] |
| 819 | ours = ["O1\n", "O2\n"] |
| 820 | theirs = ["T1\n", "T2\n"] |
| 821 | merged, _ = three_way_merge_lines(base, ours, theirs) |
| 822 | opens = sum(1 for ln in merged if ln.startswith("<<<<<<<")) |
| 823 | closes = sum(1 for ln in merged if ln.startswith(">>>>>>>")) |
| 824 | assert opens == closes |
| 825 | assert opens >= 1 |
| 826 | |
| 827 | |
| 828 | # ────────────────────────────────────────────────────────────────────────────── |
| 829 | # TestSecurity |
| 830 | # ────────────────────────────────────────────────────────────────────────────── |
| 831 | |
| 832 | class TestSecurity: |
| 833 | def test_ansi_in_content_does_not_spoof_markers(self) -> None: |
| 834 | """ANSI sequences in file content must not produce fake conflict markers.""" |
| 835 | base = ["normal\n"] |
| 836 | ours = ["\x1b[31m<<<<<<< fake ours\x1b[0m\n"] |
| 837 | theirs = ["other\n"] |
| 838 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 839 | # The ANSI-containing line is just content — the conflict marker check |
| 840 | # should find real markers (7 < chars), not ANSI-faked ones. |
| 841 | real_opens = sum(1 for ln in merged if ln.startswith("<<<<<<<")) |
| 842 | # There may be a real conflict (ours ≠ theirs), but the ANSI line itself |
| 843 | # should appear as content, not as an extra conflict opener. |
| 844 | assert isinstance(merged, list) |
| 845 | assert isinstance(conflict, bool) |
| 846 | |
| 847 | def test_null_bytes_in_content_handled(self) -> None: |
| 848 | """Null bytes in text content must not crash the merge.""" |
| 849 | base = ["a\x00b\n"] |
| 850 | ours = ["a\x00B\n"] |
| 851 | theirs = ["a\x00b\n"] |
| 852 | merged, conflict = three_way_merge_lines(base, ours, theirs) |
| 853 | assert not conflict # only ours changed |
| 854 | |
| 855 | def test_very_long_label_no_overflow(self) -> None: |
| 856 | long_label = "x" * 2000 |
| 857 | base = ["a\n"] |
| 858 | ours = ["O\n"] |
| 859 | theirs = ["T\n"] |
| 860 | merged, conflict = three_way_merge_lines( |
| 861 | base, ours, theirs, label_ours=long_label, label_theirs=long_label |
| 862 | ) |
| 863 | assert conflict |
| 864 | assert any(long_label in ln for ln in merged) |
| 865 | |
| 866 | def test_label_injection_via_newline(self) -> None: |
| 867 | """Newline in a label must not inject extra lines into markers.""" |
| 868 | evil = "branch\n>>>>>>> evil injection" |
| 869 | base = ["a\n"] |
| 870 | ours = ["O\n"] |
| 871 | theirs = ["T\n"] |
| 872 | merged, _ = three_way_merge_lines(base, ours, theirs, label_ours=evil) |
| 873 | # The evil label appears in the <<<<<<< line; verify no phantom >>>>>>> before the real one |
| 874 | positions_of_end = [i for i, ln in enumerate(merged) if ">>>>>>> end conflict" in ln] |
| 875 | positions_of_open = [i for i, ln in enumerate(merged) if ln.startswith("<<<<<<<")] |
| 876 | assert len(positions_of_end) >= 1 |
| 877 | assert len(positions_of_open) >= 1 |
| 878 | |
| 879 | def test_format_conflict_diff_ansi_in_path_sanitized( |
| 880 | self, tmp_path: pathlib.Path |
| 881 | ) -> None: |
| 882 | """ANSI codes in the file path argument must not appear in rendered output.""" |
| 883 | evil_path = "\x1b[31minjected\x1b[0m/file.py" |
| 884 | lines = format_conflict_diff( |
| 885 | evil_path, tmp_path, {}, {}, {}, |
| 886 | lambda root, oid: None, |
| 887 | ) |
| 888 | output = "\n".join(lines) |
| 889 | assert "\x1b[31m" not in output |
File History
4 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
48 days ago