tests/core/test_channel_local.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 """LocalChannel tests: the in-process wire must behave like the socket one.16 17 Same rubric (the member joins through ``attach_local`` with a REGISTER18 frame), same envelopes (a CALL is answered with a REPLY reusing its id, via19 ``send_frame``), same death semantics (hub stop → orphan, deliberate member20 close → channel lost with no orphan) and — the point of the phase — the same21 bytes: a payload mutated after ``send`` cannot reach the peer.22 """23 24 from __future__ import annotations25 26 import asyncio27 import os28 import struct29 from typing import Any30 31 import pytest32 33 from genro_asgi.channel import (34 CALL_METHOD,35 EVENT_METHOD,36 REGISTER_METHOD,37 REPLY_METHOD,38 ChannelHub,39 Frame,40 LocalChannel,41 LocalFrameStream,42 )43 from genro_asgi.channel.control import ControlPayload44 45 CONTROL = ControlPayload()46 47 48 def control_frame(*, data=None, **kwargs):49 return Frame(payload=CONTROL.encode(data), **kwargs)50 51 52 def data_of(frame):53 return CONTROL.decode(frame.payload)54 55 56 class LocalPeer:57 """An in-process member: records frames, answers CALLs with a REPLY."""58 59 def __init__(self, name: str) -> None:60 self.name = name61 self.received: list[Frame] = []62 self.orphaned = 063 self.reply_result: Any = None64 self.reply_events: list[dict[str, Any]] = []65 self.reply_error: Any = None66 self.channel = LocalChannel(name, on_message=self._on_message, on_orphan=self._on_orphan)67 68 async def join(self, hub: ChannelHub) -> None:69 await self.channel.connect()70 await hub.attach_local(self.channel)71 72 async def wait_frames(self, count: int, timeout: float = 5.0) -> None:73 deadline = asyncio.get_running_loop().time() + timeout74 while len(self.received) < count:75 if asyncio.get_running_loop().time() >= deadline:76 raise TimeoutError(f"{self.name} got {len(self.received)}/{count} frames")77 await asyncio.sleep(0.01)78 79 async def _on_message(self, frame: Frame) -> None:80 self.received.append(frame)81 if frame.method == CALL_METHOD:82 data: dict[str, Any] = {"events": list(self.reply_events)}83 if self.reply_error is not None:84 data["error"] = self.reply_error85 else:86 data["result"] = self.reply_result87 await self.channel.send_frame(88 control_frame(id=frame.id, method=REPLY_METHOD, path=frame.path, data=data)89 )90 91 def _on_orphan(self, channel: LocalChannel) -> None:92 self.orphaned += 193 94 95 class LocalHarness:96 """A started hub plus the callback log its tests assert on."""97 98 def __init__(self) -> None:99 self.joined: list[str] = []100 self.lost: list[str] = []101 self.events: list[tuple[str, Frame]] = []102 self.hub = ChannelHub(103 on_member_joined=lambda member: self.joined.append(member.name),104 on_channel_lost=lambda member: self.lost.append(member.name),105 on_event=lambda member, frame: self.events.append((member.name, frame)),106 )107 108 async def wait_lost(self, count: int, timeout: float = 5.0) -> None:109 deadline = asyncio.get_running_loop().time() + timeout110 while len(self.lost) < count:111 if asyncio.get_running_loop().time() >= deadline:112 raise TimeoutError(f"hub saw {len(self.lost)}/{count} losses")113 await asyncio.sleep(0.01)114 115 116 @pytest.fixture117 async def harness():118 harness = LocalHarness()119 await harness.hub.start()120 yield harness121 await harness.hub.stop()122 123 124 async def test_register_handshake_lands_in_the_rubric(harness):125 peer = LocalPeer("W:local-1")126 await peer.join(harness.hub)127 member = harness.hub.resolve("W:local-1")128 assert member is not None129 assert member.pid == os.getpid()130 assert harness.joined == ["W:local-1"]131 assert peer.channel.connected is True132 await peer.channel.close()133 134 135 async def test_event_from_the_hub_reaches_the_member(harness):136 peer = LocalPeer("W:local-1")137 await peer.join(harness.hub)138 frame_id = await harness.hub.post("W:local-1", "/occupancy", {"users": 3})139 await peer.wait_frames(1)140 frame = peer.received[0]141 assert (frame.method, frame.path, frame.id) == (EVENT_METHOD, "/occupancy", frame_id)142 assert data_of(frame) == {"users": 3}143 await peer.channel.close()144 145 146 async def test_event_from_the_member_reaches_the_hub(harness):147 peer = LocalPeer("W:local-1")148 await peer.join(harness.hub)149 await peer.channel.send(method=EVENT_METHOD, path="/op/new_user", data={"seq": 1})150 deadline = asyncio.get_running_loop().time() + 5.0151 while not harness.events:152 assert asyncio.get_running_loop().time() < deadline, "hub saw no event"153 await asyncio.sleep(0.01)154 name, frame = harness.events[0]155 assert (name, frame.path, data_of(frame)) == ("W:local-1", "/op/new_user", {"seq": 1})156 await peer.channel.close()157 158 159 async def test_payload_mutated_after_send_does_not_reach_the_peer(harness):160 peer = LocalPeer("W:local-1")161 await peer.join(harness.hub)162 payload = {"users": 1}163 frame = control_frame(method=EVENT_METHOD, path="/occupancy", data=payload)164 await harness.hub.resolve("W:local-1").write(frame)165 payload["users"] = 999166 await peer.wait_frames(1)167 assert data_of(peer.received[0]) == {"users": 1}168 169 outbound = {"seq": 1}170 await peer.channel.send(method=EVENT_METHOD, path="/op/new_user", data=outbound)171 outbound["seq"] = 999172 deadline = asyncio.get_running_loop().time() + 5.0173 while not harness.events:174 assert asyncio.get_running_loop().time() < deadline, "hub saw no event"175 await asyncio.sleep(0.01)176 assert data_of(harness.events[0][1]) == {"seq": 1}177 await peer.channel.close()178 179 180 async def test_call_reply_delivers_the_payload_verbatim(harness):181 peer = LocalPeer("W:local-1")182 peer.reply_result = {"ok": True}183 peer.reply_events = [{"op": "new_user", "seq": 1}]184 await peer.join(harness.hub)185 186 payload = await asyncio.wait_for(187 harness.hub.call("W:local-1", "/op/new_user", {"identity": "u1"}), timeout=5.0188 )189 190 assert payload == {"result": {"ok": True}, "events": [{"op": "new_user", "seq": 1}]}191 assert peer.received[0].method == CALL_METHOD192 assert data_of(peer.received[0]) == {"identity": "u1"}193 await peer.channel.close()194 195 196 async def test_error_reply_rides_the_payload(harness):197 peer = LocalPeer("W:local-1")198 peer.reply_error = "unsupported until phase B"199 await peer.join(harness.hub)200 payload = await harness.hub.call("W:local-1", "/op/new_user", {"identity": "u1"}, timeout=5.0)201 assert payload == {"error": "unsupported until phase B", "events": []}202 await peer.channel.close()203 204 205 async def test_member_close_is_a_channel_loss_without_orphan(harness):206 peer = LocalPeer("W:local-1")207 await peer.join(harness.hub)208 await peer.channel.close()209 await harness.wait_lost(1)210 assert harness.lost == ["W:local-1"]211 assert peer.orphaned == 0212 assert peer.channel.connected is False213 assert peer.channel.closed is True214 assert harness.hub.resolve("W:local-1") is None215 216 217 async def test_hub_stop_orphans_the_member(harness):218 peer = LocalPeer("W:local-1")219 await peer.join(harness.hub)220 await harness.hub.stop()221 await asyncio.wait_for(peer.channel.wait_closed(), timeout=5.0)222 assert peer.orphaned == 1223 assert harness.lost == []224 225 226 async def test_send_before_connect_is_refused():227 channel = LocalChannel("W:local-1")228 with pytest.raises(ConnectionError):229 await channel.send(path="/op/new_user")230 231 232 async def test_connect_twice_is_refused():233 channel = LocalChannel("W:local-1")234 await channel.connect()235 with pytest.raises(RuntimeError, match="already connected"):236 await channel.connect()237 await channel.close()238 239 240 async def test_decode_error_is_a_protocol_violation():241 inbound: asyncio.Queue[bytes | None] = asyncio.Queue()242 outbound: asyncio.Queue[bytes | None] = asyncio.Queue()243 stream = LocalFrameStream(inbound, outbound)244 await inbound.put(struct.pack("!4sBII", b"NOPE", 1, 2, 0) + b"{}")245 with pytest.raises(ValueError, match="invalid frame magic"):246 await stream.read()247 248 249 async def test_oversized_frame_is_refused_both_ways():250 inbound: asyncio.Queue[bytes | None] = asyncio.Queue()251 outbound: asyncio.Queue[bytes | None] = asyncio.Queue()252 stream = LocalFrameStream(inbound, outbound, max_size=64)253 with pytest.raises(ValueError, match="exceeds max_size"):254 await stream.write(255 control_frame(method=EVENT_METHOD, path="/big", data={"blob": "x" * 200})256 )257 await inbound.put(struct.pack("!4sBII", b"GNRF", 1, 10, 190) + b"x" * 200)258 with pytest.raises(ValueError, match="exceeds max_size"):259 await stream.read()260 261 262 async def test_closing_one_end_ends_both_reads():263 channel = LocalChannel("W:local-1")264 hub_stream = channel.hub_stream265 await channel.connect()266 register = await hub_stream.read()267 assert register.method == REGISTER_METHOD268 await hub_stream.close()269 assert await hub_stream.read() is None270 await asyncio.wait_for(channel.wait_closed(), timeout=5.0)271 with pytest.raises(ConnectionError):272 await channel.send(path="/late")273 274 275 async def test_local_queue_admission_is_bounded_and_close_never_waits():276 inbound, outbound = asyncio.Queue(), asyncio.Queue()277 stream = LocalFrameStream(inbound, outbound)278 for _ in range(16):279 await stream.write(Frame(payload=b"opaque"))280 with pytest.raises(ConnectionError, match="not sent"):281 await stream.write(Frame(payload=b"overflow"))282 assert outbound.qsize() == 16283 await asyncio.wait_for(stream.close(), 1)284 285 286 async def test_local_queue_limit_is_configurable():287 inbound, outbound = asyncio.Queue(), asyncio.Queue()288 stream = LocalFrameStream(inbound, outbound, max_queue_size=1)289 await stream.write(Frame(payload=b"first"))290 with pytest.raises(ConnectionError, match="not sent"):291 await stream.write(Frame(payload=b"second"))