200 lines
9.5 KiB
Python
200 lines
9.5 KiB
Python
"""Backend contract checks with local HTTP fixtures and a loopback voice server.
|
|
|
|
Run from the repository root: .venv/bin/python -m unittest discover -s s2s-client/tests
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import httpx
|
|
from fastapi.testclient import TestClient
|
|
from websockets.sync.server import serve
|
|
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
# Never consume deployment credentials or contact services from a developer's .env.
|
|
with patch.dict(os.environ, {}, clear=True), patch("dotenv.load_dotenv", return_value=False):
|
|
import server
|
|
|
|
|
|
GRANT = {
|
|
"session_id": "local-session",
|
|
"connect_url": "ws://voice.test/v1/realtime?token=fixture",
|
|
"pending_timeout_s": 60,
|
|
}
|
|
TICKET = {"state": "queued", "queue_id": "local-ticket", "position": 2, "poll_interval_s": 1}
|
|
ASYNC_CLIENT = httpx.AsyncClient
|
|
|
|
|
|
class BackendContractTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.upstream = Mock(side_effect=AssertionError("Unexpected upstream HTTP request"))
|
|
self.enterContext(patch.object(server, "LOAD_BALANCER_URL", "http://lb.test"))
|
|
self.enterContext(patch.object(
|
|
server.httpx, "AsyncClient",
|
|
side_effect=lambda **kwargs: ASYNC_CLIENT(
|
|
transport=httpx.MockTransport(self.upstream), **kwargs,
|
|
),
|
|
))
|
|
self.client = self.enterContext(TestClient(server.app))
|
|
|
|
def respond(self, *responses):
|
|
self.upstream.side_effect = iter(responses)
|
|
|
|
def test_bootstrap_and_removed_account_contract(self):
|
|
response = self.client.get("/api/config")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()["lb"])
|
|
self.assertFalse(response.json()["allowDirect"])
|
|
self.assertNotIn("auth", response.json())
|
|
paths = self.client.get("/openapi.json").json()["paths"]
|
|
for path in ("/api/me", "/api/session/heartbeat", "/api/session/end"):
|
|
self.assertNotIn(path, paths)
|
|
self.assertFalse(any(getattr(route, "path", "").startswith("/oauth") for route in server.app.routes))
|
|
self.upstream.assert_not_called()
|
|
|
|
def test_session_grant_passes_through_without_extra_fields_or_cookies(self):
|
|
self.respond(httpx.Response(200, json=GRANT))
|
|
response = self.client.post("/api/session", json={})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json(), GRANT)
|
|
self.assertNotIn("set-cookie", response.headers)
|
|
request = self.upstream.call_args.args[0]
|
|
self.assertEqual((request.method, str(request.url)), ("POST", "http://lb.test/session"))
|
|
self.assertEqual(json.loads(request.content), {})
|
|
|
|
def test_queue_polling_returns_positions_and_then_grant(self):
|
|
self.respond(
|
|
httpx.Response(200, json=TICKET),
|
|
httpx.Response(200, json={**TICKET, "position": 1}),
|
|
httpx.Response(200, json=GRANT),
|
|
)
|
|
self.assertEqual(self.client.post("/api/session").json(), TICKET)
|
|
self.assertEqual(self.client.get("/api/queue/local-ticket").json(), {**TICKET, "position": 1})
|
|
response = self.client.get("/api/queue/local-ticket")
|
|
self.assertEqual(response.json(), GRANT)
|
|
self.assertNotIn("set-cookie", response.headers)
|
|
self.assertEqual(
|
|
[(call.args[0].method, call.args[0].url.path) for call in self.upstream.call_args_list],
|
|
[("POST", "/session"), ("GET", "/queue/local-ticket"), ("GET", "/queue/local-ticket")],
|
|
)
|
|
|
|
def test_capacity_and_expired_queue_responses_remain(self):
|
|
self.respond(httpx.Response(503, json={"state": "at_capacity"}), httpx.Response(404))
|
|
response = self.client.post("/api/session")
|
|
self.assertEqual((response.status_code, response.json()), (503, {"state": "at_capacity"}))
|
|
response = self.client.get("/api/queue/local-ticket")
|
|
self.assertEqual((response.status_code, response.json()), (404, {"state": "expired"}))
|
|
|
|
def test_explicit_and_beacon_queue_cancellation(self):
|
|
self.respond(httpx.Response(204), httpx.Response(204))
|
|
self.assertEqual(self.client.delete("/api/queue/local-ticket").json(), {"ok": True})
|
|
# sendBeacon posts a JSON blob, not a form or a session identifier.
|
|
response = self.client.post(
|
|
"/api/queue/end", content=json.dumps({"queueId": "local-ticket"}),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.json(), {"ok": True})
|
|
for call in self.upstream.call_args_list:
|
|
self.assertEqual((call.args[0].method, call.args[0].url.path), ("DELETE", "/queue/local-ticket"))
|
|
self.assertEqual(self.upstream.call_count, 2)
|
|
|
|
def test_empty_beacon_is_harmless_and_cancellation_is_best_effort(self):
|
|
self.assertEqual(self.client.post("/api/queue/end", content="").json(), {"ok": True})
|
|
self.upstream.assert_not_called()
|
|
self.upstream.side_effect = httpx.ConnectError("Local fixture unavailable")
|
|
self.assertEqual(self.client.delete("/api/queue/local-ticket").json(), {"ok": True})
|
|
|
|
def test_upstream_failures_are_still_reported(self):
|
|
self.upstream.side_effect = httpx.ConnectError("Local fixture unavailable")
|
|
self.assertEqual(self.client.post("/api/session").status_code, 502)
|
|
self.assertEqual(self.client.get("/api/queue/local-ticket").status_code, 502)
|
|
self.respond(httpx.Response(429), httpx.Response(500))
|
|
self.assertEqual(self.client.post("/api/session").status_code, 502)
|
|
self.assertEqual(self.client.get("/api/queue/local-ticket").status_code, 502)
|
|
|
|
def test_direct_mode_disables_lb_and_preserves_pinned_voice_url(self):
|
|
with patch.object(server, "LOAD_BALANCER_URL", ""), patch.object(
|
|
server, "SPEECH_TO_SPEECH_URL", GRANT["connect_url"],
|
|
):
|
|
config = self.client.get("/api/config").json()
|
|
self.assertEqual(config["s2sUrl"], GRANT["connect_url"])
|
|
self.assertFalse(config["lb"])
|
|
self.assertTrue(config["allowDirect"])
|
|
for method, path in (
|
|
("POST", "/api/session"), ("GET", "/api/queue/local-ticket"),
|
|
("DELETE", "/api/queue/local-ticket"), ("POST", "/api/queue/end"),
|
|
):
|
|
self.assertEqual(self.client.request(method, path).status_code, 404)
|
|
self.upstream.assert_not_called()
|
|
|
|
def test_caller_profile_still_uses_upstream_login_token(self):
|
|
profile = {"result": {"name": "Local Caller"}}
|
|
self.respond(
|
|
httpx.Response(200, json={"token": "fixture-token"}),
|
|
httpx.Response(200, json=profile),
|
|
)
|
|
with patch.object(server, "S2S_LOGIN_BY_PHONE", "http://profile.test/login"), patch.object(
|
|
server, "VOCERO_USER_PHONE_AUTH_URL", "http://profile.test/user",
|
|
):
|
|
response = self.client.get("/api/user/phone", params={"phoneNumber": "+56000000000"})
|
|
self.assertEqual(response.json(), profile)
|
|
login, lookup = [call.args[0] for call in self.upstream.call_args_list]
|
|
self.assertEqual(json.loads(login.content), {"phoneNumber": "+56000000000"})
|
|
self.assertEqual(lookup.headers["authorization"], "Bearer fixture-token")
|
|
self.assertEqual(lookup.url.params["phoneNumber"], "+56000000000")
|
|
|
|
def test_caller_auth_headers_remain_independent(self):
|
|
request = server.Request({
|
|
"type": "http", "headers": [
|
|
(b"authorization", b"Bearer fixture-token"), (b"x-demo-session", b"fixture-demo"),
|
|
],
|
|
})
|
|
self.assertEqual(server._apipilot_parse_authorization(request), ("Bearer fixture-token", "fixture-demo"))
|
|
|
|
|
|
class VoiceProxyTests(unittest.TestCase):
|
|
def test_proxy_voice_frames_keepalive_and_client_disconnect(self):
|
|
closed = threading.Event()
|
|
received = []
|
|
keepalive = threading.Event()
|
|
|
|
def voice(socket):
|
|
try:
|
|
socket.send(json.dumps({"type": "session.created"}))
|
|
if socket.ping().wait(3):
|
|
keepalive.set()
|
|
for frame in socket:
|
|
received.append(frame)
|
|
socket.send(frame)
|
|
finally:
|
|
closed.set()
|
|
|
|
with serve(voice, "127.0.0.1", 0) as upstream:
|
|
thread = threading.Thread(target=upstream.serve_forever, daemon=True)
|
|
thread.start()
|
|
self.addCleanup(thread.join, 5)
|
|
target = f"ws://127.0.0.1:{upstream.socket.getsockname()[1]}/v1/realtime"
|
|
with patch.object(server, "S2S_PROXY_TARGET", target), TestClient(server.app) as client:
|
|
config = client.get("/api/config", headers={"host": "localhost:19420"}).json()
|
|
self.assertEqual(config["s2sUrl"], "ws://localhost:19420/v1/realtime")
|
|
with client.websocket_connect("/v1/realtime") as socket:
|
|
self.assertEqual(socket.receive_json(), {"type": "session.created"})
|
|
update = {"type": "session.update", "session": {"instructions": "Local fixture"}}
|
|
socket.send_json(update)
|
|
self.assertEqual(socket.receive_json(), update)
|
|
socket.send_bytes(b"\x00\x01\x02\x03")
|
|
self.assertEqual(socket.receive_bytes(), b"\x00\x01\x02\x03")
|
|
self.assertTrue(keepalive.wait(3))
|
|
self.assertTrue(closed.wait(3), "Client disconnect must close the upstream WebSocket")
|
|
self.assertEqual(len(received), 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|