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)
163 lines
4.8 KiB
Python
163 lines
4.8 KiB
Python
"""Tests for WebSocket manager."""
|
|
|
|
import asyncio
|
|
import json
|
|
import pytest
|
|
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.network.websocket import WebSocketManager
|
|
from src.network.schemas import (
|
|
MixerStateSchema,
|
|
ChannelSchema,
|
|
WSMessage,
|
|
WSMessageType,
|
|
)
|
|
from src.network.auth import APIKeyAuth
|
|
from src.network.rate_limiter import RateLimiter
|
|
|
|
|
|
class TestWebSocketManager:
|
|
"""Tests for WebSocketManager (unit tests for the manager class)."""
|
|
|
|
@pytest.fixture
|
|
def manager(self):
|
|
return WebSocketManager(max_connections=10)
|
|
|
|
def test_initial_state(self, manager):
|
|
assert manager.connected_clients == 0
|
|
stats = manager.stats
|
|
assert stats["connected_clients"] == 0
|
|
assert stats["total_connections"] == 0
|
|
assert stats["total_broadcasts"] == 0
|
|
|
|
def test_set_state_provider(self, manager):
|
|
called = []
|
|
|
|
def get_state():
|
|
called.append(True)
|
|
return MixerStateSchema(
|
|
channels=[ChannelSchema(channel=0)],
|
|
)
|
|
|
|
manager.set_state_provider(get_state)
|
|
assert manager._state_provider is not None
|
|
|
|
def test_set_parameter_handler(self, manager):
|
|
calls = []
|
|
|
|
def handler(pt, v, ch):
|
|
calls.append((pt, v, ch))
|
|
|
|
manager.set_parameter_handler(handler)
|
|
assert manager._parameter_handler is not None
|
|
|
|
def test_set_transport_handler(self, manager):
|
|
calls = []
|
|
|
|
def handler(cmd):
|
|
calls.append(cmd)
|
|
|
|
manager.set_transport_handler(handler)
|
|
assert manager._transport_handler is not None
|
|
|
|
def test_max_connections_property(self, manager):
|
|
assert manager._max_connections == 10
|
|
|
|
|
|
class TestWebSocketBroadcast:
|
|
"""Tests for WebSocket broadcasting logic."""
|
|
|
|
@pytest.fixture
|
|
def manager(self):
|
|
mgr = WebSocketManager(max_connections=10)
|
|
return mgr
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_no_clients(self, manager):
|
|
msg = WSMessage(
|
|
type=WSMessageType.PARAMETER_UPDATE,
|
|
payload={"param_type": "volume", "value": -3.0, "channel": 0},
|
|
)
|
|
count = await manager.broadcast(msg)
|
|
assert count == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_parameter_update_no_clients(self, manager):
|
|
count = await manager.broadcast_parameter_update("volume", -3.0, 0)
|
|
assert count == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_transport_no_clients(self, manager):
|
|
count = await manager.broadcast_transport_update({"command": "play"})
|
|
assert count == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_add_remove(self, manager):
|
|
"""Test client bookkeeping (without actual WebSocket)."""
|
|
assert manager.connected_clients == 0
|
|
|
|
# Simulate adding a client
|
|
async with manager._lock:
|
|
manager._clients["test-client"] = "mock-ws"
|
|
manager._total_connections += 1
|
|
|
|
assert manager.connected_clients == 1
|
|
assert manager.stats["total_connections"] == 1
|
|
|
|
# Remove client
|
|
await manager._remove_client("test-client")
|
|
assert manager.connected_clients == 0
|
|
|
|
|
|
class TestWebSocketStats:
|
|
"""Tests for WebSocket statistics."""
|
|
|
|
def test_stats_structure(self):
|
|
manager = WebSocketManager(max_connections=5)
|
|
stats = manager.stats
|
|
assert "connected_clients" in stats
|
|
assert "total_connections" in stats
|
|
assert "total_messages" in stats
|
|
assert "total_broadcasts" in stats
|
|
assert "uptime_seconds" in stats
|
|
assert "max_connections" in stats
|
|
assert stats["max_connections"] == 5
|
|
|
|
def test_uptime_increases(self):
|
|
manager = WebSocketManager()
|
|
stats1 = manager.stats
|
|
# Stats should have a reasonable uptime
|
|
assert stats1["uptime_seconds"] >= 0
|
|
|
|
|
|
class TestWebSocketMessageTypes:
|
|
"""Tests for WebSocket message type handling."""
|
|
|
|
def test_ws_message_serialization(self):
|
|
msg = WSMessage(
|
|
type=WSMessageType.FULL_STATE,
|
|
payload={"channels": [], "master": {"volume_db": 0.0}},
|
|
)
|
|
data = msg.model_dump()
|
|
assert data["type"] == "full_state"
|
|
|
|
def test_all_message_types(self):
|
|
for msg_type in WSMessageType:
|
|
msg = WSMessage(type=msg_type, payload={})
|
|
assert msg.type == msg_type
|
|
assert msg.payload == {}
|
|
|
|
def test_parameter_update_message(self):
|
|
msg = WSMessage(
|
|
type=WSMessageType.PARAMETER_UPDATE,
|
|
payload={
|
|
"param_type": "comp_threshold",
|
|
"value": -15.0,
|
|
"channel": 2,
|
|
},
|
|
)
|
|
assert msg.payload["param_type"] == "comp_threshold"
|
|
assert msg.payload["value"] == -15.0
|