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)
978 lines
34 KiB
Python
978 lines
34 KiB
Python
"""Tests for the mixer DSP engine components.
|
||
|
||
Tests are designed to run without JACK or Carla — they validate
|
||
the logic, state management, and parameter dispatch without
|
||
requiring actual audio hardware.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
import time
|
||
import math
|
||
|
||
from src.mixer.osc_client import (
|
||
encode_osc,
|
||
decode_osc,
|
||
CarlaOSCClient,
|
||
CarlaPluginInfo,
|
||
DEFAULT_PLUGIN_LAYOUT,
|
||
linear_to_db,
|
||
db_to_linear,
|
||
freq_to_normalized,
|
||
normalized_to_freq,
|
||
time_ms_to_normalized,
|
||
mix_to_pan,
|
||
)
|
||
from src.mixer.channel_strip import (
|
||
ChannelStrip,
|
||
ChannelState,
|
||
_apply_state,
|
||
_transform_value,
|
||
_get_osc_commands,
|
||
_normalize_param,
|
||
)
|
||
from src.mixer.routing_matrix import (
|
||
RoutingMatrix,
|
||
RouteNode,
|
||
RoutingEdge,
|
||
NodeType,
|
||
)
|
||
from src.mixer.bus_manager import (
|
||
BusManager,
|
||
AuxBus,
|
||
SubgroupBus,
|
||
VCAGroup,
|
||
MasterBus,
|
||
BusType,
|
||
)
|
||
from src.mixer.fader_automation import (
|
||
FaderAutomation,
|
||
AutomationLane,
|
||
AutomationPoint,
|
||
Scene,
|
||
InterpolationMode,
|
||
_interpolate,
|
||
)
|
||
from src.mixer.dsp_engine import (
|
||
DSPEngine,
|
||
DSPEngineConfig,
|
||
create_default_engine,
|
||
_make_param_key,
|
||
_parse_param_key,
|
||
)
|
||
from src.midi.types import (
|
||
ParameterType,
|
||
ParameterCategory,
|
||
MixerParameter,
|
||
)
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# OSC Protocol Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestOSCEncoding:
|
||
"""Test OSC message encoding/decoding."""
|
||
|
||
def test_encode_float(self):
|
||
packet = encode_osc("/test", 1.0)
|
||
assert b"/test" in packet
|
||
assert b",f" in packet
|
||
|
||
def test_encode_int(self):
|
||
packet = encode_osc("/test", 42)
|
||
assert b"/test" in packet
|
||
assert b",i" in packet
|
||
|
||
def test_encode_string(self):
|
||
packet = encode_osc("/test", "hello")
|
||
assert b"/test" in packet
|
||
assert b",s" in packet
|
||
assert b"hello" in packet
|
||
|
||
def test_encode_multiple_args(self):
|
||
packet = encode_osc("/Carla/1/set_parameter_value", 3, 0, 0.5)
|
||
assert b"/Carla/1/set_parameter_value" in packet
|
||
assert b",iif" in packet
|
||
|
||
def test_decode_float(self):
|
||
packet = encode_osc("/test", 3.14)
|
||
addr, args = decode_osc(packet)
|
||
assert addr == "/test"
|
||
assert len(args) == 1
|
||
assert abs(args[0] - 3.14) < 0.001
|
||
|
||
def test_decode_int(self):
|
||
packet = encode_osc("/test", 99)
|
||
addr, args = decode_osc(packet)
|
||
assert addr == "/test"
|
||
assert args[0] == 99
|
||
|
||
def test_decode_string(self):
|
||
packet = encode_osc("/test", "world")
|
||
addr, args = decode_osc(packet)
|
||
assert addr == "/test"
|
||
assert args[0] == "world"
|
||
|
||
def test_decode_multiple(self):
|
||
packet = encode_osc("/path", 1.0, 42, "abc")
|
||
addr, args = decode_osc(packet)
|
||
assert addr == "/path"
|
||
assert len(args) == 3
|
||
assert abs(args[0] - 1.0) < 0.001
|
||
assert args[1] == 42
|
||
assert args[2] == "abc"
|
||
|
||
def test_roundtrip(self):
|
||
"""Encode then decode should return the same values."""
|
||
original_addr = "/some/path"
|
||
original_args = [0.75, 128, "test_value"]
|
||
packet = encode_osc(original_addr, *original_args)
|
||
addr, args = decode_osc(packet)
|
||
assert addr == original_addr
|
||
assert len(args) == len(original_args)
|
||
for a, b in zip(args, original_args):
|
||
if isinstance(a, float) and isinstance(b, float):
|
||
assert abs(a - b) < 0.001
|
||
else:
|
||
assert a == b
|
||
|
||
|
||
class TestCarlaOSCClient:
|
||
"""Test the OSC client (without actual network)."""
|
||
|
||
def test_client_creation(self):
|
||
client = CarlaOSCClient()
|
||
assert client.host == "127.0.0.1"
|
||
assert client.port == 22752
|
||
client.close()
|
||
|
||
def test_client_context_manager(self):
|
||
with CarlaOSCClient() as client:
|
||
assert client.host == "127.0.0.1"
|
||
|
||
def test_stats(self):
|
||
client = CarlaOSCClient()
|
||
stats = client.stats
|
||
assert stats["host"] == "127.0.0.1"
|
||
assert stats["sends"] == 0
|
||
assert stats["errors"] == 0
|
||
client.close()
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# Value Conversion Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestValueConversions:
|
||
|
||
def test_linear_to_db_silence(self):
|
||
# With default min_db=-60.0, linear=0 returns -inf (true digital silence)
|
||
assert linear_to_db(0.0) == float("-inf")
|
||
|
||
def test_linear_to_db_unity(self):
|
||
assert abs(linear_to_db(0.5) - (-24.0)) < 1.0
|
||
|
||
def test_linear_to_db_max(self):
|
||
assert abs(linear_to_db(1.0) - 12.0) < 0.1
|
||
|
||
def test_db_to_linear_roundtrip(self):
|
||
# Skip -60 (floor): mathematically db_to_linear(-60)=0.0, linear_to_db(0.0)=-inf
|
||
for db in [-40, -20, 0, 6, 12]:
|
||
lin = db_to_linear(db)
|
||
db2 = linear_to_db(lin)
|
||
assert abs(db - db2) < 0.5, f"Roundtrip failed at {db} dB"
|
||
|
||
def test_freq_to_normalized(self):
|
||
assert freq_to_normalized(20) < 0.01
|
||
assert freq_to_normalized(20000) > 0.99
|
||
assert 0.4 < freq_to_normalized(1000) < 0.6
|
||
|
||
def test_normalized_to_freq(self):
|
||
for hz in [50, 200, 1000, 5000, 15000]:
|
||
norm = freq_to_normalized(hz)
|
||
hz2 = normalized_to_freq(norm)
|
||
ratio = hz2 / hz
|
||
assert 0.8 < ratio < 1.25, f"Failed at {hz} Hz: got {hz2:.0f}"
|
||
|
||
def test_time_ms_to_normalized(self):
|
||
assert time_ms_to_normalized(0.1) < 0.01
|
||
assert time_ms_to_normalized(2000) > 0.99
|
||
assert 0.0 < time_ms_to_normalized(100) < 0.5
|
||
|
||
def test_mix_to_pan_center(self):
|
||
l, r = mix_to_pan(1.0, 0.0) # mix full, pan center
|
||
assert abs(l - r) < 0.001
|
||
assert abs(l - 0.707) < 0.01 # cos(pi/4)
|
||
|
||
def test_mix_to_pan_left(self):
|
||
l, r = mix_to_pan(1.0, -1.0)
|
||
assert l > 0.99
|
||
assert r < 0.01
|
||
|
||
def test_mix_to_pan_right(self):
|
||
l, r = mix_to_pan(1.0, 1.0)
|
||
assert l < 0.01
|
||
assert r > 0.99
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# Channel Strip Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestChannelStrip:
|
||
|
||
def test_default_state(self):
|
||
strip = ChannelStrip(index=0)
|
||
assert strip.index == 0
|
||
assert strip.state.volume == 0.0
|
||
assert strip.state.mute is False
|
||
assert strip.state.pan == 0.0
|
||
|
||
def test_set_volume(self):
|
||
strip = ChannelStrip(index=0)
|
||
strip.set_parameter(ParameterType.VOLUME, -6.0)
|
||
assert strip.state.volume == -6.0
|
||
|
||
def test_set_mute(self):
|
||
strip = ChannelStrip(index=0)
|
||
strip.set_parameter(ParameterType.MUTE, 1.0)
|
||
assert strip.state.mute is True
|
||
strip.set_parameter(ParameterType.MUTE, 0.0)
|
||
assert strip.state.mute is False
|
||
|
||
def test_set_eq(self):
|
||
strip = ChannelStrip(index=3)
|
||
strip.set_parameter(ParameterType.EQ_LOW_FREQ, 120.0)
|
||
strip.set_parameter(ParameterType.EQ_LOW_GAIN, 3.0)
|
||
strip.set_parameter(ParameterType.EQ_LOW_Q, 1.5)
|
||
assert strip.state.eq_low_freq == 120.0
|
||
assert strip.state.eq_low_gain == 3.0
|
||
assert strip.state.eq_low_q == 1.5
|
||
|
||
def test_set_compressor(self):
|
||
strip = ChannelStrip(index=1)
|
||
strip.set_parameter(ParameterType.COMP_THRESHOLD, -25.0)
|
||
strip.set_parameter(ParameterType.COMP_RATIO, 4.0)
|
||
strip.set_parameter(ParameterType.COMP_ATTACK, 5.0)
|
||
assert strip.state.comp_threshold == -25.0
|
||
assert strip.state.comp_ratio == 4.0
|
||
assert strip.state.comp_attack == 5.0
|
||
|
||
def test_set_gate(self):
|
||
strip = ChannelStrip(index=2)
|
||
strip.set_parameter(ParameterType.GATE_THRESHOLD, -50.0)
|
||
strip.set_parameter(ParameterType.GATE_RANGE, -70.0)
|
||
assert strip.state.gate_threshold == -50.0
|
||
assert strip.state.gate_range == -70.0
|
||
|
||
def test_set_fx_sends(self):
|
||
strip = ChannelStrip(index=0)
|
||
strip.set_parameter(ParameterType.FX_SEND_A, -10.0)
|
||
strip.set_parameter(ParameterType.FX_SEND_B, -20.0)
|
||
assert strip.state.fx_send_a == -10.0
|
||
assert strip.state.fx_send_b == -20.0
|
||
|
||
def test_snapshot_restore(self):
|
||
strip = ChannelStrip(index=5)
|
||
strip.set_parameter(ParameterType.VOLUME, -12.0)
|
||
strip.set_parameter(ParameterType.PAN, 0.5)
|
||
|
||
snap = strip.snapshot()
|
||
assert snap.volume == -12.0
|
||
assert snap.pan == 0.5
|
||
|
||
# Modify
|
||
strip.set_parameter(ParameterType.VOLUME, 0.0)
|
||
assert strip.state.volume == 0.0
|
||
|
||
# Restore
|
||
strip.restore(snap, send_osc=False)
|
||
assert strip.state.volume == -12.0
|
||
assert strip.state.pan == 0.5
|
||
|
||
def test_reset(self):
|
||
strip = ChannelStrip(index=0)
|
||
strip.set_parameter(ParameterType.VOLUME, 6.0)
|
||
strip.set_parameter(ParameterType.MUTE, 1.0)
|
||
strip.reset()
|
||
assert strip.state.volume == 0.0
|
||
assert strip.state.mute is False
|
||
|
||
def test_has_dsp(self):
|
||
strip = ChannelStrip(index=0)
|
||
assert strip.has_dsp is False
|
||
|
||
plugin = CarlaPluginInfo(1, "Test Gate", "gate", 0, {"threshold": 0})
|
||
strip.register_plugin(plugin)
|
||
assert strip.has_dsp is True
|
||
|
||
def test_set_many(self):
|
||
strip = ChannelStrip(index=0)
|
||
strip.set_many({
|
||
ParameterType.VOLUME: -3.0,
|
||
ParameterType.PAN: 0.5,
|
||
ParameterType.MUTE: 1.0,
|
||
})
|
||
assert strip.state.volume == -3.0
|
||
assert strip.state.pan == 0.5
|
||
assert strip.state.mute is True
|
||
|
||
def test_plugin_registration(self):
|
||
strip = ChannelStrip(index=3)
|
||
plugin = CarlaPluginInfo(10, "Ch4 EQ", "eq", 3, {"low_freq": 0, "low_gain": 1})
|
||
strip.register_plugin(plugin)
|
||
assert "eq" in strip._plugins
|
||
assert strip._plugins["eq"].plugin_id == 10
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# Routing Matrix Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestRoutingMatrix:
|
||
|
||
def test_add_node(self):
|
||
rm = RoutingMatrix()
|
||
node = rm.add_node(RouteNode("test", NodeType.CHANNEL_INPUT, "Test"))
|
||
assert rm.get_node("test") is node
|
||
|
||
def test_connect(self):
|
||
rm = RoutingMatrix()
|
||
src = rm.add_node(RouteNode("src", NodeType.SYSTEM_INPUT))
|
||
dst = rm.add_node(RouteNode("dst", NodeType.CHANNEL_INPUT))
|
||
edge = rm.connect("src", "dst", gain_db=-6.0, apply_jack=False)
|
||
assert edge is not None
|
||
assert edge.gain_db == -6.0
|
||
|
||
def test_connect_nonexistent(self):
|
||
rm = RoutingMatrix()
|
||
edge = rm.connect("a", "b")
|
||
assert edge is None
|
||
|
||
def test_disconnect(self):
|
||
rm = RoutingMatrix()
|
||
rm.add_node(RouteNode("src", NodeType.SYSTEM_INPUT))
|
||
rm.add_node(RouteNode("dst", NodeType.CHANNEL_INPUT))
|
||
rm.connect("src", "dst", apply_jack=False)
|
||
assert len(rm._edges) == 1
|
||
assert rm.disconnect("src", "dst", apply_jack=False)
|
||
assert len(rm._edges) == 0
|
||
|
||
def test_set_gain(self):
|
||
rm = RoutingMatrix()
|
||
rm.add_node(RouteNode("src", NodeType.SYSTEM_INPUT))
|
||
rm.add_node(RouteNode("dst", NodeType.CHANNEL_INPUT))
|
||
rm.connect("src", "dst", apply_jack=False)
|
||
assert rm.set_gain("src", "dst", -12.0)
|
||
edge = rm.get_edges("src")[0]
|
||
assert edge.gain_db == -12.0
|
||
|
||
def test_set_mute(self):
|
||
rm = RoutingMatrix()
|
||
rm.add_node(RouteNode("src", NodeType.SYSTEM_INPUT))
|
||
rm.add_node(RouteNode("dst", NodeType.CHANNEL_INPUT))
|
||
rm.connect("src", "dst", apply_jack=False)
|
||
rm.set_mute("src", "dst", True)
|
||
edge = rm.get_edges("src")[0]
|
||
assert edge.muted is True
|
||
|
||
def test_solo_logic(self):
|
||
rm = RoutingMatrix()
|
||
src1 = rm.add_node(RouteNode("src1", NodeType.CHANNEL_INPUT))
|
||
src2 = rm.add_node(RouteNode("src2", NodeType.CHANNEL_INPUT))
|
||
dst = rm.add_node(RouteNode("dst", NodeType.MASTER_INPUT))
|
||
rm.connect("src1", "dst", apply_jack=False)
|
||
rm.connect("src2", "dst", apply_jack=False)
|
||
|
||
rm.set_solo("src1", True)
|
||
edges = rm.get_edges()
|
||
edge1 = [e for e in edges if e.source.node_id == "src1"][0]
|
||
edge2 = [e for e in edges if e.source.node_id == "src2"][0]
|
||
assert edge1.soloed is True
|
||
assert edge2.soloed is False
|
||
|
||
rm.clear_solo()
|
||
edges = rm.get_edges()
|
||
assert all(not e.soloed for e in edges)
|
||
|
||
def test_remove_node_cleans_edges(self):
|
||
rm = RoutingMatrix()
|
||
rm.add_node(RouteNode("src", NodeType.SYSTEM_INPUT))
|
||
rm.add_node(RouteNode("dst", NodeType.CHANNEL_INPUT))
|
||
rm.connect("src", "dst", apply_jack=False)
|
||
assert len(rm._edges) == 1
|
||
rm.remove_node("src")
|
||
assert len(rm._edges) == 0
|
||
assert rm.get_node("src") is None
|
||
|
||
def test_build_default_8ch(self):
|
||
rm = RoutingMatrix()
|
||
rm.build_default_8ch_matrix()
|
||
# Should have system inputs (8), system outputs (2), channel nodes (16),
|
||
# aux nodes (8: 4 buses × 2), subgroups (2), master (3)
|
||
assert len(rm._nodes) >= 8 + 2
|
||
# Should have connections: 8 sys_in → channels, 8 channels → master, 2 master → sys_out
|
||
assert len(rm._edges) >= 8 + 8 + 2
|
||
|
||
def test_serialization_roundtrip(self):
|
||
rm = RoutingMatrix()
|
||
rm.add_node(RouteNode("src", NodeType.SYSTEM_INPUT, "Input"))
|
||
rm.add_node(RouteNode("dst", NodeType.CHANNEL_INPUT, "Channel"))
|
||
rm.connect("src", "dst", gain_db=-3.0, apply_jack=False)
|
||
|
||
data = rm.to_dict()
|
||
rm2 = RoutingMatrix()
|
||
rm2.from_dict(data)
|
||
|
||
assert len(rm2._nodes) == 2
|
||
assert len(rm2._edges) == 1
|
||
assert rm2.get_node("src") is not None
|
||
assert rm2._edges[0].gain_db == -3.0
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# Bus Manager Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestBusManager:
|
||
|
||
def test_default_initialization(self):
|
||
bm = BusManager(num_channels=16, num_aux=4, num_subgroups=2, num_vca=2)
|
||
assert len(bm.aux_buses) == 4
|
||
assert len(bm.subgroups) == 2
|
||
assert len(bm.vca_groups) == 2
|
||
assert bm.master.volume_db == 0.0
|
||
|
||
def test_aux_send(self):
|
||
bm = BusManager(num_channels=8, num_aux=4)
|
||
bm.set_aux_send(0, 3, -10.0)
|
||
aux = bm.get_aux(0)
|
||
assert aux.get_send(3) == -10.0
|
||
|
||
def test_aux_return(self):
|
||
bm = BusManager(num_channels=8, num_aux=4)
|
||
bm.set_aux_return(0, -5.0)
|
||
assert bm.get_aux(0).return_gain_db == -5.0
|
||
|
||
def test_aux_mute(self):
|
||
bm = BusManager(num_channels=8, num_aux=4)
|
||
bm.set_aux_mute(1, True)
|
||
assert bm.get_aux(1).muted is True
|
||
|
||
def test_aux_out_of_range(self):
|
||
bm = BusManager(num_channels=8, num_aux=2)
|
||
assert bm.get_aux(5) is None
|
||
|
||
def test_subgroup_membership(self):
|
||
bm = BusManager(num_channels=8, num_aux=2, num_subgroups=2)
|
||
assert bm.subgroup_add_channel(0, 3)
|
||
assert 3 in bm.subgroups[0].members
|
||
assert bm.subgroup_remove_channel(0, 3)
|
||
assert 3 not in bm.subgroups[0].members
|
||
|
||
def test_vca_membership(self):
|
||
bm = BusManager(num_channels=8, num_aux=2, num_vca=2)
|
||
bm.vca_add_channel(0, 5)
|
||
bm.vca_add_channel(1, 5)
|
||
vcas = bm.get_channel_vcas(5)
|
||
assert len(vcas) == 2
|
||
|
||
def test_vca_offset(self):
|
||
bm = BusManager(num_channels=8, num_vca=2)
|
||
bm.vca_add_channel(0, 2)
|
||
vca = bm.get_vca(0)
|
||
vca.master_db = -5.0
|
||
assert bm.get_channel_vca_offset(2) == -5.0
|
||
|
||
def test_vca_multiple_offset(self):
|
||
bm = BusManager(num_channels=8, num_vca=2)
|
||
bm.vca_add_channel(0, 2)
|
||
bm.vca_add_channel(1, 2)
|
||
bm.get_vca(0).master_db = -3.0
|
||
bm.get_vca(1).master_db = -2.0
|
||
assert bm.get_channel_vca_offset(2) == -5.0
|
||
|
||
def test_vca_muted_offset(self):
|
||
bm = BusManager(num_channels=8, num_vca=2)
|
||
bm.vca_add_channel(0, 2)
|
||
bm.get_vca(0).master_db = -5.0
|
||
bm.get_vca(0).muted = True
|
||
assert bm.get_channel_vca_offset(2) == 0.0 # muted VCA doesn't contribute
|
||
|
||
def test_master_operations(self):
|
||
bm = BusManager()
|
||
bm.set_master_volume(-10.0)
|
||
bm.set_master_mute(True)
|
||
bm.set_master_dim(True)
|
||
assert bm.master.volume_db == -10.0
|
||
assert bm.master.muted is True
|
||
assert bm.master.dim_active is True
|
||
# Effective gain with mute = 0
|
||
assert bm.master.effective_gain == 0.0
|
||
|
||
def test_master_dim_gain(self):
|
||
bm = BusManager()
|
||
bm.master.dim_db = -20.0
|
||
bm.master.dim_active = True
|
||
# -20 dB dim
|
||
assert abs(bm.master.effective_gain - 0.1) < 0.001
|
||
|
||
def test_serialization_roundtrip(self):
|
||
bm = BusManager(num_channels=4, num_aux=2, num_subgroups=1, num_vca=1)
|
||
bm.set_aux_send(0, 2, -6.0)
|
||
bm.subgroup_add_channel(0, 1)
|
||
bm.vca_add_channel(0, 0)
|
||
bm.get_vca(0).master_db = -3.0
|
||
bm.set_master_volume(-12.0)
|
||
|
||
data = bm.to_dict()
|
||
bm2 = BusManager(num_channels=4, num_aux=2, num_subgroups=1, num_vca=1)
|
||
bm2.from_dict(data)
|
||
|
||
assert bm2.get_aux(0).get_send(2) == -6.0
|
||
assert 1 in bm2.subgroups[0].members
|
||
assert 0 in bm2.vca_groups[0].members
|
||
assert bm2.master.volume_db == -12.0
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# Fader Automation Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestInterpolation:
|
||
|
||
def test_linear(self):
|
||
assert _interpolate(0.0, 0, 10, InterpolationMode.LINEAR) == 0
|
||
assert _interpolate(0.5, 0, 10, InterpolationMode.LINEAR) == 5
|
||
assert _interpolate(1.0, 0, 10, InterpolationMode.LINEAR) == 10
|
||
|
||
def test_s_curve(self):
|
||
assert _interpolate(0.0, 0, 10, InterpolationMode.S_CURVE) == 0
|
||
assert _interpolate(0.5, 0, 10, InterpolationMode.S_CURVE) == 5 # smoothstep at 0.5 = 0.5
|
||
assert _interpolate(1.0, 0, 10, InterpolationMode.S_CURVE) == 10
|
||
|
||
def test_logarithmic(self):
|
||
v = _interpolate(0.5, 1, 100, InterpolationMode.LOGARITHMIC)
|
||
assert 5 < v < 15 # roughly sqrt(100) = 10
|
||
|
||
def test_clamp(self):
|
||
assert _interpolate(-0.5, 0, 10, InterpolationMode.LINEAR) == 0
|
||
assert _interpolate(1.5, 0, 10, InterpolationMode.LINEAR) == 10
|
||
|
||
|
||
class TestAutomationLane:
|
||
|
||
def test_add_point_sorted(self):
|
||
lane = AutomationLane("test")
|
||
lane.add_point(2.0, 0.5)
|
||
lane.add_point(1.0, 0.2)
|
||
lane.add_point(3.0, 0.8)
|
||
assert [p.time_sec for p in lane.points] == [1.0, 2.0, 3.0]
|
||
|
||
def test_get_value_at_exact(self):
|
||
lane = AutomationLane("test")
|
||
lane.add_point(1.0, 0.2)
|
||
lane.add_point(2.0, 0.8)
|
||
assert abs(lane.get_value_at(1.5) - 0.5) < 0.01
|
||
|
||
def test_get_value_at_before(self):
|
||
lane = AutomationLane("test")
|
||
lane.add_point(1.0, 0.2)
|
||
assert abs(lane.get_value_at(0.0) - 0.2) < 0.01
|
||
|
||
def test_get_value_at_after(self):
|
||
lane = AutomationLane("test")
|
||
lane.add_point(1.0, 0.2)
|
||
lane.add_point(2.0, 0.8)
|
||
assert abs(lane.get_value_at(3.0) - 0.8) < 0.01
|
||
|
||
def test_get_value_empty(self):
|
||
lane = AutomationLane("test")
|
||
assert lane.get_value_at(1.0, default=-1.0) == -1.0
|
||
|
||
def test_clone(self):
|
||
lane = AutomationLane("test")
|
||
lane.add_point(1.0, 0.5)
|
||
clone = lane.clone()
|
||
clone.points[0].value = 0.9
|
||
assert lane.points[0].value == 0.5 # original unchanged
|
||
|
||
|
||
class TestFaderAutomation:
|
||
|
||
def test_recording(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
assert fa.is_recording
|
||
|
||
time.sleep(0.01)
|
||
fa.record("test_param", 0.5)
|
||
fa.stop_recording()
|
||
assert not fa.is_recording
|
||
|
||
assert "test_param" in fa._lanes
|
||
assert len(fa._lanes["test_param"].points) > 0
|
||
|
||
def test_not_recording_when_stopped(self):
|
||
fa = FaderAutomation()
|
||
fa.record("test", 0.5)
|
||
assert "test" not in fa._lanes
|
||
|
||
def test_playback(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("vol", 0.0)
|
||
time.sleep(0.02)
|
||
fa.record("vol", 1.0)
|
||
fa.stop_recording()
|
||
|
||
lane = fa._lanes["vol"]
|
||
assert len(lane.points) == 2
|
||
|
||
def test_scene_save_load(self):
|
||
fa = FaderAutomation()
|
||
scene = fa.save_scene("TestScene", {0: {"volume": -6.0}}, {}, {})
|
||
assert "TestScene" in fa.list_scenes()
|
||
|
||
loaded = fa.load_scene("TestScene")
|
||
assert loaded.name == "TestScene"
|
||
|
||
def test_scene_delete(self):
|
||
fa = FaderAutomation()
|
||
fa.save_scene("DeleteMe", {}, {}, {})
|
||
assert fa.delete_scene("DeleteMe")
|
||
assert "DeleteMe" not in fa.list_scenes()
|
||
|
||
def test_scene_recall(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("param", 0.25)
|
||
fa.stop_recording()
|
||
|
||
fa.save_scene("S1", {0: {"volume": -6.0}}, {}, {})
|
||
fa.recall_scene("S1")
|
||
assert fa._current_scene == "S1"
|
||
|
||
def test_crossfade(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("param", 0.0)
|
||
fa.stop_recording()
|
||
|
||
fa.save_scene("Target", {}, {}, {})
|
||
|
||
# Start very short crossfade
|
||
result = fa.crossfade_to("Target", 0.05)
|
||
assert result is True
|
||
assert fa._crossfade_active
|
||
|
||
# Let it complete
|
||
time.sleep(0.1)
|
||
fa.update_crossfade()
|
||
assert not fa._crossfade_active
|
||
|
||
def test_crossfade_missing_scene(self):
|
||
fa = FaderAutomation()
|
||
assert fa.crossfade_to("Nonexistent", 1.0) is False
|
||
|
||
def test_cancel_crossfade(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("param", 0.0)
|
||
fa.stop_recording()
|
||
fa.save_scene("Target", {}, {}, {})
|
||
|
||
fa.crossfade_to("Target", 10.0)
|
||
assert fa._crossfade_active
|
||
fa.cancel_crossfade()
|
||
assert not fa._crossfade_active
|
||
|
||
def test_seek(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("param", 0.0)
|
||
time.sleep(0.02)
|
||
fa.record("param", 1.0)
|
||
fa.stop_recording()
|
||
|
||
fa.start_playback()
|
||
fa.seek(1000.0) # seek way past
|
||
val = fa.get_value("param")
|
||
assert abs(val - 1.0) < 0.01
|
||
|
||
def test_get_duration(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("a", 0.0)
|
||
time.sleep(0.05)
|
||
fa.record("a", 0.5)
|
||
time.sleep(0.05)
|
||
fa.record("b", 1.0)
|
||
fa.stop_recording()
|
||
|
||
duration = fa.get_duration()
|
||
assert duration > 0.08 # at least 100ms total
|
||
|
||
def test_serialization_roundtrip(self):
|
||
fa = FaderAutomation()
|
||
fa.start_recording()
|
||
fa.record("param1", 0.3)
|
||
fa.stop_recording()
|
||
fa.save_scene("Test", {}, {}, {})
|
||
|
||
data = fa.to_dict()
|
||
fa2 = FaderAutomation()
|
||
fa2.from_dict(data)
|
||
|
||
assert len(fa2._lanes) == 1
|
||
assert "Test" in fa2.list_scenes()
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# DSP Engine Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestDSPEngine:
|
||
|
||
def test_create_default_engine(self):
|
||
engine = create_default_engine(num_channels=8, osc_enabled=False)
|
||
assert len(engine.channels) == 8
|
||
assert engine.config.num_aux == 4
|
||
assert not engine.running
|
||
|
||
def test_start_stop(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
assert engine.running
|
||
engine.stop()
|
||
assert not engine.running
|
||
|
||
def test_handle_channel_volume(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
param = MixerParameter(
|
||
param_type=ParameterType.VOLUME,
|
||
category=ParameterCategory.CHANNEL,
|
||
channel=2,
|
||
)
|
||
engine.handle_parameter(param, -12.0)
|
||
assert engine.channels[2].state.volume == -12.0
|
||
|
||
def test_handle_mute(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
param = MixerParameter(
|
||
param_type=ParameterType.MUTE,
|
||
category=ParameterCategory.CHANNEL,
|
||
channel=0,
|
||
)
|
||
engine.handle_parameter(param, 1.0)
|
||
assert engine.channels[0].state.mute is True
|
||
|
||
def test_handle_master_volume(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
param = MixerParameter(
|
||
param_type=ParameterType.MASTER_VOLUME,
|
||
category=ParameterCategory.MASTER,
|
||
channel=-1,
|
||
)
|
||
engine.handle_parameter(param, -20.0)
|
||
assert engine.buses.master.volume_db == -20.0
|
||
|
||
def test_handle_fx_return(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
param = MixerParameter(
|
||
param_type=ParameterType.FX_RETURN_A,
|
||
category=ParameterCategory.FX,
|
||
channel=-1,
|
||
)
|
||
engine.handle_parameter(param, -6.0)
|
||
assert engine.buses.get_aux(0).return_gain_db == -6.0
|
||
|
||
def test_handle_eq_params(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
param = MixerParameter(
|
||
param_type=ParameterType.EQ_MID_GAIN,
|
||
category=ParameterCategory.CHANNEL,
|
||
channel=1,
|
||
)
|
||
engine.handle_parameter(param, 4.0)
|
||
assert engine.channels[1].state.eq_mid_gain == 4.0
|
||
|
||
def test_snapshot_save_load(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
# Set some state
|
||
engine.channels[0].set_parameter(ParameterType.VOLUME, -6.0)
|
||
engine.buses.set_master_volume(-3.0)
|
||
|
||
# Save
|
||
engine.save_snapshot("test_snap")
|
||
|
||
# Modify
|
||
engine.channels[0].set_parameter(ParameterType.VOLUME, 0.0)
|
||
engine.buses.set_master_volume(0.0)
|
||
|
||
# Load
|
||
assert engine.load_snapshot("test_snap")
|
||
assert engine.channels[0].state.volume == -6.0
|
||
assert engine.buses.master.volume_db == -3.0
|
||
|
||
def test_param_key_helpers(self):
|
||
key = _make_param_key(ParameterCategory.CHANNEL, 3, ParameterType.VOLUME)
|
||
cat, ch, pt = _parse_param_key(key)
|
||
assert cat == ParameterCategory.CHANNEL
|
||
assert ch == 3
|
||
assert pt == ParameterType.VOLUME
|
||
|
||
def test_param_key_master(self):
|
||
key = _make_param_key(ParameterCategory.MASTER, -1, ParameterType.MASTER_VOLUME)
|
||
cat, ch, pt = _parse_param_key(key)
|
||
assert cat == ParameterCategory.MASTER
|
||
assert ch == -1
|
||
assert pt == ParameterType.MASTER_VOLUME
|
||
|
||
def test_stats(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
stats = engine.stats
|
||
assert stats["running"]
|
||
assert stats["num_channels"] == 4
|
||
assert not stats["osc_connected"]
|
||
|
||
def test_to_dict_from_dict(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
engine.channels[0].set_parameter(ParameterType.VOLUME, -10.0)
|
||
|
||
data = engine.to_dict()
|
||
engine2 = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine2.from_dict(data)
|
||
|
||
assert engine2.channels[0].state.volume == -10.0
|
||
|
||
def test_next_prev_scene(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
engine.save_snapshot("A")
|
||
engine.save_snapshot("B")
|
||
engine.save_snapshot("C")
|
||
|
||
assert engine.load_snapshot("B")
|
||
assert engine.automation._current_scene == "B"
|
||
|
||
# Bound to the scenes
|
||
assert engine.next_scene() # goes to C
|
||
assert engine.next_scene() # wraps to A
|
||
assert engine.prev_scene() # goes to C
|
||
|
||
def test_automation_recording_handling(self):
|
||
engine = create_default_engine(num_channels=4, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
engine.automation.start_recording()
|
||
|
||
param = MixerParameter(
|
||
param_type=ParameterType.VOLUME,
|
||
category=ParameterCategory.CHANNEL,
|
||
channel=0,
|
||
)
|
||
engine.handle_parameter(param, -5.0)
|
||
|
||
key = _make_param_key(ParameterCategory.CHANNEL, 0, ParameterType.VOLUME)
|
||
assert key in engine.automation._lanes
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# Integration Tests
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestIntegration:
|
||
|
||
def test_full_signal_chain_state(self):
|
||
"""Test that the full signal chain can be configured through the engine."""
|
||
engine = create_default_engine(num_channels=16, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
# Configure channel 0
|
||
params_to_set = [
|
||
(ParameterType.GAIN, 20.0),
|
||
(ParameterType.EQ_LOW_FREQ, 80.0),
|
||
(ParameterType.EQ_LOW_GAIN, 3.0),
|
||
(ParameterType.EQ_MID_FREQ, 2500.0),
|
||
(ParameterType.EQ_MID_GAIN, -2.0),
|
||
(ParameterType.EQ_HIGH_FREQ, 8000.0),
|
||
(ParameterType.EQ_HIGH_GAIN, 1.5),
|
||
(ParameterType.COMP_THRESHOLD, -18.0),
|
||
(ParameterType.COMP_RATIO, 3.0),
|
||
(ParameterType.GATE_THRESHOLD, -45.0),
|
||
(ParameterType.FX_SEND_A, -8.0),
|
||
(ParameterType.VOLUME, -3.0),
|
||
(ParameterType.PAN, 0.2),
|
||
]
|
||
for pt, val in params_to_set:
|
||
param = MixerParameter(pt, ParameterCategory.CHANNEL, 0)
|
||
engine.handle_parameter(param, val)
|
||
|
||
ch0 = engine.channels[0]
|
||
assert ch0.state.gain == 20.0
|
||
assert ch0.state.eq_low_freq == 80.0
|
||
assert ch0.state.comp_threshold == -18.0
|
||
assert ch0.state.volume == -3.0
|
||
|
||
def test_multiple_channels_independent(self):
|
||
"""Test that channels maintain independent state."""
|
||
engine = create_default_engine(num_channels=8, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
# Set different volumes on different channels
|
||
for ch, vol in [(0, -6.0), (3, 0.0), (7, 6.0)]:
|
||
param = MixerParameter(ParameterType.VOLUME, ParameterCategory.CHANNEL, ch)
|
||
engine.handle_parameter(param, vol)
|
||
|
||
assert engine.channels[0].state.volume == -6.0
|
||
assert engine.channels[3].state.volume == 0.0
|
||
assert engine.channels[7].state.volume == 6.0
|
||
assert engine.channels[1].state.volume == 0.0 # unchanged
|
||
|
||
def test_bus_routing_persistence(self):
|
||
"""Test that bus state persists across snapshot save/load."""
|
||
engine = create_default_engine(num_channels=8, osc_enabled=False)
|
||
engine.start(connect_osc=False)
|
||
|
||
# Set up complex bus state
|
||
engine.buses.set_aux_send(0, 1, -12.0)
|
||
engine.buses.set_aux_send(2, 3, -4.0)
|
||
engine.buses.subgroup_add_channel(0, 0)
|
||
engine.buses.subgroup_add_channel(0, 4)
|
||
engine.buses.vca_add_channel(0, 2)
|
||
engine.buses.get_vca(0).master_db = -5.0
|
||
|
||
# Save and verify restore
|
||
snap = engine.save_snapshot("complex")
|
||
engine.load_snapshot("complex")
|
||
|
||
assert engine.buses.get_aux(0).get_send(1) == -12.0
|
||
assert engine.buses.get_aux(2).get_send(3) == -4.0
|
||
assert 0 in engine.buses.subgroups[0].members
|
||
assert 2 in engine.buses.vca_groups[0].members
|