tests/core/test_channel.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 """Channel tests (SPECIFICATION.md §4): the child connects to a fake hub over16 UDS, REGISTERs, and EOF is the death signal; the CommunicationMixin arms the17 parent side from ``parent=`` and hooks the lifespan cooperatively (D16/D17).18 19 The fake hub is a plain asyncio UDS server built in the test: it decodes20 frames with the package's own ``FrameStream`` (so both ends of the protocol21 are exercised) and records what it receives.22 """23 24 from __future__ import annotations25 26 import asyncio27 import json28 import os29 import shutil30 import socket31 import struct32 import tempfile33 34 import pytest35 36 from genro_asgi import BaseApplication, BaseServer37 from genro_asgi.channel import (38 MAX_FRAME_SIZE,39 REGISTER_METHOD,40 REGISTER_PATH,41 ChannelClient,42 Frame,43 FrameStream,44 )45 from genro_asgi.communication import CommunicationMixin46 from genro_asgi.channel.control import ControlPayload47 from genro_asgi.channel.frame import FrameCodec48 49 CONTROL = ControlPayload()50 51 52 def control_frame(*, data=None, **kwargs):53 return Frame(payload=CONTROL.encode(data), **kwargs)54 55 56 def data_of(frame):57 return CONTROL.decode(frame.payload)58 59 60 class FakeHub:61 """A plain asyncio UDS server standing in for the orchestration hub."""62 63 def __init__(self, path: str) -> None:64 self.path = path65 self.frames: list[Frame] = []66 self.streams: list[FrameStream] = []67 self.eofs: list[FrameStream] = []68 self._server: asyncio.Server | None = None69 70 async def start(self) -> None:71 self._server = await asyncio.start_unix_server(self._serve, path=self.path)72 73 async def stop(self) -> None:74 for stream in self.streams:75 await stream.close()76 self._server.close()77 await self._server.wait_closed()78 79 async def wait_frames(self, count: int, timeout: float = 5.0) -> None:80 deadline = asyncio.get_running_loop().time() + timeout81 while len(self.frames) < count:82 if asyncio.get_running_loop().time() >= deadline:83 raise TimeoutError(f"hub received {len(self.frames)}/{count} frames")84 await asyncio.sleep(0.01)85 86 async def wait_eofs(self, count: int, timeout: float = 5.0) -> None:87 deadline = asyncio.get_running_loop().time() + timeout88 while len(self.eofs) < count:89 if asyncio.get_running_loop().time() >= deadline:90 raise TimeoutError(f"hub saw {len(self.eofs)}/{count} EOFs")91 await asyncio.sleep(0.01)92 93 async def _serve(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:94 stream = FrameStream(reader, writer)95 self.streams.append(stream)96 while True:97 frame = await stream.read()98 if frame is None:99 break100 self.frames.append(frame)101 self.eofs.append(stream)102 103 104 class ChannelServer(CommunicationMixin, BaseServer):105 """The composition under test: communication capability over the base."""106 107 108 class RecordingApp(BaseApplication):109 """Minimal app recording lifecycle hook calls to a shared list.110 111 Constructor kwargs peeled here (cooperative chain): ``events`` — the112 shared list the hooks append to.113 """114 115 def __init__(self, **kwargs: object) -> None:116 self.events: list[str] = kwargs.pop("events")117 super().__init__(**kwargs)118 119 def on_startup(self) -> None:120 self.events.append("on_startup")121 122 def on_shutdown(self) -> None:123 self.events.append("on_shutdown")124 125 126 @pytest.fixture127 def hub_path():128 tmpdir = tempfile.mkdtemp(prefix="gnrchan_")129 yield os.path.join(tmpdir, "hub.sock")130 shutil.rmtree(tmpdir, ignore_errors=True)131 132 133 @pytest.fixture134 async def hub(hub_path):135 fake = FakeHub(hub_path)136 await fake.start()137 yield fake138 await fake.stop()139 140 141 async def stream_pair(max_size: int = MAX_FRAME_SIZE) -> tuple[FrameStream, FrameStream]:142 """Two FrameStreams over a connected socketpair (an in-process wire)."""143 left, right = socket.socketpair()144 reader_l, writer_l = await asyncio.open_connection(sock=left)145 reader_r, writer_r = await asyncio.open_connection(sock=right)146 one = FrameStream(reader_l, writer_l, max_size=max_size)147 two = FrameStream(reader_r, writer_r, max_size=max_size)148 return one, two149 150 151 class TestFrameProtocol:152 def test_frame_id_generated_when_not_given(self) -> None:153 one, two = control_frame(), control_frame()154 assert one.id and two.id and one.id != two.id155 assert control_frame(id="fixed").id == "fixed"156 157 async def test_round_trip(self) -> None:158 one, two = await stream_pair()159 sent = control_frame(method="POST", path="/events/ready", data={"n": 1})160 await one.write(sent)161 received = await two.read()162 assert received is not None163 assert received.id == sent.id164 assert received.method == "POST"165 assert received.path == "/events/ready"166 assert data_of(received) == {"n": 1}167 await one.close()168 await two.close()169 170 async def test_eof_reads_none(self) -> None:171 one, two = await stream_pair()172 await one.close()173 assert await two.read() is None174 await two.close()175 176 async def test_oversized_write_raises(self) -> None:177 one, two = await stream_pair(max_size=32)178 with pytest.raises(ValueError, match="exceeds max_size"):179 await one.write(control_frame(data={"blob": "x" * 100}))180 await one.close()181 await two.close()182 183 async def test_envelope_missing_method_raises(self) -> None:184 one, two = await stream_pair()185 info = json.dumps({"id": "x", "path": "/foo"}).encode("utf-8")186 one.writer.write(struct.pack("!4sBII", b"GNRF", 1, len(info), 0) + info)187 await one.writer.drain()188 with pytest.raises(ValueError, match="missing 'method'"):189 await two.read()190 await one.close()191 await two.close()192 193 async def test_envelope_non_dict_payload_raises(self) -> None:194 one, two = await stream_pair()195 info = json.dumps(["a", "b"]).encode("utf-8")196 one.writer.write(struct.pack("!4sBII", b"GNRF", 1, len(info), 0) + info)197 await one.writer.drain()198 with pytest.raises(ValueError, match="must be a JSON object"):199 await two.read()200 await one.close()201 await two.close()202 203 def test_payload_is_opaque_and_info_is_separate(self) -> None:204 frame = Frame(info={"format": "application/x-test"}, payload=b"\x00\xffnot-json")205 received = FrameCodec().get_frame(frame.encode())206 assert received.info == {"format": "application/x-test"}207 assert received.payload == b"\x00\xffnot-json"208 209 def test_reserved_info_keys_are_rejected(self) -> None:210 with pytest.raises(ValueError, match="reserved keys"):211 Frame(info={"method": "EVENT"})212 213 def test_explicit_empty_id_and_unknown_method_are_rejected(self) -> None:214 with pytest.raises(ValueError, match="frame id"):215 Frame(id="")216 with pytest.raises(ValueError, match="unsupported frame method"):217 Frame(method="UNKNOWN")218 219 def test_nested_info_cannot_be_mutated_after_validation(self) -> None:220 source = {"route": {"parts": ["one"]}}221 frame = Frame(info=source)222 source["route"]["parts"].append(float("nan"))223 exposed = frame.info224 exposed["route"]["parts"].append("two")225 assert FrameCodec().get_frame(frame.encode()).info == {"route": {"parts": ["one"]}}226 227 def test_duplicate_info_keys_are_rejected(self) -> None:228 info = b'{"id":"x","method":"POST","path":"/","path":"/again"}'229 wire = struct.pack("!4sBII", b"GNRF", 1, len(info), 0) + info230 with pytest.raises(ValueError, match="duplicate JSON key"):231 FrameCodec().get_frame(wire)232 233 def test_unknown_version_is_rejected(self) -> None:234 with pytest.raises(ValueError, match="unsupported frame version"):235 FrameCodec().get_header_lengths(struct.pack("!4sBII", b"GNRF", 2, 0, 0))236 237 def test_control_payload_is_explicit_and_strict(self) -> None:238 codec = ControlPayload()239 assert codec.decode(codec.encode({"value": None})) == {"value": None}240 with pytest.raises(ValueError, match="non-finite"):241 codec.decode(b'{"value":NaN}')242 243 async def test_concurrent_writes_remain_complete_frames(self) -> None:244 one, two = await stream_pair()245 frames = [Frame(path=f"/{index}", payload=bytes([index])) for index in range(100)]246 writes = asyncio.gather(*(one.write(frame) for frame in frames))247 received = [await two.read() for _ in frames]248 await writes249 assert {(frame.path, frame.payload) for frame in received} == {250 (frame.path, frame.payload) for frame in frames251 }252 await one.close()253 await two.close()254 255 256 class TestChannelClient:257 async def test_connect_presents_register_frame(self, hub, hub_path) -> None:258 client = ChannelClient(f"uds:{hub_path}", "child_01")259 await client.connect()260 assert client.connected is True261 await hub.wait_frames(1)262 register = hub.frames[0]263 assert register.method == REGISTER_METHOD264 assert register.path == REGISTER_PATH265 assert data_of(register) == {"name": "child_01", "pid": os.getpid()}266 await client.close()267 assert client.connected is False268 assert client.closed is True269 270 async def test_connect_twice_is_refused(self, hub, hub_path) -> None:271 client = ChannelClient(f"uds:{hub_path}", "child_01")272 await client.connect()273 with pytest.raises(RuntimeError, match="already connected"):274 await client.connect()275 await client.close()276 277 async def test_reconnect_after_link_loss_keeps_the_new_generation(self, hub, hub_path) -> None:278 client = ChannelClient(f"uds:{hub_path}", "child_01")279 await client.connect()280 await hub.wait_frames(1)281 await hub.streams[0].close()282 await asyncio.wait_for(client.wait_closed(), timeout=5)283 284 await client.connect()285 await hub.wait_frames(2)286 await asyncio.sleep(0)287 assert client.connected is True288 await client.send(path="/new-generation", data={"ok": True})289 await hub.wait_frames(3)290 assert hub.frames[-1].path == "/new-generation"291 await client.close()292 293 async def test_send_relays_frames_to_the_hub(self, hub, hub_path) -> None:294 client = ChannelClient(f"uds:{hub_path}", "child_01")295 await client.connect()296 frame_id = await client.send(path="/events/ready", data={"n": 1})297 await hub.wait_frames(2)298 event = hub.frames[1]299 assert event.id == frame_id300 assert event.method == "POST"301 assert event.path == "/events/ready"302 assert data_of(event) == {"n": 1}303 await client.close()304 305 async def test_send_before_connect_raises(self, hub_path) -> None:306 client = ChannelClient(f"uds:{hub_path}", "child_01")307 with pytest.raises(ConnectionError, match="not connected"):308 await client.send(path="/events/ready")309 310 async def test_hub_eof_orphans_the_client(self, hub, hub_path) -> None:311 orphaned: list[ChannelClient] = []312 client = ChannelClient(f"uds:{hub_path}", "child_01", on_orphan=orphaned.append)313 await client.connect()314 await hub.wait_frames(1)315 await hub.stop() # the hub side goes away: EOF is the death signal316 await asyncio.wait_for(client.wait_closed(), timeout=5)317 assert client.connected is False318 assert client.closed is True319 assert orphaned == [client]320 await client.close() # a deliberate close after orphan stays safe321 322 async def test_protocol_violation_is_a_clean_death(self, hub, hub_path, caplog) -> None:323 orphaned: list[ChannelClient] = []324 client = ChannelClient(f"uds:{hub_path}", "child_01", on_orphan=orphaned.append)325 await client.connect()326 await hub.wait_frames(1)327 bogus = b"BOGUS: not a wsx envelope" # valid length prefix, invalid payload328 hub.streams[0].writer.write(len(bogus).to_bytes(4, "big") + bogus)329 await hub.streams[0].writer.drain()330 await asyncio.wait_for(client.wait_closed(), timeout=5)331 assert client.connected is False332 assert client.closed is True333 assert orphaned == [client]334 # the ValueError was caught and logged: the loop task ends clean,335 # nothing stays unretrieved336 assert any("Protocol violation" in record.getMessage() for record in caplog.records)337 # the client closed its writer on the way out: the hub reads EOF338 await hub.wait_eofs(1)339 340 async def test_malformed_envelope_is_a_clean_death(self, hub, hub_path, caplog) -> None:341 orphaned: list[ChannelClient] = []342 client = ChannelClient(f"uds:{hub_path}", "child_01", on_orphan=orphaned.append)343 await client.connect()344 await hub.wait_frames(1)345 payload = b"WSX://" + json.dumps({"id": "x", "path": "/foo"}).encode("utf-8")346 hub.streams[0].writer.write(len(payload).to_bytes(4, "big") + payload)347 await hub.streams[0].writer.drain()348 await asyncio.wait_for(client.wait_closed(), timeout=5)349 assert client.connected is False350 assert client.closed is True351 assert orphaned == [client]352 # the ValueError was caught and logged: the loop task ends clean,353 # nothing stays unretrieved354 assert any("Protocol violation" in record.getMessage() for record in caplog.records)355 # the client closed its writer on the way out: the hub reads EOF356 await hub.wait_eofs(1)357 358 async def test_deliberate_close_fires_no_orphan(self, hub, hub_path) -> None:359 orphaned: list[ChannelClient] = []360 client = ChannelClient(f"uds:{hub_path}", "child_01", on_orphan=orphaned.append)361 await client.connect()362 await client.close()363 assert orphaned == []364 assert client.closed is True365 366 async def test_connect_retries_until_the_hub_binds(self, hub_path) -> None:367 client = ChannelClient(f"uds:{hub_path}", "late_child")368 task = asyncio.create_task(client.connect())369 await asyncio.sleep(0.15) # a few retry rounds before the hub exists370 late = FakeHub(hub_path)371 await late.start()372 await asyncio.wait_for(task, timeout=5)373 assert client.connected is True374 await late.wait_frames(1)375 assert late.frames[0].method == REGISTER_METHOD376 await client.close()377 await late.stop()378 379 async def test_connect_timeout_raises_connection_error(self, hub_path) -> None:380 client = ChannelClient(f"uds:{hub_path}", "child_01", connect_timeout=0.2)381 with pytest.raises(ConnectionError, match="not reachable"):382 await client.connect()383 384 def test_invalid_addresses_raise(self) -> None:385 with pytest.raises(ValueError, match="invalid channel address"):386 ChannelClient("bogus:/x", "child_01")387 with pytest.raises(ValueError, match="invalid channel address"):388 ChannelClient("uds:", "child_01")389 with pytest.raises(ValueError, match="invalid tcp address"):390 ChannelClient("tcp:127.0.0.1", "child_01")391 392 393 class TestCommunicationMixin:394 def test_plain_base_server_lacks_the_attributes(self) -> None:395 server = BaseServer(applications=[BaseApplication(mount="")])396 assert hasattr(server, "parent_channel") is False397 assert hasattr(server, "children_channel") is False398 399 def test_unarmed_parent_channel_raises(self) -> None:400 server = ChannelServer(applications=[BaseApplication(mount="")])401 assert server.parent_armed is False402 with pytest.raises(RuntimeError, match="not armed"):403 server.parent_channel404 405 def test_children_channel_is_unarmed_in_the_minimal_package(self, hub_path) -> None:406 server = ChannelServer(applications=[BaseApplication(mount="")], parent=f"uds:{hub_path}")407 with pytest.raises(RuntimeError, match="not armed"):408 server.children_channel409 410 def test_armed_parent_channel_is_a_channel_client(self, hub_path) -> None:411 server = ChannelServer(applications=[BaseApplication(mount="")], parent=f"uds:{hub_path}")412 assert server.parent_armed is True413 assert isinstance(server.parent_channel, ChannelClient)414 assert server.parent_channel.address == f"uds:{hub_path}"415 416 def test_cooperative_chain_names_leftover_kwargs(self, hub_path) -> None:417 with pytest.raises(TypeError, match="bogus"):418 ChannelServer(419 applications=[BaseApplication(mount="")], parent=f"uds:{hub_path}", bogus=1420 )421 422 async def test_armed_parent_connects_at_startup_disconnects_at_shutdown(423 self, hub, hub_path424 ) -> None:425 events: list[str] = []426 server = ChannelServer(427 applications=[RecordingApp(mount="", events=events)], parent=f"uds:{hub_path}"428 )429 gate = asyncio.Event()430 queue = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]431 sent: list[dict[str, object]] = []432 433 async def receive() -> dict[str, object]:434 message = queue.pop(0)435 if message["type"] == "lifespan.shutdown":436 await gate.wait() # hold the running server between startup and shutdown437 return message438 439 async def send(message: dict[str, object]) -> None:440 sent.append(message)441 442 task = asyncio.create_task(server({"type": "lifespan"}, receive, send))443 await hub.wait_frames(1) # REGISTER reached the hub while the server runs444 register = hub.frames[0]445 assert register.method == REGISTER_METHOD446 assert data_of(register)["name"] == server.parent_channel.name447 assert data_of(register)["pid"] == os.getpid()448 assert server.parent_channel.connected is True449 gate.set()450 await asyncio.wait_for(task, timeout=5)451 assert server.parent_channel.connected is False452 assert server.parent_channel.closed is True453 assert {"type": "lifespan.startup.complete"} in sent454 assert {"type": "lifespan.shutdown.complete"} in sent455 assert events == ["on_startup", "on_shutdown"] # app hooks ran normally456 457 async def test_unreachable_hub_fails_startup_and_no_hook_runs(self, hub_path) -> None:458 # hub_path exists but nothing is bound there: connect retries then fails459 events: list[str] = []460 server = ChannelServer(461 applications=[RecordingApp(mount="", events=events)], parent=f"uds:{hub_path}"462 )463 server.parent_channel.connect_timeout = 0.2464 queue = [{"type": "lifespan.startup"}]465 sent: list[dict[str, object]] = []466 467 async def receive() -> dict[str, object]:468 return queue.pop(0)469 470 async def send(message: dict[str, object]) -> None:471 sent.append(message)472 473 with pytest.raises(ConnectionError, match="not reachable"):474 await server({"type": "lifespan"}, receive, send)475 476 assert len(sent) == 1477 assert sent[0]["type"] == "lifespan.startup.failed"478 assert "not reachable" in str(sent[0]["message"])479 assert events == [] # the child died before any app hook ran480 481 async def test_unarmed_composition_passes_lifespan_straight_through(self) -> None:482 server = ChannelServer(applications=[BaseApplication(mount="")])483 queue = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]484 sent: list[dict[str, object]] = []485 486 async def receive() -> dict[str, object]:487 return queue.pop(0)488 489 async def send(message: dict[str, object]) -> None:490 sent.append(message)491 492 await server({"type": "lifespan"}, receive, send)493 assert {"type": "lifespan.startup.complete"} in sent494 assert {"type": "lifespan.shutdown.complete"} in sent