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)
272 lines
8.0 KiB
Python
272 lines
8.0 KiB
Python
"""Tests for JSON schemas / Pydantic models."""
|
|
|
|
import json
|
|
import pytest
|
|
from datetime import datetime
|
|
|
|
from src.network.schemas import (
|
|
ChannelSchema,
|
|
MasterBusSchema,
|
|
MixerStateSchema,
|
|
PluginSchema,
|
|
TransportSchema,
|
|
RoutingSchema,
|
|
AuxBusSchema,
|
|
SubgroupSchema,
|
|
VCASchema,
|
|
EQSchema,
|
|
EQBandSchema,
|
|
CompressorSchema,
|
|
GateSchema,
|
|
FXSendsSchema,
|
|
ParameterUpdateSchema,
|
|
ParameterUpdateBatchSchema,
|
|
ParameterSetRequest,
|
|
WSMessage,
|
|
WSMessageType,
|
|
MixerParameterSchema,
|
|
ParameterCategoryEnum,
|
|
ParameterTypeEnum,
|
|
RouteNodeSchema,
|
|
RoutingEdgeSchema,
|
|
NodeTypeEnum,
|
|
APIResponse,
|
|
PluginParamMapSchema,
|
|
)
|
|
|
|
|
|
class TestChannelSchema:
|
|
"""Tests for ChannelSchema."""
|
|
|
|
def test_default_channel(self):
|
|
ch = ChannelSchema(channel=0)
|
|
assert ch.channel == 0
|
|
assert ch.volume_db == 0.0
|
|
assert ch.pan == 0.0
|
|
assert not ch.mute
|
|
assert not ch.solo
|
|
|
|
def test_channel_with_eq(self):
|
|
ch = ChannelSchema(
|
|
channel=3,
|
|
volume_db=-6.0,
|
|
eq=EQSchema(
|
|
enabled=True,
|
|
low=EQBandSchema(freq_hz=80.0, gain_db=3.0, q=1.0),
|
|
),
|
|
)
|
|
assert ch.eq.enabled
|
|
assert ch.eq.low.freq_hz == 80.0
|
|
assert ch.eq.low.gain_db == 3.0
|
|
|
|
def test_serialize_deserialize(self):
|
|
ch = ChannelSchema(channel=5, volume_db=-3.0, pan=0.5)
|
|
data = ch.model_dump()
|
|
ch2 = ChannelSchema.model_validate(data)
|
|
assert ch2.channel == 5
|
|
assert ch2.volume_db == -3.0
|
|
assert ch2.pan == 0.5
|
|
|
|
def test_json_roundtrip(self):
|
|
ch = ChannelSchema(channel=0, mute=True)
|
|
json_str = ch.model_dump_json()
|
|
data = json.loads(json_str)
|
|
assert data["channel"] == 0
|
|
assert data["mute"] is True
|
|
|
|
|
|
class TestMasterBusSchema:
|
|
"""Tests for MasterBusSchema."""
|
|
|
|
def test_default_master(self):
|
|
m = MasterBusSchema()
|
|
assert m.volume_db == 0.0
|
|
assert not m.muted
|
|
assert not m.dim_active
|
|
assert m.aux_buses == []
|
|
|
|
def test_with_buses(self):
|
|
m = MasterBusSchema(
|
|
volume_db=-10.0,
|
|
aux_buses=[
|
|
AuxBusSchema(index=0, label="Reverb"),
|
|
AuxBusSchema(index=1, label="Delay"),
|
|
],
|
|
subgroups=[
|
|
SubgroupSchema(index=0, label="Drums", members=[0, 1, 2]),
|
|
],
|
|
vca_groups=[
|
|
VCASchema(index=0, label="All Vocals", master_db=-3.0, members=[3, 4, 5]),
|
|
],
|
|
)
|
|
assert len(m.aux_buses) == 2
|
|
assert m.aux_buses[0].label == "Reverb"
|
|
assert len(m.subgroups) == 1
|
|
assert m.subgroups[0].members == [0, 1, 2]
|
|
assert m.vca_groups[0].master_db == -3.0
|
|
|
|
|
|
class TestTransportSchema:
|
|
"""Tests for TransportSchema."""
|
|
|
|
def test_defaults(self):
|
|
t = TransportSchema()
|
|
assert not t.playing
|
|
assert not t.recording
|
|
assert t.tempo_bpm == 120.0
|
|
assert t.position_sec == 0.0
|
|
|
|
def test_playing_state(self):
|
|
t = TransportSchema(playing=True, tempo_bpm=142.0, position_sec=35.2)
|
|
assert t.playing
|
|
assert not t.recording
|
|
assert t.tempo_bpm == 142.0
|
|
|
|
|
|
class TestMixerStateSchema:
|
|
"""Tests for full MixerStateSchema."""
|
|
|
|
def test_empty_state(self):
|
|
state = MixerStateSchema()
|
|
assert state.channels == []
|
|
assert state.master.volume_db == 0.0
|
|
assert state.transport.tempo_bpm == 120.0
|
|
|
|
def test_full_state(self):
|
|
state = MixerStateSchema(
|
|
channels=[
|
|
ChannelSchema(channel=0, volume_db=-6.0),
|
|
ChannelSchema(channel=1, mute=True),
|
|
],
|
|
master=MasterBusSchema(volume_db=-3.0),
|
|
transport=TransportSchema(playing=True),
|
|
scenes=["Intro", "Verse", "Chorus"],
|
|
uptime_seconds=3600.0,
|
|
)
|
|
assert len(state.channels) == 2
|
|
assert state.scenes == ["Intro", "Verse", "Chorus"]
|
|
assert state.uptime_seconds == 3600.0
|
|
|
|
def test_json_roundtrip(self):
|
|
state = MixerStateSchema(
|
|
channels=[ChannelSchema(channel=0)],
|
|
uptime_seconds=10.0,
|
|
)
|
|
json_str = state.model_dump_json()
|
|
data = json.loads(json_str)
|
|
assert data["uptime_seconds"] == 10.0
|
|
assert len(data["channels"]) == 1
|
|
|
|
|
|
class TestParameterUpdate:
|
|
"""Tests for parameter update schemas."""
|
|
|
|
def test_single_update(self):
|
|
update = ParameterUpdateSchema(
|
|
param_type=ParameterTypeEnum.VOLUME,
|
|
value=-3.0,
|
|
channel=2,
|
|
)
|
|
assert update.param_type == ParameterTypeEnum.VOLUME
|
|
assert update.value == -3.0
|
|
assert update.channel == 2
|
|
|
|
def test_batch_update(self):
|
|
batch = ParameterUpdateBatchSchema(updates=[
|
|
ParameterUpdateSchema(param_type=ParameterTypeEnum.VOLUME, value=-6.0, channel=0),
|
|
ParameterUpdateSchema(param_type=ParameterTypeEnum.MUTE, value=1.0, channel=0),
|
|
ParameterUpdateSchema(param_type=ParameterTypeEnum.MASTER_VOLUME, value=-3.0),
|
|
])
|
|
assert len(batch.updates) == 3
|
|
assert batch.updates[2].param_type == ParameterTypeEnum.MASTER_VOLUME
|
|
|
|
|
|
class TestWSMessage:
|
|
"""Tests for WebSocket message schemas."""
|
|
|
|
def test_parameter_update_message(self):
|
|
msg = WSMessage(
|
|
type=WSMessageType.PARAMETER_UPDATE,
|
|
payload={"param_type": "volume", "value": -3.0, "channel": 1},
|
|
)
|
|
assert msg.type == WSMessageType.PARAMETER_UPDATE
|
|
assert msg.payload["param_type"] == "volume"
|
|
|
|
def test_full_state_message(self):
|
|
msg = WSMessage(
|
|
type=WSMessageType.FULL_STATE,
|
|
payload={"channels": []},
|
|
)
|
|
data = msg.model_dump()
|
|
assert data["type"] == "full_state"
|
|
|
|
def test_error_message(self):
|
|
msg = WSMessage(
|
|
type=WSMessageType.ERROR,
|
|
payload={"message": "Something went wrong"},
|
|
)
|
|
assert msg.payload["message"] == "Something went wrong"
|
|
|
|
|
|
class TestParameterSetRequest:
|
|
"""Tests for ParameterSetRequest."""
|
|
|
|
def test_default_param_type(self):
|
|
req = ParameterSetRequest(value=-3.0)
|
|
assert req.param_type == "volume"
|
|
assert req.value == -3.0
|
|
|
|
def test_explicit_param_type(self):
|
|
req = ParameterSetRequest(param_type="mute", value=1.0)
|
|
assert req.param_type == "mute"
|
|
|
|
def test_json_parse(self):
|
|
json_str = '{"param_type": "pan", "value": 0.5}'
|
|
req = ParameterSetRequest.model_validate_json(json_str)
|
|
assert req.param_type == "pan"
|
|
assert req.value == 0.5
|
|
|
|
|
|
class TestPluginSchema:
|
|
"""Tests for PluginSchema."""
|
|
|
|
def test_plugin(self):
|
|
p = PluginSchema(
|
|
plugin_id=1,
|
|
name="Ch1 Gate",
|
|
role="gate",
|
|
channel=0,
|
|
params=[
|
|
PluginParamMapSchema(name="threshold", index=0, value=0.5),
|
|
PluginParamMapSchema(name="ratio", index=1, value=0.3),
|
|
],
|
|
)
|
|
assert p.plugin_id == 1
|
|
assert len(p.params) == 2
|
|
assert p.params[0].name == "threshold"
|
|
|
|
|
|
class TestRoutingSchema:
|
|
"""Tests for RoutingSchema."""
|
|
|
|
def test_empty_routing(self):
|
|
r = RoutingSchema()
|
|
assert r.nodes == []
|
|
assert r.edges == []
|
|
|
|
def test_with_nodes_and_edges(self):
|
|
r = RoutingSchema(
|
|
nodes=[
|
|
RouteNodeSchema(node_id="sys_in_1", type=NodeTypeEnum.SYSTEM_INPUT, jack_port="system:capture_1"),
|
|
RouteNodeSchema(node_id="ch_0_input", type=NodeTypeEnum.CHANNEL_INPUT, channel=0),
|
|
],
|
|
edges=[
|
|
RoutingEdgeSchema(source="sys_in_1", dest="ch_0_input"),
|
|
],
|
|
solo_nodes=["ch_0_input"],
|
|
)
|
|
assert len(r.nodes) == 2
|
|
assert len(r.edges) == 1
|
|
assert r.edges[0].source == "sys_in_1"
|
|
assert r.solo_nodes == ["ch_0_input"]
|