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)
242 lines
8.9 KiB
Python
242 lines
8.9 KiB
Python
"""Tests for the REST API endpoints."""
|
|
|
|
import json
|
|
import pytest
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.network.auth import APIKeyAuth, API_KEY_HEADER
|
|
from src.network.rate_limiter import RateLimiter
|
|
from src.network.rest_api import MixerStateProvider, create_router
|
|
from src.network.schemas import (
|
|
ChannelSchema,
|
|
MasterBusSchema,
|
|
MixerStateSchema,
|
|
TransportSchema,
|
|
PluginSchema,
|
|
RoutingSchema,
|
|
EQSchema,
|
|
CompressorSchema,
|
|
GateSchema,
|
|
FXSendsSchema,
|
|
AuxBusSchema,
|
|
)
|
|
|
|
|
|
class TestRESTAPI:
|
|
"""Integration tests for the REST API."""
|
|
|
|
@pytest.fixture
|
|
def state_provider(self):
|
|
"""Create a mock state provider."""
|
|
sp = MixerStateProvider()
|
|
|
|
# Mock data
|
|
channels = [
|
|
ChannelSchema(channel=i, label=f"CH{i+1}", volume_db=0.0)
|
|
for i in range(16)
|
|
]
|
|
sp.get_all_channels = lambda: channels
|
|
sp.get_channel = lambda idx: channels[idx] if 0 <= idx < len(channels) else None
|
|
sp.get_master = lambda: MasterBusSchema(
|
|
volume_db=0.0,
|
|
aux_buses=[AuxBusSchema(index=0, label="Reverb")],
|
|
)
|
|
sp.get_transport = lambda: TransportSchema(playing=False)
|
|
sp.get_plugins = lambda: [
|
|
PluginSchema(plugin_id=1, name="Ch1 Gate", role="gate", channel=0),
|
|
]
|
|
sp.get_routing = lambda: RoutingSchema()
|
|
sp.list_scenes = lambda: ["Scene A", "Scene B"]
|
|
sp.load_scene = lambda name: name in ["Scene A", "Scene B"]
|
|
sp.save_scene = lambda name: True
|
|
|
|
# Parameter handler captures
|
|
sp._param_calls = []
|
|
sp.handle_parameter = lambda pt, v, ch: sp._param_calls.append((pt, v, ch))
|
|
|
|
# Transport handler captures
|
|
sp._transport_calls = []
|
|
sp.handle_transport = lambda cmd: sp._transport_calls.append(cmd)
|
|
|
|
return sp
|
|
|
|
@pytest.fixture
|
|
def auth(self):
|
|
return APIKeyAuth(keys=["test-api-key"])
|
|
|
|
@pytest.fixture
|
|
def rate_limiter(self):
|
|
return RateLimiter(rate=1000.0, capacity=2000)
|
|
|
|
@pytest.fixture
|
|
def app(self, state_provider, auth, rate_limiter):
|
|
app = FastAPI()
|
|
router = create_router(
|
|
state=state_provider,
|
|
auth=auth,
|
|
rate_limiter=rate_limiter,
|
|
prefix="/api/v1",
|
|
)
|
|
app.include_router(router)
|
|
|
|
# Health endpoint (no auth)
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
# Full state endpoint on app directly (no auth for test)
|
|
@app.get("/api/v1/state")
|
|
async def full_state():
|
|
return state_provider.get_full_state() if state_provider.get_full_state else {}
|
|
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def client(self, app):
|
|
return TestClient(app)
|
|
|
|
@pytest.fixture
|
|
def auth_headers(self):
|
|
return {API_KEY_HEADER: "test-api-key"}
|
|
|
|
# ── Auth ────────────────────────────────────────────────────────────
|
|
|
|
def test_health_no_auth(self, client):
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
|
|
def test_channels_no_auth(self, client):
|
|
resp = client.get("/api/v1/channels")
|
|
assert resp.status_code == 401
|
|
|
|
def test_channels_wrong_key(self, client):
|
|
resp = client.get("/api/v1/channels", headers={API_KEY_HEADER: "wrong"})
|
|
assert resp.status_code == 403
|
|
|
|
# ── Channels ────────────────────────────────────────────────────────
|
|
|
|
def test_list_channels(self, client, auth_headers):
|
|
resp = client.get("/api/v1/channels", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 16
|
|
assert data[0]["channel"] == 0
|
|
|
|
def test_get_channel_valid(self, client, auth_headers):
|
|
resp = client.get("/api/v1/channels/3", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["channel"] == 3
|
|
|
|
def test_get_channel_not_found(self, client, auth_headers):
|
|
resp = client.get("/api/v1/channels/99", headers=auth_headers)
|
|
assert resp.status_code == 404
|
|
|
|
def test_set_channel_parameter(self, client, auth_headers, state_provider):
|
|
body = {"param_type": "volume", "value": -3.0}
|
|
resp = client.put(
|
|
"/api/v1/channels/0/parameter",
|
|
json=body,
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
assert len(state_provider._param_calls) == 1
|
|
assert state_provider._param_calls[0] == ("volume", -3.0, 0)
|
|
|
|
def test_set_channel_parameters_batch(self, client, auth_headers, state_provider):
|
|
body = {
|
|
"updates": [
|
|
{"param_type": "volume", "value": -6.0, "channel": 0},
|
|
{"param_type": "pan", "value": 0.5, "channel": 0},
|
|
]
|
|
}
|
|
resp = client.put(
|
|
"/api/v1/channels/0/parameters",
|
|
json=body,
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
assert len(state_provider._param_calls) == 2
|
|
|
|
# ── Mixes / Master ───────────────────────────────────────────────────
|
|
|
|
def test_get_mixes(self, client, auth_headers):
|
|
resp = client.get("/api/v1/mixes", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "volume_db" in data
|
|
assert len(data["aux_buses"]) == 1
|
|
|
|
def test_set_master_parameter(self, client, auth_headers, state_provider):
|
|
body = {"param_type": "master_volume", "value": -6.0}
|
|
resp = client.put(
|
|
"/api/v1/mixes/parameter",
|
|
json=body,
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert len(state_provider._param_calls) == 1
|
|
assert state_provider._param_calls[0] == ("master_volume", -6.0, -1)
|
|
|
|
# ── Plugins ──────────────────────────────────────────────────────────
|
|
|
|
def test_list_plugins(self, client, auth_headers):
|
|
resp = client.get("/api/v1/plugins", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "Ch1 Gate"
|
|
|
|
# ── Transport ────────────────────────────────────────────────────────
|
|
|
|
def test_get_transport(self, client, auth_headers):
|
|
resp = client.get("/api/v1/transport", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "playing" in data
|
|
assert "tempo_bpm" in data
|
|
|
|
def test_transport_command(self, client, auth_headers, state_provider):
|
|
resp = client.put(
|
|
"/api/v1/transport/command?command=play",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert len(state_provider._transport_calls) == 1
|
|
assert state_provider._transport_calls[0] == "play"
|
|
|
|
# ── Routing ──────────────────────────────────────────────────────────
|
|
|
|
def test_get_routing(self, client, auth_headers):
|
|
resp = client.get("/api/v1/routing", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "nodes" in data
|
|
assert "edges" in data
|
|
|
|
# ── Scenes ───────────────────────────────────────────────────────────
|
|
|
|
def test_list_scenes(self, client, auth_headers):
|
|
resp = client.get("/api/v1/scenes", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data == ["Scene A", "Scene B"]
|
|
|
|
def test_load_scene_valid(self, client, auth_headers):
|
|
resp = client.post("/api/v1/scenes/Scene A/load", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
|
|
def test_load_scene_not_found(self, client, auth_headers):
|
|
resp = client.post("/api/v1/scenes/Nonexistent/load", headers=auth_headers)
|
|
assert resp.status_code == 404
|
|
|
|
def test_save_scene(self, client, auth_headers):
|
|
resp = client.post("/api/v1/scenes/New Scene/save", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|