cca423dac3
PiperTTSProvider ruft das piper-Binary (--output-raw) async auf, liest die native Sample-Rate aus der .onnx.json und resampelt per ffmpeg auf 24000 Hz (Gateway-Norm). Nicht passende Stimmen (z. B. Cloud-Stimme 'Zephyr' aus der Route) fallen auf die konfigurierte Default-Stimme zurueck. Damit ist eine voll-lokale Konstellation (faster-whisper + Ollama + piper) moeglich -> keine API-Kosten, max. Datenschutz. - config: PIPER_BIN/PIPER_VOICES_DIR/PIPER_VOICE/TTS_SAMPLE_RATE (+ .env.example) - dependencies: piper-Factory mit Settings verdrahtet - tests: tests/test_piper_tts.py (offline, Fake-Binary; Resample-Test skippt ohne ffmpeg); e2e/auth-Tests nutzen jetzt einen Stub-TTS statt 'piper' als Pseudo-Stub - docs: README, BEDIENUNGSANLEITUNG (voll-lokal-Beispiel), Architektur-Roadmap Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
124 lines
4 KiB
Python
124 lines
4 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
import app.dependencies as deps
|
|
from app.main import app
|
|
from app.config import settings
|
|
|
|
client = TestClient(app)
|
|
|
|
ADMIN = "admin-secret"
|
|
|
|
|
|
def _enable_auth(monkeypatch):
|
|
monkeypatch.setattr(settings, "auth_enabled", True)
|
|
monkeypatch.setattr(settings, "admin_api_key", ADMIN)
|
|
|
|
|
|
def _create_user(name: str) -> str:
|
|
resp = client.post(
|
|
"/api/admin/users", headers={"X-Admin-Key": ADMIN}, json={"display_name": name}
|
|
)
|
|
assert resp.status_code == 200
|
|
return resp.json()["token"]
|
|
|
|
|
|
def test_protected_endpoint_requires_token(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
resp = client.post("/api/speak", json={"text": "x", "tts_provider": "piper"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_admin_requires_key(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
resp = client.post("/api/admin/users", json={"display_name": "Anna"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_create_user_and_call_me(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
token = _create_user("Anna")
|
|
auth = {"Authorization": f"Bearer {token}"}
|
|
|
|
me = client.get("/api/me", headers=auth)
|
|
assert me.status_code == 200
|
|
assert me.json()["display_name"] == "Anna"
|
|
|
|
bad = client.get("/api/me", headers={"Authorization": "Bearer nope"})
|
|
assert bad.status_code == 401
|
|
|
|
|
|
def test_tenant_isolation_returns_403(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
token_a = _create_user("A")
|
|
token_b = _create_user("B")
|
|
|
|
client.post(
|
|
"/api/sessions/shared/route",
|
|
headers={"Authorization": f"Bearer {token_a}"},
|
|
json={"tts_provider": "piper"},
|
|
)
|
|
# B versucht die Session von A zu nutzen.
|
|
resp = client.post(
|
|
"/api/speak?session_id=shared",
|
|
headers={"Authorization": f"Bearer {token_b}"},
|
|
json={"text": "x"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_user_prefs_applied_to_route(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
|
|
class StubTTS:
|
|
async def synthesize(self, text, voice=None, audio_format="pcm"):
|
|
return b""
|
|
monkeypatch.setitem(deps.TTS_REGISTRY, "stub-tts", lambda s: StubTTS())
|
|
|
|
token = _create_user("Pref")
|
|
auth = {"Authorization": f"Bearer {token}"}
|
|
|
|
client.put(
|
|
"/api/me/prefs",
|
|
headers=auth,
|
|
json={"tts_provider": "stub-tts", "output_endpoint": "loopback"},
|
|
)
|
|
resp = client.post("/api/speak", headers=auth, json={"text": "hallo"})
|
|
assert resp.status_code == 200
|
|
assert resp.headers["X-TTS-Provider"] == "stub-tts"
|
|
assert resp.headers["X-Output-Endpoint"] == "loopback"
|
|
|
|
|
|
def test_memories_crud(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
auth = {"Authorization": f"Bearer {_create_user('Mem')}"}
|
|
|
|
assert client.get("/api/me/memories", headers=auth).json() == []
|
|
|
|
created = client.post("/api/me/memories", headers=auth, json={"content": "mag Tee"})
|
|
assert created.status_code == 200
|
|
mid = created.json()["id"]
|
|
|
|
listed = client.get("/api/me/memories", headers=auth).json()
|
|
assert len(listed) == 1 and listed[0]["content"] == "mag Tee"
|
|
|
|
assert client.delete(f"/api/me/memories/{mid}", headers=auth).status_code == 200
|
|
assert client.get("/api/me/memories", headers=auth).json() == []
|
|
assert client.delete("/api/me/memories/9999", headers=auth).status_code == 404
|
|
|
|
|
|
def test_memories_are_per_user(monkeypatch):
|
|
_enable_auth(monkeypatch)
|
|
auth_a = {"Authorization": f"Bearer {_create_user('A')}"}
|
|
auth_b = {"Authorization": f"Bearer {_create_user('B')}"}
|
|
client.post("/api/me/memories", headers=auth_a, json={"content": "geheim A"})
|
|
assert client.get("/api/me/memories", headers=auth_b).json() == []
|
|
|
|
|
|
def test_admin_unconfigured_returns_503(monkeypatch):
|
|
# Auth an, aber kein Admin-Key gesetzt.
|
|
monkeypatch.setattr(settings, "auth_enabled", True)
|
|
monkeypatch.setattr(settings, "admin_api_key", "")
|
|
resp = client.post(
|
|
"/api/admin/users", headers={"X-Admin-Key": "irgendwas"}, json={"display_name": "X"}
|
|
)
|
|
assert resp.status_code == 503
|