test_knowtation_rerere.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Tests for ``muse/plugins/knowtation/rerere.py`` — Phase 2.5. |
| 2 | |
| 3 | Covers all seven Rule #0 tiers for the knowtation domain rerere plugin: |
| 4 | |
| 5 | 1. **Unit — fingerprint distinguishability** |
| 6 | Ten conflict pairs that *must* produce different fingerprints (different |
| 7 | conflicting content, different conflicting sections, different conflict |
| 8 | counts, etc.) and five pairs that *must* produce identical fingerprints |
| 9 | (cosmetic whitespace, frontmatter tag additions, frontmatter key |
| 10 | reorderings, non-conflicting section reorderings, CRLF / LF differences). |
| 11 | |
| 12 | 2. **Unit — record / lookup round-trip** |
| 13 | Hash-only resolution storage round-trips through ``record_resolution`` |
| 14 | → ``lookup_resolution``; missing fingerprints surface as ``None``; |
| 15 | distinct fingerprints do not collide on disk. |
| 16 | |
| 17 | 3. **Unit — record_full_resolution / replay_full_resolution** |
| 18 | Bytes-preserving storage round-trips; unknown fingerprint returns |
| 19 | ``None``; canonical form is what's stored on disk. |
| 20 | |
| 21 | 4. **Integration — replay after cosmetic rewrite** |
| 22 | Record a full resolution for conflict A, rewrite the surrounding note |
| 23 | cosmetically (add a tag, reorder non-conflicting sections), confirm |
| 24 | the fingerprint still matches and ``replay_full_resolution`` succeeds. |
| 25 | |
| 26 | 5. **Integration — rerere CLI hook (compute_fingerprint dispatch)** |
| 27 | The ``KnowtationPlugin`` registered in ``muse.plugins.registry`` is |
| 28 | detected as a ``RererePlugin`` by ``muse.core.rerere.compute_fingerprint`` |
| 29 | and the resulting fingerprint matches the bytes-level fingerprint, so |
| 30 | ``muse rerere`` will index its rr-cache by knowtation-aware |
| 31 | fingerprints. |
| 32 | |
| 33 | 6. **Data-integrity — replay across 50 historical conflict stubs** |
| 34 | Synthesise 50 distinct conflict pairs. Record full resolutions for |
| 35 | the first 25 — the lookup must succeed for those 25 and miss for the |
| 36 | other 25. No collisions across the 50 fingerprints. |
| 37 | |
| 38 | 7. **Security** |
| 39 | * Fingerprint cannot be forged by injecting ``"|"`` or ``":"`` |
| 40 | separators into a section title or body. |
| 41 | * Path-traversal fingerprints (``"/"`` or ``".."``) raise ``ValueError`` |
| 42 | before any filesystem access. |
| 43 | * 1 MiB-per-section conflicting input does not crash and produces a |
| 44 | valid fingerprint within a sane time budget. |
| 45 | * NUL bytes embedded in section bodies survive end-to-end. |
| 46 | * Oversized inputs (> 16 MiB) raise ``ValueError`` before parsing. |
| 47 | |
| 48 | Conventions |
| 49 | ----------- |
| 50 | * Helpers build canonical-form note bytes via the differ's |
| 51 | :func:`canonicalize` so byte-stable assertions are reliable. |
| 52 | * No real Muse repository is required for the unit tests; integration |
| 53 | tests construct a minimal repo via ``muse init --domain knowtation`` |
| 54 | through ``CliRunner``. |
| 55 | """ |
| 56 | from __future__ import annotations |
| 57 | |
| 58 | import hashlib |
| 59 | import json |
| 60 | import pathlib |
| 61 | import time |
| 62 | from typing import Any |
| 63 | |
| 64 | import pytest |
| 65 | import yaml |
| 66 | |
| 67 | from muse.plugins.knowtation.differ import canonicalize |
| 68 | from muse.plugins.knowtation.rerere import ( |
| 69 | DOMAIN, |
| 70 | KnowtationRererePlugin, |
| 71 | _FINGERPRINT_RE, |
| 72 | _MAX_NOTE_BYTES, |
| 73 | conflict_fingerprint, |
| 74 | lookup_resolution, |
| 75 | plugin as knowtation_rerere_plugin, |
| 76 | record_full_resolution, |
| 77 | record_resolution, |
| 78 | replay_full_resolution, |
| 79 | replay_resolution, |
| 80 | ) |
| 81 | |
| 82 | |
| 83 | # ────────────────────────────────────────────────────────────────────────────── |
| 84 | # Fixtures and helpers |
| 85 | # ────────────────────────────────────────────────────────────────────────────── |
| 86 | |
| 87 | |
| 88 | def _note(frontmatter: dict[str, Any] | None, body: str) -> bytes: |
| 89 | """Build canonical-form note bytes for a frontmatter mapping + body. |
| 90 | |
| 91 | The result is what :func:`canonicalize` produces, so two notes that |
| 92 | differ only in YAML key order or list quoting compare equal at the |
| 93 | byte level after this helper. |
| 94 | """ |
| 95 | if not frontmatter: |
| 96 | return canonicalize(body.encode("utf-8")) |
| 97 | yaml_text = yaml.safe_dump( |
| 98 | frontmatter, |
| 99 | sort_keys=False, |
| 100 | default_flow_style=False, |
| 101 | allow_unicode=True, |
| 102 | width=10000, |
| 103 | ) |
| 104 | return canonicalize(f"---\n{yaml_text}---\n{body}".encode("utf-8")) |
| 105 | |
| 106 | |
| 107 | def _h64(b: bytes) -> str: |
| 108 | return hashlib.sha256(b).hexdigest() |
| 109 | |
| 110 | |
| 111 | @pytest.fixture() |
| 112 | def tmp_root(tmp_path: pathlib.Path) -> pathlib.Path: |
| 113 | """Return a tmp directory acting as a Muse repo root. |
| 114 | |
| 115 | The unit and security tiers do not need a fully-initialised repo — |
| 116 | only ``.muse/`` writes are exercised, and the rerere code creates |
| 117 | that directory on demand. |
| 118 | """ |
| 119 | return tmp_path |
| 120 | |
| 121 | |
| 122 | # ────────────────────────────────────────────────────────────────────────────── |
| 123 | # Tier 1 — Unit: fingerprint collision / distinguishability table |
| 124 | # ────────────────────────────────────────────────────────────────────────────── |
| 125 | |
| 126 | |
| 127 | class TestFingerprintDistinguishability: |
| 128 | """Ten pairs that must DIFFER, five pairs that must MATCH.""" |
| 129 | |
| 130 | # -- Pairs that MUST differ ------------------------------------------------ |
| 131 | |
| 132 | def test_different_ours_body_changes_fingerprint(self) -> None: |
| 133 | ours_a = _note({"title": "T"}, "# §Summary\n\nours-A\n") |
| 134 | ours_b = _note({"title": "T"}, "# §Summary\n\nours-B\n") |
| 135 | theirs = _note({"title": "T"}, "# §Summary\n\ntheirs\n") |
| 136 | base = _note({"title": "T"}, "# §Summary\n\nbase\n") |
| 137 | assert ( |
| 138 | conflict_fingerprint(ours_a, theirs, base) |
| 139 | != conflict_fingerprint(ours_b, theirs, base) |
| 140 | ) |
| 141 | |
| 142 | def test_different_theirs_body_changes_fingerprint(self) -> None: |
| 143 | ours = _note(None, "# H\n\nours\n") |
| 144 | theirs_a = _note(None, "# H\n\ntheirs-A\n") |
| 145 | theirs_b = _note(None, "# H\n\ntheirs-B\n") |
| 146 | base = _note(None, "# H\n\nbase\n") |
| 147 | assert ( |
| 148 | conflict_fingerprint(ours, theirs_a, base) |
| 149 | != conflict_fingerprint(ours, theirs_b, base) |
| 150 | ) |
| 151 | |
| 152 | def test_different_base_body_changes_fingerprint(self) -> None: |
| 153 | ours = _note(None, "# H\n\nours\n") |
| 154 | theirs = _note(None, "# H\n\ntheirs\n") |
| 155 | base_a = _note(None, "# H\n\nbase-A\n") |
| 156 | base_b = _note(None, "# H\n\nbase-B\n") |
| 157 | assert ( |
| 158 | conflict_fingerprint(ours, theirs, base_a) |
| 159 | != conflict_fingerprint(ours, theirs, base_b) |
| 160 | ) |
| 161 | |
| 162 | def test_different_section_id_changes_fingerprint(self) -> None: |
| 163 | ours_a = _note(None, "## A\n\nours\n") |
| 164 | theirs_a = _note(None, "## A\n\ntheirs\n") |
| 165 | ours_b = _note(None, "## B\n\nours\n") |
| 166 | theirs_b = _note(None, "## B\n\ntheirs\n") |
| 167 | base = b"" |
| 168 | assert ( |
| 169 | conflict_fingerprint(ours_a, theirs_a, base) |
| 170 | != conflict_fingerprint(ours_b, theirs_b, base) |
| 171 | ) |
| 172 | |
| 173 | def test_different_section_level_changes_fingerprint(self) -> None: |
| 174 | ours_h2 = _note(None, "## A\n\nours\n") |
| 175 | theirs_h2 = _note(None, "## A\n\ntheirs\n") |
| 176 | ours_h3 = _note(None, "### A\n\nours\n") |
| 177 | theirs_h3 = _note(None, "### A\n\ntheirs\n") |
| 178 | assert ( |
| 179 | conflict_fingerprint(ours_h2, theirs_h2, b"") |
| 180 | != conflict_fingerprint(ours_h3, theirs_h3, b"") |
| 181 | ) |
| 182 | |
| 183 | def test_one_conflict_vs_two_conflicts_different(self) -> None: |
| 184 | ours_one = _note(None, "## A\n\nours\n\n## B\n\nshared\n") |
| 185 | theirs_one = _note(None, "## A\n\ntheirs\n\n## B\n\nshared\n") |
| 186 | ours_two = _note(None, "## A\n\nours\n\n## B\n\nours-too\n") |
| 187 | theirs_two = _note(None, "## A\n\ntheirs\n\n## B\n\ntheirs-too\n") |
| 188 | assert ( |
| 189 | conflict_fingerprint(ours_one, theirs_one, b"") |
| 190 | != conflict_fingerprint(ours_two, theirs_two, b"") |
| 191 | ) |
| 192 | |
| 193 | def test_swap_ours_theirs_changes_fingerprint(self) -> None: |
| 194 | """The fingerprint encodes an ordered (ours, theirs) pair.""" |
| 195 | ours = _note(None, "## A\n\nours\n") |
| 196 | theirs = _note(None, "## A\n\ntheirs\n") |
| 197 | assert ( |
| 198 | conflict_fingerprint(ours, theirs, b"") |
| 199 | != conflict_fingerprint(theirs, ours, b"") |
| 200 | ) |
| 201 | |
| 202 | def test_duplicate_heading_occurrences_distinguished(self) -> None: |
| 203 | ours_first = _note( |
| 204 | None, "## A\n\nours\n\n## A\n\nshared\n" |
| 205 | ) |
| 206 | theirs_first = _note( |
| 207 | None, "## A\n\ntheirs\n\n## A\n\nshared\n" |
| 208 | ) |
| 209 | ours_second = _note( |
| 210 | None, "## A\n\nshared\n\n## A\n\nours\n" |
| 211 | ) |
| 212 | theirs_second = _note( |
| 213 | None, "## A\n\nshared\n\n## A\n\ntheirs\n" |
| 214 | ) |
| 215 | assert ( |
| 216 | conflict_fingerprint(ours_first, theirs_first, b"") |
| 217 | != conflict_fingerprint(ours_second, theirs_second, b"") |
| 218 | ) |
| 219 | |
| 220 | def test_no_conflict_vs_real_conflict_different(self) -> None: |
| 221 | ours = _note(None, "## A\n\nshared\n") |
| 222 | theirs = _note(None, "## A\n\nshared\n") # identical → no conflict |
| 223 | ours_real = _note(None, "## A\n\nA\n") |
| 224 | theirs_real = _note(None, "## A\n\nB\n") |
| 225 | assert ( |
| 226 | conflict_fingerprint(ours, theirs, b"") |
| 227 | != conflict_fingerprint(ours_real, theirs_real, b"") |
| 228 | ) |
| 229 | |
| 230 | def test_section_addition_does_not_collide_with_section_modification( |
| 231 | self, |
| 232 | ) -> None: |
| 233 | # Side A: ours adds a section that theirs lacks → not a conflict per spec |
| 234 | ours_add = _note(None, "## A\n\nshared\n\n## B\n\nnew\n") |
| 235 | theirs_add = _note(None, "## A\n\nshared\n") |
| 236 | # Side B: real conflict on existing section |
| 237 | ours_mod = _note(None, "## A\n\nours\n") |
| 238 | theirs_mod = _note(None, "## A\n\ntheirs\n") |
| 239 | assert ( |
| 240 | conflict_fingerprint(ours_add, theirs_add, b"") |
| 241 | != conflict_fingerprint(ours_mod, theirs_mod, b"") |
| 242 | ) |
| 243 | |
| 244 | # -- Pairs that MUST match ------------------------------------------------- |
| 245 | |
| 246 | def test_cosmetic_crlf_line_endings_dont_change_fingerprint(self) -> None: |
| 247 | ours_lf = _note(None, "## A\n\nours\n") |
| 248 | theirs_lf = _note(None, "## A\n\ntheirs\n") |
| 249 | ours_crlf = ours_lf.replace(b"\n", b"\r\n") |
| 250 | theirs_crlf = theirs_lf.replace(b"\n", b"\r\n") |
| 251 | assert ( |
| 252 | conflict_fingerprint(ours_lf, theirs_lf, b"") |
| 253 | == conflict_fingerprint(ours_crlf, theirs_crlf, b"") |
| 254 | ) |
| 255 | |
| 256 | def test_frontmatter_tag_addition_doesnt_change_fingerprint(self) -> None: |
| 257 | ours_no_tags = _note({"title": "T"}, "## A\n\nours\n") |
| 258 | theirs_no_tags = _note({"title": "T"}, "## A\n\ntheirs\n") |
| 259 | ours_with_tags = _note( |
| 260 | {"title": "T", "tags": ["alpha"]}, "## A\n\nours\n" |
| 261 | ) |
| 262 | theirs_with_tags = _note( |
| 263 | {"title": "T", "tags": ["alpha"]}, "## A\n\ntheirs\n" |
| 264 | ) |
| 265 | assert ( |
| 266 | conflict_fingerprint(ours_no_tags, theirs_no_tags, b"") |
| 267 | == conflict_fingerprint(ours_with_tags, theirs_with_tags, b"") |
| 268 | ) |
| 269 | |
| 270 | def test_frontmatter_key_reordering_doesnt_change_fingerprint(self) -> None: |
| 271 | ours_a = _note( |
| 272 | {"title": "T", "project": "P", "tags": ["a"]}, "## H\n\nours\n" |
| 273 | ) |
| 274 | theirs_a = _note( |
| 275 | {"title": "T", "project": "P", "tags": ["a"]}, "## H\n\ntheirs\n" |
| 276 | ) |
| 277 | ours_b = _note( |
| 278 | {"project": "P", "tags": ["a"], "title": "T"}, "## H\n\nours\n" |
| 279 | ) |
| 280 | theirs_b = _note( |
| 281 | {"project": "P", "tags": ["a"], "title": "T"}, "## H\n\ntheirs\n" |
| 282 | ) |
| 283 | assert ( |
| 284 | conflict_fingerprint(ours_a, theirs_a, b"") |
| 285 | == conflict_fingerprint(ours_b, theirs_b, b"") |
| 286 | ) |
| 287 | |
| 288 | def test_non_conflicting_section_reorder_doesnt_change_fingerprint( |
| 289 | self, |
| 290 | ) -> None: |
| 291 | # Both sides have the SAME ## B body → not a conflict; only ## A |
| 292 | # differs → only ## A enters the fingerprint. Reordering A vs B |
| 293 | # in the document changes the section ordering but the conflict |
| 294 | # set, sorted by section_id, is identical. |
| 295 | ours_ab = _note(None, "## A\n\nours\n\n## B\n\nshared\n") |
| 296 | theirs_ab = _note(None, "## A\n\ntheirs\n\n## B\n\nshared\n") |
| 297 | ours_ba = _note(None, "## B\n\nshared\n\n## A\n\nours\n") |
| 298 | theirs_ba = _note(None, "## B\n\nshared\n\n## A\n\ntheirs\n") |
| 299 | assert ( |
| 300 | conflict_fingerprint(ours_ab, theirs_ab, b"") |
| 301 | == conflict_fingerprint(ours_ba, theirs_ba, b"") |
| 302 | ) |
| 303 | |
| 304 | def test_preamble_whitespace_doesnt_change_fingerprint(self) -> None: |
| 305 | # When the only difference is the body of a section that both sides |
| 306 | # share identically, no conflict is registered for that section, |
| 307 | # so the only conflict comes from ## A. Adding extra blank lines |
| 308 | # to the matching ## B is absorbed by the splitter. |
| 309 | ours = _note(None, "## A\n\nours\n\n## B\n\nshared\n") |
| 310 | theirs = _note(None, "## A\n\ntheirs\n\n## B\n\nshared\n") |
| 311 | ours_extra_blank = _note( |
| 312 | None, "## A\n\nours\n\n## B\n\nshared\n\n" |
| 313 | ) |
| 314 | theirs_extra_blank = _note( |
| 315 | None, "## A\n\ntheirs\n\n## B\n\nshared\n\n" |
| 316 | ) |
| 317 | assert ( |
| 318 | conflict_fingerprint(ours, theirs, b"") |
| 319 | == conflict_fingerprint(ours_extra_blank, theirs_extra_blank, b"") |
| 320 | ) |
| 321 | |
| 322 | # -- Format guarantees ----------------------------------------------------- |
| 323 | |
| 324 | def test_fingerprint_is_64_lowercase_hex(self) -> None: |
| 325 | fp = conflict_fingerprint( |
| 326 | _note(None, "## A\n\nours\n"), |
| 327 | _note(None, "## A\n\ntheirs\n"), |
| 328 | b"", |
| 329 | ) |
| 330 | assert _FINGERPRINT_RE.match(fp), f"not 64-char hex: {fp!r}" |
| 331 | |
| 332 | def test_fingerprint_is_deterministic(self) -> None: |
| 333 | ours = _note(None, "## A\n\nours\n") |
| 334 | theirs = _note(None, "## A\n\ntheirs\n") |
| 335 | assert ( |
| 336 | conflict_fingerprint(ours, theirs, b"") |
| 337 | == conflict_fingerprint(ours, theirs, b"") |
| 338 | ) |
| 339 | |
| 340 | def test_no_conflicting_sections_returns_empty_sha256(self) -> None: |
| 341 | # When ours == theirs (no conflicting sections) the fingerprint |
| 342 | # is the SHA-256 of the empty byte sequence. |
| 343 | n = _note(None, "## A\n\nshared\n") |
| 344 | empty_sha = hashlib.sha256(b"").hexdigest() |
| 345 | assert conflict_fingerprint(n, n, b"") == empty_sha |
| 346 | |
| 347 | |
| 348 | # ────────────────────────────────────────────────────────────────────────────── |
| 349 | # Tier 2 — Unit: hash-only record / lookup round-trip |
| 350 | # ────────────────────────────────────────────────────────────────────────────── |
| 351 | |
| 352 | |
| 353 | class TestHashRecordRoundtrip: |
| 354 | def test_record_then_lookup_returns_resolved_hash( |
| 355 | self, tmp_root: pathlib.Path |
| 356 | ) -> None: |
| 357 | fp = conflict_fingerprint( |
| 358 | _note(None, "## A\n\nours\n"), |
| 359 | _note(None, "## A\n\ntheirs\n"), |
| 360 | b"", |
| 361 | ) |
| 362 | resolved = _note(None, "## A\n\nresolved\n") |
| 363 | record_resolution(tmp_root, fp, resolved) |
| 364 | assert lookup_resolution(tmp_root, fp) == _h64(canonicalize(resolved)) |
| 365 | |
| 366 | def test_lookup_unknown_fingerprint_returns_none( |
| 367 | self, tmp_root: pathlib.Path |
| 368 | ) -> None: |
| 369 | assert lookup_resolution(tmp_root, "f" * 64) is None |
| 370 | |
| 371 | def test_two_distinct_fingerprints_dont_collide( |
| 372 | self, tmp_root: pathlib.Path |
| 373 | ) -> None: |
| 374 | fp_a = conflict_fingerprint( |
| 375 | _note(None, "## A\n\nours-A\n"), |
| 376 | _note(None, "## A\n\ntheirs-A\n"), |
| 377 | b"", |
| 378 | ) |
| 379 | fp_b = conflict_fingerprint( |
| 380 | _note(None, "## B\n\nours-B\n"), |
| 381 | _note(None, "## B\n\ntheirs-B\n"), |
| 382 | b"", |
| 383 | ) |
| 384 | assert fp_a != fp_b |
| 385 | record_resolution(tmp_root, fp_a, _note(None, "## A\n\nA\n")) |
| 386 | record_resolution(tmp_root, fp_b, _note(None, "## B\n\nB\n")) |
| 387 | assert lookup_resolution(tmp_root, fp_a) == _h64( |
| 388 | canonicalize(_note(None, "## A\n\nA\n")) |
| 389 | ) |
| 390 | assert lookup_resolution(tmp_root, fp_b) == _h64( |
| 391 | canonicalize(_note(None, "## B\n\nB\n")) |
| 392 | ) |
| 393 | assert lookup_resolution(tmp_root, fp_a) != lookup_resolution( |
| 394 | tmp_root, fp_b |
| 395 | ) |
| 396 | |
| 397 | def test_record_creates_sharded_path(self, tmp_root: pathlib.Path) -> None: |
| 398 | fp = "abcdef0123456789" * 4 # exactly 64 lowercase hex chars |
| 399 | record_resolution(tmp_root, fp, _note(None, "## A\n\nx\n")) |
| 400 | sharded = tmp_root / ".muse" / "rerere" / "ab" / f"{fp[2:]}.json" |
| 401 | assert sharded.exists() |
| 402 | |
| 403 | def test_replay_resolution_matches_when_current_already_resolved( |
| 404 | self, tmp_root: pathlib.Path |
| 405 | ) -> None: |
| 406 | fp = conflict_fingerprint( |
| 407 | _note(None, "## A\n\nours\n"), |
| 408 | _note(None, "## A\n\ntheirs\n"), |
| 409 | b"", |
| 410 | ) |
| 411 | resolved = _note(None, "## A\n\nresolved\n") |
| 412 | record_resolution(tmp_root, fp, resolved) |
| 413 | assert replay_resolution(tmp_root, fp, resolved) == resolved |
| 414 | |
| 415 | def test_replay_resolution_returns_none_when_diverged( |
| 416 | self, tmp_root: pathlib.Path |
| 417 | ) -> None: |
| 418 | fp = conflict_fingerprint( |
| 419 | _note(None, "## A\n\nours\n"), |
| 420 | _note(None, "## A\n\ntheirs\n"), |
| 421 | b"", |
| 422 | ) |
| 423 | record_resolution( |
| 424 | tmp_root, fp, _note(None, "## A\n\nresolved\n") |
| 425 | ) |
| 426 | diverged = _note(None, "## A\n\nsomething-else\n") |
| 427 | assert replay_resolution(tmp_root, fp, diverged) is None |
| 428 | |
| 429 | def test_replay_resolution_returns_none_when_no_record( |
| 430 | self, tmp_root: pathlib.Path |
| 431 | ) -> None: |
| 432 | assert ( |
| 433 | replay_resolution( |
| 434 | tmp_root, "0" * 64, _note(None, "## A\n\nx\n") |
| 435 | ) |
| 436 | is None |
| 437 | ) |
| 438 | |
| 439 | def test_record_is_idempotent(self, tmp_root: pathlib.Path) -> None: |
| 440 | fp = "1" * 64 |
| 441 | resolved = _note(None, "## A\n\nA\n") |
| 442 | record_resolution(tmp_root, fp, resolved) |
| 443 | first_hash = lookup_resolution(tmp_root, fp) |
| 444 | record_resolution(tmp_root, fp, resolved) |
| 445 | assert lookup_resolution(tmp_root, fp) == first_hash |
| 446 | |
| 447 | def test_record_overwrites_atomically( |
| 448 | self, tmp_root: pathlib.Path |
| 449 | ) -> None: |
| 450 | fp = "2" * 64 |
| 451 | record_resolution(tmp_root, fp, _note(None, "## A\n\nfirst\n")) |
| 452 | record_resolution(tmp_root, fp, _note(None, "## A\n\nsecond\n")) |
| 453 | # Latest record wins. |
| 454 | assert lookup_resolution(tmp_root, fp) == _h64( |
| 455 | canonicalize(_note(None, "## A\n\nsecond\n")) |
| 456 | ) |
| 457 | |
| 458 | |
| 459 | # ────────────────────────────────────────────────────────────────────────────── |
| 460 | # Tier 3 — Unit: full-bytes record / replay |
| 461 | # ────────────────────────────────────────────────────────────────────────────── |
| 462 | |
| 463 | |
| 464 | class TestFullResolutionRoundtrip: |
| 465 | def test_record_full_then_replay_returns_canonical_bytes( |
| 466 | self, tmp_root: pathlib.Path |
| 467 | ) -> None: |
| 468 | fp = "3" * 64 |
| 469 | resolved = _note(None, "## A\n\nresolved\n") |
| 470 | record_full_resolution(tmp_root, fp, resolved) |
| 471 | replayed = replay_full_resolution( |
| 472 | tmp_root, fp, _note(None, "## A\n\nany\n") |
| 473 | ) |
| 474 | assert replayed == canonicalize(resolved) |
| 475 | |
| 476 | def test_replay_full_returns_none_for_missing_fingerprint( |
| 477 | self, tmp_root: pathlib.Path |
| 478 | ) -> None: |
| 479 | assert ( |
| 480 | replay_full_resolution( |
| 481 | tmp_root, "4" * 64, _note(None, "## A\n\nx\n") |
| 482 | ) |
| 483 | is None |
| 484 | ) |
| 485 | |
| 486 | def test_record_full_uses_full_suffix( |
| 487 | self, tmp_root: pathlib.Path |
| 488 | ) -> None: |
| 489 | fp = "5" * 64 |
| 490 | record_full_resolution(tmp_root, fp, _note(None, "## A\n\nx\n")) |
| 491 | full_path = ( |
| 492 | tmp_root / ".muse" / "rerere" / "55" / f"{fp[2:]}.full" |
| 493 | ) |
| 494 | assert full_path.exists() |
| 495 | # Hash-only suffix path must NOT have been created by full record. |
| 496 | json_path = ( |
| 497 | tmp_root / ".muse" / "rerere" / "55" / f"{fp[2:]}.json" |
| 498 | ) |
| 499 | assert not json_path.exists() |
| 500 | |
| 501 | def test_full_and_hash_records_coexist( |
| 502 | self, tmp_root: pathlib.Path |
| 503 | ) -> None: |
| 504 | fp = "6" * 64 |
| 505 | resolved = _note(None, "## A\n\nresolved\n") |
| 506 | record_resolution(tmp_root, fp, resolved) |
| 507 | record_full_resolution(tmp_root, fp, resolved) |
| 508 | assert lookup_resolution(tmp_root, fp) == _h64(canonicalize(resolved)) |
| 509 | assert replay_full_resolution(tmp_root, fp, b"") == canonicalize( |
| 510 | resolved |
| 511 | ) |
| 512 | |
| 513 | def test_full_replay_returns_canonical_form( |
| 514 | self, tmp_root: pathlib.Path |
| 515 | ) -> None: |
| 516 | """The stored bytes are canonical — non-canonical input is normalised.""" |
| 517 | fp = "7" * 64 |
| 518 | # Non-canonical: CRLF and unsorted YAML keys |
| 519 | non_canonical = ( |
| 520 | b"---\r\nproject: P\r\ntitle: T\r\n---\r\n## A\r\n\r\nx\r\n" |
| 521 | ) |
| 522 | record_full_resolution(tmp_root, fp, non_canonical) |
| 523 | replayed = replay_full_resolution( |
| 524 | tmp_root, fp, _note(None, "## A\n\nx\n") |
| 525 | ) |
| 526 | assert replayed == canonicalize(non_canonical) |
| 527 | # Canonical form has LF endings. |
| 528 | assert replayed is not None and b"\r\n" not in replayed |
| 529 | |
| 530 | |
| 531 | # ────────────────────────────────────────────────────────────────────────────── |
| 532 | # Tier 4 — Integration: replay after cosmetic rewrite |
| 533 | # ────────────────────────────────────────────────────────────────────────────── |
| 534 | |
| 535 | |
| 536 | class TestCosmeticRewriteReplay: |
| 537 | def test_replay_succeeds_after_tag_addition( |
| 538 | self, tmp_root: pathlib.Path |
| 539 | ) -> None: |
| 540 | # Original conflict |
| 541 | ours_v1 = _note( |
| 542 | {"title": "Note"}, "## §Summary\n\nours-summary\n" |
| 543 | ) |
| 544 | theirs_v1 = _note( |
| 545 | {"title": "Note"}, "## §Summary\n\ntheirs-summary\n" |
| 546 | ) |
| 547 | resolved_v1 = _note( |
| 548 | {"title": "Note"}, "## §Summary\n\nmerged-summary\n" |
| 549 | ) |
| 550 | fp_v1 = conflict_fingerprint(ours_v1, theirs_v1, b"") |
| 551 | record_full_resolution(tmp_root, fp_v1, resolved_v1) |
| 552 | |
| 553 | # Same conflict after the user reformats the note: adds a tag |
| 554 | # and a project key. The conflicting section is unchanged. |
| 555 | ours_v2 = _note( |
| 556 | {"title": "Note", "project": "P", "tags": ["new"]}, |
| 557 | "## §Summary\n\nours-summary\n", |
| 558 | ) |
| 559 | theirs_v2 = _note( |
| 560 | {"title": "Note", "project": "P", "tags": ["new"]}, |
| 561 | "## §Summary\n\ntheirs-summary\n", |
| 562 | ) |
| 563 | fp_v2 = conflict_fingerprint(ours_v2, theirs_v2, b"") |
| 564 | |
| 565 | assert fp_v1 == fp_v2, ( |
| 566 | "fingerprint must be stable across cosmetic frontmatter rewrites" |
| 567 | ) |
| 568 | replayed = replay_full_resolution(tmp_root, fp_v2, ours_v2) |
| 569 | assert replayed == canonicalize(resolved_v1) |
| 570 | |
| 571 | def test_replay_succeeds_after_non_conflicting_section_reorder( |
| 572 | self, tmp_root: pathlib.Path |
| 573 | ) -> None: |
| 574 | # Two-section note; only ## §Summary differs across branches. |
| 575 | ours = _note( |
| 576 | None, "## §Summary\n\nours\n\n## Notes\n\nshared\n" |
| 577 | ) |
| 578 | theirs = _note( |
| 579 | None, "## §Summary\n\ntheirs\n\n## Notes\n\nshared\n" |
| 580 | ) |
| 581 | resolved = _note( |
| 582 | None, "## §Summary\n\nmerged\n\n## Notes\n\nshared\n" |
| 583 | ) |
| 584 | fp = conflict_fingerprint(ours, theirs, b"") |
| 585 | record_full_resolution(tmp_root, fp, resolved) |
| 586 | |
| 587 | # Reorder the non-conflicting ## Notes section to come first. |
| 588 | ours_reordered = _note( |
| 589 | None, "## Notes\n\nshared\n\n## §Summary\n\nours\n" |
| 590 | ) |
| 591 | theirs_reordered = _note( |
| 592 | None, "## Notes\n\nshared\n\n## §Summary\n\ntheirs\n" |
| 593 | ) |
| 594 | fp_reordered = conflict_fingerprint( |
| 595 | ours_reordered, theirs_reordered, b"" |
| 596 | ) |
| 597 | assert fp == fp_reordered |
| 598 | replayed = replay_full_resolution( |
| 599 | tmp_root, fp_reordered, ours_reordered |
| 600 | ) |
| 601 | assert replayed == canonicalize(resolved) |
| 602 | |
| 603 | |
| 604 | # ────────────────────────────────────────────────────────────────────────────── |
| 605 | # Tier 5 — Integration: rerere CLI hook (compute_fingerprint dispatch) |
| 606 | # ────────────────────────────────────────────────────────────────────────────── |
| 607 | |
| 608 | |
| 609 | class TestRerereCliHook: |
| 610 | """Verify the CLI dispatch sees the knowtation plugin as a RererePlugin.""" |
| 611 | |
| 612 | def test_knowtation_plugin_is_recognised_as_rereeplugin(self) -> None: |
| 613 | from muse.domain import RererePlugin |
| 614 | from muse.plugins.knowtation.plugin import plugin as kn_plugin |
| 615 | |
| 616 | assert isinstance(kn_plugin, RererePlugin) |
| 617 | |
| 618 | def test_compute_fingerprint_uses_knowtation_aware_fingerprint( |
| 619 | self, tmp_path: pathlib.Path |
| 620 | ) -> None: |
| 621 | """``muse.core.rerere.compute_fingerprint`` must call into the |
| 622 | knowtation plugin for repos with ``domain="knowtation"``. |
| 623 | |
| 624 | We exercise the bytes path by storing two note blobs in the local |
| 625 | object store and calling ``compute_fingerprint`` with their IDs. |
| 626 | The result must equal the bytes-level fingerprint computed by |
| 627 | :func:`conflict_fingerprint` (with empty base since the protocol |
| 628 | does not pass one). |
| 629 | """ |
| 630 | from muse.core.object_store import write_object |
| 631 | from muse.core.rerere import compute_fingerprint |
| 632 | from muse.plugins.knowtation.plugin import plugin as kn_plugin |
| 633 | |
| 634 | ours = _note(None, "## §Summary\n\nours-text\n") |
| 635 | theirs = _note(None, "## §Summary\n\ntheirs-text\n") |
| 636 | ours_id = _h64(ours) |
| 637 | theirs_id = _h64(theirs) |
| 638 | |
| 639 | # write_object stores under .muse/objects/. |
| 640 | (tmp_path / ".muse").mkdir(parents=True, exist_ok=True) |
| 641 | write_object(tmp_path, ours_id, ours) |
| 642 | write_object(tmp_path, theirs_id, theirs) |
| 643 | |
| 644 | cli_fp = compute_fingerprint( |
| 645 | "notes/test.md", ours_id, theirs_id, kn_plugin, tmp_path |
| 646 | ) |
| 647 | domain_fp = conflict_fingerprint(ours, theirs, b"") |
| 648 | assert cli_fp == domain_fp, ( |
| 649 | "muse.core.rerere.compute_fingerprint must dispatch to the " |
| 650 | "knowtation plugin's RererePlugin protocol method" |
| 651 | ) |
| 652 | |
| 653 | def test_compute_fingerprint_falls_back_when_blob_missing( |
| 654 | self, tmp_path: pathlib.Path |
| 655 | ) -> None: |
| 656 | """When ours_id or theirs_id is not in the local object store the |
| 657 | plugin's protocol method raises and ``compute_fingerprint`` falls |
| 658 | back to the default content fingerprint. |
| 659 | """ |
| 660 | from muse.core.rerere import ( |
| 661 | compute_fingerprint, |
| 662 | conflict_fingerprint as default_fp, |
| 663 | ) |
| 664 | from muse.plugins.knowtation.plugin import plugin as kn_plugin |
| 665 | |
| 666 | # Two valid 64-char hex IDs that are NOT in the object store. |
| 667 | ours_id = "a" * 64 |
| 668 | theirs_id = "b" * 64 |
| 669 | (tmp_path / ".muse").mkdir(parents=True, exist_ok=True) |
| 670 | |
| 671 | result = compute_fingerprint( |
| 672 | "notes/x.md", ours_id, theirs_id, kn_plugin, tmp_path |
| 673 | ) |
| 674 | # Falls back to the default ID-based fingerprint. |
| 675 | assert result == default_fp(ours_id, theirs_id) |
| 676 | |
| 677 | |
| 678 | # ────────────────────────────────────────────────────────────────────────────── |
| 679 | # Tier 6 — Data-integrity: 50 historical conflict stubs |
| 680 | # ────────────────────────────────────────────────────────────────────────────── |
| 681 | |
| 682 | |
| 683 | class TestFiftyHistoricalConflicts: |
| 684 | """Synthesise 50 distinct conflicts; record half; verify lookups.""" |
| 685 | |
| 686 | def _make_pair(self, idx: int) -> tuple[bytes, bytes, bytes]: |
| 687 | """Return ``(ours, theirs, resolved)`` for the *idx*-th synthetic conflict.""" |
| 688 | ours = _note( |
| 689 | {"title": f"N-{idx}"}, |
| 690 | f"## §Summary\n\nours body {idx}\n", |
| 691 | ) |
| 692 | theirs = _note( |
| 693 | {"title": f"N-{idx}"}, |
| 694 | f"## §Summary\n\ntheirs body {idx}\n", |
| 695 | ) |
| 696 | resolved = _note( |
| 697 | {"title": f"N-{idx}"}, |
| 698 | f"## §Summary\n\nmerged body {idx}\n", |
| 699 | ) |
| 700 | return ours, theirs, resolved |
| 701 | |
| 702 | def test_25_recorded_25_missing_no_collisions( |
| 703 | self, tmp_root: pathlib.Path |
| 704 | ) -> None: |
| 705 | fingerprints: list[str] = [] |
| 706 | for i in range(50): |
| 707 | ours, theirs, resolved = self._make_pair(i) |
| 708 | fp = conflict_fingerprint(ours, theirs, b"") |
| 709 | fingerprints.append(fp) |
| 710 | if i < 25: |
| 711 | record_full_resolution(tmp_root, fp, resolved) |
| 712 | |
| 713 | assert len(set(fingerprints)) == 50, ( |
| 714 | f"Expected 50 unique fingerprints, got {len(set(fingerprints))} — " |
| 715 | "collisions detected across synthetic stubs" |
| 716 | ) |
| 717 | |
| 718 | for i, fp in enumerate(fingerprints): |
| 719 | ours, _theirs, resolved = self._make_pair(i) |
| 720 | replayed = replay_full_resolution(tmp_root, fp, ours) |
| 721 | if i < 25: |
| 722 | assert replayed == canonicalize(resolved), ( |
| 723 | f"Stub {i}: full replay missed a recorded resolution" |
| 724 | ) |
| 725 | else: |
| 726 | assert replayed is None, ( |
| 727 | f"Stub {i}: full replay returned data for an unrecorded " |
| 728 | "fingerprint" |
| 729 | ) |
| 730 | |
| 731 | |
| 732 | # ────────────────────────────────────────────────────────────────────────────── |
| 733 | # Tier 7 — Security |
| 734 | # ────────────────────────────────────────────────────────────────────────────── |
| 735 | |
| 736 | |
| 737 | class TestSecurity: |
| 738 | # -- Separator-injection forgery defence ----------------------------------- |
| 739 | |
| 740 | def test_pipe_in_section_title_does_not_collide(self) -> None: |
| 741 | """A section title containing ``|`` must not let a hostile note |
| 742 | produce the same fingerprint as a *different* logical conflict.""" |
| 743 | ours_a = _note(None, "## A\n\nours-A\n\n## B\n\nours-B\n") |
| 744 | theirs_a = _note(None, "## A\n\ntheirs-A\n\n## B\n\ntheirs-B\n") |
| 745 | # Adversarial: title containing the joiner character that an |
| 746 | # unsafe stringly fingerprint encoder might use as a delimiter. |
| 747 | ours_b = _note( |
| 748 | None, "## A|<malicious tuple>\n\nours-A\n\n## B\n\nours-B\n" |
| 749 | ) |
| 750 | theirs_b = _note( |
| 751 | None, |
| 752 | "## A|<malicious tuple>\n\ntheirs-A\n\n## B\n\ntheirs-B\n", |
| 753 | ) |
| 754 | fp_a = conflict_fingerprint(ours_a, theirs_a, b"") |
| 755 | fp_b = conflict_fingerprint(ours_b, theirs_b, b"") |
| 756 | assert fp_a != fp_b, ( |
| 757 | "fingerprint must distinguish the original conflict from the " |
| 758 | "pipe-injected adversarial title" |
| 759 | ) |
| 760 | |
| 761 | def test_colon_in_section_title_does_not_collide(self) -> None: |
| 762 | ours_a = _note(None, "## A\n\nours\n") |
| 763 | theirs_a = _note(None, "## A\n\ntheirs\n") |
| 764 | ours_b = _note(None, "## A:section:1:B#0\n\nours\n") |
| 765 | theirs_b = _note(None, "## A:section:1:B#0\n\ntheirs\n") |
| 766 | fp_a = conflict_fingerprint(ours_a, theirs_a, b"") |
| 767 | fp_b = conflict_fingerprint(ours_b, theirs_b, b"") |
| 768 | assert fp_a != fp_b |
| 769 | |
| 770 | # -- Path-traversal defence ------------------------------------------------ |
| 771 | |
| 772 | def test_path_traversal_fingerprint_rejected( |
| 773 | self, tmp_root: pathlib.Path |
| 774 | ) -> None: |
| 775 | for bad in ("../evil", "/abs/path", "..", "abc/def", "X" * 64): |
| 776 | with pytest.raises(ValueError): |
| 777 | record_resolution( |
| 778 | tmp_root, bad, _note(None, "## A\n\nx\n") |
| 779 | ) |
| 780 | with pytest.raises(ValueError): |
| 781 | lookup_resolution(tmp_root, bad) |
| 782 | with pytest.raises(ValueError): |
| 783 | record_full_resolution( |
| 784 | tmp_root, bad, _note(None, "## A\n\nx\n") |
| 785 | ) |
| 786 | with pytest.raises(ValueError): |
| 787 | replay_full_resolution( |
| 788 | tmp_root, bad, _note(None, "## A\n\nx\n") |
| 789 | ) |
| 790 | |
| 791 | def test_uppercase_hex_fingerprint_rejected( |
| 792 | self, tmp_root: pathlib.Path |
| 793 | ) -> None: |
| 794 | # Strict lowercase-only contract. |
| 795 | with pytest.raises(ValueError): |
| 796 | record_resolution( |
| 797 | tmp_root, "A" * 64, _note(None, "## A\n\nx\n") |
| 798 | ) |
| 799 | |
| 800 | def test_whitespace_in_fingerprint_rejected( |
| 801 | self, tmp_root: pathlib.Path |
| 802 | ) -> None: |
| 803 | with pytest.raises(ValueError): |
| 804 | lookup_resolution(tmp_root, "a" * 63 + " ") |
| 805 | |
| 806 | # -- 1 MiB conflicting section --------------------------------------------- |
| 807 | |
| 808 | def test_one_mib_conflicting_section_does_not_crash(self) -> None: |
| 809 | big_ours = "x" * (1024 * 1024) |
| 810 | big_theirs = "y" * (1024 * 1024) |
| 811 | ours = _note(None, f"## §Summary\n\n{big_ours}\n") |
| 812 | theirs = _note(None, f"## §Summary\n\n{big_theirs}\n") |
| 813 | start = time.perf_counter() |
| 814 | fp = conflict_fingerprint(ours, theirs, b"") |
| 815 | elapsed = time.perf_counter() - start |
| 816 | assert _FINGERPRINT_RE.match(fp) |
| 817 | # Generous bound — pure SHA-256 of 1 MiB takes ~10 ms; canonicalize |
| 818 | # adds parse overhead but should still finish in well under 5 s. |
| 819 | assert elapsed < 5.0, f"1 MiB fingerprint took {elapsed:.2f}s" |
| 820 | |
| 821 | # -- NUL bytes in section bodies ------------------------------------------- |
| 822 | |
| 823 | def test_nul_bytes_in_section_bodies_handled( |
| 824 | self, tmp_root: pathlib.Path |
| 825 | ) -> None: |
| 826 | ours = b"## A\n\nours\x00with\x00nul\n" |
| 827 | theirs = b"## A\n\ntheirs\x00with\x00nul\n" |
| 828 | fp = conflict_fingerprint(ours, theirs, b"") |
| 829 | assert _FINGERPRINT_RE.match(fp) |
| 830 | # Round-trip a hash record on the same NUL-containing data. |
| 831 | record_resolution(tmp_root, fp, ours) |
| 832 | assert lookup_resolution(tmp_root, fp) == _h64(canonicalize(ours)) |
| 833 | |
| 834 | # -- Oversized inputs ------------------------------------------------------ |
| 835 | |
| 836 | def test_oversize_ours_raises_value_error(self) -> None: |
| 837 | big = b"# H\n\n" + b"a" * (_MAX_NOTE_BYTES + 1) |
| 838 | with pytest.raises(ValueError): |
| 839 | conflict_fingerprint(big, b"# H\n\nb\n", b"") |
| 840 | |
| 841 | def test_oversize_theirs_raises_value_error(self) -> None: |
| 842 | big = b"# H\n\n" + b"a" * (_MAX_NOTE_BYTES + 1) |
| 843 | with pytest.raises(ValueError): |
| 844 | conflict_fingerprint(b"# H\n\nb\n", big, b"") |
| 845 | |
| 846 | def test_oversize_base_raises_value_error(self) -> None: |
| 847 | big = b"# H\n\n" + b"a" * (_MAX_NOTE_BYTES + 1) |
| 848 | with pytest.raises(ValueError): |
| 849 | conflict_fingerprint(b"# H\n\nb\n", b"# H\n\nc\n", big) |
| 850 | |
| 851 | def test_oversize_resolved_record_raises_value_error( |
| 852 | self, tmp_root: pathlib.Path |
| 853 | ) -> None: |
| 854 | big = b"# H\n\n" + b"a" * (_MAX_NOTE_BYTES + 1) |
| 855 | with pytest.raises(ValueError): |
| 856 | record_resolution(tmp_root, "0" * 64, big) |
| 857 | with pytest.raises(ValueError): |
| 858 | record_full_resolution(tmp_root, "0" * 64, big) |
| 859 | |
| 860 | # -- Malformed stored record returns None gracefully ----------------------- |
| 861 | |
| 862 | def test_malformed_json_record_treated_as_missing( |
| 863 | self, tmp_root: pathlib.Path |
| 864 | ) -> None: |
| 865 | fp = "9" * 64 |
| 866 | shard = tmp_root / ".muse" / "rerere" / "99" |
| 867 | shard.mkdir(parents=True, exist_ok=True) |
| 868 | (shard / f"{fp[2:]}.json").write_text("not-valid-json{", encoding="utf-8") |
| 869 | assert lookup_resolution(tmp_root, fp) is None |
| 870 | |
| 871 | def test_record_with_non_string_resolved_hash_treated_as_missing( |
| 872 | self, tmp_root: pathlib.Path |
| 873 | ) -> None: |
| 874 | fp = "8" * 64 |
| 875 | shard = tmp_root / ".muse" / "rerere" / "88" |
| 876 | shard.mkdir(parents=True, exist_ok=True) |
| 877 | (shard / f"{fp[2:]}.json").write_text( |
| 878 | json.dumps({"resolved_hash": 12345}), encoding="utf-8" |
| 879 | ) |
| 880 | assert lookup_resolution(tmp_root, fp) is None |
| 881 | |
| 882 | def test_yaml_safe_load_used(self) -> None: |
| 883 | """The module must NOT call yaml.unsafe_load anywhere.""" |
| 884 | from muse.plugins.knowtation import rerere as rr_mod |
| 885 | import inspect |
| 886 | |
| 887 | src = inspect.getsource(rr_mod) |
| 888 | assert "unsafe_load" not in src |
| 889 | assert "Loader" not in src or "safe" in src |
| 890 | |
| 891 | |
| 892 | # ────────────────────────────────────────────────────────────────────────────── |
| 893 | # Plugin façade |
| 894 | # ────────────────────────────────────────────────────────────────────────────── |
| 895 | |
| 896 | |
| 897 | class TestPluginFacade: |
| 898 | def test_module_singleton_is_a_plugin_instance(self) -> None: |
| 899 | assert isinstance(knowtation_rerere_plugin, KnowtationRererePlugin) |
| 900 | assert knowtation_rerere_plugin.domain == DOMAIN == "knowtation" |
| 901 | |
| 902 | def test_plugin_methods_delegate(self, tmp_root: pathlib.Path) -> None: |
| 903 | p = KnowtationRererePlugin() |
| 904 | ours = _note(None, "## A\n\nours\n") |
| 905 | theirs = _note(None, "## A\n\ntheirs\n") |
| 906 | resolved = _note(None, "## A\n\nresolved\n") |
| 907 | |
| 908 | fp_method = p.conflict_fingerprint(ours, theirs, b"") |
| 909 | fp_module = conflict_fingerprint(ours, theirs, b"") |
| 910 | assert fp_method == fp_module |
| 911 | |
| 912 | p.record_resolution(tmp_root, fp_method, resolved) |
| 913 | assert p.lookup_resolution(tmp_root, fp_method) == _h64( |
| 914 | canonicalize(resolved) |
| 915 | ) |
| 916 | assert p.replay_resolution(tmp_root, fp_method, resolved) == resolved |
| 917 | |
| 918 | p.record_full_resolution(tmp_root, fp_method, resolved) |
| 919 | assert p.replay_full_resolution(tmp_root, fp_method, ours) == ( |
| 920 | canonicalize(resolved) |
| 921 | ) |