217 lines
7.6 KiB
Python
217 lines
7.6 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
_TEST_USER = {"username": "testuser", "output_dir": "/tmp", "is_admin": False}
|
|
|
|
|
|
def make_app():
|
|
from fastapi import FastAPI
|
|
from api.router import router, current_user
|
|
app = FastAPI()
|
|
# Override auth for tests — no real credentials needed
|
|
app.dependency_overrides[current_user] = lambda: _TEST_USER
|
|
app.include_router(router)
|
|
return app
|
|
|
|
|
|
def test_status_returns_idle():
|
|
client = TestClient(make_app())
|
|
r = client.get("/status")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "idle"
|
|
assert r.json()["username"] == "testuser"
|
|
|
|
|
|
def test_config_get_returns_dict():
|
|
client = TestClient(make_app())
|
|
r = client.get("/config")
|
|
assert r.status_code == 200
|
|
assert "ollama" in r.json()
|
|
|
|
|
|
def test_transcripts_returns_list():
|
|
client = TestClient(make_app())
|
|
r = client.get("/transcripts")
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json(), list)
|
|
|
|
|
|
def test_status_requires_auth():
|
|
from fastapi import FastAPI
|
|
from api.router import router
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
r = client.get("/status")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def make_app_for_dir(output_dir: str):
|
|
from fastapi import FastAPI
|
|
from api.router import router, current_user
|
|
app = FastAPI()
|
|
app.dependency_overrides[current_user] = lambda: {"username": "", "output_dir": output_dir, "is_admin": False}
|
|
app.include_router(router)
|
|
return app
|
|
|
|
|
|
def test_get_transcript_returns_content(tmp_path):
|
|
f = tmp_path / "2026-01-01-0900-test.md"
|
|
f.write_text("# Hello\n\ncontent here\n")
|
|
client = TestClient(make_app_for_dir(str(tmp_path)))
|
|
r = client.get("/transcripts/2026-01-01-0900-test.md")
|
|
assert r.status_code == 200
|
|
assert "Hello" in r.text
|
|
|
|
|
|
def test_get_transcript_rejects_path_traversal(tmp_path):
|
|
client = TestClient(make_app_for_dir(str(tmp_path)))
|
|
r = client.get("/transcripts/..%2Fsecret.md")
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_delete_transcript_removes_file(tmp_path):
|
|
f = tmp_path / "2026-01-01-0900-test.md"
|
|
f.write_text("content")
|
|
client = TestClient(make_app_for_dir(str(tmp_path)))
|
|
r = client.delete("/transcripts/2026-01-01-0900-test.md")
|
|
assert r.status_code == 200
|
|
assert not f.exists()
|
|
|
|
|
|
def test_delete_transcript_rejects_path_traversal(tmp_path):
|
|
client = TestClient(make_app_for_dir(str(tmp_path)))
|
|
r = client.delete("/transcripts/..%2Fsecret.md")
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_login_rejects_wrong_credentials():
|
|
import tempfile, os
|
|
from unittest.mock import patch
|
|
from fastapi import FastAPI
|
|
from api.router import router
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
users_path = os.path.join(tmpdir, "users.toml")
|
|
with patch("auth.USERS_PATH", users_path):
|
|
r = client.post("/login", json={"username": "nobody", "password": "wrong"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_audio_devices_returns_list(monkeypatch):
|
|
import sounddevice as sd
|
|
from main import app
|
|
from api.router import current_user
|
|
fake_devices = [
|
|
{"name": "Fake Mic", "max_input_channels": 1, "max_output_channels": 0},
|
|
{"name": "Fake Monitor", "max_input_channels": 2, "max_output_channels": 0},
|
|
{"name": "Fake Speaker", "max_input_channels": 0, "max_output_channels": 2},
|
|
]
|
|
monkeypatch.setattr(sd, "query_devices", lambda: fake_devices)
|
|
app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": True}
|
|
try:
|
|
client = TestClient(app)
|
|
r = client.get("/audio/devices", headers={"Authorization": "Bearer fake"})
|
|
assert r.status_code == 200
|
|
devices = r.json()
|
|
assert len(devices) == 2 # only input devices
|
|
assert devices[0]["name"] == "Fake Mic"
|
|
finally:
|
|
app.dependency_overrides.pop(current_user, None)
|
|
|
|
|
|
def test_audio_devices_forbidden_for_non_admin():
|
|
from main import app
|
|
from api.router import current_user
|
|
app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": False}
|
|
try:
|
|
client = TestClient(app)
|
|
r = client.get("/audio/devices", headers={"Authorization": "Bearer fake"})
|
|
assert r.status_code == 403
|
|
finally:
|
|
app.dependency_overrides.pop(current_user, None)
|
|
|
|
|
|
def test_audio_combined_forbidden_for_non_admin():
|
|
from main import app
|
|
from api.router import current_user
|
|
app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": False}
|
|
try:
|
|
from fastapi.testclient import TestClient
|
|
client = TestClient(app)
|
|
r = client.post("/audio/combined", json={"mic": "x", "monitor": "y"},
|
|
headers={"Authorization": "Bearer fake"})
|
|
assert r.status_code == 403
|
|
finally:
|
|
app.dependency_overrides.pop(current_user, None)
|
|
|
|
|
|
def test_status_includes_is_admin():
|
|
from main import app
|
|
from api.router import current_user
|
|
app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": True}
|
|
try:
|
|
from fastapi.testclient import TestClient
|
|
client = TestClient(app)
|
|
r = client.get("/status", headers={"Authorization": "Bearer fake"})
|
|
assert r.status_code == 200
|
|
assert r.json()["is_admin"] is True
|
|
finally:
|
|
app.dependency_overrides.pop(current_user, None)
|
|
|
|
|
|
def test_state_has_speaker_fields():
|
|
from api.state import AppState
|
|
s = AppState()
|
|
assert hasattr(s, "_speakers_event")
|
|
assert hasattr(s, "_pending_aligned_segments")
|
|
assert hasattr(s, "_speaker_names")
|
|
assert s._speakers_event is None
|
|
assert s._pending_aligned_segments is None
|
|
assert s._speaker_names is None
|
|
|
|
|
|
def test_post_speakers_resolves_pipeline_pause():
|
|
import asyncio
|
|
from main import app
|
|
from api.router import current_user
|
|
from api.state import state
|
|
|
|
state._speakers_event = asyncio.Event()
|
|
state._speaker_names = None
|
|
|
|
app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": False}
|
|
try:
|
|
from fastapi.testclient import TestClient
|
|
client = TestClient(app)
|
|
r = client.post("/speakers", json={"SPEAKER_00": "Thomas", "SPEAKER_01": "Möller"})
|
|
assert r.status_code == 200
|
|
assert state._speaker_names == {"SPEAKER_00": "Thomas", "SPEAKER_01": "Möller"}
|
|
assert state._speakers_event.is_set()
|
|
finally:
|
|
app.dependency_overrides.pop(current_user, None)
|
|
state._speakers_event = None
|
|
state._speaker_names = None
|
|
|
|
|
|
def test_put_config_deep_merges(tmp_path, monkeypatch):
|
|
import config as cfg_mod
|
|
monkeypatch.setattr(cfg_mod, "CONFIG_PATH", str(tmp_path / "config.toml"))
|
|
from main import app
|
|
from api.router import current_user
|
|
app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": True}
|
|
try:
|
|
from fastapi.testclient import TestClient
|
|
client = TestClient(app)
|
|
r = client.put("/config",
|
|
json={"whisper": {"base_url": "http://beastix:8000"}},
|
|
headers={"Authorization": "Bearer fake"})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
# base_url updated, model preserved
|
|
assert data["whisper"]["base_url"] == "http://beastix:8000"
|
|
assert data["whisper"]["model"] == "large-v3"
|
|
finally:
|
|
app.dependency_overrides.pop(current_user, None)
|