test_crdt.py
python
| 1 | """Tests for muse.plugins.midi._crdt_notes — NotePosition, RGANoteEntry, MidiRGA. |
| 2 | |
| 3 | Verifies all three CRDT laws: |
| 4 | 1. Commutativity: merge(a, b) == merge(b, a) |
| 5 | 2. Associativity: merge(merge(a, b), c) == merge(a, merge(b, c)) |
| 6 | 3. Idempotency: merge(a, a) == a |
| 7 | """ |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | from muse.plugins.midi._crdt_notes import ( |
| 12 | MidiRGA, |
| 13 | NotePosition, |
| 14 | RGANoteEntry, |
| 15 | _pitch_to_voice_lane, |
| 16 | ) |
| 17 | from muse.plugins.midi.midi_diff import NoteKey |
| 18 | |
| 19 | |
| 20 | def _key(pitch: int = 60, velocity: int = 80, start_tick: int = 0, |
| 21 | duration_ticks: int = 480, channel: int = 0) -> NoteKey: |
| 22 | return NoteKey(pitch=pitch, velocity=velocity, start_tick=start_tick, |
| 23 | duration_ticks=duration_ticks, channel=channel) |
| 24 | |
| 25 | |
| 26 | # --------------------------------------------------------------------------- |
| 27 | # NotePosition ordering |
| 28 | # --------------------------------------------------------------------------- |
| 29 | |
| 30 | |
| 31 | class TestNotePosition: |
| 32 | def test_ordered_by_measure_first(self) -> None: |
| 33 | p1 = NotePosition(measure=1, beat_sub=100, voice_lane=3, op_id="zzz") |
| 34 | p2 = NotePosition(measure=2, beat_sub=0, voice_lane=0, op_id="aaa") |
| 35 | assert p1 < p2 |
| 36 | |
| 37 | def test_ordered_by_beat_sub_within_measure(self) -> None: |
| 38 | p1 = NotePosition(measure=1, beat_sub=100, voice_lane=3, op_id="zzz") |
| 39 | p2 = NotePosition(measure=1, beat_sub=200, voice_lane=0, op_id="aaa") |
| 40 | assert p1 < p2 |
| 41 | |
| 42 | def test_ordered_by_voice_lane_at_same_beat(self) -> None: |
| 43 | # Bass (lane 0) should come before soprano (lane 3). |
| 44 | p_bass = NotePosition(measure=1, beat_sub=0, voice_lane=0, op_id="zzz") |
| 45 | p_soprano = NotePosition(measure=1, beat_sub=0, voice_lane=3, op_id="aaa") |
| 46 | assert p_bass < p_soprano |
| 47 | |
| 48 | def test_tie_broken_by_op_id(self) -> None: |
| 49 | p1 = NotePosition(measure=1, beat_sub=0, voice_lane=0, op_id="aaa") |
| 50 | p2 = NotePosition(measure=1, beat_sub=0, voice_lane=0, op_id="bbb") |
| 51 | assert p1 < p2 |
| 52 | |
| 53 | |
| 54 | # --------------------------------------------------------------------------- |
| 55 | # _pitch_to_voice_lane |
| 56 | # --------------------------------------------------------------------------- |
| 57 | |
| 58 | |
| 59 | class TestPitchToVoiceLane: |
| 60 | def test_bass_range(self) -> None: |
| 61 | assert _pitch_to_voice_lane(24) == 0 |
| 62 | assert _pitch_to_voice_lane(47) == 0 |
| 63 | |
| 64 | def test_tenor_range(self) -> None: |
| 65 | assert _pitch_to_voice_lane(48) == 1 |
| 66 | assert _pitch_to_voice_lane(59) == 1 |
| 67 | |
| 68 | def test_alto_range(self) -> None: |
| 69 | assert _pitch_to_voice_lane(60) == 2 |
| 70 | assert _pitch_to_voice_lane(71) == 2 |
| 71 | |
| 72 | def test_soprano_range(self) -> None: |
| 73 | assert _pitch_to_voice_lane(72) == 3 |
| 74 | assert _pitch_to_voice_lane(108) == 3 |
| 75 | |
| 76 | |
| 77 | # --------------------------------------------------------------------------- |
| 78 | # MidiRGA — basic insert / delete |
| 79 | # --------------------------------------------------------------------------- |
| 80 | |
| 81 | |
| 82 | class TestMidiRGAInsertDelete: |
| 83 | def test_single_insert_visible(self) -> None: |
| 84 | seq = MidiRGA("agent-a") |
| 85 | seq.insert(_key(60)) |
| 86 | notes = seq.to_sequence() |
| 87 | assert len(notes) == 1 |
| 88 | assert notes[0]["pitch"] == 60 |
| 89 | |
| 90 | def test_multiple_inserts_ordered_by_position(self) -> None: |
| 91 | seq = MidiRGA("agent-a") |
| 92 | # Insert soprano (pitch 72 = lane 3) before bass (pitch 36 = lane 0) |
| 93 | # at the same beat — bass should appear first in output. |
| 94 | seq.insert(_key(pitch=72, start_tick=0)) |
| 95 | seq.insert(_key(pitch=36, start_tick=0)) |
| 96 | notes = seq.to_sequence() |
| 97 | assert notes[0]["pitch"] == 36 # bass first |
| 98 | assert notes[1]["pitch"] == 72 # soprano second |
| 99 | |
| 100 | def test_delete_removes_note(self) -> None: |
| 101 | seq = MidiRGA("agent-a") |
| 102 | entry = seq.insert(_key(60)) |
| 103 | seq.delete(entry["op_id"]) |
| 104 | assert seq.to_sequence() == [] |
| 105 | |
| 106 | def test_delete_nonexistent_raises(self) -> None: |
| 107 | seq = MidiRGA("agent-a") |
| 108 | with pytest.raises(KeyError): |
| 109 | seq.delete("nonexistent-op-id") |
| 110 | |
| 111 | def test_tombstoned_entries_counted(self) -> None: |
| 112 | seq = MidiRGA("agent-a") |
| 113 | e = seq.insert(_key(60)) |
| 114 | seq.delete(e["op_id"]) |
| 115 | assert seq.entry_count() == 1 |
| 116 | assert seq.live_count() == 0 |
| 117 | |
| 118 | |
| 119 | # --------------------------------------------------------------------------- |
| 120 | # CRDT merge — commutativity, associativity, idempotency |
| 121 | # --------------------------------------------------------------------------- |
| 122 | |
| 123 | |
| 124 | class TestMidiRGACRDTLaws: |
| 125 | def _make_replicas(self) -> tuple[MidiRGA, MidiRGA, MidiRGA]: |
| 126 | a = MidiRGA("agent-a") |
| 127 | b = MidiRGA("agent-b") |
| 128 | c = MidiRGA("agent-c") |
| 129 | |
| 130 | a.insert(_key(60, start_tick=0)) |
| 131 | a.insert(_key(64, start_tick=480)) |
| 132 | |
| 133 | b.insert(_key(67, start_tick=0)) |
| 134 | b.insert(_key(71, start_tick=480)) |
| 135 | |
| 136 | c.insert(_key(72, start_tick=0)) |
| 137 | |
| 138 | # Propagate a's ops to b and c (simulating gossip). |
| 139 | for entry in list(a._entries.values()): |
| 140 | b._entries.setdefault(entry["op_id"], entry) |
| 141 | c._entries.setdefault(entry["op_id"], entry) |
| 142 | |
| 143 | return a, b, c |
| 144 | |
| 145 | def test_commutativity(self) -> None: |
| 146 | a, b, _ = self._make_replicas() |
| 147 | ab = MidiRGA.merge(a, b) |
| 148 | ba = MidiRGA.merge(b, a) |
| 149 | assert ab.to_sequence() == ba.to_sequence() |
| 150 | |
| 151 | def test_associativity(self) -> None: |
| 152 | a, b, c = self._make_replicas() |
| 153 | ab_c = MidiRGA.merge(MidiRGA.merge(a, b), c) |
| 154 | a_bc = MidiRGA.merge(a, MidiRGA.merge(b, c)) |
| 155 | assert ab_c.to_sequence() == a_bc.to_sequence() |
| 156 | |
| 157 | def test_idempotency(self) -> None: |
| 158 | a, _, _ = self._make_replicas() |
| 159 | aa = MidiRGA.merge(a, a) |
| 160 | assert aa.to_sequence() == a.to_sequence() |
| 161 | |
| 162 | def test_merge_contains_all_inserts(self) -> None: |
| 163 | a = MidiRGA("agent-a") |
| 164 | b = MidiRGA("agent-b") |
| 165 | for i in range(5): |
| 166 | a.insert(_key(60 + i, start_tick=i * 480)) |
| 167 | for i in range(5): |
| 168 | b.insert(_key(72 + i, start_tick=i * 480)) |
| 169 | merged = MidiRGA.merge(a, b) |
| 170 | assert merged.live_count() == 10 |
| 171 | |
| 172 | def test_tombstone_wins_in_merge(self) -> None: |
| 173 | a = MidiRGA("agent-a") |
| 174 | b = MidiRGA("agent-b") |
| 175 | |
| 176 | entry = a.insert(_key(60)) |
| 177 | # Share the insert with b. |
| 178 | b._entries[entry["op_id"]] = entry |
| 179 | # b deletes the shared note; a does not. |
| 180 | b.delete(entry["op_id"]) |
| 181 | |
| 182 | merged = MidiRGA.merge(a, b) |
| 183 | # Tombstone wins — note should be absent in merged result. |
| 184 | assert merged.live_count() == 0 |
| 185 | |
| 186 | |
| 187 | # --------------------------------------------------------------------------- |
| 188 | # to_domain_ops |
| 189 | # --------------------------------------------------------------------------- |
| 190 | |
| 191 | |
| 192 | class TestToDomainOps: |
| 193 | def test_empty_base_and_live_produces_no_ops(self) -> None: |
| 194 | seq = MidiRGA("agent-a") |
| 195 | ops = seq.to_domain_ops([]) |
| 196 | assert ops == [] |
| 197 | |
| 198 | def test_added_notes_produce_insert_ops(self) -> None: |
| 199 | seq = MidiRGA("agent-a") |
| 200 | seq.insert(_key(60)) |
| 201 | seq.insert(_key(64)) |
| 202 | ops = seq.to_domain_ops([]) |
| 203 | assert len(ops) == 2 |
| 204 | assert all(o["op"] == "insert" for o in ops) |
| 205 | |
| 206 | def test_removed_notes_produce_delete_ops(self) -> None: |
| 207 | base_notes = [_key(60), _key(64)] |
| 208 | seq = MidiRGA("agent-a") |
| 209 | # Add only the first note back. |
| 210 | seq.insert(_key(60)) |
| 211 | ops = seq.to_domain_ops(base_notes) |
| 212 | op_types = [o["op"] for o in ops] |
| 213 | assert "delete" in op_types |
| 214 | |
| 215 | def test_unchanged_notes_produce_no_ops(self) -> None: |
| 216 | note = _key(60) |
| 217 | seq = MidiRGA("agent-a") |
| 218 | seq.insert(note) |
| 219 | ops = seq.to_domain_ops([note]) |
| 220 | assert ops == [] |
| 221 | |
| 222 | def test_voice_ordering_preserved_in_sequence(self) -> None: |
| 223 | seq = MidiRGA("agent-a") |
| 224 | # Insert in reverse voice order; output should be ordered bass→soprano. |
| 225 | seq.insert(_key(pitch=84, start_tick=0)) # soprano |
| 226 | seq.insert(_key(pitch=60, start_tick=0)) # alto |
| 227 | seq.insert(_key(pitch=48, start_tick=0)) # tenor |
| 228 | seq.insert(_key(pitch=36, start_tick=0)) # bass |
| 229 | notes = seq.to_sequence() |
| 230 | pitches = [n["pitch"] for n in notes] |
| 231 | assert pitches == sorted(pitches) # bass < tenor < alto < soprano |