b8950b38a4
Xiaozhi WebSocket endpoint with handshake/auth, mock STT/TTS/Hermes backends, Thai chunker, barge-in queue, latency logger, GPU planner, voice-profile guard (reasoning=none, session_search only). CON-002: no GPU/audio libs loaded at import. MUST-NOT-001: all model names/tokens from config.
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
"""AC-002: app imports and the /ws/xiaozhi endpoint is live (handshake + turn)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import Config
|
|
from app.main import make_app
|
|
from app.security import DeviceAuth
|
|
|
|
TOKEN = "test-device-token"
|
|
|
|
|
|
def _cfg() -> Config:
|
|
return Config.from_dict(
|
|
{
|
|
"security": {"enabled": True, "devices": [
|
|
{"device_id": "d1", "token_hash": DeviceAuth.hash_token(TOKEN)}
|
|
]},
|
|
}
|
|
)
|
|
|
|
|
|
def test_health():
|
|
client = TestClient(make_app(_cfg()))
|
|
r = client.get("/health")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "ok"
|
|
|
|
|
|
def test_ws_handshake_and_turn():
|
|
client = TestClient(make_app(_cfg()))
|
|
with client.websocket_connect("/ws/xiaozhi") as ws:
|
|
# hello frame
|
|
ws.send_text(json.dumps({"device_id": "d1", "authorization": TOKEN}))
|
|
reply = json.loads(ws.receive_text())
|
|
assert reply["type"] == "hello"
|
|
assert reply["session_id"]
|
|
# one utterance -> expect stt + text + audio back
|
|
ws.send_bytes(b"\x00\x00" * 64)
|
|
got_text = got_audio = False
|
|
for _ in range(12):
|
|
try:
|
|
kind = ws.receive()
|
|
except Exception:
|
|
break
|
|
if kind.get("text") is not None:
|
|
got_text = True
|
|
else:
|
|
got_audio = True
|
|
assert got_text and got_audio
|