kunthawat
a582c1ff36
- auth from Authorization/Device-Id headers (firmware v2.5.0), frame fallback kept
- server hello carries transport=websocket + session_id (firmware ParseServerHello)
- /ota GET+POST returns {websocket:{url,token,version}, server_time:{timestamp,timezone_offset}, firmware:{version}}
- config.yaml: device registered by MAC, ota block added
- 21 tests pass (added OTA + header-auth coverage)
101 lines
3.0 KiB
Python
101 lines
3.0 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_ota_endpoint_delivers_server():
|
|
"""FIRMWARE: device reads websocket.url/token from /ota POST."""
|
|
cfg = _cfg()
|
|
from app.config import OtaConfig
|
|
|
|
cfg.ota = OtaConfig(
|
|
server_ws_url="wss://zhi.moreminimore.com/ws/xiaozhi",
|
|
server_token=TOKEN,
|
|
)
|
|
client = TestClient(make_app(cfg))
|
|
r = client.post("/ota", json={"device_id": "d1"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["websocket"]["url"] == "wss://zhi.moreminimore.com/ws/xiaozhi"
|
|
assert body["websocket"]["token"] == TOKEN
|
|
assert "server_time" in body
|
|
|
|
|
|
def test_ws_auth_via_headers():
|
|
"""FIRMWARE: token in Authorization header, Device-Id = MAC."""
|
|
client = TestClient(make_app(_cfg()))
|
|
with client.websocket_connect(
|
|
"/ws/xiaozhi",
|
|
headers={
|
|
"Device-Id": "d1",
|
|
"Authorization": f"Bearer {TOKEN}",
|
|
},
|
|
) as ws:
|
|
# hello frame WITHOUT auth fields — auth came in headers
|
|
ws.send_text(json.dumps({"device_id": "d1"}))
|
|
reply = json.loads(ws.receive_text())
|
|
assert reply["type"] == "hello"
|
|
assert reply["transport"] == "websocket"
|
|
assert reply["session_id"]
|
|
|
|
|
|
def test_ws_header_auth_rejects_bad_token():
|
|
client = TestClient(make_app(_cfg()))
|
|
with client.websocket_connect(
|
|
"/ws/xiaozhi",
|
|
headers={"Device-Id": "d1", "Authorization": "Bearer wrong"},
|
|
) as ws:
|
|
ws.send_text(json.dumps({"device_id": "d1"}))
|
|
reply = json.loads(ws.receive_text())
|
|
assert reply["type"] == "error"
|
|
|
|
|
|
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
|