test_music_plugin.py
python
| 1 | """Tests for muse.plugins.midi.plugin — the MuseDomainPlugin reference implementation.""" |
| 2 | |
| 3 | import pathlib |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from muse.domain import DriftReport, MergeResult, MuseDomainPlugin, SnapshotManifest |
| 8 | from muse.plugins.midi.plugin import MidiPlugin, content_hash, plugin |
| 9 | |
| 10 | |
| 11 | def _snap(files: dict[str, str]) -> SnapshotManifest: |
| 12 | return SnapshotManifest(files=files, domain="midi") |
| 13 | |
| 14 | |
| 15 | class TestProtocolConformance: |
| 16 | def test_plugin_satisfies_protocol(self) -> None: |
| 17 | assert isinstance(plugin, MuseDomainPlugin) |
| 18 | |
| 19 | def test_module_singleton_is_music_plugin(self) -> None: |
| 20 | assert isinstance(plugin, MidiPlugin) |
| 21 | |
| 22 | |
| 23 | class TestSnapshot: |
| 24 | def test_from_dict_passthrough(self) -> None: |
| 25 | snap = SnapshotManifest(files={"a.mid": "h1"}, domain="midi") |
| 26 | assert plugin.snapshot(snap) is snap |
| 27 | |
| 28 | def test_from_workdir(self, tmp_path: pathlib.Path) -> None: |
| 29 | workdir = tmp_path / "muse-work" |
| 30 | workdir.mkdir() |
| 31 | (workdir / "beat.mid").write_bytes(b"drums") |
| 32 | snap = plugin.snapshot(workdir) |
| 33 | assert "files" in snap |
| 34 | assert "beat.mid" in snap["files"] |
| 35 | assert snap["domain"] == "midi" |
| 36 | |
| 37 | def test_empty_workdir(self, tmp_path: pathlib.Path) -> None: |
| 38 | workdir = tmp_path / "muse-work" |
| 39 | workdir.mkdir() |
| 40 | snap = plugin.snapshot(workdir) |
| 41 | assert snap["files"] == {} |
| 42 | |
| 43 | |
| 44 | class TestDiff: |
| 45 | """diff() returns a StructuredDelta with typed ops — not the old DeltaManifest.""" |
| 46 | |
| 47 | def test_no_change_empty_ops(self) -> None: |
| 48 | snap = _snap({"a.mid": "h1"}) |
| 49 | delta = plugin.diff(snap, snap) |
| 50 | assert delta["ops"] == [] |
| 51 | |
| 52 | def test_no_change_summary(self) -> None: |
| 53 | snap = _snap({"a.mid": "h1"}) |
| 54 | delta = plugin.diff(snap, snap) |
| 55 | assert delta["summary"] == "no changes" |
| 56 | |
| 57 | def test_added_file_produces_insert_op(self) -> None: |
| 58 | base = _snap({}) |
| 59 | target = _snap({"new.mid": "h1"}) |
| 60 | delta = plugin.diff(base, target) |
| 61 | insert_ops = [op for op in delta["ops"] if op["op"] == "insert"] |
| 62 | assert any(op["address"] == "new.mid" for op in insert_ops) |
| 63 | |
| 64 | def test_removed_file_produces_delete_op(self) -> None: |
| 65 | base = _snap({"old.mid": "h1"}) |
| 66 | target = _snap({}) |
| 67 | delta = plugin.diff(base, target) |
| 68 | delete_ops = [op for op in delta["ops"] if op["op"] == "delete"] |
| 69 | assert any(op["address"] == "old.mid" for op in delete_ops) |
| 70 | |
| 71 | def test_modified_file_produces_replace_or_patch_op(self) -> None: |
| 72 | base = _snap({"f.mid": "old"}) |
| 73 | target = _snap({"f.mid": "new"}) |
| 74 | delta = plugin.diff(base, target) |
| 75 | changed_ops = [op for op in delta["ops"] if op["op"] in ("replace", "patch")] |
| 76 | assert any(op["address"] == "f.mid" for op in changed_ops) |
| 77 | |
| 78 | def test_domain_is_music(self) -> None: |
| 79 | snap = _snap({"a.mid": "h"}) |
| 80 | delta = plugin.diff(snap, snap) |
| 81 | assert delta["domain"] == "midi" |
| 82 | |
| 83 | |
| 84 | class TestMerge: |
| 85 | def test_clean_merge(self) -> None: |
| 86 | base = _snap({"a.mid": "h0", "b.mid": "h0"}) |
| 87 | left = _snap({"a.mid": "h_left", "b.mid": "h0"}) |
| 88 | right = _snap({"a.mid": "h0", "b.mid": "h_right"}) |
| 89 | result = plugin.merge(base, left, right) |
| 90 | assert isinstance(result, MergeResult) |
| 91 | assert result.is_clean |
| 92 | assert result.merged["files"]["a.mid"] == "h_left" |
| 93 | assert result.merged["files"]["b.mid"] == "h_right" |
| 94 | |
| 95 | def test_conflict_detected(self) -> None: |
| 96 | base = _snap({"a.mid": "h0"}) |
| 97 | left = _snap({"a.mid": "h_left"}) |
| 98 | right = _snap({"a.mid": "h_right"}) |
| 99 | result = plugin.merge(base, left, right) |
| 100 | assert not result.is_clean |
| 101 | assert result.conflicts == ["a.mid"] |
| 102 | |
| 103 | def test_both_delete_same_file(self) -> None: |
| 104 | base = _snap({"a.mid": "h0", "b.mid": "h0"}) |
| 105 | left = _snap({"b.mid": "h0"}) |
| 106 | right = _snap({"b.mid": "h0"}) |
| 107 | result = plugin.merge(base, left, right) |
| 108 | assert result.is_clean |
| 109 | assert "a.mid" not in result.merged["files"] |
| 110 | |
| 111 | |
| 112 | class TestDrift: |
| 113 | def test_no_drift(self) -> None: |
| 114 | snap = _snap({"a.mid": "h1"}) |
| 115 | report = plugin.drift(snap, snap) |
| 116 | assert isinstance(report, DriftReport) |
| 117 | assert not report.has_drift |
| 118 | |
| 119 | def test_has_drift_on_addition(self) -> None: |
| 120 | committed = _snap({"a.mid": "h1"}) |
| 121 | live = _snap({"a.mid": "h1", "b.mid": "h2"}) |
| 122 | report = plugin.drift(committed, live) |
| 123 | assert report.has_drift |
| 124 | assert "added" in report.summary |
| 125 | |
| 126 | def test_drift_delta_is_structured(self) -> None: |
| 127 | snap = _snap({"a.mid": "h1"}) |
| 128 | report = plugin.drift(snap, snap) |
| 129 | assert "ops" in report.delta |
| 130 | assert "summary" in report.delta |
| 131 | assert "domain" in report.delta |
| 132 | |
| 133 | |
| 134 | class TestContentHash: |
| 135 | def test_deterministic(self) -> None: |
| 136 | snap = SnapshotManifest(files={"a.mid": "h1"}, domain="midi") |
| 137 | assert content_hash(snap) == content_hash(snap) |
| 138 | |
| 139 | def test_different_content_different_hash(self) -> None: |
| 140 | a = SnapshotManifest(files={"a.mid": "h1"}, domain="midi") |
| 141 | b = SnapshotManifest(files={"a.mid": "h2"}, domain="midi") |
| 142 | assert content_hash(a) != content_hash(b) |