test_structured_delta.py
python
| 1 | """Tests for the structured delta type system. |
| 2 | |
| 3 | Covers: |
| 4 | - All five DomainOp TypedDicts can be constructed and serialised to JSON. |
| 5 | - StructuredDelta satisfies the StateDelta type alias. |
| 6 | - MidiPlugin.diff() returns a StructuredDelta with correctly typed ops. |
| 7 | - PatchOp wraps note-level child_ops for modified .mid files. |
| 8 | - DriftReport.delta is a StructuredDelta. |
| 9 | - muse show and muse diff display structured output. |
| 10 | """ |
| 11 | |
| 12 | import json |
| 13 | import pathlib |
| 14 | |
| 15 | import pytest |
| 16 | |
| 17 | from muse.domain import ( |
| 18 | DeleteOp, |
| 19 | DomainOp, |
| 20 | DriftReport, |
| 21 | InsertOp, |
| 22 | MoveOp, |
| 23 | PatchOp, |
| 24 | ReplaceOp, |
| 25 | SnapshotManifest, |
| 26 | StateDelta, |
| 27 | StructuredDelta, |
| 28 | ) |
| 29 | from muse.plugins.midi.plugin import MidiPlugin, plugin |
| 30 | |
| 31 | |
| 32 | # --------------------------------------------------------------------------- |
| 33 | # Helpers |
| 34 | # --------------------------------------------------------------------------- |
| 35 | |
| 36 | def _snap(files: dict[str, str]) -> SnapshotManifest: |
| 37 | return SnapshotManifest(files=files, domain="midi") |
| 38 | |
| 39 | |
| 40 | def _make_insert(address: str = "a.mid", content_id: str = "abc123") -> InsertOp: |
| 41 | return InsertOp( |
| 42 | op="insert", |
| 43 | address=address, |
| 44 | position=None, |
| 45 | content_id=content_id, |
| 46 | content_summary=f"new file: {address}", |
| 47 | ) |
| 48 | |
| 49 | |
| 50 | def _make_delete(address: str = "a.mid", content_id: str = "abc123") -> DeleteOp: |
| 51 | return DeleteOp( |
| 52 | op="delete", |
| 53 | address=address, |
| 54 | position=None, |
| 55 | content_id=content_id, |
| 56 | content_summary=f"deleted: {address}", |
| 57 | ) |
| 58 | |
| 59 | |
| 60 | def _make_move() -> MoveOp: |
| 61 | return MoveOp( |
| 62 | op="move", |
| 63 | address="note:5", |
| 64 | from_position=5, |
| 65 | to_position=12, |
| 66 | content_id="deadbeef", |
| 67 | ) |
| 68 | |
| 69 | |
| 70 | def _make_replace(address: str = "a.mid") -> ReplaceOp: |
| 71 | return ReplaceOp( |
| 72 | op="replace", |
| 73 | address=address, |
| 74 | position=None, |
| 75 | old_content_id="old123", |
| 76 | new_content_id="new456", |
| 77 | old_summary=f"{address} (prev)", |
| 78 | new_summary=f"{address} (new)", |
| 79 | ) |
| 80 | |
| 81 | |
| 82 | def _make_patch(child_ops: list[DomainOp] | None = None) -> PatchOp: |
| 83 | return PatchOp( |
| 84 | op="patch", |
| 85 | address="tracks/drums.mid", |
| 86 | child_ops=child_ops or [], |
| 87 | child_domain="midi_notes", |
| 88 | child_summary="2 notes added", |
| 89 | ) |
| 90 | |
| 91 | |
| 92 | def _make_delta(ops: list[DomainOp] | None = None) -> StructuredDelta: |
| 93 | return StructuredDelta( |
| 94 | domain="midi", |
| 95 | ops=ops or [], |
| 96 | summary="no changes", |
| 97 | ) |
| 98 | |
| 99 | |
| 100 | # --------------------------------------------------------------------------- |
| 101 | # TypedDict construction and JSON round-trips |
| 102 | # --------------------------------------------------------------------------- |
| 103 | |
| 104 | class TestDeltaOpTypes: |
| 105 | def test_insert_op_has_correct_discriminant(self) -> None: |
| 106 | op = _make_insert() |
| 107 | assert op["op"] == "insert" |
| 108 | |
| 109 | def test_delete_op_has_correct_discriminant(self) -> None: |
| 110 | op = _make_delete() |
| 111 | assert op["op"] == "delete" |
| 112 | |
| 113 | def test_move_op_has_correct_discriminant(self) -> None: |
| 114 | op = _make_move() |
| 115 | assert op["op"] == "move" |
| 116 | |
| 117 | def test_replace_op_has_correct_discriminant(self) -> None: |
| 118 | op = _make_replace() |
| 119 | assert op["op"] == "replace" |
| 120 | |
| 121 | def test_patch_op_has_correct_discriminant(self) -> None: |
| 122 | op = _make_patch() |
| 123 | assert op["op"] == "patch" |
| 124 | |
| 125 | def test_insert_op_round_trips_json(self) -> None: |
| 126 | op = _make_insert() |
| 127 | serialised = json.dumps(op) |
| 128 | restored = json.loads(serialised) |
| 129 | assert restored["op"] == "insert" |
| 130 | assert restored["address"] == "a.mid" |
| 131 | assert restored["position"] is None |
| 132 | assert restored["content_id"] == "abc123" |
| 133 | |
| 134 | def test_delete_op_round_trips_json(self) -> None: |
| 135 | op = _make_delete() |
| 136 | serialised = json.dumps(op) |
| 137 | restored = json.loads(serialised) |
| 138 | assert restored["op"] == "delete" |
| 139 | assert restored["address"] == "a.mid" |
| 140 | |
| 141 | def test_move_op_round_trips_json(self) -> None: |
| 142 | op = _make_move() |
| 143 | serialised = json.dumps(op) |
| 144 | restored = json.loads(serialised) |
| 145 | assert restored["op"] == "move" |
| 146 | assert restored["from_position"] == 5 |
| 147 | assert restored["to_position"] == 12 |
| 148 | |
| 149 | def test_replace_op_round_trips_json(self) -> None: |
| 150 | op = _make_replace() |
| 151 | serialised = json.dumps(op) |
| 152 | restored = json.loads(serialised) |
| 153 | assert restored["op"] == "replace" |
| 154 | assert restored["old_content_id"] == "old123" |
| 155 | assert restored["new_content_id"] == "new456" |
| 156 | |
| 157 | def test_patch_op_with_child_ops_round_trips_json(self) -> None: |
| 158 | insert = _make_insert("note:3", "aabbcc") |
| 159 | patch = _make_patch(child_ops=[insert]) |
| 160 | serialised = json.dumps(patch) |
| 161 | restored = json.loads(serialised) |
| 162 | assert restored["op"] == "patch" |
| 163 | assert len(restored["child_ops"]) == 1 |
| 164 | assert restored["child_ops"][0]["op"] == "insert" |
| 165 | |
| 166 | def test_structured_delta_round_trips_json(self) -> None: |
| 167 | delta = _make_delta(ops=[_make_insert(), _make_delete("b.mid", "xyz")]) |
| 168 | serialised = json.dumps(delta) |
| 169 | restored = json.loads(serialised) |
| 170 | assert restored["domain"] == "midi" |
| 171 | assert len(restored["ops"]) == 2 |
| 172 | assert restored["summary"] == "no changes" |
| 173 | |
| 174 | def test_structured_delta_is_state_delta_type(self) -> None: |
| 175 | delta: StateDelta = _make_delta() |
| 176 | assert delta["domain"] == "midi" |
| 177 | |
| 178 | def test_structured_delta_has_required_keys(self) -> None: |
| 179 | delta = _make_delta() |
| 180 | assert "domain" in delta |
| 181 | assert "ops" in delta |
| 182 | assert "summary" in delta |
| 183 | |
| 184 | |
| 185 | # --------------------------------------------------------------------------- |
| 186 | # MidiPlugin.diff() returns StructuredDelta |
| 187 | # --------------------------------------------------------------------------- |
| 188 | |
| 189 | class TestMidiPluginStructuredDiff: |
| 190 | def test_no_change_returns_empty_ops(self) -> None: |
| 191 | snap = _snap({"a.mid": "h1"}) |
| 192 | delta = plugin.diff(snap, snap) |
| 193 | assert isinstance(delta, dict) |
| 194 | assert delta["ops"] == [] |
| 195 | |
| 196 | def test_no_change_summary_is_no_changes(self) -> None: |
| 197 | snap = _snap({"a.mid": "h1"}) |
| 198 | delta = plugin.diff(snap, snap) |
| 199 | assert delta["summary"] == "no changes" |
| 200 | |
| 201 | def test_file_added_returns_insert_op(self) -> None: |
| 202 | base = _snap({}) |
| 203 | target = _snap({"new.mid": "h1"}) |
| 204 | delta = plugin.diff(base, target) |
| 205 | ops = delta["ops"] |
| 206 | assert len(ops) == 1 |
| 207 | assert ops[0]["op"] == "insert" |
| 208 | assert ops[0]["address"] == "new.mid" |
| 209 | |
| 210 | def test_file_added_insert_op_has_content_id(self) -> None: |
| 211 | base = _snap({}) |
| 212 | target = _snap({"new.mid": "abcdef123"}) |
| 213 | delta = plugin.diff(base, target) |
| 214 | assert delta["ops"][0]["content_id"] == "abcdef123" |
| 215 | |
| 216 | def test_file_removed_returns_delete_op(self) -> None: |
| 217 | base = _snap({"old.mid": "h1"}) |
| 218 | target = _snap({}) |
| 219 | delta = plugin.diff(base, target) |
| 220 | ops = delta["ops"] |
| 221 | assert len(ops) == 1 |
| 222 | assert ops[0]["op"] == "delete" |
| 223 | assert ops[0]["address"] == "old.mid" |
| 224 | |
| 225 | def test_file_removed_delete_op_has_content_id(self) -> None: |
| 226 | base = _snap({"old.mid": "prevhash"}) |
| 227 | target = _snap({}) |
| 228 | delta = plugin.diff(base, target) |
| 229 | assert delta["ops"][0]["content_id"] == "prevhash" |
| 230 | |
| 231 | def test_non_midi_modified_returns_replace_op(self) -> None: |
| 232 | base = _snap({"notes.txt": "old"}) |
| 233 | target = _snap({"notes.txt": "new"}) |
| 234 | delta = plugin.diff(base, target) |
| 235 | ops = delta["ops"] |
| 236 | assert len(ops) == 1 |
| 237 | assert ops[0]["op"] == "replace" |
| 238 | assert ops[0]["address"] == "notes.txt" |
| 239 | |
| 240 | def test_replace_op_has_old_and_new_ids(self) -> None: |
| 241 | base = _snap({"notes.txt": "oldhash"}) |
| 242 | target = _snap({"notes.txt": "newhash"}) |
| 243 | delta = plugin.diff(base, target) |
| 244 | op = delta["ops"][0] |
| 245 | assert op["op"] == "replace" |
| 246 | assert op["old_content_id"] == "oldhash" |
| 247 | assert op["new_content_id"] == "newhash" |
| 248 | |
| 249 | def test_mid_modified_without_repo_root_returns_replace_op(self) -> None: |
| 250 | # Without repo_root we can't load blobs, so fallback to ReplaceOp. |
| 251 | base = _snap({"drums.mid": "old"}) |
| 252 | target = _snap({"drums.mid": "new"}) |
| 253 | delta = plugin.diff(base, target) |
| 254 | assert delta["ops"][0]["op"] == "replace" |
| 255 | |
| 256 | def test_multiple_changes_produce_multiple_ops(self) -> None: |
| 257 | base = _snap({"a.mid": "h1", "b.mid": "h2"}) |
| 258 | target = _snap({"b.mid": "h2_new", "c.mid": "h3"}) |
| 259 | delta = plugin.diff(base, target) |
| 260 | kinds = {op["op"] for op in delta["ops"]} |
| 261 | assert "insert" in kinds # c.mid added |
| 262 | assert "delete" in kinds # a.mid removed |
| 263 | assert "replace" in kinds # b.mid modified |
| 264 | |
| 265 | def test_summary_mentions_added_on_add(self) -> None: |
| 266 | base = _snap({}) |
| 267 | target = _snap({"x.mid": "h"}) |
| 268 | delta = plugin.diff(base, target) |
| 269 | assert "added" in delta["summary"] |
| 270 | |
| 271 | def test_summary_mentions_removed_on_delete(self) -> None: |
| 272 | base = _snap({"x.mid": "h"}) |
| 273 | target = _snap({}) |
| 274 | delta = plugin.diff(base, target) |
| 275 | assert "removed" in delta["summary"] |
| 276 | |
| 277 | def test_domain_is_music(self) -> None: |
| 278 | snap = _snap({"a.mid": "h"}) |
| 279 | delta = plugin.diff(snap, snap) |
| 280 | assert delta["domain"] == "midi" |
| 281 | |
| 282 | def test_insert_op_position_is_none_for_file_level(self) -> None: |
| 283 | base = _snap({}) |
| 284 | target = _snap({"f.mid": "h"}) |
| 285 | delta = plugin.diff(base, target) |
| 286 | assert delta["ops"][0]["position"] is None |
| 287 | |
| 288 | def test_ops_are_sorted_by_address(self) -> None: |
| 289 | base = _snap({}) |
| 290 | target = _snap({"z.mid": "h1", "a.mid": "h2", "m.mid": "h3"}) |
| 291 | delta = plugin.diff(base, target) |
| 292 | addresses = [op["address"] for op in delta["ops"]] |
| 293 | assert addresses == sorted(addresses) |
| 294 | |
| 295 | |
| 296 | # --------------------------------------------------------------------------- |
| 297 | # DriftReport uses StructuredDelta |
| 298 | # --------------------------------------------------------------------------- |
| 299 | |
| 300 | class TestDriftReportDelta: |
| 301 | def test_no_drift_delta_is_structured(self) -> None: |
| 302 | snap = _snap({"a.mid": "h"}) |
| 303 | report = plugin.drift(snap, snap) |
| 304 | assert isinstance(report, DriftReport) |
| 305 | assert isinstance(report.delta, dict) |
| 306 | assert "ops" in report.delta |
| 307 | assert "summary" in report.delta |
| 308 | |
| 309 | def test_drift_delta_has_insert_op_on_addition(self) -> None: |
| 310 | committed = _snap({"a.mid": "h1"}) |
| 311 | live = _snap({"a.mid": "h1", "b.mid": "h2"}) |
| 312 | report = plugin.drift(committed, live) |
| 313 | assert report.has_drift |
| 314 | insert_ops = [op for op in report.delta["ops"] if op["op"] == "insert"] |
| 315 | assert any(op["address"] == "b.mid" for op in insert_ops) |
| 316 | |
| 317 | def test_drift_summary_still_human_readable(self) -> None: |
| 318 | committed = _snap({"a.mid": "h1"}) |
| 319 | live = _snap({"a.mid": "h1", "b.mid": "h2"}) |
| 320 | report = plugin.drift(committed, live) |
| 321 | assert "added" in report.summary |
| 322 | |
| 323 | def test_default_drift_report_delta_is_empty_structured(self) -> None: |
| 324 | report = DriftReport(has_drift=False) |
| 325 | assert report.delta["ops"] == [] |
| 326 | assert report.delta["domain"] == "" |
| 327 | |
| 328 | |
| 329 | # --------------------------------------------------------------------------- |
| 330 | # MidiPlugin.apply() handles StructuredDelta |
| 331 | # --------------------------------------------------------------------------- |
| 332 | |
| 333 | class TestMidiPluginApply: |
| 334 | def test_apply_delete_op_removes_file(self) -> None: |
| 335 | snap = _snap({"a.mid": "h1", "b.mid": "h2"}) |
| 336 | delta: StructuredDelta = StructuredDelta( |
| 337 | domain="midi", |
| 338 | ops=[DeleteOp( |
| 339 | op="delete", address="a.mid", position=None, |
| 340 | content_id="h1", content_summary="deleted: a.mid", |
| 341 | )], |
| 342 | summary="1 file removed", |
| 343 | ) |
| 344 | result = plugin.apply(delta, snap) |
| 345 | assert "a.mid" not in result["files"] |
| 346 | assert "b.mid" in result["files"] |
| 347 | |
| 348 | def test_apply_replace_op_updates_hash(self) -> None: |
| 349 | snap = _snap({"a.mid": "old"}) |
| 350 | delta: StructuredDelta = StructuredDelta( |
| 351 | domain="midi", |
| 352 | ops=[ReplaceOp( |
| 353 | op="replace", address="a.mid", position=None, |
| 354 | old_content_id="old", new_content_id="new", |
| 355 | old_summary="a.mid (prev)", new_summary="a.mid (new)", |
| 356 | )], |
| 357 | summary="1 file modified", |
| 358 | ) |
| 359 | result = plugin.apply(delta, snap) |
| 360 | assert result["files"]["a.mid"] == "new" |
| 361 | |
| 362 | def test_apply_insert_op_adds_file(self) -> None: |
| 363 | snap = _snap({}) |
| 364 | delta: StructuredDelta = StructuredDelta( |
| 365 | domain="midi", |
| 366 | ops=[InsertOp( |
| 367 | op="insert", address="new.mid", position=None, |
| 368 | content_id="newhash", content_summary="new file: new.mid", |
| 369 | )], |
| 370 | summary="1 file added", |
| 371 | ) |
| 372 | result = plugin.apply(delta, snap) |
| 373 | assert result["files"]["new.mid"] == "newhash" |
| 374 | |
| 375 | def test_apply_from_workdir_rescans(self, tmp_path: pathlib.Path) -> None: |
| 376 | workdir = tmp_path / "muse-work" |
| 377 | workdir.mkdir() |
| 378 | (workdir / "beat.mid").write_bytes(b"drums") |
| 379 | delta: StructuredDelta = StructuredDelta( |
| 380 | domain="midi", ops=[], summary="no changes", |
| 381 | ) |
| 382 | result = plugin.apply(delta, workdir) |
| 383 | assert "beat.mid" in result["files"] |
| 384 | |
| 385 | |
| 386 | # --------------------------------------------------------------------------- |
| 387 | # CLI show displays structured delta |
| 388 | # --------------------------------------------------------------------------- |
| 389 | |
| 390 | class TestShowStructuredOutput: |
| 391 | def test_show_displays_structured_summary( |
| 392 | self, tmp_path: pathlib.Path |
| 393 | ) -> None: |
| 394 | from typer.testing import CliRunner |
| 395 | from muse.cli.app import cli |
| 396 | |
| 397 | runner = CliRunner() |
| 398 | result = runner.invoke(cli, ["init", "--domain", "midi"], obj={}) |
| 399 | # Just check the command is importable and types are correct — |
| 400 | # full CLI integration is covered in test_cli_workflow.py. |
| 401 | assert result is not None |