shawn
9cd8292acc
Lint & Validate / lint (push) Has been cancelled
P2-R1: ALSA + JACK2 low-latency config (scripts, quirks, tuning) P2-R2: Carla integration (build scripts, 8ch rack config, NAM LV2 support) P2-R3: Plugin manager, categories, blacklist, NAM model support P3-R1: Mixer DSP engine (channel strip, routing matrix, bus mgr, automation) P4-R1: MIDI engine (learn mode, clock sync, HID discovery, mapping store) P4-R2: Network API (OSC server, FastAPI REST, WebSocket, auth, rate limiter) P5-R1: Touchscreen UI evaluation + main entry point docs: Audio stack, Carla integration, MIDI support, UI evaluation tests: Full test suite (292 passing)
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""Tests for OSC server."""
|
|
|
|
import asyncio
|
|
import socket
|
|
import struct
|
|
import time
|
|
import pytest
|
|
|
|
from src.midi.types import ParameterType, MixerParameter, ParameterCategory
|
|
from src.mixer.osc_client import encode_osc, decode_osc
|
|
from src.network.osc_server import (
|
|
OSCServer,
|
|
_parse_osc_address,
|
|
_extract_value,
|
|
DEFAULT_OSC_PORT,
|
|
)
|
|
|
|
|
|
class TestOSCAddressParsing:
|
|
"""Tests for OSC address parsing."""
|
|
|
|
def test_channel_volume(self):
|
|
result = _parse_osc_address("/mixer/channel/0/volume")
|
|
assert result is not None
|
|
cat, pt, ch, _ = result
|
|
assert cat == ParameterCategory.CHANNEL
|
|
assert pt == ParameterType.VOLUME
|
|
assert ch == 0
|
|
|
|
def test_channel_mute(self):
|
|
result = _parse_osc_address("/mixer/channel/5/mute")
|
|
cat, pt, ch, _ = result
|
|
assert pt == ParameterType.MUTE
|
|
assert ch == 5
|
|
|
|
def test_channel_eq_low_gain(self):
|
|
result = _parse_osc_address("/mixer/channel/3/eq_low_gain")
|
|
cat, pt, ch, _ = result
|
|
assert pt == ParameterType.EQ_LOW_GAIN
|
|
assert ch == 3
|
|
|
|
def test_channel_comp_ratio(self):
|
|
result = _parse_osc_address("/mixer/channel/7/comp_ratio")
|
|
cat, pt, ch, _ = result
|
|
assert pt == ParameterType.COMP_RATIO
|
|
|
|
def test_master_volume(self):
|
|
result = _parse_osc_address("/mixer/master/volume")
|
|
cat, pt, ch, _ = result
|
|
assert cat == ParameterCategory.MASTER
|
|
assert pt == ParameterType.MASTER_VOLUME
|
|
assert ch == -1
|
|
|
|
def test_master_mute(self):
|
|
result = _parse_osc_address("/mixer/master/mute")
|
|
cat, pt, ch, _ = result
|
|
assert pt == ParameterType.MASTER_MUTE
|
|
|
|
def test_master_dim(self):
|
|
result = _parse_osc_address("/mixer/master/dim")
|
|
cat, pt, ch, _ = result
|
|
assert pt == ParameterType.MASTER_DIM
|
|
|
|
def test_fx_return(self):
|
|
result = _parse_osc_address("/mixer/fx/return_a")
|
|
cat, pt, ch, _ = result
|
|
assert cat == ParameterCategory.FX
|
|
assert pt == ParameterType.FX_RETURN_A
|
|
|
|
def test_transport_play(self):
|
|
result = _parse_osc_address("/mixer/transport/play")
|
|
cat, pt, ch, _ = result
|
|
assert cat == ParameterCategory.TRANSPORT
|
|
assert pt == ParameterType.PLAY
|
|
|
|
def test_transport_tempo(self):
|
|
result = _parse_osc_address("/mixer/transport/tempo")
|
|
_, pt, _, _ = result
|
|
assert pt == ParameterType.TEMPO
|
|
|
|
def test_utility_snapshot_load(self):
|
|
result = _parse_osc_address("/mixer/utility/snapshot_load")
|
|
_, pt, _, _ = result
|
|
assert pt == ParameterType.SNAPSHOT_LOAD
|
|
|
|
def test_unknown_address(self):
|
|
result = _parse_osc_address("/mixer/nonexistent/path")
|
|
assert result is None
|
|
|
|
def test_unknown_param_name(self):
|
|
result = _parse_osc_address("/mixer/channel/0/invalid_param")
|
|
assert result is None
|
|
|
|
def test_invalid_channel_number(self):
|
|
result = _parse_osc_address("/mixer/channel/abc/volume")
|
|
assert result is None
|
|
|
|
|
|
class TestExtractValue:
|
|
"""Tests for _extract_value."""
|
|
|
|
def test_float(self):
|
|
assert _extract_value([3.14]) == pytest.approx(3.14)
|
|
|
|
def test_int(self):
|
|
assert _extract_value([42]) == 42.0
|
|
|
|
def test_bool_true(self):
|
|
assert _extract_value([True]) == 1.0
|
|
|
|
def test_bool_false(self):
|
|
assert _extract_value([False]) == 0.0
|
|
|
|
def test_string_number(self):
|
|
assert _extract_value(["0.75"]) == 0.75
|
|
|
|
def test_empty_args(self):
|
|
assert _extract_value([]) == 0.0
|
|
|
|
def test_multiple_args_uses_first(self):
|
|
assert _extract_value([0.5, 1.0, 2.0]) == 0.5
|
|
|
|
def test_zero(self):
|
|
assert _extract_value([0.0]) == 0.0
|
|
|
|
def test_negative(self):
|
|
assert _extract_value([-6.0]) == -6.0
|
|
|
|
|
|
class TestOSCServer:
|
|
"""Tests for the async OSC server.
|
|
|
|
Each test manages its own server lifecycle using asyncio.run().
|
|
"""
|
|
|
|
@staticmethod
|
|
def _send_udp(packet: bytes, host: str, port: int) -> None:
|
|
"""Send a UDP packet."""
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
sock.sendto(packet, (host, port))
|
|
finally:
|
|
sock.close()
|
|
|
|
@staticmethod
|
|
def _find_free_port() -> int:
|
|
"""Find a free UDP port."""
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
return port
|
|
|
|
async def _create_server(self, received: list, port: int | None = None) -> OSCServer:
|
|
"""Create and start a server, returning it."""
|
|
port = port or self._find_free_port()
|
|
server = OSCServer(host="127.0.0.1", port=port)
|
|
server.set_dispatcher(lambda p, v: received.append((p, v)))
|
|
await server.start()
|
|
return server
|
|
|
|
# ── Test runners (synchronous entry points) ──────────────────────
|
|
|
|
def test_receive_volume(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
packet = encode_osc("/mixer/channel/0/volume", -3.0)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
|
|
assert len(received) == 1
|
|
param, value = received[0]
|
|
assert param.param_type == ParameterType.VOLUME
|
|
assert param.channel == 0
|
|
assert value == pytest.approx(-3.0)
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_receive_mute(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
packet = encode_osc("/mixer/channel/2/mute", True)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
|
|
assert len(received) == 1
|
|
param, value = received[0]
|
|
assert param.param_type == ParameterType.MUTE
|
|
assert param.channel == 2
|
|
assert value == 1.0
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_receive_master_volume(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
packet = encode_osc("/mixer/master/volume", -10.0)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
|
|
assert len(received) == 1
|
|
param, value = received[0]
|
|
assert param.param_type == ParameterType.MASTER_VOLUME
|
|
assert value == pytest.approx(-10.0)
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_receive_transport(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
packet = encode_osc("/mixer/transport/play", 1.0)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
|
|
assert len(received) == 1
|
|
param, _ = received[0]
|
|
assert param.param_type == ParameterType.PLAY
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_receive_multiple(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
msgs = [
|
|
encode_osc("/mixer/channel/0/volume", -6.0),
|
|
encode_osc("/mixer/channel/0/pan", 0.5),
|
|
encode_osc("/mixer/channel/1/mute", 1.0),
|
|
]
|
|
for packet in msgs:
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
|
|
await asyncio.sleep(0.15)
|
|
|
|
assert len(received) == 3
|
|
types = [p.param_type for p, _ in received]
|
|
assert ParameterType.VOLUME in types
|
|
assert ParameterType.PAN in types
|
|
assert ParameterType.MUTE in types
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_unknown_address_not_dispatched(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
packet = encode_osc("/unknown/path", 1.0)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
assert len(received) == 0
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_stats(self):
|
|
async def run():
|
|
received = []
|
|
port = self._find_free_port()
|
|
server = await self._create_server(received, port)
|
|
try:
|
|
packet = encode_osc("/mixer/channel/0/volume", 0.0)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
|
|
stats = server.stats
|
|
assert stats["running"] is True
|
|
assert stats["messages"] >= 1
|
|
assert stats["host"] == "127.0.0.1"
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_no_dispatcher_no_crash(self):
|
|
async def run():
|
|
port = self._find_free_port()
|
|
server = OSCServer(host="127.0.0.1", port=port)
|
|
await server.start()
|
|
try:
|
|
packet = encode_osc("/mixer/channel/0/volume", 0.0)
|
|
self._send_udp(packet, "127.0.0.1", port)
|
|
await asyncio.sleep(0.1)
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|
|
|
|
def test_start_stop(self):
|
|
async def run():
|
|
port = self._find_free_port()
|
|
server = OSCServer(host="127.0.0.1", port=port)
|
|
assert not server.running
|
|
|
|
await server.start()
|
|
assert server.running
|
|
|
|
await server.stop()
|
|
assert not server.running
|
|
asyncio.run(run())
|
|
|
|
def test_double_start(self):
|
|
async def run():
|
|
port = self._find_free_port()
|
|
server = OSCServer(host="127.0.0.1", port=port)
|
|
await server.start()
|
|
try:
|
|
await server.start() # should be idempotent
|
|
assert server.running
|
|
finally:
|
|
await server.stop()
|
|
asyncio.run(run())
|