Spaces:
Running
Running
File size: 5,680 Bytes
c349d51 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | 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
|