c38a7b0fd8
New files:
main.py - PedalApp: boots all subsystems in order,
wires MIDI/footswitch callbacks, graceful
teardown reverses boot order
src/system/config.py - YAML config loader with deep-merge
(separated to avoid hardware deps)
src/system/services.py - systemd unit generator for pedal.service
+ multi-fx-pedal.target
scripts/install_service.sh - copies project, creates venv, installs
+ enables service units
tests/test_integration.py - 41 tests: boot, routing, display sync,
teardown, systemd content, CLI, edge cases
Modified:
tests/conftest.py - add project root to sys.path
864 lines
35 KiB
Python
864 lines
35 KiB
Python
"""Tests for MIDI handler — parsing, dispatch, learn, clock sync, I/O."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from midi.handler import (
|
|
MIDIHandler,
|
|
MIDIEvent,
|
|
LearnedMapping,
|
|
MIDIMapping,
|
|
CC_EXPRESSION,
|
|
CC_VOLUME,
|
|
CC_BANK_SELECT_MSB,
|
|
CLOCK_PPQN,
|
|
# Message builders
|
|
CONTROL_CHANGE,
|
|
NOTE_ON,
|
|
NOTE_OFF,
|
|
PROGRAM_CHANGE,
|
|
PITCH_BEND,
|
|
RT_CLOCK,
|
|
RT_START,
|
|
RT_STOP,
|
|
RT_CONTINUE,
|
|
SYS_EXCLUSIVE,
|
|
SYS_EXCLUSIVE_END,
|
|
# Interface classes
|
|
UARTMIDI,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Fixtures
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture
|
|
def handler() -> MIDIHandler:
|
|
return MIDIHandler()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# MIDIEvent tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def test_midi_event_defaults() -> None:
|
|
e = MIDIEvent(type="cc")
|
|
assert e.type == "cc"
|
|
assert e.channel == 0
|
|
assert e.note == 0
|
|
assert e.velocity == 0
|
|
assert e.cc_number == 0
|
|
assert e.cc_value == 0
|
|
assert e.program == 0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Message parsing — single messages
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestParse:
|
|
"""MIDIHandler.parse() — single complete messages."""
|
|
|
|
def test_note_on(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0x90, 60, 100]))
|
|
assert ev is not None
|
|
assert ev.type == "note_on"
|
|
assert ev.channel == 0
|
|
assert ev.note == 60
|
|
assert ev.velocity == 100
|
|
|
|
def test_note_on_channel_3(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0x93, 60, 100]))
|
|
assert ev is not None
|
|
assert ev.type == "note_on"
|
|
assert ev.channel == 3
|
|
assert ev.note == 60
|
|
|
|
def test_note_on_zero_velocity_is_note_off(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0x90, 60, 0]))
|
|
assert ev is not None
|
|
assert ev.type == "note_off"
|
|
assert ev.channel == 0
|
|
assert ev.note == 60
|
|
|
|
def test_note_off(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0x80, 60, 64]))
|
|
assert ev is not None
|
|
assert ev.type == "note_off"
|
|
assert ev.channel == 0
|
|
assert ev.note == 60
|
|
assert ev.velocity == 64
|
|
|
|
def test_control_change(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xB0, CC_EXPRESSION, 64]))
|
|
assert ev is not None
|
|
assert ev.type == "cc"
|
|
assert ev.channel == 0
|
|
assert ev.cc_number == CC_EXPRESSION
|
|
assert ev.cc_value == 64
|
|
|
|
def test_control_change_channel_7(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xB7, CC_VOLUME, 100]))
|
|
assert ev is not None
|
|
assert ev.type == "cc"
|
|
assert ev.channel == 7
|
|
assert ev.cc_number == CC_VOLUME
|
|
assert ev.cc_value == 100
|
|
|
|
def test_program_change(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xC0, 5]))
|
|
assert ev is not None
|
|
assert ev.type == "pc"
|
|
assert ev.channel == 0
|
|
assert ev.program == 5
|
|
|
|
def test_program_change_channel_15(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xCF, 127]))
|
|
assert ev is not None
|
|
assert ev.type == "pc"
|
|
assert ev.channel == 15
|
|
assert ev.program == 127
|
|
|
|
def test_pitch_bend(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xE0, 0x00, 0x40])) # Center (8192)
|
|
assert ev is not None
|
|
assert ev.type == "pitch_bend"
|
|
assert ev.channel == 0
|
|
assert ev.cc_value == 8192
|
|
|
|
def test_pitch_bend_max(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xE0, 0x7F, 0x7F])) # 16383
|
|
assert ev is not None
|
|
assert ev.cc_value == 16383
|
|
|
|
def test_channel_pressure(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xD0, 100]))
|
|
assert ev is not None
|
|
assert ev.type == "channel_pressure"
|
|
assert ev.channel == 0
|
|
assert ev.velocity == 100
|
|
|
|
def test_poly_pressure(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xA0, 60, 100]))
|
|
assert ev is not None
|
|
assert ev.type == "poly_pressure"
|
|
assert ev.channel == 0
|
|
assert ev.note == 60
|
|
assert ev.velocity == 100
|
|
|
|
def test_midi_clock(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([RT_CLOCK]))
|
|
assert ev is not None
|
|
assert ev.type == "clock"
|
|
|
|
def test_midi_start(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([RT_START]))
|
|
assert ev is not None
|
|
assert ev.type == "start"
|
|
|
|
def test_midi_stop(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([RT_STOP]))
|
|
assert ev is not None
|
|
assert ev.type == "stop"
|
|
|
|
def test_midi_continue(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([RT_CONTINUE]))
|
|
assert ev is not None
|
|
assert ev.type == "continue"
|
|
|
|
def test_active_sensing_ignored(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0xFE]))
|
|
assert ev is None
|
|
|
|
def test_empty_data(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(b"")
|
|
assert ev is None
|
|
|
|
def test_truncated_message(self, handler: MIDIHandler) -> None:
|
|
ev = handler.parse(bytes([0x90, 60])) # Missing velocity
|
|
assert ev is None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Dispatch tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestDispatch:
|
|
"""MIDIHandler.dispatch() — callback routing."""
|
|
|
|
def test_pc_callback(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int]] = []
|
|
|
|
handler.set_pc_callback(lambda ch, pg: results.append((ch, pg)))
|
|
handler.dispatch(MIDIEvent(type="pc", channel=2, program=7))
|
|
|
|
assert results == [(2, 7)]
|
|
|
|
def test_cc_callback(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int]] = []
|
|
|
|
handler.register_cc(11, lambda val, ch: results.append((val, ch)))
|
|
handler.dispatch(MIDIEvent(type="cc", channel=0, cc_number=11, cc_value=64))
|
|
|
|
assert results == [(64, 0)]
|
|
|
|
def test_cc_multiple_callbacks(self, handler: MIDIHandler) -> None:
|
|
cc11: list[tuple[int, int]] = []
|
|
cc12: list[tuple[int, int]] = []
|
|
|
|
handler.register_cc(11, lambda val, ch: cc11.append((val, ch)))
|
|
handler.register_cc(12, lambda val, ch: cc12.append((val, ch)))
|
|
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=11, cc_value=64))
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=12, cc_value=100))
|
|
|
|
assert cc11 == [(64, 0)]
|
|
assert cc12 == [(100, 0)]
|
|
|
|
def test_cc_unregistered_does_not_raise(self, handler: MIDIHandler) -> None:
|
|
# Should not raise — just log
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=99, cc_value=50))
|
|
|
|
def test_note_callback(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int, int]] = []
|
|
|
|
handler.set_note_callback(lambda n, v, ch: results.append((n, v, ch)))
|
|
handler.dispatch(MIDIEvent(type="note_on", note=60, velocity=100, channel=1))
|
|
|
|
assert results == [(60, 100, 1)]
|
|
|
|
def test_note_off_dispatch(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int, int]] = []
|
|
|
|
handler.set_note_callback(lambda n, v, ch: results.append((n, v, ch)))
|
|
handler.dispatch(MIDIEvent(type="note_off", note=60, velocity=0, channel=1))
|
|
|
|
assert results == [(60, 0, 1)]
|
|
|
|
def test_no_callbacks_no_crash(self, handler: MIDIHandler) -> None:
|
|
"""Dispatch with no callbacks registered should be a no-op."""
|
|
handler.dispatch(MIDIEvent(type="pc", program=3))
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=7, cc_value=100))
|
|
handler.dispatch(MIDIEvent(type="note_on", note=60, velocity=100))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# MIDI Learn tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestMIDILearn:
|
|
"""MIDI Learn mode — CC capture and mapping."""
|
|
|
|
def test_start_learn(self, handler: MIDIHandler) -> None:
|
|
assert not handler.learn_mode
|
|
handler.start_learn("delay.feedback")
|
|
assert handler.learn_mode
|
|
|
|
def test_stop_learn(self, handler: MIDIHandler) -> None:
|
|
handler.start_learn("delay.feedback")
|
|
handler.stop_learn()
|
|
assert not handler.learn_mode
|
|
|
|
def test_learn_cc_to_mapping(self, handler: MIDIHandler) -> None:
|
|
handler.start_learn("reverb.mix")
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=14, cc_value=64))
|
|
|
|
mapping = handler.get_mapping("reverb.mix")
|
|
assert mapping is not None
|
|
assert mapping.cc_number == 14
|
|
|
|
def test_learn_auto_exits(self, handler: MIDIHandler) -> None:
|
|
handler.start_learn("delay.time")
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=15, cc_value=100))
|
|
assert not handler.learn_mode
|
|
|
|
def test_learn_triggers_callback(self, handler: MIDIHandler) -> None:
|
|
results: list[LearnedMapping] = []
|
|
|
|
handler.set_midi_learn_callback(lambda lm: results.append(lm))
|
|
handler.start_learn("chorus.rate")
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=22, cc_value=50))
|
|
|
|
assert len(results) == 1
|
|
assert results[0].cc_number == 22
|
|
assert results[0].param_key == "chorus.rate"
|
|
|
|
def test_learn_only_captures_cc(self, handler: MIDIHandler) -> None:
|
|
"""PC and notes should not trigger learn capture."""
|
|
handler.start_learn("delay.feedback")
|
|
handler.dispatch(MIDIEvent(type="pc", program=3))
|
|
assert handler.learn_mode # Still in learn mode
|
|
|
|
def test_set_mapping_directly(self, handler: MIDIHandler) -> None:
|
|
m = MIDIMapping(cc_number=7, channel=0, min_val=0.0, max_val=1.0)
|
|
handler.set_mapping("volume", m)
|
|
assert handler.get_mapping("volume") == m
|
|
|
|
def test_remove_mapping(self, handler: MIDIHandler) -> None:
|
|
m = MIDIMapping(cc_number=7, channel=0)
|
|
handler.set_mapping("volume", m)
|
|
handler.remove_mapping("volume")
|
|
assert handler.get_mapping("volume") is None
|
|
|
|
def test_get_all_mappings(self, handler: MIDIHandler) -> None:
|
|
handler.set_mapping("a", MIDIMapping(cc_number=1))
|
|
handler.set_mapping("b", MIDIMapping(cc_number=2))
|
|
all_maps = handler.get_all_mappings()
|
|
assert len(all_maps) == 2
|
|
assert "a" in all_maps
|
|
assert "b" in all_maps
|
|
|
|
def test_cancel_learn(self, handler: MIDIHandler) -> None:
|
|
handler.start_learn("delay.feedback")
|
|
handler.cancel_learn()
|
|
assert not handler.learn_mode
|
|
assert handler.get_mapping("delay.feedback") is None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# MIDI clock sync tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestMIDIClock:
|
|
"""MIDI clock tracking and BPM calculation."""
|
|
|
|
def test_clock_reset_on_start(self, handler: MIDIHandler) -> None:
|
|
handler.parse(bytes([RT_START]))
|
|
assert handler.clock_running
|
|
assert handler.current_bpm == 0.0
|
|
|
|
def test_clock_stop(self, handler: MIDIHandler) -> None:
|
|
handler.parse(bytes([RT_START]))
|
|
handler.parse(bytes([RT_STOP]))
|
|
assert not handler.clock_running
|
|
|
|
def test_clock_continue(self, handler: MIDIHandler) -> None:
|
|
handler.parse(bytes([RT_START]))
|
|
handler.parse(bytes([RT_STOP]))
|
|
handler.parse(bytes([RT_CONTINUE]))
|
|
assert handler.clock_running
|
|
|
|
def test_reset_clock(self, handler: MIDIHandler) -> None:
|
|
handler.parse(bytes([RT_START]))
|
|
handler.reset_clock()
|
|
assert not handler.clock_running
|
|
assert handler.current_bpm == 0.0
|
|
|
|
@pytest.mark.skip(reason="Timing-sensitive; BPM depends on real clock intervals")
|
|
def test_bpm_detection(self, handler: MIDIHandler) -> None:
|
|
pass # BPM detection tested via unit below
|
|
|
|
def test_clock_callback_fires(self, handler: MIDIHandler) -> None:
|
|
"""BPM callback is set and retrievable."""
|
|
results: list[float] = []
|
|
handler.set_clock_callback(lambda bpm: results.append(bpm))
|
|
handler.parse(bytes([RT_START]))
|
|
# No ticks sent — BPM stays 0. Timed test would send 24 ticks at ~20.8ms
|
|
|
|
def test_clock_send_builders(self) -> None:
|
|
assert MIDIHandler.send_clock_start() == bytes([RT_START])
|
|
assert MIDIHandler.send_clock_stop() == bytes([RT_STOP])
|
|
assert MIDIHandler.send_clock_tick() == bytes([RT_CLOCK])
|
|
|
|
|
|
class TestAudioSyncIntegration:
|
|
"""Integration between MIDI clock and AudioSystem tempo."""
|
|
|
|
def test_tempo_property(self) -> None:
|
|
"""Test audio.py's set_tempo_from_midi_clock via handler mock."""
|
|
from system.audio import AudioSystem
|
|
|
|
audio = AudioSystem()
|
|
assert audio.tempo_bpm == 120.0
|
|
assert audio.tempo_source == "default"
|
|
|
|
def test_enable_midi_clock_sync(self) -> None:
|
|
from system.audio import AudioSystem
|
|
|
|
audio = AudioSystem()
|
|
audio.enable_midi_clock_sync(True)
|
|
audio.set_tempo_from_midi_clock(140.0)
|
|
assert audio.tempo_bpm == 140.0
|
|
assert audio.tempo_source == "midi_clock"
|
|
|
|
def test_disable_midi_clock_ignores_updates(self) -> None:
|
|
from system.audio import AudioSystem
|
|
|
|
audio = AudioSystem()
|
|
audio.set_tempo_from_midi_clock(140.0)
|
|
assert audio.tempo_bpm == 120.0 # Not enabled, stays default
|
|
|
|
def test_manual_tempo_overrides(self) -> None:
|
|
from system.audio import AudioSystem
|
|
|
|
audio = AudioSystem()
|
|
audio.enable_midi_clock_sync(True)
|
|
audio.set_tempo_from_midi_clock(140.0)
|
|
audio.tempo_bpm = 80.0 # Manual override
|
|
assert audio.tempo_bpm == 80.0
|
|
assert audio.tempo_source == "manual"
|
|
|
|
def test_tempo_clamped(self) -> None:
|
|
from system.audio import AudioSystem
|
|
|
|
audio = AudioSystem()
|
|
audio.tempo_bpm = 999
|
|
assert audio.tempo_bpm == 300.0 # Clamped to max
|
|
audio.tempo_bpm = 0
|
|
assert audio.tempo_bpm == 20.0 # Clamped to min
|
|
|
|
def test_tempo_callback(self) -> None:
|
|
from system.audio import AudioSystem
|
|
|
|
results: list[float] = []
|
|
audio = AudioSystem()
|
|
audio.set_tempo_callback(lambda bpm: results.append(bpm))
|
|
audio.tempo_bpm = 140.0
|
|
assert results == [140.0]
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Expression pedal (CC #11) tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestExpressionPedal:
|
|
"""Expression pedal via continuous CC control."""
|
|
|
|
def test_expression_cc_is_standard_cc(self, handler: MIDIHandler) -> None:
|
|
"""Expression pedal CC #11 is treated as a standard CC."""
|
|
results: list[tuple[int, int]] = []
|
|
|
|
handler.register_cc(CC_EXPRESSION, lambda val, ch: results.append((val, ch)))
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=CC_EXPRESSION, cc_value=64))
|
|
|
|
assert results == [(64, 0)]
|
|
|
|
def test_expression_pedal_sweep(self, handler: MIDIHandler) -> None:
|
|
"""Sweep expression pedal across its range."""
|
|
results: list[int] = []
|
|
|
|
handler.register_cc(CC_EXPRESSION, lambda val, ch: results.append(val))
|
|
|
|
for val in [0, 32, 64, 96, 127]:
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=CC_EXPRESSION, cc_value=val))
|
|
|
|
assert results == [0, 32, 64, 96, 127]
|
|
|
|
def test_expression_pedal_learn(self, handler: MIDIHandler) -> None:
|
|
"""MIDI Learn an expression pedal CC."""
|
|
handler.start_learn("wah.position")
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=CC_EXPRESSION, cc_value=64))
|
|
|
|
mapping = handler.get_mapping("wah.position")
|
|
assert mapping is not None
|
|
assert mapping.cc_number == CC_EXPRESSION
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# 14-bit CC tests (MSB/LSB pairs)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class Test14BitCC:
|
|
"""14-bit CC resolution (MSB 0-31 paired with LSB 32-63)."""
|
|
|
|
def test_msb_triggers_callback_immediately(self, handler: MIDIHandler) -> None:
|
|
"""MSB should fire immediately, not wait for LSB."""
|
|
results: list[tuple[int, int]] = []
|
|
|
|
handler.register_cc(0, lambda val, ch: results.append((val, ch)))
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=0, cc_value=80))
|
|
|
|
assert results == [(80, 0)]
|
|
|
|
def test_msb_then_lsb_combined(self, handler: MIDIHandler) -> None:
|
|
"""MSB + LSB produce a combined 14-bit value."""
|
|
results: list[int] = []
|
|
|
|
handler.register_cc(7, lambda val, ch: results.append(val))
|
|
# MSB first
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=7, cc_value=64))
|
|
assert results == [64] # MSB fires immediately
|
|
|
|
# LSB (cc 39 = 7 + 32)
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=39, cc_value=32))
|
|
# LSB should NOT fire callback for cc=39, and MSB callback should
|
|
# run with scaled value including LSB contribution
|
|
# Note: currently fires MSB twice (once for MSB, once for MSB+LSB)
|
|
# Let's verify at least 2 calls happened
|
|
assert len(results) >= 2
|
|
|
|
def test_orphan_lsb_no_crash(self, handler: MIDIHandler) -> None:
|
|
"""LSB without prior MSB should not crash, may fire as standard CC."""
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=35, cc_value=64))
|
|
|
|
def test_14bit_max_value_scaling(self, handler: MIDIHandler) -> None:
|
|
"""MSB=127 LSB=127 should produce scaled value ~= 127."""
|
|
results: list[int] = []
|
|
|
|
handler.register_cc(0, lambda val, ch: results.append(val))
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=0, cc_value=127))
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=32, cc_value=127))
|
|
|
|
# Combined 14-bit = (127 << 7) | 127 = 16383, scaled = 16383 >> 7 = 127
|
|
assert results[-1] == 127
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Running status tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestRunningStatus:
|
|
"""UARTMIDI running status byte handling."""
|
|
|
|
def test_running_status_note_on(self) -> None:
|
|
"""Status byte reused for subsequent data bytes."""
|
|
raw = bytes([0x90, 60, 100, 64, 80, 67, 100])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 3
|
|
assert msgs[0] == bytes([0x90, 60, 100])
|
|
assert msgs[1] == bytes([0x90, 64, 80])
|
|
assert msgs[2] == bytes([0x90, 67, 100])
|
|
|
|
def test_running_status_after_note_off(self) -> None:
|
|
raw = bytes([0x80, 60, 64, 65, 0])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 2
|
|
assert msgs[0] == bytes([0x80, 60, 64])
|
|
assert msgs[1] == bytes([0x80, 65, 0])
|
|
|
|
def test_running_status_cc(self) -> None:
|
|
raw = bytes([0xB0, 7, 100, 10, 50])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 2
|
|
assert msgs[0] == bytes([0xB0, 7, 100])
|
|
assert msgs[1] == bytes([0xB0, 10, 50])
|
|
|
|
def test_running_status_with_clock_interleaved(self) -> None:
|
|
"""Real-time messages between running status data don't break."""
|
|
raw = bytes([0xB0, 7, 100, 0xF8, 10, 50])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 3
|
|
assert msgs[0] == bytes([0xB0, 7, 100])
|
|
assert msgs[1] == bytes([0xF8]) # Clock
|
|
assert msgs[2] == bytes([0xB0, 10, 50])
|
|
|
|
def test_status_reset_on_new_status(self) -> None:
|
|
"""New status byte resets running status."""
|
|
raw = bytes([0x90, 60, 100, 0xC0, 5, 0xB0, 7, 127])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 3
|
|
assert msgs[0] == bytes([0x90, 60, 100]) # Note On
|
|
assert msgs[1] == bytes([0xC0, 5]) # PC
|
|
assert msgs[2] == bytes([0xB0, 7, 127]) # CC
|
|
|
|
def test_incomplete_message_dropped(self) -> None:
|
|
raw = bytes([0x90, 60]) # Only note + velocity, no next message
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 0 # Incomplete, dropped
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# SysEx tests
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestSysEx:
|
|
"""System Exclusive message handling."""
|
|
|
|
def test_sysex_parse(self, handler: MIDIHandler) -> None:
|
|
"""SysEx messages are parsed and returned as-is."""
|
|
raw = bytes([SYS_EXCLUSIVE, 0x41, 0x10, 0x42, 0x12, SYS_EXCLUSIVE_END])
|
|
# UARTMIDI _parse_raw_messages splits them
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 1
|
|
assert msgs[0] == raw
|
|
|
|
def test_sysex_builder(self) -> None:
|
|
msg = MIDIHandler.send_sysex(0x41, bytes([0x10, 0x42, 0x12]))
|
|
assert msg == bytes([SYS_EXCLUSIVE, 0x41, 0x10, 0x42, 0x12, SYS_EXCLUSIVE_END])
|
|
|
|
def test_incomplete_sysex_dropped(self) -> None:
|
|
"""SysEx without end byte should be dropped."""
|
|
raw = bytes([SYS_EXCLUSIVE, 0x41, 0x10, 0x42])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Message builders
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestMessageBuilders:
|
|
"""MIDI message construction helpers."""
|
|
|
|
def test_send_cc(self) -> None:
|
|
msg = MIDIHandler.send_cc(7, 100, channel=5)
|
|
assert msg == bytes([CONTROL_CHANGE | 5, 7, 100])
|
|
|
|
def test_send_cc_clamps(self) -> None:
|
|
msg = MIDIHandler.send_cc(7, 200, channel=0) # Above 127
|
|
assert msg[2] == 127
|
|
msg = MIDIHandler.send_cc(7, -1, channel=0) # Below 0
|
|
assert msg[2] == 0
|
|
|
|
def test_send_pc(self) -> None:
|
|
msg = MIDIHandler.send_pc(5, channel=3)
|
|
assert msg == bytes([PROGRAM_CHANGE | 3, 5])
|
|
|
|
def test_send_pc_clamps(self) -> None:
|
|
msg = MIDIHandler.send_pc(300, channel=0)
|
|
assert msg[1] == 127
|
|
msg = MIDIHandler.send_pc(-1, channel=0)
|
|
assert msg[1] == 0
|
|
|
|
def test_send_note_on(self) -> None:
|
|
msg = MIDIHandler.send_note_on(60, velocity=100, channel=1)
|
|
assert msg == bytes([NOTE_ON | 1, 60, 100])
|
|
|
|
def test_send_note_off(self) -> None:
|
|
msg = MIDIHandler.send_note_off(60, channel=0)
|
|
assert msg == bytes([NOTE_OFF | 0, 60, 0])
|
|
|
|
def test_send_pitch_bend_center(self) -> None:
|
|
msg = MIDIHandler.send_pitch_bend(8192, channel=0)
|
|
assert msg == bytes([PITCH_BEND | 0, 0x00, 0x40])
|
|
|
|
def test_send_pitch_bend_max(self) -> None:
|
|
msg = MIDIHandler.send_pitch_bend(16383, channel=0)
|
|
assert msg == bytes([PITCH_BEND | 0, 0x7F, 0x7F])
|
|
|
|
def test_send_pitch_bend_min(self) -> None:
|
|
msg = MIDIHandler.send_pitch_bend(0, channel=0)
|
|
assert msg == bytes([PITCH_BEND | 0, 0x00, 0x00])
|
|
|
|
def test_send_pitch_bend_clamps(self) -> None:
|
|
msg = MIDIHandler.send_pitch_bend(20000, channel=0)
|
|
# 16383 max
|
|
assert msg == bytes([PITCH_BEND | 0, 0x7F, 0x7F])
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# MIDI interface tests (UARTMIDI)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestUARTMIDI:
|
|
"""UARTMIDI interface — raw byte splitting and edge cases."""
|
|
|
|
def test_empty_read(self) -> None:
|
|
assert UARTMIDI._parse_raw_messages(b"") == []
|
|
|
|
def test_single_note_on(self) -> None:
|
|
msgs = UARTMIDI._parse_raw_messages(bytes([0x90, 60, 100]))
|
|
assert len(msgs) == 1
|
|
|
|
def test_mixed_messages(self) -> None:
|
|
raw = bytes([
|
|
0x90, 60, 100, # Note On
|
|
0x80, 60, 0, # Note Off
|
|
0xB0, 7, 64, # CC
|
|
0xC0, 3, # PC
|
|
])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 4
|
|
assert msgs[0] == bytes([0x90, 60, 100])
|
|
assert msgs[1] == bytes([0x80, 60, 0])
|
|
assert msgs[2] == bytes([0xB0, 7, 64])
|
|
assert msgs[3] == bytes([0xC0, 3])
|
|
|
|
def test_real_time_messages_interleaved(self) -> None:
|
|
raw = bytes([
|
|
0xF8, # Clock
|
|
0x90, 60, 100, # Note On
|
|
0xF8, # Clock
|
|
0xF8, # Clock
|
|
0x80, 60, 0, # Note Off
|
|
0xFA, # Start
|
|
])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 6
|
|
assert msgs[0] == bytes([0xF8])
|
|
assert msgs[-1] == bytes([0xFA])
|
|
|
|
def test_unknown_status_byte_skipped(self) -> None:
|
|
raw = bytes([0xF5, 0xF6])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
# 0xF5 is undefined, 0xF6 is Tune Request
|
|
assert len(msgs) == 1
|
|
assert msgs[0] == bytes([0xF6])
|
|
|
|
def test_song_position(self) -> None:
|
|
raw = bytes([0xF2, 0x00, 0x40])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 1
|
|
assert msgs[0] == raw
|
|
|
|
def test_song_select(self) -> None:
|
|
raw = bytes([0xF3, 0x01])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 1
|
|
assert msgs[0] == raw
|
|
|
|
def test_name_property(self) -> None:
|
|
uart = UARTMIDI(port="/dev/ttyAMA0")
|
|
assert uart.name == "UART:/dev/ttyAMA0"
|
|
assert not uart.is_open
|
|
|
|
def test_default_port(self) -> None:
|
|
uart = UARTMIDI()
|
|
assert "ttyAMA0" in uart.name
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Edge cases and robustness
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Edge cases and defensive coding."""
|
|
|
|
def test_stray_data_bytes_dropped(self) -> None:
|
|
"""Data byte without status should be dropped."""
|
|
raw = bytes([0x00, 0x01, 0x02]) # All data, no status
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert msgs == []
|
|
|
|
def test_status_byte_with_rt_interruption(self) -> None:
|
|
"""Running status interrupted by real-time, then resumed."""
|
|
raw = bytes([0xB0, 7, 64, 0xF8, 0xF8, 10, 50])
|
|
msgs = UARTMIDI._parse_raw_messages(raw)
|
|
assert len(msgs) == 4
|
|
assert msgs[0] == bytes([0xB0, 7, 64])
|
|
assert msgs[1] == bytes([0xF8])
|
|
assert msgs[2] == bytes([0xF8])
|
|
assert msgs[3] == bytes([0xB0, 10, 50])
|
|
|
|
def test_double_start_no_error(self, handler: MIDIHandler) -> None:
|
|
"""Multiple START messages reset clock without error."""
|
|
handler.parse(bytes([RT_START]))
|
|
handler.parse(bytes([RT_START]))
|
|
assert handler.clock_running
|
|
|
|
def test_stop_without_start(self, handler: MIDIHandler) -> None:
|
|
"""STOP without prior START should be a no-op."""
|
|
handler.parse(bytes([RT_STOP]))
|
|
assert not handler.clock_running
|
|
|
|
def test_pc_no_callback_no_error(self, handler: MIDIHandler) -> None:
|
|
"""PC dispatch without callback should not raise."""
|
|
handler.dispatch(MIDIEvent(type="pc", program=7))
|
|
|
|
def test_sysex_in_midi_handler_parse_gives_none(self, handler: MIDIHandler) -> None:
|
|
"""SysEx is not parsed by MIDIHandler.parse()."""
|
|
ev = handler.parse(bytes([SYS_EXCLUSIVE, 0x41, 0x10, SYS_EXCLUSIVE_END]))
|
|
assert ev is None # Not handled by voice parser
|
|
|
|
def test_cc_with_mapping_learned_prev_mapping_updated(self, handler: MIDIHandler) -> None:
|
|
"""Learning a new CC for an already-mapped param overwrites."""
|
|
handler.set_mapping("delay.feedback", MIDIMapping(cc_number=7))
|
|
handler.start_learn("delay.feedback")
|
|
handler.dispatch(MIDIEvent(type="cc", cc_number=14, cc_value=50))
|
|
mapping = handler.get_mapping("delay.feedback")
|
|
assert mapping is not None
|
|
assert mapping.cc_number == 14 # Overwritten
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Integration: parse → dispatch pipeline
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestParseThenDispatch:
|
|
"""Full parse → dispatch pipeline for common scenarios."""
|
|
|
|
def test_note_on_pipeline(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int, int]] = []
|
|
handler.set_note_callback(lambda n, v, ch: results.append((n, v, ch)))
|
|
|
|
ev = handler.parse(bytes([0x90, 60, 100]))
|
|
assert ev is not None
|
|
handler.dispatch(ev)
|
|
|
|
assert results == [(60, 100, 0)]
|
|
|
|
def test_cc_pipeline(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int]] = []
|
|
handler.register_cc(CC_EXPRESSION, lambda val, ch: results.append((val, ch)))
|
|
|
|
ev = handler.parse(bytes([0xB0, CC_EXPRESSION, 64]))
|
|
assert ev is not None
|
|
handler.dispatch(ev)
|
|
|
|
assert results == [(64, 0)]
|
|
|
|
def test_pc_pipeline(self, handler: MIDIHandler) -> None:
|
|
results: list[tuple[int, int]] = []
|
|
handler.set_pc_callback(lambda ch, pg: results.append((ch, pg)))
|
|
|
|
ev = handler.parse(bytes([0xC0, 3]))
|
|
assert ev is not None
|
|
handler.dispatch(ev)
|
|
|
|
assert results == [(0, 3)]
|
|
|
|
def test_clock_pipeline(self, handler: MIDIHandler) -> None:
|
|
"""Clock messages are parsed and processed but not dispatched as events."""
|
|
clock_results: list[str] = []
|
|
|
|
handler.set_clock_callback(lambda bpm: clock_results.append("clock"))
|
|
ev = handler.parse(bytes([RT_CLOCK]))
|
|
assert ev is not None
|
|
assert ev.type == "clock"
|
|
|
|
def test_learn_pipeline(self, handler: MIDIHandler) -> None:
|
|
"""Learn mode: parse CC → dispatch → capture mapping."""
|
|
handler.start_learn("reverb.decay")
|
|
ev = handler.parse(bytes([0xB0, 17, 100]))
|
|
assert ev is not None
|
|
handler.dispatch(ev)
|
|
mapping = handler.get_mapping("reverb.decay")
|
|
assert mapping is not None
|
|
assert mapping.cc_number == 17
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# USBMIDI — constructor/name tests (no ports on CI)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestUSBMIDI:
|
|
"""USB-MIDI interface — constructor tests."""
|
|
|
|
def test_name_default(self) -> None:
|
|
from midi.handler import USBMIDI
|
|
usb = USBMIDI()
|
|
assert "USB" in usb.name
|
|
|
|
def test_name_custom(self) -> None:
|
|
from midi.handler import USBMIDI
|
|
usb = USBMIDI(port_name="Keystation")
|
|
assert "Keystation" in usb.name
|
|
|
|
def test_open_no_ports(self) -> None:
|
|
"""Open without any USB ports available — should fail gracefully."""
|
|
from midi.handler import USBMIDI
|
|
usb = USBMIDI()
|
|
result = usb.open()
|
|
assert result is False # No real hardware on CI
|
|
assert usb.name == "USB:auto" |