tests/spa/orchestration/test_orchestration_worker_connector.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 """WorkerConnector tests: one socket, one child, the handshake and the end.16 17 Everything runs on a real UDS: the wire is the one place that has to behave18 like the kernel behaves, so a fake transport would assert nothing. The child is19 a ``ChildPeer`` over the package's own ``FrameStream`` — it presents itself,20 answers CALLs reusing their id, and dies by closing the socket, which is21 exactly what a worker process does.22 23 The sockets live under a short ``mkdtemp`` root and not under ``tmp_path``:24 the system caps a UDS path at about a hundred characters, which is the very25 reason worker names are short — pytest's own directory is already past it.26 """27 28 from __future__ import annotations29 30 import asyncio31 import os32 from typing import Any33 34 import pytest35 36 from tests.spa.orchestration.frame_helpers import control_frame, read_control37 38 from genro_asgi.channel.frame import REGISTER_METHOD, REGISTER_PATH, Frame, FrameStream39 from genro_asgi_multiworker_spa.orchestration import WorkerConnector40 from genro_asgi_multiworker_spa.orchestration.worker_connector import (41 CALL_METHOD,42 REPLY_METHOD,43 )44 45 from .conftest import wait_for46 47 48 class ChildPeer:49 """The worker process, seen from the other end of the socket."""50 51 def __init__(self, socket_path: str) -> None:52 self.socket_path = socket_path53 self.received: list[Frame] = []54 self.handshake_reply: Frame | None = None55 self.reply_result: Any = None56 self.answer_calls = True57 self.stream: FrameStream | None = None58 self._task: asyncio.Task[None] | None = None59 60 async def present(self, config: Any = None) -> Frame:61 """Connect, send the presentation, return the REPLY that came back."""62 reader, writer = await asyncio.open_unix_connection(self.socket_path)63 self.stream = FrameStream(reader, writer)64 await self.stream.write(65 control_frame(66 method=REGISTER_METHOD,67 path=REGISTER_PATH,68 data={"pid": os.getpid(), "config": config},69 )70 )71 self.handshake_reply = await self.stream.read()72 self._task = asyncio.create_task(self._receive_loop())73 return self.handshake_reply74 75 async def connect_without_presenting(self) -> None:76 reader, writer = await asyncio.open_unix_connection(self.socket_path)77 self.stream = FrameStream(reader, writer)78 79 async def close(self) -> None:80 if self._task is not None:81 self._task.cancel()82 try:83 await self._task84 except asyncio.CancelledError:85 pass86 await self.stream.close()87 88 async def send(self, method: str, path: str, data: Any = None) -> None:89 await self.stream.write(control_frame(method=method, path=path, data=data))90 91 async def wait_frames(self, count: int, timeout: float = 5.0) -> None:92 deadline = asyncio.get_running_loop().time() + timeout93 while len(self.received) < count:94 if asyncio.get_running_loop().time() >= deadline:95 raise TimeoutError(f"the child got {len(self.received)}/{count} frames")96 await asyncio.sleep(0.01)97 98 async def _receive_loop(self) -> None:99 while True:100 frame = await self.stream.read()101 if frame is None:102 return103 self.received.append(frame)104 if frame.method == CALL_METHOD and self.answer_calls:105 await self.stream.write(106 control_frame(107 id=frame.id,108 method=REPLY_METHOD,109 path=frame.path,110 data={"result": self.reply_result, "worker_events": []},111 )112 )113 114 115 class HandlerStub:116 """The WorkerHandler seen by its wire: the envelopes it takes, what it is told.117 118 The wire reads nothing of what arrives: it hands the envelope over whole and119 writes back down whatever comes out. So the handler of these tests is the fold120 — it keeps the envelopes it was given, and answers with a payload of its own,121 whatever the real chain happens to compose today.122 """123 124 def __init__(self, name: str = "standard_0001") -> None:125 self.name = name126 self.descent: dict[str, Any] = {"descending": "what the chain composed"}127 self.envelopes: list[dict[str, Any]] = []128 self.losses = 0129 130 def read_envelope(self, envelope: dict[str, Any]) -> dict[str, Any]:131 """Keep what arrived, and answer with what goes down."""132 self.envelopes.append(envelope)133 return dict(self.descent)134 135 def on_child_lost(self) -> None:136 self.losses += 1137 138 139 @pytest.fixture140 def handler():141 return HandlerStub()142 143 144 @pytest.fixture145 async def connector(short_root, handler):146 wire = WorkerConnector(handler, short_root / "i" / f"{handler.name}.sock")147 await wire.start()148 yield wire149 await wire.stop()150 151 152 async def test_the_socket_is_bound_in_a_private_directory(connector):153 assert connector.socket_path.exists()154 assert os.stat(connector.socket_path.parent).st_mode & 0o777 == 0o700155 assert connector.address == f"uds:{connector.socket_path}"156 assert connector.connected is False157 158 159 async def test_a_stale_socket_is_unlinked_before_the_bind(short_root, handler):160 socket_path = short_root / "i" / f"{handler.name}.sock"161 socket_path.parent.mkdir(mode=0o700, parents=True)162 socket_path.write_bytes(b"what the crash left behind")163 164 wire = WorkerConnector(handler, socket_path)165 await wire.start()166 try:167 child = ChildPeer(str(socket_path))168 await child.present()169 await child.close()170 finally:171 await wire.stop()172 173 174 async def test_the_presentation_is_answered_with_what_the_chain_composed(connector, handler):175 child = ChildPeer(str(connector.socket_path))176 reply = await child.present(config={"pool_size": 4})177 178 assert reply.method == REPLY_METHOD179 assert read_control(reply) == handler.descent180 await connector.wait_connected()181 assert connector.connected is True182 183 await child.close()184 185 186 async def test_a_call_travels_and_its_reply_comes_back(connector):187 child = ChildPeer(str(connector.socket_path))188 await child.present()189 child.reply_result = {"alive": True}190 191 payload = await connector.call("/probe", {"kwargs": {}})192 193 assert payload == {"result": {"alive": True}, "worker_events": []}194 assert child.received[0].method == CALL_METHOD195 assert child.received[0].path == "/probe"196 assert read_control(child.received[0]) == {"kwargs": {}}197 198 await child.close()199 200 201 async def test_timed_out_id_waits_for_its_late_reply_before_reuse(connector):202 child = ChildPeer(str(connector.socket_path))203 await child.present()204 child.answer_calls = False205 request = control_frame(id="reserved", method=CALL_METHOD, path="/probe", data={})206 207 with pytest.raises(TimeoutError):208 await connector.call_frame(request, timeout=0.02)209 with pytest.raises(ValueError, match="duplicate in-flight correlation id"):210 await connector.call_frame(request, timeout=0.02)211 212 await child.stream.write(control_frame(213 id="reserved", method=REPLY_METHOD, path="/probe", data={"result": "late"}214 ))215 await wait_for(lambda: "reserved" not in connector._abandoned)216 child.answer_calls = True217 child.reply_result = "fresh"218 reply = await connector.call_frame(request, timeout=1)219 assert read_control(reply)["result"] == "fresh"220 await child.close()221 222 223 async def test_wrong_route_reply_is_a_link_protocol_violation(connector, handler):224 child = ChildPeer(str(connector.socket_path))225 await child.present()226 child.answer_calls = False227 request = control_frame(id="owned", method=CALL_METHOD, path="/right", data={})228 pending = asyncio.create_task(connector.call_frame(request, timeout=2))229 await child.wait_frames(1)230 folded_before = len(handler.envelopes)231 await child.stream.write(control_frame(232 id="owned", method=REPLY_METHOD, path="/wrong", data={"result": "misrouted"}233 ))234 235 with pytest.raises(ConnectionError, match="wire.*down"):236 await pending237 await wait_for(lambda: not connector.connected)238 assert handler.losses == 1239 assert len(handler.envelopes) == folded_before240 await child.close()241 242 243 async def test_an_envelope_that_is_neither_of_the_two_lanes_is_denounced(244 connector, handler, caplog245 ):246 """A REPLY resolves and a CALL is served: what is left has nowhere to go."""247 child = ChildPeer(str(connector.socket_path))248 await child.present()249 250 with caplog.at_level("WARNING"):251 await child.send("POST", "/lock_taken", {"user": "mario"})252 await wait_for(lambda: "Unexpected envelope" in caplog.text)253 254 assert "Unexpected envelope POST" in caplog.text255 assert handler.losses == 0256 257 await child.close()258 259 260 async def test_a_call_the_handler_has_no_hook_for_comes_back_as_an_error(connector, handler):261 """This handler serves no calls at all — the child is answered all the same."""262 child = ChildPeer(str(connector.socket_path))263 await child.present()264 child.answer_calls = False265 266 await child.send(CALL_METHOD, "/op/ask", {"user": "mario"})267 await child.wait_frames(1)268 269 answer = child.received[0]270 assert answer.method == REPLY_METHOD271 assert "AttributeError" in read_control(answer)["error"]272 assert handler.losses == 0273 274 await child.close()275 276 277 async def test_an_unencodable_child_call_result_comes_back_as_an_error(connector, handler):278 child = ChildPeer(str(connector.socket_path))279 await child.present()280 child.answer_calls = False281 handler.serve_child_call = lambda path, payload: object()282 283 await child.send(CALL_METHOD, "/op/unencodable", {"value": 1})284 await child.wait_frames(1)285 286 answer = read_control(child.received[0])287 assert "not encodable" in answer["error"]288 assert connector.connected is True289 await child.close()290 291 292 async def test_the_death_of_the_child_is_a_local_event(connector, handler):293 child = ChildPeer(str(connector.socket_path))294 await child.present()295 await connector.wait_connected()296 297 await child.close()298 await wait_for(lambda: handler.losses == 1)299 300 assert connector.connected is False301 302 303 async def test_a_call_in_flight_dies_with_the_child(connector):304 child = ChildPeer(str(connector.socket_path))305 await child.present()306 await connector.wait_connected()307 child.answer_calls = False308 309 async def kill_the_child() -> None:310 await asyncio.sleep(0.05)311 await child.close()312 313 asyncio.create_task(kill_the_child())314 with pytest.raises(ConnectionError):315 await connector.call("/freeze_everybody")316 317 318 async def test_a_deliberate_stop_announces_no_death(short_root, handler):319 wire = WorkerConnector(handler, short_root / "i" / f"{handler.name}.sock")320 await wire.start()321 child = ChildPeer(str(wire.socket_path))322 await child.present()323 await wire.wait_connected()324 325 await wire.stop()326 327 assert handler.losses == 0328 assert wire.connected is False329 assert not wire.socket_path.exists()330 331 332 async def test_the_successor_finds_the_same_socket(connector, handler):333 first = ChildPeer(str(connector.socket_path))334 await first.present()335 await connector.wait_connected()336 await first.close()337 await wait_for(lambda: connector.connected is False)338 339 handler.descent = {"descending": "what the chain composes now"}340 successor = ChildPeer(str(connector.socket_path))341 reply = await successor.present()342 await connector.wait_connected()343 344 assert read_control(reply) == {"descending": "what the chain composes now"}345 346 await successor.close()347 348 349 async def test_a_second_child_on_a_taken_wire_is_refused(connector):350 resident = ChildPeer(str(connector.socket_path))351 await resident.present()352 await connector.wait_connected()353 354 intruder = ChildPeer(str(connector.socket_path))355 await intruder.connect_without_presenting()356 assert await intruder.stream.read() is None357 358 resident.reply_result = "still here"359 assert await connector.call("/probe") == {"result": "still here", "worker_events": []}360 361 await resident.close()362 363 364 async def test_a_child_that_does_not_present_itself_is_refused(connector):365 intruder = ChildPeer(str(connector.socket_path))366 await intruder.connect_without_presenting()367 await intruder.stream.write(control_frame(method=CALL_METHOD, path="/whatever"))368 369 assert await intruder.stream.read() is None370 assert connector.connected is False371 372 373 async def test_calling_a_wire_with_no_child_is_an_error(connector):374 with pytest.raises(ConnectionError):375 await connector.call("/probe")