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)
319 lines
9.7 KiB
Python
319 lines
9.7 KiB
Python
"""Tests for web application routes — login, settings, file management."""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
import pytest
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.network.auth import APIKeyAuth, API_KEY_HEADER
|
|
from src.network.session import SessionManager
|
|
from src.network.web_routes import create_web_routes, get_static_files_app
|
|
|
|
|
|
class TestLoginRoutes:
|
|
"""Tests for session auth login/logout endpoints."""
|
|
|
|
@pytest.fixture
|
|
def auth(self):
|
|
return APIKeyAuth(keys=["test-key-123"])
|
|
|
|
@pytest.fixture
|
|
def session_mgr(self):
|
|
return SessionManager(ttl=3600, max_sessions=10)
|
|
|
|
@pytest.fixture
|
|
def app(self, auth, session_mgr):
|
|
app = FastAPI()
|
|
routers = create_web_routes(
|
|
auth=auth,
|
|
session_manager=session_mgr,
|
|
settings_getter=lambda: {},
|
|
settings_setter=lambda _: True,
|
|
)
|
|
for router in routers:
|
|
app.include_router(router)
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def client(self, app):
|
|
return TestClient(app)
|
|
|
|
def test_login_with_header(self, client):
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
headers={API_KEY_HEADER: "test-key-123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "token" in data
|
|
assert "expires_in" in data
|
|
assert len(data["token"]) == 64
|
|
|
|
def test_login_with_json_body(self, client):
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"api_key": "test-key-123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "token" in data
|
|
|
|
def test_login_invalid_key(self, client):
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"api_key": "wrong-key"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_login_missing_key(self, client):
|
|
resp = client.post("/api/v1/auth/login")
|
|
assert resp.status_code == 403
|
|
|
|
def test_login_no_body_or_header(self, client):
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
headers={},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_login_creates_session(self, client, session_mgr):
|
|
assert session_mgr.active_sessions == 0
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
headers={API_KEY_HEADER: "test-key-123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert session_mgr.active_sessions == 1
|
|
|
|
def test_logout_valid_token(self, client, session_mgr):
|
|
# Create a session first
|
|
token = session_mgr.create_session("browser")
|
|
assert session_mgr.active_sessions == 1
|
|
|
|
resp = client.post(
|
|
"/api/v1/auth/logout",
|
|
json={"token": token},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
assert session_mgr.active_sessions == 0
|
|
|
|
def test_logout_invalid_token(self, client, session_mgr):
|
|
resp = client.post(
|
|
"/api/v1/auth/logout",
|
|
json={"token": "not-a-real-token"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is False
|
|
|
|
def test_logout_missing_token(self, client):
|
|
resp = client.post("/api/v1/auth/logout")
|
|
assert resp.status_code == 400
|
|
|
|
def test_login_without_content_type(self, client):
|
|
"""Test login without JSON content-type header."""
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
content="not json",
|
|
headers={API_KEY_HEADER: "test-key-123"},
|
|
)
|
|
assert resp.status_code == 200 # Falls back to header auth
|
|
|
|
|
|
class TestSettingsRoutes:
|
|
"""Tests for settings REST endpoints."""
|
|
|
|
@pytest.fixture
|
|
def settings_state(self):
|
|
return {"sample_rate": 48000, "buffer_size": 256, "volume_limit": -3.0}
|
|
|
|
@pytest.fixture
|
|
def auth(self):
|
|
return APIKeyAuth(keys=["test-key"])
|
|
|
|
@pytest.fixture
|
|
def session_mgr(self):
|
|
return SessionManager()
|
|
|
|
@pytest.fixture
|
|
def app(self, auth, session_mgr, settings_state):
|
|
app = FastAPI()
|
|
|
|
def getter():
|
|
return dict(settings_state)
|
|
|
|
def setter(updates):
|
|
valid = set(settings_state.keys())
|
|
for k in updates:
|
|
if k not in valid:
|
|
return False
|
|
settings_state.update(updates)
|
|
return True
|
|
|
|
routers = create_web_routes(
|
|
auth=auth,
|
|
session_manager=session_mgr,
|
|
settings_getter=getter,
|
|
settings_setter=setter,
|
|
)
|
|
for router in routers:
|
|
app.include_router(router)
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def client(self, app):
|
|
return TestClient(app)
|
|
|
|
def test_get_settings(self, client, settings_state):
|
|
resp = client.get("/api/v1/settings")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data == settings_state
|
|
|
|
def test_update_settings_valid(self, client, settings_state):
|
|
resp = client.put(
|
|
"/api/v1/settings",
|
|
json={"buffer_size": 512},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
assert settings_state["buffer_size"] == 512
|
|
|
|
def test_update_settings_invalid_key(self, client, settings_state):
|
|
resp = client.put(
|
|
"/api/v1/settings",
|
|
json={"not_a_setting": 123},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_update_settings_invalid_body(self, client):
|
|
resp = client.put(
|
|
"/api/v1/settings",
|
|
content="not json",
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_update_settings_not_a_dict(self, client):
|
|
resp = client.put(
|
|
"/api/v1/settings",
|
|
json=[1, 2, 3],
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
class TestFileRoutes:
|
|
"""Tests for file management REST endpoints."""
|
|
|
|
@pytest.fixture
|
|
def audio_dir(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
audio_path = Path(tmp)
|
|
# Create some test audio files
|
|
(audio_path / "track1.wav").write_text("fake wav data")
|
|
(audio_path / "track2.mp3").write_text("fake mp3 data")
|
|
(audio_path / "readme.txt").write_text("not audio")
|
|
(audio_path / "track3.flac").write_text("fake flac data")
|
|
yield audio_path
|
|
|
|
@pytest.fixture
|
|
def auth(self):
|
|
return APIKeyAuth(keys=["test-key"])
|
|
|
|
@pytest.fixture
|
|
def session_mgr(self):
|
|
return SessionManager()
|
|
|
|
@pytest.fixture
|
|
def app(self, auth, session_mgr, audio_dir):
|
|
app = FastAPI()
|
|
routers = create_web_routes(
|
|
auth=auth,
|
|
session_manager=session_mgr,
|
|
settings_getter=lambda: {},
|
|
settings_setter=lambda _: True,
|
|
audio_dir=audio_dir,
|
|
)
|
|
for router in routers:
|
|
app.include_router(router)
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def client(self, app):
|
|
return TestClient(app)
|
|
|
|
def test_list_files(self, client):
|
|
resp = client.get("/api/v1/files")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["exists"] is True
|
|
assert data["count"] == 3 # wav, mp3, flac (not .txt)
|
|
names = {f["name"] for f in data["files"]}
|
|
assert "track1.wav" in names
|
|
assert "track2.mp3" in names
|
|
assert "track3.flac" in names
|
|
assert "readme.txt" not in names
|
|
|
|
def test_download_audio_file(self, client):
|
|
resp = client.get("/api/v1/files/track1.wav")
|
|
assert resp.status_code == 200
|
|
assert resp.content == b"fake wav data"
|
|
|
|
def test_download_nonexistent_file(self, client):
|
|
resp = client.get("/api/v1/files/nope.wav")
|
|
assert resp.status_code == 404
|
|
|
|
def test_download_non_audio_file(self, client):
|
|
resp = client.get("/api/v1/files/readme.txt")
|
|
assert resp.status_code == 403
|
|
|
|
def test_path_traversal_prevented(self, client):
|
|
resp = client.get("/api/v1/files/../../../etc/passwd")
|
|
assert resp.status_code in (403, 404)
|
|
|
|
def test_nonexistent_directory(self):
|
|
"""Test listing files when directory doesn't exist."""
|
|
app = FastAPI()
|
|
auth = APIKeyAuth(keys=["test-key"])
|
|
session_mgr = SessionManager()
|
|
routers = create_web_routes(
|
|
auth=auth,
|
|
session_manager=session_mgr,
|
|
settings_getter=lambda: {},
|
|
settings_setter=lambda _: True,
|
|
audio_dir=Path("/tmp/nonexistent-audio-dir-xyz"),
|
|
)
|
|
for router in routers:
|
|
app.include_router(router)
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/files")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["exists"] is False
|
|
assert data["files"] == []
|
|
|
|
|
|
class TestStaticFiles:
|
|
"""Tests for static file serving."""
|
|
|
|
def test_get_static_files_app(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
web_dir = Path(tmp)
|
|
(web_dir / "index.html").write_text("<h1>Hello</h1>")
|
|
app = get_static_files_app(web_dir)
|
|
assert app is not None
|
|
|
|
def test_placeholder_created(self):
|
|
"""Test that placeholder UI is created when dir doesn't exist."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
web_dir = Path(tmp) / "nonexistent-web"
|
|
app = get_static_files_app(web_dir)
|
|
assert web_dir.exists()
|
|
index = web_dir / "index.html"
|
|
assert index.exists()
|
|
content = index.read_text()
|
|
assert "RPi Audio Mixer" in content
|