diff --git a/tests/test_admin_auth.py b/tests/test_admin_auth.py new file mode 100644 index 0000000..80a0f69 --- /dev/null +++ b/tests/test_admin_auth.py @@ -0,0 +1,56 @@ +"""Tests for admin JWT issue/verify and the ``require_admin`` HTTP guard. + +``test_auth_route`` already covers the /admin/login endpoint (correct password → token, +wrong → 401, brute-force limiting). This module covers the token itself: the round-trip, +expiry, and how ``require_admin`` maps a missing/invalid/expired token to a 401. The +guard is tested directly (no TestClient) since it's pure dependency logic. +""" + +import jwt +import pytest +from fastapi import HTTPException +from fastapi.security import HTTPAuthorizationCredentials + +from app.core import security +from app.core.security import create_token, decode_token +from app.dependencies import require_admin + + +def _creds(token: str) -> HTTPAuthorizationCredentials: + return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) + + +def test_token_round_trip() -> None: + assert decode_token(create_token()) == "admin" + + +def test_expired_token_is_rejected(monkeypatch) -> None: + # Mint a token that expired an hour ago, then confirm decode refuses it. + monkeypatch.setattr(security.settings, "JWT_EXPIRE_HOURS", -1) + token = create_token() + with pytest.raises(jwt.ExpiredSignatureError): + decode_token(token) + + +def test_require_admin_accepts_valid_token() -> None: + assert require_admin(_creds(create_token())) == "admin" + + +def test_require_admin_rejects_missing_token() -> None: + with pytest.raises(HTTPException) as exc: + require_admin(None) + assert exc.value.status_code == 401 + + +def test_require_admin_rejects_garbage_token() -> None: + with pytest.raises(HTTPException) as exc: + require_admin(_creds("not-a-jwt")) + assert exc.value.status_code == 401 + + +def test_require_admin_rejects_expired_token(monkeypatch) -> None: + monkeypatch.setattr(security.settings, "JWT_EXPIRE_HOURS", -1) + expired = create_token() + with pytest.raises(HTTPException) as exc: + require_admin(_creds(expired)) + assert exc.value.status_code == 401 diff --git a/tests/test_admin_route.py b/tests/test_admin_route.py new file mode 100644 index 0000000..66b827e --- /dev/null +++ b/tests/test_admin_route.py @@ -0,0 +1,139 @@ +"""Integration tests for the /admin observability & config routes. + +One test per route in ``api/routers/admin.py``, driven through ``TestClient`` with +``require_admin`` overridden (the same pattern ``test_sessions_route`` uses for the +/sessions routes). The two routes that would do real work — Whisper transcription and +the model-test API call — have their service functions mocked, so nothing hits ffmpeg +or the Anthropic API. +""" + +import pytest +from fastapi.testclient import TestClient + +from app.api.routers import admin +from app.config import settings +from app.dependencies import require_admin +from app.main import app + +client = TestClient(app) + + +@pytest.fixture(autouse=True) +def _override_admin() -> None: + # Bypass the JWT check per test; another module may pop this, so set it fresh. + app.dependency_overrides[require_admin] = lambda: "test-admin" + yield + app.dependency_overrides.pop(require_admin, None) + + +def test_routes_require_admin() -> None: + # Drop the override to confirm the router really is gated. + app.dependency_overrides.pop(require_admin, None) + assert client.get("/v1/admin/health").status_code == 401 + assert client.get("/v1/admin/config").status_code == 401 + + +def test_health() -> None: + resp = client.get("/v1/admin/health") + assert resp.status_code == 200 + body = resp.json() + assert "uptime_seconds" in body + assert body["python_version"] + + +def test_prompt() -> None: + resp = client.get("/v1/admin/prompt") + assert resp.status_code == 200 + body = resp.json() + assert body["system_prompt"] + assert body["min_words"] >= 1 + assert isinstance(body["valid_statuses"], list) + + +def test_config_lists_blocks() -> None: + resp = client.get("/v1/admin/config") + assert resp.status_code == 200 + blocks = resp.json()["blocks"] + keys = {f["key"] for b in blocks for f in b["fields"]} + assert "ANTHROPIC_MODEL" in keys + # A secret-status field reports whether it's set, never its raw value. + api_key = next( + f for b in blocks for f in b["fields"] if f["key"] == "ANTHROPIC_API_KEY" + ) + assert "value" not in api_key or api_key.get("value") is None + assert "configured" in api_key + + +def test_patch_config_updates_editable_field() -> None: + original: str = settings.LOG_LEVEL + try: + resp = client.patch( + "/v1/admin/config", json={"updates": {"LOG_LEVEL": "DEBUG"}} + ) + assert resp.status_code == 200 + assert resp.json()["changed"] == {"LOG_LEVEL": "DEBUG"} + assert settings.LOG_LEVEL == "DEBUG" + finally: + settings.LOG_LEVEL = original + + +def test_patch_config_rejects_non_editable_field() -> None: + resp = client.patch( + "/v1/admin/config", json={"updates": {"DATABASE_URL": "sqlite://x"}} + ) + assert resp.status_code == 422 + + +def test_logs() -> None: + resp = client.get("/v1/admin/logs") + assert resp.status_code == 200 + assert "entries" in resp.json() + + +def test_ws_status() -> None: + resp = client.get("/v1/admin/ws/status") + assert resp.status_code == 200 + body = resp.json() + assert body["active"] == [] + assert "total_since_start" in body + + +def test_whisper_transcribe(monkeypatch) -> None: + monkeypatch.setattr( + admin, "transcribe_with_detail", lambda audio: {"segments": ["bonjour"]} + ) + resp = client.post( + "/v1/admin/whisper/transcribe", + files={"file": ("clip.wav", b"\x00\x01", "audio/wav")}, + ) + assert resp.status_code == 200 + assert resp.json() == {"segments": ["bonjour"]} + + +def test_whisper_transcribe_rejects_empty_file() -> None: + resp = client.post( + "/v1/admin/whisper/transcribe", files={"file": ("clip.wav", b"", "audio/wav")} + ) + assert resp.status_code == 422 + + +def test_model_test(monkeypatch) -> None: + async def _fake_extract(text, web_search): + return { + "claims": [], + "turns": 1, + "usage": {"input_tokens": 10, "output_tokens": 5}, + "model": "test-model", + "web_search_enabled": web_search, + "web_search_called": False, + } + + monkeypatch.setattr(admin, "debug_extract", _fake_extract) + monkeypatch.setattr(admin, "estimate_cost", lambda *a, **k: 0.0) + + resp = client.post("/v1/admin/model-test", json={"text": "La Terre est plate."}) + assert resp.status_code == 200 + body = resp.json() + assert body["turns"] == 1 + assert body["model"] == "test-model" + assert body["estimated_cost_usd"] == 0.0 diff --git a/tests/test_session_cycle.py b/tests/test_session_cycle.py new file mode 100644 index 0000000..6ed3702 --- /dev/null +++ b/tests/test_session_cycle.py @@ -0,0 +1,178 @@ +"""Tests for the per-utterance claim cycle in session.py. + +Covers the pieces the dedup/persistence tests don't: +- ``_make_claim``: result dict → ``Claim`` mapping. +- ``_spawn_claims``: skips a transcript under ``MIN_WORDS``, otherwise fires a task. +- ``_process_claims``: the pending → claim(s) / remove_claim lifecycle. + +Everything runs in isolation — ``extract_and_verify`` is mocked, the WebSocket is a +fake that records ``send_json`` payloads, and persistence is turned off so no DB or +webhook is touched. +""" + +import asyncio +from collections import OrderedDict + +import pytest + +from app.services import session +from app.services.claim_extractor import MIN_WORDS, ExtractResult +from app.services.session import _make_claim, _process_claims, _spawn_claims + + +class FakeWebSocket: + """Minimal stand-in: records every JSON frame the session sends.""" + + def __init__(self) -> None: + self.sent: list[dict] = [] + + async def send_json(self, message: dict) -> None: + self.sent.append(message) + + +@pytest.fixture(autouse=True) +def _no_persistence(monkeypatch) -> None: + # Keep _process_claims off the DB: _persist short-circuits when this is False. + monkeypatch.setattr(session.settings, "PERSIST_SESSIONS", False) + + +def _mock_extract(monkeypatch, claims: list[dict]) -> None: + async def _fake(transcript, context, web_search): + return ExtractResult(claims=claims) + + monkeypatch.setattr(session, "extract_and_verify", _fake) + + +# --- _make_claim ---------------------------------------------------------------- + + +def test_make_claim_maps_all_fields() -> None: + result = { + "text": "La Terre est plate.", + "status": "false", + "explanation": "La Terre est un géoïde.", + "sources": ["https://example.com"], + "category": "science", + "confidence": 6, + "counter_claim": "La Terre est sphérique.", + "web_search_used": True, + } + claim = _make_claim(result, "claim-1", 1234) + + assert claim.id == "claim-1" + assert claim.text == "La Terre est plate." + assert claim.status.value == "false" + assert claim.timestamp == 1234 + assert claim.sources == ["https://example.com"] + assert claim.category == "science" + assert claim.confidence == 6 + assert claim.counter_claim == "La Terre est sphérique." + assert claim.web_search_used is True + + +def test_make_claim_uses_defaults_for_optional_fields() -> None: + claim = _make_claim({"text": "x", "status": "verified", "explanation": ""}, "id", 0) + assert claim.sources == [] + assert claim.category == "" + assert claim.confidence == 0 + assert claim.web_search_used is False + + +# --- _spawn_claims -------------------------------------------------------------- + + +def _session_info() -> dict: + return { + "id": "sess-1", + "claims_spawned": 0, + "verification_level": session.VerificationLevel.FAST, + "seen_claims": OrderedDict(), + "webhooks": [], + } + + +def test_spawn_skips_transcript_under_min_words() -> None: + info = _session_info() + tasks: set = set() + short = " ".join(["mot"] * (MIN_WORDS - 1)) + + _spawn_claims(FakeWebSocket(), short, [], tasks, info, "seg-1") + + assert info["claims_spawned"] == 0 + assert tasks == set() + + +def test_spawn_fires_task_when_long_enough(monkeypatch) -> None: + _mock_extract(monkeypatch, claims=[]) + info = _session_info() + long = " ".join(["mot"] * MIN_WORDS) + + async def _run() -> None: + tasks: set = set() + _spawn_claims(FakeWebSocket(), long, [], tasks, info, "seg-1") + assert info["claims_spawned"] == 1 + assert len(tasks) == 1 + await asyncio.gather(*tasks) + + asyncio.run(_run()) + + +# --- _process_claims lifecycle -------------------------------------------------- + + +def _process(ws: FakeWebSocket) -> None: + asyncio.run( + _process_claims( + ws, + transcript="La Terre est plate.", + context=[], + web_search=False, + session_id="sess-1", + segment_id="seg-1", + seen_claims=OrderedDict(), + webhooks=[], + ) + ) + + +def _verified(text: str) -> dict: + return {"text": text, "status": "false", "explanation": "..."} + + +def test_pending_then_verified_claim_reuses_id(monkeypatch) -> None: + _mock_extract(monkeypatch, claims=[_verified("La Terre est plate.")]) + ws = FakeWebSocket() + _process(ws) + + assert [m["type"] for m in ws.sent] == ["claim", "claim"] + pending, verified = ws.sent + assert pending["claim"]["status"] == "pending" + # The verified claim replaces the placeholder in-place: same id, real status. + assert verified["claim"]["id"] == pending["claim"]["id"] + assert verified["claim"]["status"] == "false" + + +def test_no_claim_removes_pending(monkeypatch) -> None: + _mock_extract(monkeypatch, claims=[]) + ws = FakeWebSocket() + _process(ws) + + assert [m["type"] for m in ws.sent] == ["claim", "remove_claim"] + pending, removal = ws.sent + assert pending["claim"]["status"] == "pending" + assert removal["id"] == pending["claim"]["id"] + + +def test_extra_claims_get_fresh_ids(monkeypatch) -> None: + _mock_extract( + monkeypatch, claims=[_verified("Premier fait."), _verified("Second fait.")] + ) + ws = FakeWebSocket() + _process(ws) + + # pending, first (reusing the pending id), then one extra with a new id. + assert [m["type"] for m in ws.sent] == ["claim", "claim", "claim"] + pending_id = ws.sent[0]["claim"]["id"] + assert ws.sent[1]["claim"]["id"] == pending_id + assert ws.sent[2]["claim"]["id"] != pending_id + assert ws.sent[2]["claim"]["text"] == "Second fait."