tests/test_admin.py

import json

import pytest

pytest.importorskip("fastapi")
pytest.importorskip("httpx")

from fastapi import FastAPI  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402


class VozFalsa:
    frecuencia_destino = 16000

    def sintetizar(self, texto):
        import numpy as np

        yield np.zeros(1600, dtype=np.int16)


class TranscriptorFalso:
    def transcribe(self, muestras):
        return "¿tengo que avisar de que uso un chatbot?"


class OrquestadorFalso:
    def ask(self, pregunta):
        from types import SimpleNamespace

        respuesta = SimpleNamespace(text="Sí, debe informarlo.", citations=[])
        return SimpleNamespace(answer=respuesta, topic="TRANSPARENCIA")


@pytest.fixture()
def entorno(tmp_path, monkeypatch):
    monkeypatch.setenv("LEXIA_RUNTIME", str(tmp_path / "rt"))
    import importlib
    from pathlib import Path

    from lexia.core import paths as modulo_paths

    importlib.reload(modulo_paths)
    for nombre in ("lexia.core.config", "lexia.core.logging", "lexia.admin.security"):
        importlib.reload(importlib.import_module(nombre))
    from lexia.admin import routes as modulo_rutas
    from lexia.admin import security
    from lexia.web import remote_voice
    from lexia.web.camera import CameraStream

    importlib.reload(modulo_rutas)
    modulo_paths.ensure_runtime()

    camara = CameraStream()
    sesion = remote_voice.RemoteVoiceSession()
    consola = remote_voice.RemoteVoiceConsole(
        OrquestadorFalso(), TranscriptorFalso(), VozFalsa(), sesion
    )
    app = FastAPI()
    panel = Path(modulo_rutas.__file__).resolve().parent.parent / "static" / "admin.html"
    modulo_rutas.register_admin(
        app, panel=panel, camera=camara, session=sesion, console_provider=lambda: consola
    )
    return TestClient(app), security.ensure_token(), camara, sesion


def test_sin_token_no_se_ve_nada(entorno):
    cliente, *_ = entorno
    for ruta in ("/api/admin/overview", "/api/admin/logs", "/api/admin/camera/stream"):
        assert cliente.get(ruta).status_code == 401, ruta


def test_un_token_incorrecto_no_entra(entorno):
    cliente, *_ = entorno
    assert cliente.post("/api/admin/login", json={"token": "0" * 40}).status_code == 401


def test_con_token_se_entra_y_se_ve_el_estado(entorno):
    cliente, token, *_ = entorno
    assert cliente.post("/api/admin/login", json={"token": token}).status_code == 200
    datos = cliente.get("/api/admin/overview").json()
    assert "…" in datos["token"], "el token nunca se muestra entero"
    assert datos["camera"]["enabled"] is False, "la cámara arranca apagada"
    assert datos["camera"]["recording"] is False
    assert "admin" not in datos["settings"], "los ajustes públicos no llevan el token"


def test_la_camara_apagada_no_emite(entorno):
    cliente, token, camara, _ = entorno
    cliente.post("/api/admin/login", json={"token": token})
    assert cliente.get("/api/admin/camera/stream").status_code == 409
    cliente.post("/api/admin/camera", json={"enabled": True})
    assert camara.enabled is True
    cliente.post("/api/admin/camera", json={"enabled": False})
    assert camara.enabled is False


def test_el_token_no_se_puede_escribir_desde_los_ajustes(entorno):
    cliente, token, *_ = entorno
    cliente.post("/api/admin/login", json={"token": token})
    respuesta = cliente.patch("/api/admin/settings", json={"admin": {"token": "x" * 40}})
    assert respuesta.status_code == 400


def test_un_ajuste_invalido_no_se_aplica(entorno):
    cliente, token, *_ = entorno
    cliente.post("/api/admin/login", json={"token": token})
    assert cliente.patch(
        "/api/admin/settings", json={"retrieval": {"require_evidence": False}}
    ).status_code == 422


def test_un_ajuste_valido_se_aplica(entorno):
    cliente, token, *_ = entorno
    cliente.post("/api/admin/login", json={"token": token})
    datos = cliente.patch("/api/admin/settings", json={"expression": {"expressiveness": 0.3}}).json()
    assert datos["expression"]["expressiveness"] == pytest.approx(0.3)


def test_hay_que_reclamar_la_sesion_antes_de_hablar(entorno):
    cliente, token, _, sesion = entorno
    cliente.post("/api/admin/login", json={"token": token})
    assert cliente.post("/api/admin/voice/ask", content=b"\x00\x00" * 800).status_code == 409
    assert cliente.post("/api/admin/voice/claim").json()["active"] is True
    respuesta = cliente.post("/api/admin/voice/ask", content=b"\x00\x00" * 8000)
    assert respuesta.status_code == 200
    assert respuesta.headers["content-type"] == "audio/wav"
    assert respuesta.content[:4] == b"RIFF"
    cliente.post("/api/admin/voice/release")
    assert sesion.active is False


def test_el_registro_se_puede_leer(entorno):
    cliente, token, *_ = entorno
    cliente.post("/api/admin/login", json={"token": token})
    from lexia.core.logging import log

    log("prueba_de_registro", valor=1)
    datos = cliente.get("/api/admin/logs?lines=50").json()
    assert any("prueba_de_registro" in linea for linea in datos["lines"])
    assert all(json.loads(linea) for linea in datos["lines"])


def test_el_panel_expone_las_metricas_de_recuperacion(entorno):
    """Separar recuperar de redactar es lo que hace accionable una queja."""
    cliente, token, *_ = entorno
    respuesta = cliente.get("/api/admin/retrieval?days=7", headers={"x-lexia-token": token})
    assert respuesta.status_code == 200
    datos = respuesta.json()
    assert {"consultas", "etapas", "abstenciones", "evidencia_media"} <= set(datos)


def test_las_metricas_de_recuperacion_piden_token(entorno):
    cliente, *_ = entorno
    assert cliente.get("/api/admin/retrieval").status_code == 401