tests/core/test_websocket_raw_seam.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 """The admitted mode: an application that wants the socket itself (#68 phase 5).16 17 Contract tests. An application that defines ``serve_websocket`` is handed the18 raw scope, receive and send: nothing of the WSX motor runs for it — no accept,19 no Origin gate, no registry, no refusal on the server's state. An application20 that takes the socket takes all of it.21 22 It is the seam a hosted framework with a websocket protocol of its own reaches23 the server by; the core builds nothing beyond the seam (decisions.md §11).24 """25 26 from __future__ import annotations27 28 from typing import Any29 30 from genro_asgi import BaseApplication, BaseServer31 from genro_asgi.lifespan import QUITTING32 from genro_asgi.types import Message, Receive, Scope, Send33 34 35 class EchoSocketApp(BaseApplication):36 """An application that speaks its own protocol on the socket."""37 38 def __init__(self, **kwargs: Any) -> None:39 super().__init__(**kwargs)40 self.seen: list[Scope] = []41 42 async def serve_websocket(self, scope: Scope, receive: Receive, send: Send) -> None:43 self.seen.append(scope)44 await receive()45 await send({"type": "websocket.accept", "subprotocol": "echo"})46 while True:47 message = await receive()48 if message["type"] == "websocket.disconnect":49 return50 await send({"type": "websocket.send", "text": f"echo:{message['text']}"})51 52 53 class PlainApp(BaseApplication):54 """An application with no socket of its own: the motor serves it."""55 56 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:57 await send({"type": "http.response.start", "status": 200, "headers": []})58 await send({"type": "http.response.body", "body": b""})59 60 61 async def drive(server: BaseServer, path: str, *texts: str) -> list[Message]:62 """Live one handshake at the ASGI level and hand back what was written."""63 incoming: list[Message] = [{"type": "websocket.connect"}]64 incoming += [{"type": "websocket.receive", "text": text} for text in texts]65 incoming.append({"type": "websocket.disconnect", "code": 1000})66 sent: list[Message] = []67 68 async def receive() -> Message:69 return incoming.pop(0) if incoming else {"type": "websocket.disconnect", "code": 1006}70 71 async def send(message: Message) -> None:72 sent.append(message)73 74 await server({"type": "websocket", "path": path, "headers": []}, receive, send)75 return sent76 77 78 class TestAnApplicationThatTakesTheSocket:79 async def test_it_accepts_and_answers_in_its_own_protocol(self) -> None:80 server = BaseServer(applications=[EchoSocketApp(mount="raw")])81 sent = await drive(server, "/raw/live", "one", "two")82 assert sent == [83 {"type": "websocket.accept", "subprotocol": "echo"},84 {"type": "websocket.send", "text": "echo:one"},85 {"type": "websocket.send", "text": "echo:two"},86 ]87 88 async def test_the_path_arrives_without_the_mount(self) -> None:89 # The same demux an HTTP request goes through: the segment that named90 # the application is off, and the rest is the application's own.91 app = EchoSocketApp(mount="raw")92 await drive(BaseServer(applications=[app]), "/raw/live")93 assert app.seen[0]["path"] == "/live"94 95 async def test_nothing_of_the_motor_runs(self) -> None:96 # No accept of ours before its own, and the socket is in no registry:97 # the core does not half-serve a connection it does not hold.98 server = BaseServer(applications=[EchoSocketApp(mount="raw")])99 sent = await drive(server, "/raw/live")100 assert sent[0]["subprotocol"] == "echo"101 assert server.websockets.snapshot() == []102 103 async def test_a_server_that_is_not_running_never_reaches_it(self) -> None:104 # The state is judged above the demux, for every websocket alike105 # (owner, 2026-09-07): the machine refuses before the application is106 # even named, and the handshake is turned away with no accept — 1013107 # exists only after one, and here the accept would be the108 # application's.109 app = EchoSocketApp(mount="raw")110 server = BaseServer(applications=[app])111 server.state = QUITTING112 sent = await drive(server, "/raw/live", "one")113 assert sent == [{"type": "websocket.close", "code": 1013, "reason": "server restarting"}]114 assert app.seen == []115 116 117 class TestAnApplicationThatDoesNot:118 async def test_the_motor_serves_it_as_ever(self) -> None:119 server = BaseServer(applications=[PlainApp(mount="")])120 sent = await drive(server, "/")121 assert sent == [{"type": "websocket.accept"}]122 123 async def test_a_server_that_is_not_running_refuses_it_the_same_way(self) -> None:124 # One refusal for both modes: the machine's state is judged before125 # anybody knows which application would have served the socket.126 server = BaseServer(applications=[PlainApp(mount="")])127 server.state = QUITTING128 sent = await drive(server, "/")129 assert sent == [{"type": "websocket.close", "code": 1013, "reason": "server restarting"}]130 131 async def test_the_base_application_defines_no_such_seam(self) -> None:132 # Absent by default, not None: what is not there cannot be called by133 # mistake, and a subclass declares the mode by defining it.134 assert not hasattr(BaseApplication(mount=""), "serve_websocket")