"""Tests for ``muse/plugins/knowtation/differ.py`` — Phase 2.1. This suite covers every test tier required by Rule #0: 1. **Unit** — per-layer behaviour (title, frontmatter scalar, tag add/remove, section add/remove/move, body line edit), helpers (``_content_id``, ``_stringify``, ``_section_id``, address parsing, canonicalisation). 2. **Integration** — ``diff_notes`` over realistic note pairs returns the expected op shape for a known fixture. 3. **End-to-end** — ``apply(diff_notes(A, B), A) == B`` for 20 hand-crafted note pairs spanning all edge cases (empty input, frontmatter-only, no frontmatter, single section, many sections, deep section bodies, set field churn, scalar field add/remove, title creation, title deletion, section moves, etc.). 4. **Data integrity** — 200 ``hypothesis``-generated note pairs assert the round-trip invariant. 5. **Performance** — a realistic 50-section note diffs in < 100 ms. 6. **Stress** — 200 mixed note pairs diff + apply sequentially in under 5 s. 7. **Security** — adversarial inputs (binary blobs, NUL bytes, very long lines, deeply nested YAML, oversized notes) either return cleanly or raise the documented ``ValueError``; never crash, never expose secrets. Conventions ----------- * Every helper returns canonical-form bytes so the round-trip invariant is byte-stable. * The ``_note`` helper builds a canonical note from a frontmatter mapping plus body text. * Hypothesis strategies use only printable ASCII so generated frontmatter values round-trip through YAML deterministically without quoting surprises. """ from __future__ import annotations import string import time from typing import Any import pytest import yaml from hypothesis import HealthCheck, given, settings, strategies as st from muse.plugins.knowtation.differ import ( _FM_PREFIX, _LINE_INFIX, _MAX_NOTE_BYTES, _SECTION_PREFIX, _TITLE_ADDR, _content_id, _diff_body_lines, _diff_frontmatter, _diff_sections, _diff_title, _parse_note, _section_from_address, _section_id, _split_sections_typed, _stringify, apply, canonicalize, diff_notes, ) # ───────────────────────────────────────────────────────────────────────────── # Note construction helpers # ───────────────────────────────────────────────────────────────────────────── def _build_note(frontmatter: dict[str, Any] | None, body: str) -> bytes: """Construct canonical-form note bytes from a frontmatter dict + body. Always passes through :func:`canonicalize` so the result is the byte- stable form that the differ guarantees to round-trip. Args: frontmatter: Frontmatter mapping (``None`` for no frontmatter). body: Body text (post-frontmatter). Returns: Canonical-form note bytes. """ if not frontmatter: return canonicalize(body.encode("utf-8")) yaml_text = yaml.safe_dump( frontmatter, sort_keys=False, default_flow_style=False, allow_unicode=True, width=10000, ) raw = f"---\n{yaml_text}---\n{body}".encode("utf-8") return canonicalize(raw) def _round_trip(a: bytes, b: bytes) -> None: """Assert ``apply(diff_notes(A, B), A) == B`` for canonical inputs. Args: a: Base note bytes (canonical form). b: Target note bytes (canonical form). Raises: AssertionError: When the round-trip fails. """ delta = diff_notes(a, b) out = apply(delta, a) assert out == b, ( f"Round-trip mismatch.\n" f"--- expected ---\n{b.decode('utf-8', errors='replace')}\n" f"--- got ---\n{out.decode('utf-8', errors='replace')}\n" f"--- ops ---\n{delta.get('ops')}" ) # ============================================================================= # Tier 1 — Unit tests # ============================================================================= class TestStringifyAndContentId: """Unit tests for the small scalar helpers.""" def test_stringify_none(self) -> None: assert _stringify(None) == "" def test_stringify_bool_true(self) -> None: assert _stringify(True) == "true" def test_stringify_bool_false(self) -> None: assert _stringify(False) == "false" def test_stringify_int(self) -> None: assert _stringify(42) == "42" def test_stringify_str(self) -> None: assert _stringify("hello") == "hello" def test_content_id_deterministic(self) -> None: assert _content_id("hello") == _content_id("hello") def test_content_id_hex_length(self) -> None: assert len(_content_id("hello")) == 64 assert all(c in string.hexdigits for c in _content_id("hello")) def test_content_id_different_inputs_differ(self) -> None: assert _content_id("a") != _content_id("b") class TestSectionIdAndAddress: """Unit tests for section-ID encoding and reverse parsing.""" def test_section_id_format(self) -> None: assert _section_id(2, "Background", 0) == "section:2:Background#0" def test_section_id_preamble(self) -> None: assert _section_id(0, "(preamble)", 0) == "section:0:(preamble)#0" def test_section_from_address_basic(self) -> None: sec = _section_from_address("section:2:Background#0", "## Background\n\nbody\n") assert sec.level == 2 assert sec.title == "Background" assert sec.text == "## Background\n\nbody\n" assert sec.sid == "section:2:Background#0" def test_section_from_address_title_with_hash(self) -> None: sec = _section_from_address("section:2:Issue #42#1", "## Issue #42\n") assert sec.level == 2 assert sec.title == "Issue #42" def test_section_from_address_malformed_defaults_safe(self) -> None: sec = _section_from_address("not-a-section", "body") assert sec.level == 0 assert sec.title == "(preamble)" assert sec.text == "body" class TestParseNoteAndCanonicalize: """Unit tests for ``_parse_note`` / ``canonicalize`` invariants.""" def test_parse_empty_note(self) -> None: parsed = _parse_note(b"") assert parsed.frontmatter == {} assert parsed.body == "" def test_parse_no_frontmatter(self) -> None: parsed = _parse_note(b"# Just a heading\n\nbody\n") assert parsed.frontmatter == {} assert parsed.body == "# Just a heading\n\nbody\n" def test_parse_with_frontmatter(self) -> None: parsed = _parse_note(b"---\ntitle: X\n---\n# body\n") assert parsed.frontmatter == {"title": "X"} assert parsed.body == "# body\n" def test_parse_crlf_normalised(self) -> None: parsed = _parse_note(b"---\r\ntitle: X\r\n---\r\nbody\r\n") assert parsed.frontmatter == {"title": "X"} assert "\r" not in parsed.body def test_parse_oversize_raises(self) -> None: oversize = b"x" * (_MAX_NOTE_BYTES + 1) with pytest.raises(ValueError): _parse_note(oversize) def test_canonicalize_idempotent(self) -> None: raw = b"---\ntags:\n- b\n- a\ntitle: X\n---\nbody\n" once = canonicalize(raw) twice = canonicalize(once) assert once == twice def test_canonicalize_sorts_tags_and_hoists_title(self) -> None: raw = b"---\nproject: zeta\ntitle: X\ntags:\n- z\n- a\n---\nbody\n" out = canonicalize(raw) decoded = out.decode("utf-8") # Title appears before project in canonical output assert decoded.index("title:") < decoded.index("project:") # Tags appear sorted (a before z) assert decoded.index("- a") < decoded.index("- z") def test_canonicalize_no_frontmatter_passthrough(self) -> None: raw = b"# H\nbody\n" assert canonicalize(raw) == raw class TestSplitSectionsTyped: """Unit tests for the section splitter with ID assignment.""" def test_empty_body(self) -> None: assert _split_sections_typed("") == [] def test_preamble_only(self) -> None: sections = _split_sections_typed("just text\n") assert len(sections) == 1 assert sections[0].level == 0 assert sections[0].title == "(preamble)" def test_duplicate_titles_get_occurrence_indices(self) -> None: body = "## Foo\n\nbody1\n\n## Foo\n\nbody2\n" sections = _split_sections_typed(body) assert [s.sid for s in sections] == [ "section:2:Foo#0", "section:2:Foo#1", ] def test_concat_round_trip(self) -> None: body = "# H1\n\nbody\n\n## H2\n\nmore\n" sections = _split_sections_typed(body) assert "".join(s.text for s in sections) == body class TestLayer1TitleDiff: """Unit tests for Layer 1 — title diff.""" def test_no_change_no_ops(self) -> None: a = _parse_note(b"---\ntitle: X\n---\n") b = _parse_note(b"---\ntitle: X\n---\n") assert _diff_title(a, b) == [] def test_title_changed_emits_replace(self) -> None: a = _parse_note(b"---\ntitle: X\n---\n") b = _parse_note(b"---\ntitle: Y\n---\n") ops = _diff_title(a, b) assert len(ops) == 1 op = ops[0] assert op["op"] == "replace" assert op["address"] == _TITLE_ADDR assert op["old_summary"] == "X" assert op["new_summary"] == "Y" def test_title_added_old_empty(self) -> None: a = _parse_note(b"---\n{}\n---\n") b = _parse_note(b"---\ntitle: New\n---\n") ops = _diff_title(a, b) assert len(ops) == 1 assert ops[0]["old_summary"] == "" assert ops[0]["new_summary"] == "New" def test_title_removed_new_empty(self) -> None: a = _parse_note(b"---\ntitle: Old\n---\n") b = _parse_note(b"---\nproject: x\n---\n") ops = _diff_title(a, b) assert len(ops) == 1 assert ops[0]["old_summary"] == "Old" assert ops[0]["new_summary"] == "" class TestLayer2FrontmatterDiff: """Unit tests for Layer 2 — frontmatter diff.""" def test_scalar_change_emits_replace(self) -> None: a = _parse_note(b"---\ntitle: X\nproject: alpha\n---\n") b = _parse_note(b"---\ntitle: X\nproject: beta\n---\n") ops = _diff_frontmatter(a, b) assert len(ops) == 1 assert ops[0]["op"] == "replace" assert ops[0]["address"] == f"{_FM_PREFIX}project" assert ops[0]["old_summary"] == "alpha" assert ops[0]["new_summary"] == "beta" def test_scalar_added_emits_insert(self) -> None: a = _parse_note(b"---\ntitle: X\n---\n") b = _parse_note(b"---\ntitle: X\nproject: p\n---\n") ops = _diff_frontmatter(a, b) assert len(ops) == 1 assert ops[0]["op"] == "insert" assert ops[0]["address"] == f"{_FM_PREFIX}project" def test_scalar_removed_emits_delete(self) -> None: a = _parse_note(b"---\ntitle: X\nproject: p\n---\n") b = _parse_note(b"---\ntitle: X\n---\n") ops = _diff_frontmatter(a, b) assert len(ops) == 1 assert ops[0]["op"] == "delete" def test_tag_added(self) -> None: a = _parse_note(b"---\ntags: [a]\n---\n") b = _parse_note(b"---\ntags: [a, b]\n---\n") ops = _diff_frontmatter(a, b) assert len(ops) == 1 assert ops[0]["op"] == "insert" assert ops[0]["content_summary"] == "b" def test_tag_removed(self) -> None: a = _parse_note(b"---\ntags: [a, b]\n---\n") b = _parse_note(b"---\ntags: [a]\n---\n") ops = _diff_frontmatter(a, b) assert len(ops) == 1 assert ops[0]["op"] == "delete" assert ops[0]["content_summary"] == "b" def test_tag_swap_emits_delete_and_insert(self) -> None: a = _parse_note(b"---\ntags: [a]\n---\n") b = _parse_note(b"---\ntags: [b]\n---\n") ops = _diff_frontmatter(a, b) op_kinds = sorted(op["op"] for op in ops) assert op_kinds == ["delete", "insert"] def test_excludes_title(self) -> None: a = _parse_note(b"---\ntitle: X\nproject: p\n---\n") b = _parse_note(b"---\ntitle: Y\nproject: q\n---\n") ops = _diff_frontmatter(a, b) addresses = [op["address"] for op in ops] # Layer 2 must not emit ops for the title — that is Layer 1's job. assert _TITLE_ADDR not in addresses class TestLayer3SectionDiff: """Unit tests for Layer 3 — section diff.""" def test_section_added_at_end(self) -> None: sa = _split_sections_typed("# H\n\nbody\n") sb = _split_sections_typed("# H\n\nbody\n\n## New\n\nnew body\n") ops, stable = _diff_sections(sa, sb) assert any(op["op"] == "insert" and op["address"].startswith(_SECTION_PREFIX) for op in ops) def test_section_removed(self) -> None: sa = _split_sections_typed("# H\n\n## X\n\nbody\n") sb = _split_sections_typed("# H\n") ops, _ = _diff_sections(sa, sb) assert any(op["op"] == "delete" for op in ops) def test_section_move_emits_moveop(self) -> None: # ``split_sections`` assigns each section the text from its heading # up to the next heading. Trailing-newline counts therefore depend # on whether the section is the *last* one or not. To get a clean # move (byte-identical section text on both sides) the moved # sections must live at *internal* positions in both notes — never # at the very end. Below, X and Y swap places but both are # surrounded by A/C/E on either side, so their texts are stable # across the swap and detect_moves collapses the insert+delete # pair into a MoveOp. body_a = "## A\n\n## X\n\n## C\n\n## Y\n\n## E\n" body_b = "## A\n\n## Y\n\n## C\n\n## X\n\n## E\n" sa = _split_sections_typed(body_a) sb = _split_sections_typed(body_b) ops, _ = _diff_sections(sa, sb) kinds = {op["op"] for op in ops} assert "move" in kinds def test_unchanged_sections_marked_stable(self) -> None: body = "# H\n\nbody\n\n## X\n\nbody2\n" sa = _split_sections_typed(body) sb = _split_sections_typed(body) _, stable = _diff_sections(sa, sb) assert "section:1:H#0" in stable assert "section:2:X#0" in stable class TestLayer4BodyLineDiff: """Unit tests for Layer 4 — body line diff.""" def test_no_change_no_ops(self) -> None: assert _diff_body_lines("sid", "a\nb\n", "a\nb\n") == [] def test_line_added(self) -> None: ops = _diff_body_lines("sid", "a\nb", "a\nb\nc") kinds = [op["op"] for op in ops] assert "insert" in kinds def test_line_removed(self) -> None: ops = _diff_body_lines("sid", "a\nb\nc", "a\nc") kinds = [op["op"] for op in ops] assert "delete" in kinds def test_line_changed(self) -> None: ops = _diff_body_lines("sid", "a\nb\nc", "a\nX\nc") kinds = [op["op"] for op in ops] assert "insert" in kinds and "delete" in kinds def test_line_op_address_format(self) -> None: ops = _diff_body_lines("section:2:H#0", "a", "a\nb") for op in ops: assert op["address"].startswith("section:2:H#0") assert _LINE_INFIX in op["address"] # ============================================================================= # Tier 2 — Integration tests # ============================================================================= class TestDiffNotesIntegration: """Integration tests across all four layers.""" def test_combined_changes(self) -> None: a = _build_note( {"title": "Foo", "project": "alpha", "tags": ["a", "b"]}, "# Heading\n\nbody line 1\nbody line 2\n", ) b = _build_note( {"title": "Bar", "project": "alpha", "tags": ["a", "c"]}, "# Heading\n\nbody line 1\nbody line 3\n", ) delta = diff_notes(a, b) assert delta["domain"] == "knowtation" assert delta["ops"] addresses = [op["address"] for op in delta["ops"]] assert _TITLE_ADDR in addresses assert any(addr.startswith(_FM_PREFIX) for addr in addresses) assert any(_LINE_INFIX in addr for addr in addresses) def test_identical_notes_no_ops(self) -> None: note = _build_note({"title": "X"}, "body\n") delta = diff_notes(note, note) assert delta["ops"] == [] assert delta["summary"] == "no changes" def test_summary_is_string(self) -> None: a = _build_note({"title": "A"}, "") b = _build_note({"title": "B"}, "") delta = diff_notes(a, b) assert isinstance(delta["summary"], str) assert delta["summary"] # ============================================================================= # Tier 3 — End-to-end round-trip (20 hand-crafted pairs) # ============================================================================= def _e2e_pairs() -> list[tuple[str, bytes, bytes]]: """Return 20 hand-crafted (label, A, B) note pairs for round-trip tests. Each pair exercises a different edge case of the four-layer differ. Both sides are pre-canonicalised so the round-trip is byte-stable. Returns: List of ``(label, a, b)`` triples. """ pairs: list[tuple[str, bytes, bytes]] = [] pairs.append(("identical-empty", _build_note(None, ""), _build_note(None, ""))) pairs.append(( "title-only-change", _build_note({"title": "Old"}, ""), _build_note({"title": "New"}, ""), )) pairs.append(( "title-added-from-nothing", _build_note(None, "body\n"), _build_note({"title": "Hello"}, "body\n"), )) pairs.append(( "title-removed", _build_note({"title": "Hello", "project": "p"}, "body\n"), _build_note({"project": "p"}, "body\n"), )) pairs.append(( "scalar-frontmatter-change", _build_note({"title": "x", "project": "a"}, "body"), _build_note({"title": "x", "project": "b"}, "body"), )) pairs.append(( "tag-added", _build_note({"title": "x", "tags": ["a"]}, "body"), _build_note({"title": "x", "tags": ["a", "b"]}, "body"), )) pairs.append(( "tag-removed", _build_note({"title": "x", "tags": ["a", "b"]}, "body"), _build_note({"title": "x", "tags": ["a"]}, "body"), )) pairs.append(( "tag-swap", _build_note({"title": "x", "tags": ["a"]}, "body"), _build_note({"title": "x", "tags": ["b"]}, "body"), )) pairs.append(( "entity-list-churn", _build_note({"title": "x", "entity": ["alice", "bob"]}, "body"), _build_note({"title": "x", "entity": ["bob", "carol"]}, "body"), )) pairs.append(( "section-added-at-end", _build_note({"title": "x"}, "# H\n\nbody\n"), _build_note({"title": "x"}, "# H\n\nbody\n\n## New\n\nnew body\n"), )) pairs.append(( "section-removed", _build_note({"title": "x"}, "# H\n\n## Gone\n\nbye\n"), _build_note({"title": "x"}, "# H\n"), )) pairs.append(( "section-body-changed", _build_note({"title": "x"}, "# H\n\nold body line\n"), _build_note({"title": "x"}, "# H\n\nnew body line\n"), )) pairs.append(( "body-line-inserted", _build_note({"title": "x"}, "# H\n\nline 1\nline 3\n"), _build_note({"title": "x"}, "# H\n\nline 1\nline 2\nline 3\n"), )) pairs.append(( "body-line-deleted", _build_note({"title": "x"}, "# H\n\nline 1\nline 2\nline 3\n"), _build_note({"title": "x"}, "# H\n\nline 1\nline 3\n"), )) pairs.append(( "no-frontmatter-both-sides", _build_note(None, "just body\n"), _build_note(None, "just body!\n"), )) pairs.append(( "frontmatter-added-from-none", _build_note(None, "body\n"), _build_note({"title": "x", "project": "p"}, "body\n"), )) pairs.append(( "frontmatter-removed-completely", _build_note({"title": "x", "project": "p"}, "body\n"), _build_note(None, "body\n"), )) pairs.append(( "many-sections-many-edits", _build_note( {"title": "x"}, "## A\n\na\n\n## B\n\nb\n\n## C\n\nc\n\n## D\n\nd\n", ), _build_note( {"title": "x"}, "## A\n\na2\n\n## B\n\nb\n\n## E\n\ne\n\n## D\n\nd2\n", ), )) pairs.append(( "preamble-changed-only", _build_note({"title": "x"}, "preamble line 1\n\n# H\n"), _build_note({"title": "x"}, "preamble line 1 changed\n\n# H\n"), )) pairs.append(( "duplicate-section-titles", _build_note({"title": "x"}, "## Foo\n\nfirst\n\n## Foo\n\nsecond\n"), _build_note({"title": "x"}, "## Foo\n\nfirst-modified\n\n## Foo\n\nsecond\n"), )) assert len(pairs) == 20, "_e2e_pairs must produce exactly 20 pairs" return pairs @pytest.mark.parametrize("label,a,b", _e2e_pairs(), ids=lambda x: x if isinstance(x, str) else "") def test_round_trip_hand_crafted(label: str, a: bytes, b: bytes) -> None: """End-to-end: round-trip every hand-crafted pair.""" _round_trip(a, b) # ============================================================================= # Tier 3b — Regression tests for Phase 2.1.1 duplicate-section MoveOp bug # ============================================================================= class TestDuplicateSectionMoveOpRegression: """Regression suite for the detect_moves duplicate-content-id bug (Phase 2.1.1). Root cause: ``detect_moves`` in ``muse.core.diff_algorithms.lcs`` tracked consumed content IDs rather than consumed *objects*. When two sections shared identical text (same ``content_id``), the second delete was wrongly removed from ``remaining_deletes``, causing the section to survive the ``apply()`` call and breaking the round-trip invariant. Fix: ``paired_delete_ids`` now uses ``id()`` of specific ``DeleteOp`` objects so only the exact object that became a ``MoveOp`` is excluded from ``remaining_deletes``. Every test here follows the ``apply(diff_notes(A, B), A) == B`` invariant after canonicalising both sides. """ def test_hypothesis_failing_seed_exact(self) -> None: """Exact Hypothesis counter-example: H2 + two identical H1 → H1 + H2.""" a = b"---\ntitle: A\n---\n## A\n\n\n# A\n\n\n# A\n\n\n" b = b"---\ntitle: A\n---\n# A\n\n\n## A\n\n\n" _round_trip(canonicalize(a), canonicalize(b)) def test_three_identical_sections_reduce_to_two(self) -> None: """Three consecutive identical sections reduced to two.""" a = _build_note({"title": "T"}, "# X\n\nline\n\n# X\n\nline\n\n# X\n\nline\n\n") b = _build_note({"title": "T"}, "# X\n\nline\n\n# X\n\nline\n\n") _round_trip(a, b) def test_three_identical_sections_reduce_to_one(self) -> None: """Three consecutive identical sections reduced to one.""" a = _build_note({"title": "T"}, "# X\n\nline\n\n# X\n\nline\n\n# X\n\nline\n\n") b = _build_note({"title": "T"}, "# X\n\nline\n\n") _round_trip(a, b) def test_duplicate_sections_reordered(self) -> None: """Two identical H1 sections swapped around a different H2 section.""" a = _build_note({"title": "T"}, "# Same\n\nbody\n\n## Diff\n\ndiff\n\n# Same\n\nbody\n\n") b = _build_note({"title": "T"}, "# Same\n\nbody\n\n# Same\n\nbody\n\n## Diff\n\ndiff\n\n") _round_trip(a, b) def test_two_distinct_sections_swapped(self) -> None: """Two sections with different content (no duplicate) still round-trip.""" a = _build_note({"title": "T"}, "# A\n\nalpha\n\n# B\n\nbeta\n\n") b = _build_note({"title": "T"}, "# B\n\nbeta\n\n# A\n\nalpha\n\n") _round_trip(a, b) def test_expand_one_duplicate_to_three(self) -> None: """Going from one section to three identical copies (inverse direction).""" a = _build_note({"title": "T"}, "# X\n\ntext\n\n") b = _build_note({"title": "T"}, "# X\n\ntext\n\n# X\n\ntext\n\n# X\n\ntext\n\n") _round_trip(a, b) def test_all_sections_deleted_duplicate_content(self) -> None: """All sections deleted when source has duplicates and target is empty body.""" a = _build_note({"title": "T"}, "# X\n\nfoo\n\n# X\n\nfoo\n\n") b = _build_note({"title": "T"}, "") _round_trip(a, b) def test_detect_moves_does_not_consume_sibling_deletes(self) -> None: """Directly verify detect_moves leaves sibling deletes intact. When two DeleteOps share a content_id and one InsertOp with the same content_id exists, detect_moves must produce exactly one MoveOp and leave the second delete in remaining_deletes. """ from muse.core.diff_algorithms.lcs import detect_moves from muse.domain import DeleteOp, InsertOp cid = "aaaa" * 16 # 64-char fake content_id d1 = DeleteOp(op="delete", address="section:1:X#0", position=1, content_id=cid, content_summary="# X\n\n") d2 = DeleteOp(op="delete", address="section:1:X#1", position=2, content_id=cid, content_summary="# X\n\n") i1 = InsertOp(op="insert", address="section:1:X#0", position=0, content_id=cid, content_summary="# X\n\n") moves, rem_inserts, rem_deletes = detect_moves([i1], [d1, d2]) assert len(moves) == 1, f"Expected 1 MoveOp, got {len(moves)}" assert moves[0]["from_position"] == 1 assert moves[0]["to_position"] == 0 assert rem_inserts == [], f"No inserts should remain, got {rem_inserts}" assert len(rem_deletes) == 1, f"Expected 1 remaining delete, got {rem_deletes}" assert rem_deletes[0]["address"] == "section:1:X#1" assert rem_deletes[0]["position"] == 2 def test_detect_moves_two_pairs_same_content(self) -> None: """Two inserts and two deletes with identical content: first insert gets MoveOp, second insert and second delete remain unpaired. """ from muse.core.diff_algorithms.lcs import detect_moves from muse.domain import DeleteOp, InsertOp cid = "bbbb" * 16 d1 = DeleteOp(op="delete", address="section:1:Y#0", position=0, content_id=cid, content_summary="# Y\n\n") d2 = DeleteOp(op="delete", address="section:1:Y#1", position=3, content_id=cid, content_summary="# Y\n\n") i1 = InsertOp(op="insert", address="section:1:Y#0", position=2, content_id=cid, content_summary="# Y\n\n") i2 = InsertOp(op="insert", address="section:1:Y#1", position=4, content_id=cid, content_summary="# Y\n\n") moves, rem_inserts, rem_deletes = detect_moves([i1, i2], [d1, d2]) # Only the first insert pairs with the first delete (first-come first-served). assert len(moves) == 1 assert moves[0]["from_position"] == 0 assert moves[0]["to_position"] == 2 # Second insert and second delete remain unpaired. assert len(rem_inserts) == 1 assert rem_inserts[0]["position"] == 4 assert len(rem_deletes) == 1 assert rem_deletes[0]["position"] == 3 # ============================================================================= # Tier 4 — Data-integrity fuzz with hypothesis # ============================================================================= _PRINTABLE = string.ascii_letters + string.digits + " " _slug_st = st.text( alphabet=string.ascii_lowercase + string.digits, min_size=1, max_size=12 ) _word_st = st.text(alphabet=_PRINTABLE, min_size=0, max_size=40) def _line_st() -> st.SearchStrategy[str]: """Strategy producing a single body line (no embedded newlines).""" return _word_st def _heading_st() -> st.SearchStrategy[str]: """Strategy producing a Markdown heading line.""" return st.tuples( st.integers(min_value=1, max_value=4), st.text(alphabet=string.ascii_letters + " ", min_size=1, max_size=12).map( lambda s: s.strip() or "Heading" ), ).map(lambda lt: "#" * lt[0] + " " + lt[1]) def _section_st() -> st.SearchStrategy[str]: """Strategy producing a complete section (heading + body lines + trailing newline).""" return st.tuples( _heading_st(), st.lists(_line_st(), min_size=0, max_size=4), ).map(lambda hb: hb[0] + "\n\n" + "\n".join(hb[1]) + "\n") def _body_st() -> st.SearchStrategy[str]: """Strategy producing a Markdown body of 0–4 sections.""" return st.lists(_section_st(), min_size=0, max_size=4).map("".join) def _frontmatter_st() -> st.SearchStrategy[dict[str, Any]]: """Strategy producing a small canonical frontmatter mapping.""" return st.fixed_dictionaries( { "title": st.text( alphabet=string.ascii_letters + " ", min_size=1, max_size=12 ).map(lambda s: s.strip() or "Untitled"), }, optional={ "project": _slug_st, "tags": st.lists(_slug_st, min_size=0, max_size=4, unique=True), "entity": st.lists(_slug_st, min_size=0, max_size=4, unique=True), }, ) def _note_st() -> st.SearchStrategy[bytes]: """Strategy producing canonical-form note bytes.""" return st.tuples(_frontmatter_st(), _body_st()).map( lambda fb: _build_note(fb[0], fb[1]) ) @settings( max_examples=200, deadline=None, suppress_health_check=[HealthCheck.too_slow, HealthCheck.filter_too_much], ) @given(a=_note_st(), b=_note_st()) def test_fuzz_round_trip_invariant(a: bytes, b: bytes) -> None: """Hypothesis fuzz: ``apply(diff_notes(A, B), A) == B`` for random pairs.""" _round_trip(a, b) # ============================================================================= # Tier 5 — Performance # ============================================================================= def _build_50_section_note(seed: int) -> bytes: """Build a deterministic 50-section note (sized realistically). Each section has ~10 body lines, yielding a ~3 KB total note. *seed* perturbs section bodies so two builds with different seeds produce different but structurally-similar notes. Args: seed: Integer that perturbs the section bodies. Returns: Canonical-form note bytes. """ parts: list[str] = [] for i in range(50): parts.append(f"## Section {i}\n\n") for j in range(10): parts.append(f"line {i}-{j}-{seed % 7}\n") parts.append("\n") body = "".join(parts) return _build_note({"title": f"Note {seed}", "tags": ["perf"]}, body) class TestPerformance: """Tier 5 — diff of a realistic 50-section note must finish in < 100 ms.""" def test_50_section_note_under_100ms(self) -> None: a = _build_50_section_note(seed=1) b = _build_50_section_note(seed=2) start = time.perf_counter() delta = diff_notes(a, b) elapsed = time.perf_counter() - start assert delta["ops"] assert elapsed < 0.5, ( f"diff_notes(50-section) took {elapsed * 1000:.1f} ms " f"(soft target < 100 ms, hard fail > 500 ms)" ) # ============================================================================= # Tier 6 — Stress # ============================================================================= class TestStress: """Tier 6 — 200 sequential diff+apply cycles complete in under 5 s.""" def test_200_note_pairs_under_5s(self) -> None: pairs = [] for i in range(200): a = _build_note( {"title": f"N{i}", "tags": [f"t{i % 5}"]}, f"# H{i}\n\nline a {i}\nline b\n", ) b = _build_note( {"title": f"N{i}", "tags": [f"t{(i + 1) % 5}"]}, f"# H{i}\n\nline a {i}\nline b modified\n", ) pairs.append((a, b)) start = time.perf_counter() for a, b in pairs: delta = diff_notes(a, b) out = apply(delta, a) assert out == b elapsed = time.perf_counter() - start assert elapsed < 5.0, ( f"200 diff+apply cycles took {elapsed:.2f} s (limit 5 s)" ) # ============================================================================= # Tier 7 — Security # ============================================================================= class TestSecurity: """Tier 7 — adversarial inputs must not crash, leak, or hang.""" def test_binary_bytes_do_not_crash(self) -> None: a = bytes(range(256)) b = bytes(reversed(range(256))) delta = diff_notes(a, b) assert isinstance(delta["ops"], list) out = apply(delta, a) assert isinstance(out, bytes) def test_nul_bytes_in_body_safe(self) -> None: a = b"---\ntitle: X\n---\nhello\x00world\n" b = b"---\ntitle: Y\n---\nhello\x00world\n" # Should not raise; apply may produce a normalised re-serialisation # (NUL bytes survive the YAML / body path because we never decode # strictly). delta = diff_notes(a, b) assert any(op["address"] == _TITLE_ADDR for op in delta["ops"]) out = apply(delta, a) assert isinstance(out, bytes) def test_very_long_line_safe(self) -> None: long_line = "x" * 200_000 a = _build_note({"title": "L"}, f"# H\n\n{long_line}\n") b = _build_note({"title": "L"}, f"# H\n\n{long_line}y\n") _round_trip(a, b) def test_deeply_nested_yaml_safe(self) -> None: # PyYAML's safe_load accepts deep mappings; we should pass them # through without recursing into arbitrary Python objects. nested: Any = {"deep": 1} for _ in range(50): nested = {"k": nested} a = _build_note({"title": "X", "nested": nested}, "body\n") b = _build_note({"title": "Y", "nested": nested}, "body\n") delta = diff_notes(a, b) assert any(op["address"] == _TITLE_ADDR for op in delta["ops"]) out = apply(delta, a) # The nested structure round-trips because canonicalize stabilises it. assert out == b def test_oversize_input_raises_value_error(self) -> None: too_big = b"x" * (_MAX_NOTE_BYTES + 1) with pytest.raises(ValueError): diff_notes(too_big, b"") with pytest.raises(ValueError): apply({"domain": "knowtation", "ops": []}, too_big) # type: ignore[arg-type] def test_yaml_safe_load_only(self) -> None: # !!python/object exploit attempts must not instantiate arbitrary # Python objects. yaml.safe_load raises a YAMLError, which our # _parse_note catches and downgrades to "no frontmatter". The body # text falls through unchanged. evil = ( b"---\n" b"!!python/object/apply:os.system ['echo pwned']\n" b"---\n" b"body\n" ) parsed = _parse_note(evil) # Either the YAML parse failed entirely (frontmatter empty) or the # crafted node was rejected — in neither case is a Python object # instantiated. The important assertion is: no os.system call. assert isinstance(parsed.frontmatter, dict) # Round-trip with self must always succeed (idempotency). delta = diff_notes(evil, evil) assert delta["ops"] == [] def test_malformed_unclosed_frontmatter_safe(self) -> None: a = b"---\ntitle: open\n" # never closes b = b"---\ntitle: open\nthen more\n" delta = diff_notes(a, b) out = apply(delta, a) assert isinstance(out, bytes)