test_addressed_merge_plugin.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Addressed-merge plugin — naming correctness and op-type structural tests. |
| 2 | |
| 3 | The code domain uses address-keyed (Map CRDT) merge semantics, not classical OT. |
| 4 | Two operations on different symbol addresses commute automatically; same-address |
| 5 | conflicts surface to Harmony. This is a Map CRDT, not an Operational |
| 6 | Transformation system. |
| 7 | |
| 8 | Coverage matrix |
| 9 | --------------- |
| 10 | A Protocol naming |
| 11 | A1 AddressedMergePlugin exists and is importable from muse.domain |
| 12 | A2 StructuredMergePlugin does NOT exist in muse.domain (no backward-compat alias) |
| 13 | A3 CodePlugin satisfies AddressedMergePlugin via isinstance |
| 14 | A4 MidiPlugin satisfies AddressedMergePlugin via isinstance |
| 15 | A5 IdentityPlugin satisfies AddressedMergePlugin via isinstance |
| 16 | |
| 17 | B AddressedInsertOp — no position field |
| 18 | B1 AddressedInsertOp is importable from muse.domain |
| 19 | B2 AddressedInsertOp instance has keys: op, address, content_id, content_summary |
| 20 | B3 AddressedInsertOp instance has NO position key |
| 21 | B4 AddressedInsertOp.__required_keys__ does not include position |
| 22 | B5 Literal["insert"] is the op value |
| 23 | |
| 24 | C AddressedDeleteOp — no position field |
| 25 | C1 AddressedDeleteOp is importable from muse.domain |
| 26 | C2 AddressedDeleteOp instance has keys: op, address, content_id, content_summary |
| 27 | C3 AddressedDeleteOp instance has NO position key |
| 28 | C4 AddressedDeleteOp.__required_keys__ does not include position |
| 29 | C5 Literal["delete"] is the op value |
| 30 | |
| 31 | D Sequence ops retain position — InsertOp / DeleteOp unchanged |
| 32 | D1 InsertOp still has position in its required keys |
| 33 | D2 DeleteOp still has position in its required keys |
| 34 | D3 InsertOp and AddressedInsertOp are distinct types |
| 35 | |
| 36 | E Code plugin emits AddressedInsertOp / AddressedDeleteOp (no position in output) |
| 37 | E1 CodePlugin.diff() file-level added op has no position key |
| 38 | E2 CodePlugin.diff() file-level removed op has no position key |
| 39 | E3 muse read --json structured delta ops contain no position key for code commits |
| 40 | """ |
| 41 | |
| 42 | from __future__ import annotations |
| 43 | |
| 44 | import json |
| 45 | import pathlib |
| 46 | |
| 47 | import pytest |
| 48 | |
| 49 | # --------------------------------------------------------------------------- |
| 50 | # A — Protocol naming |
| 51 | # --------------------------------------------------------------------------- |
| 52 | |
| 53 | class TestProtocolNaming: |
| 54 | def test_A1_addressed_merge_plugin_importable(self) -> None: |
| 55 | """A1: AddressedMergePlugin exists in muse.domain.""" |
| 56 | from muse.domain import AddressedMergePlugin # noqa: F401 |
| 57 | |
| 58 | def test_A2_structured_merge_plugin_does_not_exist(self) -> None: |
| 59 | """A2: StructuredMergePlugin is not exported — no backward-compat alias.""" |
| 60 | import muse.domain as domain_module |
| 61 | assert not hasattr(domain_module, "StructuredMergePlugin"), ( |
| 62 | "StructuredMergePlugin must not exist — use AddressedMergePlugin" |
| 63 | ) |
| 64 | |
| 65 | def test_A3_code_plugin_satisfies_addressed_merge_plugin(self) -> None: |
| 66 | """A3: CodePlugin satisfies AddressedMergePlugin at runtime.""" |
| 67 | from muse.domain import AddressedMergePlugin |
| 68 | from muse.plugins.code.plugin import plugin as code_plugin |
| 69 | assert isinstance(code_plugin, AddressedMergePlugin), ( |
| 70 | "CodePlugin must satisfy AddressedMergePlugin" |
| 71 | ) |
| 72 | |
| 73 | def test_A4_midi_plugin_satisfies_addressed_merge_plugin(self) -> None: |
| 74 | """A4: MidiPlugin satisfies AddressedMergePlugin at runtime.""" |
| 75 | from muse.domain import AddressedMergePlugin |
| 76 | from muse.plugins.midi.plugin import plugin as midi_plugin |
| 77 | assert isinstance(midi_plugin, AddressedMergePlugin), ( |
| 78 | "MidiPlugin must satisfy AddressedMergePlugin" |
| 79 | ) |
| 80 | |
| 81 | def test_A5_identity_plugin_satisfies_addressed_merge_plugin(self) -> None: |
| 82 | """A5: IdentityPlugin satisfies AddressedMergePlugin at runtime.""" |
| 83 | from muse.domain import AddressedMergePlugin |
| 84 | from muse.plugins.identity.plugin import IdentityPlugin |
| 85 | assert isinstance(IdentityPlugin(), AddressedMergePlugin), ( |
| 86 | "IdentityPlugin must satisfy AddressedMergePlugin" |
| 87 | ) |
| 88 | |
| 89 | |
| 90 | # --------------------------------------------------------------------------- |
| 91 | # B — AddressedInsertOp |
| 92 | # --------------------------------------------------------------------------- |
| 93 | |
| 94 | class TestAddressedInsertOp: |
| 95 | def test_B1_importable(self) -> None: |
| 96 | """B1: AddressedInsertOp is importable from muse.domain.""" |
| 97 | from muse.domain import AddressedInsertOp # noqa: F401 |
| 98 | |
| 99 | def test_B2_required_keys(self) -> None: |
| 100 | """B2: AddressedInsertOp has exactly the expected keys.""" |
| 101 | from muse.domain import AddressedInsertOp |
| 102 | op = AddressedInsertOp( |
| 103 | op="insert", |
| 104 | address="src/billing.py::compute_total", |
| 105 | content_id="sha256:abc123", |
| 106 | content_summary="added compute_total", |
| 107 | ) |
| 108 | assert set(op.keys()) == {"op", "address", "content_id", "content_summary"} |
| 109 | |
| 110 | def test_B3_no_position_key_in_instance(self) -> None: |
| 111 | """B3: an AddressedInsertOp instance has no 'position' key.""" |
| 112 | from muse.domain import AddressedInsertOp |
| 113 | op = AddressedInsertOp( |
| 114 | op="insert", |
| 115 | address="src/billing.py::compute_total", |
| 116 | content_id="sha256:abc123", |
| 117 | content_summary="added compute_total", |
| 118 | ) |
| 119 | assert "position" not in op, ( |
| 120 | "AddressedInsertOp must not have a position key — " |
| 121 | "code symbols are name-addressed, not position-indexed" |
| 122 | ) |
| 123 | |
| 124 | def test_B4_no_position_in_required_keys(self) -> None: |
| 125 | """B4: 'position' is absent from AddressedInsertOp.__required_keys__.""" |
| 126 | from muse.domain import AddressedInsertOp |
| 127 | all_keys = AddressedInsertOp.__required_keys__ | AddressedInsertOp.__optional_keys__ |
| 128 | assert "position" not in all_keys, ( |
| 129 | "position must not be declared on AddressedInsertOp at all" |
| 130 | ) |
| 131 | |
| 132 | def test_B5_op_literal(self) -> None: |
| 133 | """B5: op field value is 'insert'.""" |
| 134 | from muse.domain import AddressedInsertOp |
| 135 | op = AddressedInsertOp( |
| 136 | op="insert", |
| 137 | address="src/billing.py::compute_total", |
| 138 | content_id="sha256:abc123", |
| 139 | content_summary="added compute_total", |
| 140 | ) |
| 141 | assert op["op"] == "insert" |
| 142 | |
| 143 | |
| 144 | # --------------------------------------------------------------------------- |
| 145 | # C — AddressedDeleteOp |
| 146 | # --------------------------------------------------------------------------- |
| 147 | |
| 148 | class TestAddressedDeleteOp: |
| 149 | def test_C1_importable(self) -> None: |
| 150 | """C1: AddressedDeleteOp is importable from muse.domain.""" |
| 151 | from muse.domain import AddressedDeleteOp # noqa: F401 |
| 152 | |
| 153 | def test_C2_required_keys(self) -> None: |
| 154 | """C2: AddressedDeleteOp has exactly the expected keys.""" |
| 155 | from muse.domain import AddressedDeleteOp |
| 156 | op = AddressedDeleteOp( |
| 157 | op="delete", |
| 158 | address="src/billing.py::old_fn", |
| 159 | content_id="sha256:dead", |
| 160 | content_summary="removed old_fn", |
| 161 | ) |
| 162 | assert set(op.keys()) == {"op", "address", "content_id", "content_summary"} |
| 163 | |
| 164 | def test_C3_no_position_key_in_instance(self) -> None: |
| 165 | """C3: an AddressedDeleteOp instance has no 'position' key.""" |
| 166 | from muse.domain import AddressedDeleteOp |
| 167 | op = AddressedDeleteOp( |
| 168 | op="delete", |
| 169 | address="src/billing.py::old_fn", |
| 170 | content_id="sha256:dead", |
| 171 | content_summary="removed old_fn", |
| 172 | ) |
| 173 | assert "position" not in op, ( |
| 174 | "AddressedDeleteOp must not have a position key" |
| 175 | ) |
| 176 | |
| 177 | def test_C4_no_position_in_required_keys(self) -> None: |
| 178 | """C4: 'position' is absent from AddressedDeleteOp type declaration.""" |
| 179 | from muse.domain import AddressedDeleteOp |
| 180 | all_keys = AddressedDeleteOp.__required_keys__ | AddressedDeleteOp.__optional_keys__ |
| 181 | assert "position" not in all_keys |
| 182 | |
| 183 | def test_C5_op_literal(self) -> None: |
| 184 | """C5: op field value is 'delete'.""" |
| 185 | from muse.domain import AddressedDeleteOp |
| 186 | op = AddressedDeleteOp( |
| 187 | op="delete", |
| 188 | address="src/billing.py::old_fn", |
| 189 | content_id="sha256:dead", |
| 190 | content_summary="removed old_fn", |
| 191 | ) |
| 192 | assert op["op"] == "delete" |
| 193 | |
| 194 | |
| 195 | # --------------------------------------------------------------------------- |
| 196 | # D — Sequence ops (InsertOp / DeleteOp) retain position — unchanged |
| 197 | # --------------------------------------------------------------------------- |
| 198 | |
| 199 | class TestSequenceOpsRetainPosition: |
| 200 | def test_D1_insert_op_has_position(self) -> None: |
| 201 | """D1: InsertOp (for ordered sequences) still has position.""" |
| 202 | from muse.domain import InsertOp |
| 203 | all_keys = InsertOp.__required_keys__ | InsertOp.__optional_keys__ |
| 204 | assert "position" in all_keys, ( |
| 205 | "InsertOp is for ordered-sequence domains (MIDI) and must retain position" |
| 206 | ) |
| 207 | |
| 208 | def test_D2_delete_op_has_position(self) -> None: |
| 209 | """D2: DeleteOp (for ordered sequences) still has position.""" |
| 210 | from muse.domain import DeleteOp |
| 211 | all_keys = DeleteOp.__required_keys__ | DeleteOp.__optional_keys__ |
| 212 | assert "position" in all_keys, ( |
| 213 | "DeleteOp is for ordered-sequence domains (MIDI) and must retain position" |
| 214 | ) |
| 215 | |
| 216 | def test_D3_addressed_and_sequence_insert_are_distinct(self) -> None: |
| 217 | """D3: AddressedInsertOp and InsertOp are distinct types.""" |
| 218 | from muse.domain import AddressedInsertOp, InsertOp |
| 219 | assert AddressedInsertOp is not InsertOp, ( |
| 220 | "AddressedInsertOp and InsertOp must be distinct TypedDicts" |
| 221 | ) |
| 222 | |
| 223 | |
| 224 | # --------------------------------------------------------------------------- |
| 225 | # E — Code plugin emits AddressedInsertOp / AddressedDeleteOp |
| 226 | # --------------------------------------------------------------------------- |
| 227 | |
| 228 | class TestCodePluginAddressedOps: |
| 229 | @pytest.fixture() |
| 230 | def two_file_repo(self, tmp_path: pathlib.Path) -> tuple[pathlib.Path, dict, dict]: |
| 231 | """Two minimal snapshots: base has file_a, target adds file_b, removes file_a.""" |
| 232 | from muse.domain import SnapshotManifest |
| 233 | base = SnapshotManifest( |
| 234 | files={"file_a.py": "sha256:" + "a" * 64}, |
| 235 | domain="code", |
| 236 | directories=[], |
| 237 | ) |
| 238 | target = SnapshotManifest( |
| 239 | files={"file_b.py": "sha256:" + "b" * 64}, |
| 240 | domain="code", |
| 241 | directories=[], |
| 242 | ) |
| 243 | return tmp_path, base, target |
| 244 | |
| 245 | def test_E1_added_file_op_has_no_position( |
| 246 | self, two_file_repo: tuple[pathlib.Path, dict, dict] |
| 247 | ) -> None: |
| 248 | """E1: code plugin file-level insert op has no 'position' key.""" |
| 249 | from muse.plugins.code.plugin import plugin as code_plugin |
| 250 | _, base, target = two_file_repo |
| 251 | delta = code_plugin.diff(base, target) |
| 252 | insert_ops = [o for o in delta["ops"] if o.get("op") == "insert"] |
| 253 | assert insert_ops, "expected at least one insert op" |
| 254 | for op in insert_ops: |
| 255 | assert "position" not in op, ( |
| 256 | f"code insert op for '{op.get('address')}' must not have position" |
| 257 | ) |
| 258 | |
| 259 | def test_E2_removed_file_op_has_no_position( |
| 260 | self, two_file_repo: tuple[pathlib.Path, dict, dict] |
| 261 | ) -> None: |
| 262 | """E2: code plugin file-level delete op has no 'position' key.""" |
| 263 | from muse.plugins.code.plugin import plugin as code_plugin |
| 264 | _, base, target = two_file_repo |
| 265 | delta = code_plugin.diff(base, target) |
| 266 | delete_ops = [o for o in delta["ops"] if o.get("op") == "delete"] |
| 267 | assert delete_ops, "expected at least one delete op" |
| 268 | for op in delete_ops: |
| 269 | assert "position" not in op, ( |
| 270 | f"code delete op for '{op.get('address')}' must not have position" |
| 271 | ) |
| 272 | |
| 273 | def test_E3_muse_read_json_has_no_position_in_code_ops( |
| 274 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch |
| 275 | ) -> None: |
| 276 | """E3: muse read --json for a code commit has no position key in any op.""" |
| 277 | import subprocess |
| 278 | import sys |
| 279 | monkeypatch.chdir(tmp_path) |
| 280 | env = {"MUSE_REPO_ROOT": str(tmp_path)} |
| 281 | |
| 282 | from tests.cli_test_helper import CliRunner |
| 283 | runner = CliRunner() |
| 284 | |
| 285 | def run(*args: str) -> str: |
| 286 | result = runner.invoke(None, list(args), env=env) |
| 287 | assert result.exit_code == 0, f"{args} failed:\n{result.output}" |
| 288 | return result.output.strip() |
| 289 | |
| 290 | run("init", "--domain", "code") |
| 291 | (tmp_path / "hello.py").write_text("def greet(): pass\n") |
| 292 | run("code", "add", ".") |
| 293 | run("commit", "-m", "initial") |
| 294 | (tmp_path / "world.py").write_text("def world(): pass\n") |
| 295 | run("code", "add", ".") |
| 296 | run("commit", "-m", "add world") |
| 297 | |
| 298 | output = run("read", "--json") |
| 299 | data = json.loads(output) |
| 300 | |
| 301 | def collect_ops(ops: list[dict]) -> list[dict]: |
| 302 | result = [] |
| 303 | for op in ops: |
| 304 | result.append(op) |
| 305 | for child in op.get("child_ops", []): |
| 306 | result.append(child) |
| 307 | return result |
| 308 | |
| 309 | all_ops = collect_ops(data.get("structured_delta", {}).get("ops", [])) |
| 310 | for op in all_ops: |
| 311 | assert "position" not in op, ( |
| 312 | f"code-domain op '{op.get('op')} @ {op.get('address')}' " |
| 313 | f"must not have position — found: {op}" |
| 314 | ) |