tests/spa/orchestration/test_orchestration_websocket_e2e.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 """A message of a page, from the browser to its worker and back (#68 phase 4).16 17 Contract tests, entered where a real message enters: the socket. A message18 becomes a synthetic request, the front packs it, the vertex places it, the19 worker serves it on the row of the page it belongs to — and the answer comes20 back with the id the browser correlates on.21 22 ``openchannel`` is what a page must send first, and it is the one command both23 halves of the machine touch: the front validates the page against24 ``page_connection_map`` and writes the channel on its row through the worker;25 the CONNECTION binds the page to its socket, and only because the front26 answered 200 (owner, N30).27 28 The pool here is real — a commander, a group, a worker on a UDS — through the29 ``worker_commander_lane`` fixture; the front is a real ``SpaApplication`` with30 that commander in place.31 """32 33 from __future__ import annotations34 35 import asyncio36 from datetime import date37 from decimal import Decimal38 from typing import Any39 40 import pytest41 42 from genro_asgi import BaseServer, MiddlewareMixin43 from genro_asgi_multiworker_spa.spa_app import SPA_CONNECTION_ID_COOKIE, SpaApplication44 from genro_asgi.exceptions import HTTPBadRequest45 from genro_asgi.request import Request46 from genro_asgi.types import Message, Scope47 from genro_asgi.wsx import WsxConnection, WsxEnvelope48 49 PREFIX = "WSX://"50 51 52 async def no_body() -> Message:53 """A receive that hands over an empty body: the request carries none."""54 return {"type": "http.request", "body": b"", "more_body": False}55 CID = "cid-a"56 USER = "mario"57 PAGE = "page-1"58 59 60 class WsServer(MiddlewareMixin, BaseServer):61 """The composition a websocket needs."""62 63 64 class XT_Front(SpaApplication):65 """The real front, with the pool of the fixture already in place."""66 67 def __init__(self, commander: Any, **kwargs: Any) -> None:68 super().__init__(**kwargs)69 self._commander = commander70 self.mount_channel_control()71 72 73 class XT_Browser:74 """A browser that stays connected until the test lets it go."""75 76 def __init__(self) -> None:77 self.sent: list[Message] = []78 self._incoming: asyncio.Queue[Message] = asyncio.Queue()79 self._incoming.put_nowait({"type": "websocket.connect"})80 81 async def receive(self) -> Message:82 return await self._incoming.get()83 84 async def send(self, message: Message) -> None:85 self.sent.append(message)86 87 def says(self, envelope: WsxEnvelope) -> None:88 self._incoming.put_nowait({"type": "websocket.receive", "text": envelope.encode()})89 90 def leave(self) -> None:91 self._incoming.put_nowait({"type": "websocket.disconnect", "code": 1000})92 93 @property94 def answers(self) -> list[WsxEnvelope]:95 return [96 WsxEnvelope(m["text"])97 for m in self.sent98 if m["type"] == "websocket.send" and m.get("text", "").startswith(PREFIX)99 ]100 101 102 class XT_Machine:103 """The whole machine of one story: server, front, pool, and one browser."""104 105 def __init__(self, lane: Any) -> None:106 self.lane = lane107 self.commander = lane.commander108 self.front = XT_Front(lane.commander, code="site0", mount="")109 self.server = WsServer(applications=[self.front])110 self.browser = XT_Browser()111 self._task: asyncio.Task[None] | None = None112 113 async def __aenter__(self) -> XT_Machine:114 scope: Scope = {115 "type": "websocket",116 "path": "/_wsx",117 "headers": [118 (b"host", b"site.example"),119 (b"cookie", f"{SPA_CONNECTION_ID_COOKIE}={CID}".encode()),120 ],121 "subprotocols": [],122 }123 connection = WsxConnection(self.server, scope, self.browser.receive, self.browser.send)124 self._task = asyncio.create_task(connection.serve())125 await self.settle()126 return self127 128 async def __aexit__(self, *failure: Any) -> None:129 self.browser.leave()130 if self._task is not None:131 await self._task132 133 async def settle(self) -> None:134 """Let the connection's task, the lane and the worker move on."""135 for _ in range(20):136 await asyncio.sleep(0)137 138 async def message(self, envelope: WsxEnvelope) -> WsxEnvelope | None:139 """Say one message and hand back the answer, when it asked for one."""140 before = len(self.browser.answers)141 self.browser.says(envelope)142 for _ in range(40):143 await asyncio.sleep(0)144 if len(self.browser.answers) > before:145 return self.browser.answers[-1]146 return None147 148 async def open_channel(self, page_id: str = PAGE, **parameters: Any) -> WsxEnvelope | None:149 return await self.message(150 WsxEnvelope(151 id=f"open-{page_id}",152 method="WSK",153 path="/_wsx/openchannel",154 page_id=page_id,155 data={"parameters": parameters} if parameters else None,156 )157 )158 159 160 @pytest.fixture161 async def machine(worker_commander_lane):162 """The machine with one page already born on the worker, as a site does."""163 lane = worker_commander_lane164 lane.commander.connection_user_map[CID] = USER165 lane.commander.resolve_user(CID)166 await lane.verb("new_connection", CID, user=USER)167 await lane.verb("new_page", USER, PAGE, connection_id=CID)168 async with XT_Machine(lane) as running:169 yield running170 171 172 class TestTheHandshakeOfTheSpa:173 async def test_a_socket_with_no_connection_cookie_is_closed_1008(174 self, worker_commander_lane175 ) -> None:176 # What the first browser found (#70 A): the front declared no cookie,177 # so a socket opened without one stayed open for ever. The gate of178 # W-13 had been built and had no user.179 machine = XT_Machine(worker_commander_lane)180 scope: Scope = {181 "type": "websocket",182 "path": "/_wsx",183 "headers": [(b"host", b"site.example")],184 "subprotocols": [],185 }186 connection = WsxConnection(187 machine.server, scope, machine.browser.receive, machine.browser.send188 )189 # The browser leaves right away: without the gate the read loop would190 # wait for ever, which is exactly what the defect looked like.191 machine.browser.leave()192 await connection.serve()193 assert machine.browser.sent == [194 {"type": "websocket.accept"},195 {196 "type": "websocket.close",197 "code": 1008,198 "reason": "connection cookie required: spa_connection_id",199 },200 ]201 202 async def test_a_socket_with_the_cookie_is_let_in(self, machine) -> None:203 assert {"type": "websocket.accept"} in machine.browser.sent204 205 def test_the_front_names_the_cookie_its_handshake_must_carry(self) -> None:206 assert SpaApplication(code="site0", mount="").handshake_cookie == (207 SPA_CONNECTION_ID_COOKIE208 )209 210 211 class TestOpeningTheChannelOfAPage:212 async def test_the_page_is_told_its_channel_is_open(self, machine) -> None:213 answer = await machine.open_channel()214 assert answer is not None215 assert (answer.id, answer.status) == (f"open-{PAGE}", 200)216 217 async def test_the_worker_wrote_the_channel_on_the_row(self, machine) -> None:218 await machine.open_channel()219 assert machine.lane.worker.page_register.get(PAGE)["wsx"] is True220 221 async def test_the_parameters_the_page_asked_for_are_on_the_row(self, machine) -> None:222 await machine.open_channel(sequential=True)223 assert machine.lane.worker.page_register.get(PAGE)["wsx"] == {"sequential": True}224 225 async def test_the_page_now_speaks_on_this_socket(self, machine) -> None:226 assert machine.server.websockets.get_page_socket(PAGE) is None227 await machine.open_channel()228 assert machine.server.websockets.get_page_socket(PAGE) is not None229 230 async def test_saying_it_twice_changes_nothing(self, machine) -> None:231 await machine.open_channel()232 first = machine.server.websockets.get_page_socket(PAGE)233 answer = await machine.open_channel()234 assert answer.status == 200235 assert machine.server.websockets.get_page_socket(PAGE) is first236 237 238 class TestAPageThatIsNotThisConnections:239 async def test_a_page_of_another_connection_is_refused(self, machine) -> None:240 machine.commander.page_connection_map["page-elsewhere"] = "cid-b"241 answer = await machine.open_channel("page-elsewhere")242 assert answer.status == 403243 244 async def test_a_page_nobody_ever_created_is_refused(self, machine) -> None:245 answer = await machine.open_channel("page-nowhere")246 assert answer.status == 403247 248 async def test_neither_is_ever_bound_to_the_socket(self, machine) -> None:249 machine.commander.page_connection_map["page-elsewhere"] = "cid-b"250 await machine.open_channel("page-elsewhere")251 await machine.open_channel("page-nowhere")252 assert machine.server.websockets.get_page_socket("page-elsewhere") is None253 assert machine.server.websockets.get_page_socket("page-nowhere") is None254 255 async def test_a_message_that_names_no_page_is_refused(self, machine) -> None:256 answer = await machine.message(257 WsxEnvelope(id="m1", method="WSK", path="/_wsx/openchannel")258 )259 assert answer.status == 400260 261 262 class TestWhatTheWorkerRefuses:263 async def test_a_channel_for_a_page_the_worker_never_saw_is_an_error(264 self, machine265 ) -> None:266 # The vertex believes the page is this connection's — the front lets it267 # through — but the worker has no such row: the command fails there,268 # loudly, and nothing is bound.269 machine.commander.page_connection_map["ghost"] = CID270 answer = await machine.open_channel("ghost")271 assert answer.status == 500 and "never born here" in answer.data272 assert machine.server.websockets.get_page_socket("ghost") is None273 274 275 class TestACallerWithNoCookie:276 async def test_the_command_refuses_a_request_that_carries_no_connection(277 self, worker_commander_lane278 ) -> None:279 # No websocket can reach this any more — the handshake gate closes such280 # a socket first (#70 A) — but the route is a route: an HTTP caller can281 # knock on it with no cookie at all, and the answer must never be «some282 # connection».283 front = XT_Front(worker_commander_lane.commander, code="site0", mount="")284 control = front.route.node("/_wsx/openchannel")285 request = Request(286 {287 "type": "http",288 "method": "POST",289 "path": "/_wsx/openchannel",290 "headers": [],291 "genro.page_id": PAGE,292 },293 no_body,294 )295 await request.init()296 with pytest.raises(HTTPBadRequest, match="no cookie"):297 await control(parameters=None, _request=request)298 299 300 class TestAMessageBeforeTheChannelIsOpen:301 async def test_the_browser_is_told_what_is_wrong_and_not_that_the_site_broke(302 self, machine303 ) -> None:304 # What the first browser found (#70 C): the refusal was normalised into305 # the 502 of an upstream that broke, and the reason lived only in the306 # worker's log. It is a refusal of the CLIENT, not a failure of the307 # site, and it carries its own status and its own words.308 answer = await machine.message(309 WsxEnvelope(id="m1", method="WSK", path="/main/rpc", page_id=PAGE)310 )311 assert answer.status == 409312 assert "no open channel" in answer.data313 314 async def test_the_page_is_served_once_its_channel_is_open(self, machine) -> None:315 await machine.open_channel()316 answer = await machine.message(317 WsxEnvelope(id="m2", method="WSK", path="/main/rpc", page_id=PAGE)318 )319 # The lane's worker hosts no application, so what comes back now is the320 # site's own failure — a 502 with the fixed text — and no longer the321 # refusal: the channel is open and the message was passed on.322 assert answer.status == 502323 324 325 class TestTheChannelSurvivesTheDeposit:326 async def test_a_page_woken_from_the_deposit_still_has_its_channel(327 self, machine328 ) -> None:329 # W-5: a freeze does not touch the websocket, and the browser notices330 # nothing. So the row must come back with its channel — otherwise the331 # very next message of a page that is still connected would be refused.332 await machine.open_channel(sequential=True)333 worker = machine.lane.worker334 await worker.freeze_designated_user(USER)335 assert worker.page_register.get(PAGE) is None336 337 # The road a request takes when it finds him frozen: the store comes338 # home, then the connection with the pages hanging under it.339 await worker.adopt_user(USER)340 await worker.adopt_connection(USER, CID)341 342 assert worker.page_register.get(PAGE)["wsx"] == {"sequential": True}343 344 async def test_the_queue_of_a_woken_page_is_a_fresh_one(self, machine) -> None:345 # The lock itself never travels: it is an object, and the page that346 # comes back is served by a loop that never saw the old one.347 await machine.open_channel(sequential=True)348 worker = machine.lane.worker349 before = worker.page_register.get(PAGE)["call_lock"]350 await worker.freeze_designated_user(USER)351 await worker.adopt_user(USER)352 await worker.adopt_connection(USER, CID)353 after = worker.page_register.get(PAGE)["call_lock"]354 assert isinstance(after, asyncio.Lock) and after is not before355 356 357 class TestTheServerSpeaksToTheOpenedPage:358 async def test_the_worker_reaches_the_browser_through_the_vertex(self, machine) -> None:359 # The whole road back: the site's own code calls `send_message` on a360 # pool thread, the CALL climbs the lane, the front finds the socket.361 await machine.open_channel()362 delivered = await machine.lane.verb("send_message", PAGE, "/main/refresh", {"n": 1})363 await machine.settle()364 assert delivered is True365 pushed = [m for m in machine.browser.answers if m.id is None]366 assert [(m.path, m.data, m.page_id) for m in pushed] == [367 ("/main/refresh", {"n": 1}, PAGE)368 ]369 370 async def test_bytes_reach_the_browser_through_the_codec(self, machine) -> None:371 # What the first browser found (#70 B): the lane is JSON, so `data`372 # went into `json.dumps` as it was and a bytes payload died there with373 # a TypeError — the browser saw a 502 and no push. The codec that374 # carries bytes is TYTX, and the worker speaks it before the CALL.375 await machine.open_channel()376 delivered = await machine.lane.verb(377 "send_message", PAGE, "/main/blob", {"blob": b"\x00\x01\xff", "n": 1}378 )379 await machine.settle()380 assert delivered is True381 pushed = [m for m in machine.browser.answers if m.id is None]382 assert pushed[0].data == {"blob": b"\x00\x01\xff", "n": 1}383 384 async def test_what_the_codec_carries_survives_the_whole_road(self, machine) -> None:385 await machine.open_channel()386 sent = {"day": date(2026, 9, 7), "total": Decimal("9.99"), "nothing": None}387 await machine.lane.verb("send_message", PAGE, "/main/typed", sent)388 await machine.settle()389 assert [m for m in machine.browser.answers if m.id is None][0].data == sent390 391 async def test_a_page_that_never_opened_its_channel_is_reachable_by_nobody(392 self, machine393 ) -> None:394 delivered = await machine.lane.verb("send_message", PAGE, "/main/refresh")395 assert delivered is False396 397 async def test_a_page_the_vertex_no_longer_knows_is_reachable_by_nobody(398 self, machine399 ) -> None:400 # The fold dropped the page from the map: the socket may still hold a401 # stale binding, and the branch validates before it writes.402 await machine.open_channel()403 machine.commander.page_connection_map.pop(PAGE)404 delivered = await machine.lane.verb("send_message", PAGE, "/main/refresh")405 assert delivered is False