268 lines
11 KiB
Python
268 lines
11 KiB
Python
"""End-to-end self test.
|
|
|
|
Exercises the whole stack the way the browser does - REST for commands, WebSocket
|
|
for the lifecycle - and prints a pass/fail report.
|
|
|
|
1. start the server: python backend/main.py
|
|
2. in another window: python scripts/selftest.py
|
|
|
|
Works against mock mode out of the box. Against a real robot it also works, but
|
|
the robot will actually speak, so only run it when that is fine.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
try:
|
|
import httpx
|
|
except ImportError: # pragma: no cover
|
|
print("Install dependencies first: pip install -r requirements.txt")
|
|
raise SystemExit(2)
|
|
|
|
try:
|
|
from websockets.asyncio.client import connect as ws_connect
|
|
except Exception: # pragma: no cover
|
|
from websockets.client import connect as ws_connect # type: ignore
|
|
|
|
def _use_utf8_console() -> None:
|
|
"""Windows consoles default to a legacy code page; the robot speaks Chinese."""
|
|
for stream in (sys.stdout, sys.stderr):
|
|
try:
|
|
if stream is not None and hasattr(stream, "reconfigure"):
|
|
stream.reconfigure(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
_use_utf8_console()
|
|
|
|
|
|
PASS = " [PASS]"
|
|
FAIL = " [FAIL]"
|
|
INFO = " [ .. ]"
|
|
|
|
results: List[tuple] = []
|
|
|
|
|
|
def check(name: str, condition: bool, detail: str = "") -> bool:
|
|
results.append((name, condition, detail))
|
|
print("{0} {1}{2}".format(PASS if condition else FAIL, name, " - " + detail if detail else ""))
|
|
return condition
|
|
|
|
|
|
class Listener:
|
|
"""Collects WebSocket events in the background, like the dashboard does."""
|
|
|
|
def __init__(self, url: str) -> None:
|
|
self.url = url
|
|
self.events: List[Dict[str, Any]] = []
|
|
self.hello: Optional[Dict[str, Any]] = None
|
|
self._ws: Any = None
|
|
self._task: Optional[asyncio.Task] = None
|
|
|
|
async def __aenter__(self) -> "Listener":
|
|
self._ws = await ws_connect(self.url)
|
|
self._task = asyncio.create_task(self._read())
|
|
for _ in range(50):
|
|
if self.hello is not None:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
return self
|
|
|
|
async def __aexit__(self, *exc: Any) -> None:
|
|
if self._task:
|
|
self._task.cancel()
|
|
if self._ws:
|
|
await self._ws.close()
|
|
|
|
async def _read(self) -> None:
|
|
async for raw in self._ws:
|
|
message = json.loads(raw)
|
|
if message.get("type") == "hello":
|
|
self.hello = message["data"]
|
|
self.events.append(message)
|
|
|
|
def stages(self, request_id: Optional[str] = None) -> List[str]:
|
|
out = []
|
|
for event in self.events:
|
|
if event.get("type") != "speech.progress":
|
|
continue
|
|
data = event["data"]
|
|
if request_id and data.get("requestId") != request_id:
|
|
continue
|
|
out.append(data["stage"])
|
|
return out
|
|
|
|
async def wait_for_event(self, event_type: str, timeout: float = 10.0) -> bool:
|
|
"""Wait for the next event of a type to arrive (ignoring earlier ones)."""
|
|
seen = sum(1 for e in self.events if e.get("type") == event_type)
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if sum(1 for e in self.events if e.get("type") == event_type) > seen:
|
|
return True
|
|
await asyncio.sleep(0.1)
|
|
return False
|
|
|
|
async def wait_for_stage(self, stage: str, request_id: str, timeout: float = 30.0) -> bool:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if stage in self.stages(request_id):
|
|
return True
|
|
await asyncio.sleep(0.05)
|
|
return False
|
|
|
|
|
|
async def run(base: str) -> int:
|
|
ws_url = base.replace("http://", "ws://").replace("https://", "wss://") + "/ws"
|
|
|
|
async with httpx.AsyncClient(base_url=base, timeout=20.0) as client:
|
|
print("\n=== 1. Server & configuration " + "=" * 40)
|
|
try:
|
|
health = (await client.get("/api/health")).json()
|
|
except Exception as exc:
|
|
print(FAIL + " server unreachable at {0} ({1})".format(base, exc))
|
|
print("\n Start it first: python backend/main.py\n")
|
|
return 1
|
|
check("GET /api/health returns ok", health.get("status") == "ok", "mode=" + str(health.get("mode")))
|
|
|
|
config = (await client.get("/api/config")).json()
|
|
check("GET /api/config returns a mode", config.get("mode") in ("mock", "real"))
|
|
check("no blocking config errors",
|
|
not [i for i in config.get("issues", []) if i["level"] == "error"],
|
|
str(config.get("issues")))
|
|
|
|
print("\n=== 2. Dashboard is served " + "=" * 43)
|
|
page = await client.get("/")
|
|
check("GET / serves the dashboard", page.status_code == 200 and "AGIBOT" in page.text)
|
|
for asset in ("/styles/main.css", "/js/app.js", "/js/api.js", "/js/socket.js", "/js/ui.js"):
|
|
response = await client.get(asset)
|
|
check("GET {0}".format(asset), response.status_code == 200)
|
|
|
|
print("\n=== 3. Robot connection " + "=" * 46)
|
|
status = None
|
|
for _ in range(40): # allow the supervisor a moment to connect
|
|
status = (await client.get("/api/robot/status")).json()
|
|
if status.get("connected"):
|
|
break
|
|
await asyncio.sleep(0.25)
|
|
check("robot reports connected", bool(status and status.get("connected")),
|
|
"state=" + str(status.get("state") if status else "?"))
|
|
|
|
async with Listener(ws_url) as listener:
|
|
check("WebSocket accepts connection and sends hello", listener.hello is not None)
|
|
if listener.hello:
|
|
check("hello carries config + status + history",
|
|
all(k in listener.hello for k in ("config", "status", "history")))
|
|
|
|
print("\n=== 4. Speak (the main workflow) " + "=" * 37)
|
|
text = "Hello, welcome to our company."
|
|
started = time.perf_counter()
|
|
response = await client.post("/api/robot/speak", json={"text": text})
|
|
http_ms = (time.perf_counter() - started) * 1000
|
|
check("POST /api/robot/speak accepted", response.status_code == 200,
|
|
"HTTP {0}".format(response.status_code))
|
|
if response.status_code != 200:
|
|
print(" body:", response.text[:300])
|
|
return 1
|
|
|
|
body = response.json()
|
|
request_id = body["requestId"]
|
|
check("response reports success", body.get("success") is True, json.dumps(body))
|
|
check("HTTP call returns before speech ends (< 3s)", http_ms < 3000,
|
|
"{0:.0f} ms".format(http_ms))
|
|
check("ack latency reported", isinstance(body.get("ackLatencyMs"), int),
|
|
"{0} ms".format(body.get("ackLatencyMs")))
|
|
|
|
got_speaking = await listener.wait_for_stage("speaking", request_id, timeout=10)
|
|
check("lifecycle reaches 'speaking'", got_speaking)
|
|
got_completed = await listener.wait_for_stage("completed", request_id, timeout=60)
|
|
check("lifecycle reaches 'completed'", got_completed)
|
|
stages = listener.stages(request_id)
|
|
check("full stage sequence observed",
|
|
["sending", "processing", "speaking", "completed"] == [
|
|
s for s in stages if s in ("sending", "processing", "speaking", "completed")
|
|
],
|
|
" -> ".join(stages))
|
|
|
|
print("\n=== 5. Stop " + "=" * 58)
|
|
long_text = "This is a much longer sentence used to verify that the stop button " \
|
|
"interrupts an utterance while the robot is still speaking it out loud."
|
|
body2 = (await client.post("/api/robot/speak", json={"text": long_text})).json()
|
|
rid2 = body2["requestId"]
|
|
await listener.wait_for_stage("speaking", rid2, timeout=10)
|
|
stop_response = await client.post("/api/robot/stop")
|
|
check("POST /api/robot/stop accepted", stop_response.status_code == 200)
|
|
cancelled = await listener.wait_for_stage("cancelled", rid2, timeout=10)
|
|
check("utterance reports 'cancelled'", cancelled, " -> ".join(listener.stages(rid2)))
|
|
|
|
print("\n=== 6. Error handling " + "=" * 48)
|
|
empty = await client.post("/api/robot/speak", json={"text": " "})
|
|
check("empty text rejected with 400", empty.status_code == 400,
|
|
"HTTP {0} {1}".format(empty.status_code, empty.text[:120]))
|
|
|
|
too_long = await client.post(
|
|
"/api/robot/speak", json={"text": "x " * (config.get("maxLength", 1000) + 50)}
|
|
)
|
|
check("over-long text rejected with 400", too_long.status_code == 400,
|
|
"HTTP {0}".format(too_long.status_code))
|
|
|
|
missing = await client.post("/api/robot/speak", json={})
|
|
check("malformed body rejected", missing.status_code in (400, 422),
|
|
"HTTP {0}".format(missing.status_code))
|
|
|
|
print("\n=== 7. History " + "=" * 55)
|
|
history = (await client.get("/api/speech/history")).json()
|
|
check("history contains the utterances", history.get("count", 0) >= 2,
|
|
"count={0}".format(history.get("count")))
|
|
first = history["items"][0] if history.get("items") else {}
|
|
check("history entries carry stage + latency",
|
|
"stage" in first and "ackLatencyMs" in first)
|
|
|
|
cleared = (await client.delete("/api/speech/history")).json()
|
|
check("history cleared", cleared.get("success") is True,
|
|
"removed={0}".format(cleared.get("removed")))
|
|
after = (await client.get("/api/speech/history")).json()
|
|
check("history is empty afterwards", after.get("count") == 0)
|
|
|
|
print("\n=== 8. Diagnostics & recovery " + "=" * 40)
|
|
diagnostics = (await client.get("/api/robot/diagnostics")).json()
|
|
check("GET /api/robot/diagnostics works", "status" in diagnostics and "adapter" in diagnostics)
|
|
reconnect = (await client.post("/api/robot/reconnect")).json()
|
|
check("POST /api/robot/reconnect works", reconnect.get("success") is True)
|
|
|
|
# Status is pushed once per health interval, so wait for one rather
|
|
# than assuming the earlier checks took long enough to see it.
|
|
budget = float(config.get("healthInterval") or 5.0) + 4.0
|
|
saw_status_event = await listener.wait_for_event("robot.status", timeout=budget)
|
|
check("robot.status events pushed over WebSocket", saw_status_event,
|
|
"waited up to {0:.0f}s".format(budget))
|
|
|
|
passed = sum(1 for _, ok, _ in results if ok)
|
|
total = len(results)
|
|
print("\n" + "=" * 72)
|
|
print(" {0}/{1} checks passed".format(passed, total))
|
|
if passed != total:
|
|
print("\n Failures:")
|
|
for name, ok, detail in results:
|
|
if not ok:
|
|
print(" - {0} {1}".format(name, detail))
|
|
print("=" * 72 + "\n")
|
|
return 0 if passed == total else 1
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="End-to-end test for the A3 voice dashboard.")
|
|
parser.add_argument("--url", default="http://127.0.0.1:8000", help="Base URL of the running app.")
|
|
args = parser.parse_args()
|
|
raise SystemExit(asyncio.run(run(args.url.rstrip("/"))))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|