tests/core/test_wsx_connection.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """One websocket connection, from the handshake to the last message (#68 phase 2).16 17 Contract tests. ``WsxConnection.serve()`` is the whole life of a socket: the18 gate (the server's state, the Origin, the identity, the home application's19 cookie), the accept, the read loop where every message becomes a request, and20 the wait for what is still in flight.21 22 The gate answers in ONE shape: accept first, then close with a code the browser23 can read. The single exception is a hostile Origin, refused before the accept —24 there is nothing to tell a caller that was never admitted.25 """26 27 from __future__ import annotations28 29 import asyncio30 from typing import Any31 32 from genro_asgi import BaseApplication, BaseServer, MiddlewareMixin33 from genro_asgi.exceptions import HTTPForbidden, HTTPUnauthorized34 from genro_asgi.request import Request35 from genro_tytx import to_msgpack, to_tytx36 37 from genro_asgi.types import Message, Receive, Scope, Send38 from genro_asgi.wsx import WsxConnection, WsxEnvelope39 40 PREFIX = "WSX://"41 42 43 class WsServer(MiddlewareMixin, BaseServer):44 """The composition a websocket needs: the chain, for the session layer."""45 46 47 class EchoApp(BaseApplication):48 """Answers what it was asked, so a test can see the whole request."""49 50 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:51 if scope["path"] == "/boom":52 raise RuntimeError("boom")53 if scope["path"] == "/forbidden":54 raise HTTPForbidden("not yours")55 request = Request(scope, receive, server=self.server, application=self)56 await request.init()57 request.response.set_result(58 {59 "method": scope["method"],60 "path": scope["path"],61 "data": request.data,62 "identity": scope["auth"].identity if scope.get("auth") else None,63 "session": scope["session"].id if scope.get("session") else None,64 "host": request.headers.get("host"),65 }66 )67 await request.response(scope, receive, send)68 69 70 class TellingApp(EchoApp):71 """Answers with what the synthetic scope carried, and with other shapes."""72 73 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:74 if scope["path"] == "/keys":75 await send(76 {77 "type": "http.response.start",78 "status": 200,79 "headers": [(b"content-type", b"application/json")],80 }81 )82 keys = {83 "page_id": scope.get("genro.page_id"),84 "reply_path": scope.get("genro.reply_path"),85 }86 await send({"type": "http.response.body", "body": to_tytx(keys, "json").encode()})87 return88 if scope["path"] == "/plain":89 await send(90 {91 "type": "http.response.start",92 "status": 200,93 "headers": [(b"content-type", b"text/plain")],94 }95 )96 await send({"type": "http.response.body", "body": b"just text"})97 return98 if scope["path"] == "/stream":99 await send({"type": "http.response.start", "status": 200, "headers": []})100 await send({"type": "http.response.body", "body": b"first", "more_body": True})101 await send({"type": "http.response.body", "body": b"second"})102 return103 if scope["path"] == "/msgpack":104 await send(105 {106 "type": "http.response.start",107 "status": 200,108 "headers": [(b"content-type", b"application/vnd.tytx+msgpack")],109 }110 )111 await send({"type": "http.response.body", "body": to_msgpack({"n": 7})})112 return113 if scope["path"] == "/binary":114 await send(115 {116 "type": "http.response.start",117 "status": 200,118 "headers": [(b"content-type", b"image/png")],119 }120 )121 await send({"type": "http.response.body", "body": b"\x89PNG\xff"})122 return123 if scope["path"] == "/silent":124 await send({"type": "http.response.start", "status": 204, "headers": []})125 await send({"type": "http.response.body", "body": b""})126 return127 await super().__call__(scope, receive, send)128 129 130 class GatedApp(EchoApp):131 """An application that admits no handshake without its cookie."""132 133 @property134 def handshake_cookie(self) -> str | None:135 return "spa_connection_id"136 137 138 class XT_Socket:139 """One scripted browser: texts in, ASGI messages out."""140 141 def __init__(self, *texts: str) -> None:142 self.incoming: list[Message] = [{"type": "websocket.connect"}]143 self.incoming += [{"type": "websocket.receive", "text": text} for text in texts]144 self.incoming.append({"type": "websocket.disconnect", "code": 1000})145 self.sent: list[Message] = []146 147 async def receive(self) -> Message:148 if not self.incoming:149 await asyncio.sleep(0)150 return {"type": "websocket.disconnect", "code": 1006}151 message = self.incoming.pop(0)152 if message["type"] == "websocket.disconnect":153 # A real browser does not vanish the instant after it spoke: give154 # the messages in flight the turns they need to answer.155 for _ in range(10):156 await asyncio.sleep(0)157 return message158 159 async def send(self, message: Message) -> None:160 self.sent.append(message)161 162 @property163 def accepted(self) -> bool:164 return any(m["type"] == "websocket.accept" for m in self.sent)165 166 @property167 def closed(self) -> tuple[int, str] | None:168 for message in self.sent:169 if message["type"] == "websocket.close":170 return message["code"], message.get("reason", "")171 return None172 173 @property174 def answers(self) -> list[WsxEnvelope]:175 return [176 WsxEnvelope(m["text"])177 for m in self.sent178 if m["type"] == "websocket.send" and m.get("text", "").startswith(PREFIX)179 ]180 181 182 def request_message(path: str, **fields: Any) -> str:183 """One WSX request as the browser would write it."""184 return WsxEnvelope(method="WSK", path=path, **fields).encode()185 186 187 async def drive(188 server: BaseServer,189 socket: XT_Socket,190 path: str = "/echo/main",191 headers: list[tuple[bytes, bytes]] | None = None,192 ) -> XT_Socket:193 """Live one whole connection and hand back what the browser saw."""194 scope: Scope = {195 "type": "websocket",196 "path": path,197 "headers": headers if headers is not None else [(b"host", b"example.org")],198 "query_string": b"",199 "subprotocols": [],200 }201 await WsxConnection(server, scope, socket.receive, socket.send).serve()202 return socket203 204 205 def echo_server(**kwargs: Any) -> WsServer:206 return WsServer(applications=[EchoApp(mount="echo")], **kwargs)207 208 209 def closure(socket: XT_Socket) -> tuple[int, str]:210 """The close the browser saw; a socket that was never closed is a failure."""211 closed = socket.closed212 assert closed is not None, "the socket was never closed"213 return closed214 215 216 class TestTheGate:217 async def test_a_handshake_on_a_path_no_application_serves_is_closed_1008(self) -> None:218 socket = await drive(echo_server(), XT_Socket(), path="/nowhere/x")219 assert socket.accepted220 assert closure(socket)[0] == 1008 and "no application" in closure(socket)[1]221 222 async def test_the_home_application_may_demand_a_cookie(self) -> None:223 server = WsServer(applications=[GatedApp(mount="spa")])224 socket = await drive(server, XT_Socket(), path="/spa/_wsx")225 assert socket.accepted226 assert closure(socket)[0] == 1008 and "cookie" in closure(socket)[1]227 228 async def test_the_handshake_passes_when_the_cookie_is_there(self) -> None:229 # Nothing closes it: the client left on its own, and there is nobody230 # left to tell.231 server = WsServer(applications=[GatedApp(mount="spa")])232 socket = await drive(233 server,234 XT_Socket(),235 path="/spa/_wsx",236 headers=[(b"cookie", b"spa_connection_id=c1")],237 )238 assert socket.accepted and socket.closed is None239 240 async def test_an_application_that_gates_nothing_lets_every_handshake_in(self) -> None:241 socket = await drive(echo_server(), XT_Socket())242 assert socket.accepted and socket.closed is None243 244 245 class TestTheIdentity:246 async def test_an_invalid_credential_is_accepted_then_closed_1008(self) -> None:247 class RefusingServer(WsServer):248 def authenticate(self, request: Any) -> Any:249 raise HTTPUnauthorized("bad token")250 251 server = RefusingServer(applications=[EchoApp(mount="echo")])252 socket = await drive(server, XT_Socket())253 assert socket.accepted254 assert closure(socket)[0] == 1008 and "bad token" in closure(socket)[1]255 256 257 class TestWhatTheSyntheticScopeCarries:258 def telling_server(self) -> WsServer:259 return WsServer(applications=[TellingApp(mount="echo")])260 261 async def test_the_page_and_the_reply_path_reach_the_application(self) -> None:262 message = request_message("/echo/keys", id="m1", page_id="p1", reply_path="/echo/done")263 socket = await drive(self.telling_server(), XT_Socket(message))264 assert socket.answers[0].data == {"page_id": "p1", "reply_path": "/echo/done"}265 266 async def test_a_message_without_them_leaves_them_out_of_the_scope(self) -> None:267 socket = await drive(268 self.telling_server(), XT_Socket(request_message("/echo/keys", id="m1"))269 )270 assert socket.answers[0].data == {"page_id": None, "reply_path": None}271 272 async def test_a_plain_text_answer_travels_as_its_text(self) -> None:273 socket = await drive(274 self.telling_server(), XT_Socket(request_message("/echo/plain", id="m1"))275 )276 assert socket.answers[0].data == "just text"277 278 async def test_an_empty_answer_carries_no_data(self) -> None:279 socket = await drive(280 self.telling_server(), XT_Socket(request_message("/echo/silent", id="m1"))281 )282 assert (socket.answers[0].status, socket.answers[0].data) == (204, None)283 284 async def test_a_msgpack_answer_comes_back_hydrated(self) -> None:285 # The third transport a TYTX answer may use: the mirror of what286 # `Request.decode_body` understands on the way in.287 socket = await drive(288 self.telling_server(), XT_Socket(request_message("/echo/msgpack", id="m1"))289 )290 assert socket.answers[0].data == {"n": 7}291 292 async def test_a_binary_answer_travels_as_its_bytes(self) -> None:293 # Now that the codec carries bytes, an answer nobody can decode as text294 # reaches the page as the bytes it is.295 socket = await drive(296 self.telling_server(), XT_Socket(request_message("/echo/binary", id="m1"))297 )298 assert socket.answers[0].data == b"\x89PNG\xff"299 300 async def test_a_streaming_answer_is_refused_out_loud(self) -> None:301 socket = await drive(302 self.telling_server(), XT_Socket(request_message("/echo/stream", id="m1"))303 )304 assert socket.answers[0].status == 500305 assert "streaming" in socket.answers[0].data306 307 308 class TestTheDrain:309 async def test_what_is_still_in_flight_when_the_client_leaves_is_cut(310 self, monkeypatch: Any311 ) -> None:312 # The wait is bounded: a message whose handler hangs must not keep the313 # connection's own task alive for ever.314 import genro_asgi.wsx as wsx_module315 316 monkeypatch.setattr(wsx_module, "DRAIN_TIMEOUT_SECONDS", 0.01)317 started = asyncio.Event()318 319 class HangingApp(EchoApp):320 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:321 started.set()322 await asyncio.sleep(30)323 324 server = WsServer(applications=[HangingApp(mount="echo")])325 socket = await drive(server, XT_Socket(request_message("/echo/main", id="m1")))326 assert started.is_set()327 assert socket.answers == []328 329 330 class TestAnAnswerNobodyIsThereFor:331 async def test_a_handler_that_ends_after_the_client_left_writes_nothing(self) -> None:332 # The message was served, but the socket is gone: writing to it would333 # raise, and there is nobody to read the answer anyway.334 class SlowApp(EchoApp):335 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:336 await asyncio.sleep(0.02)337 await super().__call__(scope, receive, send)338 339 server = WsServer(applications=[SlowApp(mount="echo")])340 socket = await drive(server, XT_Socket(request_message("/echo/main", id="m1")))341 assert socket.answers == []342 343 344 class TestTheOrigin:345 async def test_no_origin_header_passes(self) -> None:346 socket = await drive(echo_server(websocket={"origins": ["https://app.example.org"]}), XT_Socket())347 assert socket.accepted348 349 async def test_a_listed_origin_passes(self) -> None:350 server = echo_server(websocket={"origins": ["https://app.example.org"]})351 socket = await drive(352 server,353 XT_Socket(),354 headers=[(b"host", b"example.org"), (b"origin", b"https://app.example.org")],355 )356 assert socket.accepted357 358 async def test_an_unlisted_origin_is_refused_without_an_accept(self) -> None:359 # The one refusal with no accept: nothing was admitted, so there is360 # nobody to tell.361 server = echo_server(websocket={"origins": ["https://app.example.org"]})362 socket = await drive(363 server,364 XT_Socket(),365 headers=[(b"host", b"example.org"), (b"origin", b"https://evil.example.com")],366 )367 assert not socket.accepted368 assert closure(socket)[0] == 1008 and "origin" in closure(socket)[1].lower()369 370 async def test_a_star_admits_every_origin(self) -> None:371 server = echo_server(websocket={"origins": ["*"]})372 socket = await drive(373 server,374 XT_Socket(),375 headers=[(b"host", b"example.org"), (b"origin", b"https://anywhere.example.com")],376 )377 assert socket.accepted378 379 async def test_with_no_list_the_origin_must_match_the_host(self) -> None:380 server = echo_server()381 same = await drive(382 server,383 XT_Socket(),384 headers=[(b"host", b"example.org"), (b"origin", b"https://example.org")],385 )386 other = await drive(387 server,388 XT_Socket(),389 headers=[(b"host", b"example.org"), (b"origin", b"https://evil.example.com")],390 )391 assert same.accepted392 assert not other.accepted and closure(other)[0] == 1008393 394 395 class TestAMessageIsARequest:396 async def test_the_answer_carries_the_id_of_the_message(self) -> None:397 socket = await drive(echo_server(), XT_Socket(request_message("/echo/main", id="m1")))398 assert [(a.id, a.status) for a in socket.answers] == [("m1", 200)]399 400 async def test_the_application_sees_the_method_wsk_and_the_path(self) -> None:401 socket = await drive(echo_server(), XT_Socket(request_message("/echo/main", id="m1")))402 answered = socket.answers[0].data403 assert (answered["method"], answered["path"]) == ("WSK", "/main")404 405 async def test_the_data_of_the_message_reaches_the_application_hydrated(self) -> None:406 message = request_message("/echo/main", id="m1", data={"n": 41, "text": "ok"})407 socket = await drive(echo_server(), XT_Socket(message))408 assert socket.answers[0].data["data"] == {"n": 41, "text": "ok"}409 410 async def test_the_headers_of_the_handshake_travel_with_every_message(self) -> None:411 socket = await drive(echo_server(), XT_Socket(request_message("/echo/main", id="m1")))412 assert socket.answers[0].data["host"] == "example.org"413 414 async def test_a_message_with_no_id_is_answered_by_nothing(self) -> None:415 socket = await drive(echo_server(), XT_Socket(request_message("/echo/main")))416 assert socket.answers == []417 418 async def test_a_binary_frame_is_ignored_and_the_socket_lives_on(self) -> None:419 socket = XT_Socket(request_message("/echo/main", id="m1"))420 socket.incoming.insert(1, {"type": "websocket.receive", "bytes": b"\x00\x01"})421 await drive(echo_server(), socket)422 assert [a.id for a in socket.answers] == ["m1"]423 424 async def test_a_text_that_is_not_wsx_is_ignored_and_the_socket_lives_on(self) -> None:425 socket = await drive(426 echo_server(), XT_Socket("hello", request_message("/echo/main", id="m1"))427 )428 assert [a.id for a in socket.answers] == ["m1"]429 430 async def test_two_messages_are_answered_each_with_its_own_id(self) -> None:431 socket = await drive(432 echo_server(),433 XT_Socket(request_message("/echo/main", id="m1"), request_message("/echo/main", id="m2")),434 )435 assert sorted(a.id for a in socket.answers) == ["m1", "m2"]436 437 438 class TestWhenTheApplicationRefuses:439 async def test_an_http_exception_becomes_the_status_of_the_answer(self) -> None:440 socket = await drive(echo_server(), XT_Socket(request_message("/echo/forbidden", id="m1")))441 assert socket.answers[0].status == 403442 443 async def test_any_other_failure_is_a_500_and_the_socket_survives(self) -> None:444 socket = await drive(445 echo_server(),446 XT_Socket(request_message("/echo/boom", id="m1"), request_message("/echo/main", id="m2")),447 )448 assert [(a.id, a.status) for a in socket.answers] == [("m1", 500), ("m2", 200)]449 450 async def test_a_path_no_application_serves_is_a_404(self) -> None:451 socket = await drive(echo_server(), XT_Socket(request_message("/nowhere/x", id="m1")))452 assert socket.answers[0].status == 404453 454 455 class TestTheControlPing:456 async def test_the_ping_is_answered_by_the_server_itself(self) -> None:457 socket = await drive(echo_server(), XT_Socket(request_message("/_wsx/ping", id="m1")))458 assert [(a.id, a.status) for a in socket.answers] == [("m1", 200)]459 460 async def test_it_does_not_reach_any_application(self) -> None:461 # `/_wsx` is a reserved segment: no application is mounted there, and462 # a ping must be answered anyway.463 socket = await drive(echo_server(), XT_Socket(request_message("/_wsx/ping", id="m1")))464 assert socket.answers[0].data == "pong"465 466 467 class TestWhatTheRegistryHolds:468 async def test_the_socket_is_in_the_registry_while_it_lives(self) -> None:469 server = echo_server()470 seen: list[int] = []471 472 class WatchingApp(EchoApp):473 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:474 seen.append(len(server.websockets.snapshot()))475 await super().__call__(scope, receive, send)476 477 server = WsServer(applications=[WatchingApp(mount="echo")])478 await drive(server, XT_Socket(request_message("/echo/main", id="m1")))479 assert seen == [1]480 assert server.websockets.snapshot() == []481 482 async def test_a_handshake_refused_at_the_gate_leaves_nothing_behind(self) -> None:483 server = echo_server()484 await drive(server, XT_Socket(), path="/nowhere/x")485 assert server.websockets.snapshot() == []486 487 488 class TestWhatTheRequestRegistryCounts:489 async def test_a_message_with_an_id_is_a_registered_request(self) -> None:490 server = echo_server()491 counted: list[int] = []492 493 class CountingApp(EchoApp):494 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:495 counted.append(server.requests.in_flight)496 await super().__call__(scope, receive, send)497 498 server = WsServer(applications=[CountingApp(mount="echo")])499 await drive(server, XT_Socket(request_message("/echo/main", id="m1")))500 assert counted == [1]501 assert server.requests.in_flight == 0502 503 async def test_an_event_is_served_and_counted_nowhere(self) -> None:504 server = echo_server()505 counted: list[int] = []506 507 class CountingApp(EchoApp):508 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:509 counted.append(server.requests.in_flight)510 await super().__call__(scope, receive, send)511 512 server = WsServer(applications=[CountingApp(mount="echo")])513 await drive(server, XT_Socket(request_message("/echo/main")))514 assert counted == [0]