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