182 lines
6.4 KiB
Python
182 lines
6.4 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 subprocess
|
|
from main import app
|
|
from api.router import current_user
|
|
pactl_output = (
|
|
"1\talsa_input.pci.analog-stereo\tPipeWire\ts32le 2ch 48000Hz\tRUNNING\n"
|
|
"2\talsa_output.pci.analog-stereo.monitor\tPipeWire\ts32le 2ch 48000Hz\tIDLE\n"
|
|
)
|
|
monkeypatch.setattr(subprocess, "check_output", lambda *a, **kw: pactl_output.encode())
|
|
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
|
|
assert devices[0]["name"] == "alsa_input.pci.analog-stereo"
|
|
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_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)
|