my_voice_assistant_v3_jamulix/tests/test_auth.py

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", language=None):
            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