tests/core/test_channel_hub.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 """ChannelHub tests: the rubric, the CALL/REPLY/EVENT envelopes, EOF and isolation.16 17 The member side is a ``MemberPeer`` over the package's own ``FrameStream``18 (both ends of the codec are exercised): it REGISTERs, records what it19 receives and answers CALLs with a REPLY **reusing the CALL id** — the20 correlation the hub keys its futures on, which ``ChannelClient.send`` cannot21 express since it mints a fresh id per frame. The protocol-violation and22 no-REGISTER cases write raw bytes on a plain connection.23 """24 25 from __future__ import annotations26 27 import asyncio28 import logging29 import os30 import shutil31 import tempfile32 from typing import Any33 34 import pytest35 36 from genro_asgi.channel import (37 CALL_METHOD,38 EVENT_METHOD,39 REGISTER_METHOD,40 REGISTER_PATH,41 REPLY_METHOD,42 ChannelHub,43 Frame,44 FrameStream,45 )46 from genro_asgi.channel.control import ControlPayload47 48 CONTROL = ControlPayload()49 50 51 def control_frame(*, data=None, **kwargs):52 return Frame(payload=CONTROL.encode(data), **kwargs)53 54 55 def data_of(frame):56 return CONTROL.decode(frame.payload)57 58 59 class MemberPeer:60 """A child on the channel: REPLYs to CALLs reusing their id."""61 62 def __init__(self, address: str, name: str) -> None:63 self.address = address64 self.name = name65 self.received: list[Frame] = []66 self.reply_result: Any = None67 self.reply_events: list[dict[str, Any]] = []68 self.reply_error: Any = None69 self.answer_calls = True70 self.stream: FrameStream | None = None71 self._task: asyncio.Task[None] | None = None72 73 async def connect(self) -> None:74 transport, _, rest = self.address.partition(":")75 if transport == "uds":76 reader, writer = await asyncio.open_unix_connection(rest)77 else:78 host, _, port = rest.rpartition(":")79 reader, writer = await asyncio.open_connection(host, int(port))80 self.stream = FrameStream(reader, writer)81 await self.stream.write(82 control_frame(83 method=REGISTER_METHOD,84 path=REGISTER_PATH,85 data={"name": self.name, "pid": os.getpid()},86 )87 )88 self._task = asyncio.create_task(self._receive_loop())89 90 async def close(self) -> None:91 if self._task is not None:92 self._task.cancel()93 try:94 await self._task95 except asyncio.CancelledError:96 pass97 await self.stream.close()98 99 async def send(100 self, method: str, path: str, data: Any = None, *, id: str | None = None101 ) -> None:102 kwargs = {"id": id} if id is not None else {}103 await self.stream.write(control_frame(method=method, path=path, data=data, **kwargs))104 105 async def wait_frames(self, count: int, timeout: float = 5.0) -> None:106 deadline = asyncio.get_running_loop().time() + timeout107 while len(self.received) < count:108 if asyncio.get_running_loop().time() >= deadline:109 raise TimeoutError(f"{self.name} got {len(self.received)}/{count} frames")110 await asyncio.sleep(0.01)111 112 async def _receive_loop(self) -> None:113 while True:114 frame = await self.stream.read()115 if frame is None:116 return117 self.received.append(frame)118 if frame.method == CALL_METHOD and self.answer_calls:119 await self._answer(frame)120 121 async def _answer(self, call: Frame) -> None:122 data: dict[str, Any] = {"events": list(self.reply_events)}123 if self.reply_error is not None:124 data["error"] = self.reply_error125 else:126 data["result"] = self.reply_result127 await self.stream.write(128 control_frame(id=call.id, method=REPLY_METHOD, path=call.path, data=data)129 )130 131 132 class HubHarness:133 """A started hub plus the callback log its tests assert on."""134 135 def __init__(self, **kwargs: Any) -> None:136 self.joined: list[str] = []137 self.lost: list[str] = []138 self.events: list[tuple[str, Frame]] = []139 self.hub = ChannelHub(140 on_member_joined=lambda member: self.joined.append(member.name),141 on_channel_lost=lambda member: self.lost.append(member.name),142 on_event=lambda member, frame: self.events.append((member.name, frame)),143 **kwargs,144 )145 146 async def wait_members(self, count: int, timeout: float = 5.0) -> None:147 deadline = asyncio.get_running_loop().time() + timeout148 while len(self.hub.members) < count:149 if asyncio.get_running_loop().time() >= deadline:150 raise TimeoutError(f"hub has {len(self.hub.members)}/{count} members")151 await asyncio.sleep(0.01)152 153 async def wait_lost(self, count: int, timeout: float = 5.0) -> None:154 deadline = asyncio.get_running_loop().time() + timeout155 while len(self.lost) < count:156 if asyncio.get_running_loop().time() >= deadline:157 raise TimeoutError(f"hub saw {len(self.lost)}/{count} losses")158 await asyncio.sleep(0.01)159 160 161 @pytest.fixture162 def socket_dir():163 path = tempfile.mkdtemp(prefix="gnrhubtest_")164 yield path165 shutil.rmtree(path, ignore_errors=True)166 167 168 @pytest.fixture169 async def uds_harness(socket_dir):170 harness = HubHarness(path=os.path.join(socket_dir, "hub.sock"))171 await harness.hub.start()172 yield harness173 await harness.hub.stop()174 175 176 async def test_register_lands_in_the_rubric(uds_harness):177 peer = MemberPeer(uds_harness.hub.address, "W:one")178 await peer.connect()179 await uds_harness.wait_members(1)180 member = uds_harness.hub.resolve("W:one")181 assert member is not None182 assert member.name == "W:one"183 assert member.pid == os.getpid()184 assert uds_harness.joined == ["W:one"]185 assert uds_harness.hub.resolve("W:missing") is None186 await peer.close()187 188 189 async def test_register_over_tcp():190 harness = HubHarness(host="127.0.0.1", port=0)191 await harness.hub.start()192 assert harness.hub.address.startswith("tcp:127.0.0.1:")193 peer = MemberPeer(harness.hub.address, "W:tcp")194 await peer.connect()195 await harness.wait_members(1)196 assert harness.hub.resolve("W:tcp") is not None197 await peer.close()198 await harness.hub.stop()199 200 201 async def test_owned_socket_directory_is_private_and_removed():202 harness = HubHarness()203 await harness.hub.start()204 path = str(harness.hub.path)205 owned_dir = os.path.dirname(path)206 assert os.stat(owned_dir).st_mode & 0o777 == 0o700207 await harness.hub.stop()208 assert not os.path.exists(path)209 assert not os.path.exists(owned_dir)210 211 212 async def test_call_returns_the_reply_payload_verbatim(uds_harness):213 peer = MemberPeer(uds_harness.hub.address, "W:one")214 peer.reply_result = {"ok": 1}215 peer.reply_events = [{"op": "new_user", "seq": 1}, {"op": "drop_user", "seq": 2}]216 await peer.connect()217 await uds_harness.wait_members(1)218 219 payload = await uds_harness.hub.call("W:one", "/op/new_user", {"identity": "u1"}, timeout=5.0)220 221 assert payload == {"result": {"ok": 1}, "events": peer.reply_events}222 assert peer.received[0].method == CALL_METHOD223 assert peer.received[0].path == "/op/new_user"224 assert data_of(peer.received[0]) == {"identity": "u1"}225 await peer.close()226 227 228 async def test_error_reply_is_delivered_not_raised(uds_harness):229 peer = MemberPeer(uds_harness.hub.address, "W:one")230 peer.reply_error = "unsupported until phase B"231 await peer.connect()232 await uds_harness.wait_members(1)233 234 payload = await uds_harness.hub.call("W:one", "/op/http", {"http": {}}, timeout=5.0)235 236 assert payload == {"error": "unsupported until phase B", "events": []}237 await peer.close()238 239 240 async def test_reply_without_a_parked_caller_is_dropped(uds_harness):241 peer = MemberPeer(uds_harness.hub.address, "W:one")242 peer.answer_calls = False243 await peer.connect()244 await uds_harness.wait_members(1)245 246 with pytest.raises(TimeoutError):247 await uds_harness.hub.call("W:one", "/op/slow", None, timeout=0.1)248 await peer._answer(peer.received[0])249 await asyncio.sleep(0.05)250 251 assert uds_harness.hub._pending == {}252 assert uds_harness.hub.resolve("W:one") is not None253 await peer.close()254 255 256 async def test_call_timeout_unparks_the_future(uds_harness):257 peer = MemberPeer(uds_harness.hub.address, "W:one")258 peer.answer_calls = False259 await peer.connect()260 await uds_harness.wait_members(1)261 262 with pytest.raises(TimeoutError):263 await uds_harness.hub.call("W:one", "/op/silent", None, timeout=0.1)264 assert uds_harness.hub._pending == {}265 await peer.close()266 267 268 async def test_a_call_without_timeout_waits_for_its_reply(uds_harness):269 peer = MemberPeer(uds_harness.hub.address, "W:one")270 peer.answer_calls = False271 peer.reply_result = "late"272 await peer.connect()273 await uds_harness.wait_members(1)274 275 parked = asyncio.create_task(uds_harness.hub.call("W:one", "/op/slow", None))276 await peer.wait_frames(1)277 await asyncio.sleep(0.2)278 assert not parked.done()279 280 await peer._answer(peer.received[0])281 assert (await parked)["result"] == "late"282 assert uds_harness.hub._pending == {}283 await peer.close()284 285 286 async def test_member_death_fails_its_parked_calls(uds_harness):287 peer = MemberPeer(uds_harness.hub.address, "W:one")288 other = MemberPeer(uds_harness.hub.address, "W:two")289 peer.answer_calls = False290 other.answer_calls = False291 await peer.connect()292 await other.connect()293 await uds_harness.wait_members(2)294 295 parked = asyncio.create_task(uds_harness.hub.call("W:one", "/op/slow", None))296 survivor = asyncio.create_task(uds_harness.hub.call("W:two", "/op/slow", None))297 await peer.wait_frames(1)298 await other.wait_frames(1)299 300 await peer.close()301 await uds_harness.wait_lost(1)302 303 with pytest.raises(ConnectionError, match="channel to W:one lost"):304 await parked305 assert not survivor.done()306 307 survivor.cancel()308 await other.close()309 310 311 async def test_stop_fails_every_parked_call(socket_dir):312 harness = HubHarness(path=os.path.join(socket_dir, "hub.sock"))313 await harness.hub.start()314 peer = MemberPeer(harness.hub.address, "W:one")315 peer.answer_calls = False316 await peer.connect()317 await harness.wait_members(1)318 319 parked = asyncio.create_task(harness.hub.call("W:one", "/op/slow", None))320 await peer.wait_frames(1)321 322 await harness.hub.stop()323 324 with pytest.raises(ConnectionError):325 await parked326 assert harness.lost == []327 await peer.close()328 329 330 async def test_call_on_unknown_member_raises_lookup(uds_harness):331 with pytest.raises(LookupError):332 await uds_harness.hub.call("W:ghost", "/op/new_user", None, timeout=0.5)333 334 335 async def test_call_frame_requires_call_and_rejects_duplicate_pending_id(uds_harness):336 peer = MemberPeer(uds_harness.hub.address, "W:one")337 await peer.connect()338 await uds_harness.wait_members(1)339 peer.answer_calls = False340 with pytest.raises(ValueError, match="requires a CALL"):341 await uds_harness.hub.call_frame(342 "W:one", Frame(id="event", method=EVENT_METHOD, path="/event")343 )344 first = asyncio.create_task(345 uds_harness.hub.call_frame("W:one", Frame(id="same", method=CALL_METHOD, path="/one"))346 )347 await peer.wait_frames(1)348 with pytest.raises(RuntimeError, match="already pending"):349 await uds_harness.hub.call_frame("W:one", Frame(id="same", method=CALL_METHOD, path="/two"))350 first.cancel()351 with pytest.raises(asyncio.CancelledError):352 await first353 354 355 async def test_reply_from_another_member_cannot_complete_a_call(uds_harness):356 one = MemberPeer(uds_harness.hub.address, "W:one")357 two = MemberPeer(uds_harness.hub.address, "W:two")358 await one.connect()359 await two.connect()360 await uds_harness.wait_members(2)361 one.answer_calls = False362 pending = asyncio.create_task(363 uds_harness.hub.call_frame(364 "W:one", Frame(id="owned", method=CALL_METHOD, path="/ask"), timeout=1365 )366 )367 await one.wait_frames(1)368 await two.send(REPLY_METHOD, "/ask", {"result": "spoof"}, id="owned")369 await asyncio.wait_for(two._task, timeout=1)370 assert not pending.done()371 assert uds_harness.hub.resolve("W:two") is None372 await one.send(REPLY_METHOD, "/ask", {"result": "real"}, id="owned")373 assert data_of(await pending) == {"result": "real"}374 375 376 async def test_late_reply_cannot_resolve_reused_id_or_wrong_path(uds_harness):377 peer = MemberPeer(uds_harness.hub.address, "W:one")378 peer.answer_calls = False379 await peer.connect()380 await uds_harness.wait_members(1)381 expired = Frame(id="late", method=CALL_METHOD, path="/old")382 with pytest.raises(TimeoutError):383 await uds_harness.hub.call_frame("W:one", expired, timeout=0.01)384 with pytest.raises(RuntimeError, match="awaiting a late reply"):385 await uds_harness.hub.call_frame("W:one", Frame(id="late", method=CALL_METHOD, path="/new"))386 await peer.send(REPLY_METHOD, "/wrong", {"result": "wrong"}, id="late")387 await uds_harness.wait_lost(1)388 assert uds_harness.hub.resolve("W:one") is None389 390 replacement_peer = MemberPeer(uds_harness.hub.address, "W:one")391 replacement_peer.answer_calls = False392 await replacement_peer.connect()393 await uds_harness.wait_members(1)394 replacement = asyncio.create_task(uds_harness.hub.call_frame("W:one", expired))395 await replacement_peer.wait_frames(1)396 await replacement_peer.send(REPLY_METHOD, "/old", {"result": "fresh"}, id="late")397 assert data_of(await replacement) == {"result": "fresh"}398 399 400 async def test_cancel_during_write_reserves_id_until_late_reply(uds_harness, monkeypatch):401 peer = MemberPeer(uds_harness.hub.address, "W:one")402 peer.answer_calls = False403 await peer.connect()404 await uds_harness.wait_members(1)405 member = uds_harness.hub.resolve("W:one")406 original_write = type(member).write407 transmitted = asyncio.Event()408 release = asyncio.Event()409 410 async def blocked_after_write(self, frame):411 await original_write(self, frame)412 transmitted.set()413 await release.wait()414 415 monkeypatch.setattr(type(member), "write", blocked_after_write)416 frame = Frame(id="during-write", method=CALL_METHOD, path="/old")417 call = asyncio.create_task(uds_harness.hub.call_frame("W:one", frame))418 await transmitted.wait()419 await peer.wait_frames(1)420 call.cancel()421 with pytest.raises(asyncio.CancelledError):422 await call423 424 with pytest.raises(RuntimeError, match="awaiting a late reply"):425 await uds_harness.hub.call_frame("W:one", frame)426 await peer.send(REPLY_METHOD, "/old", {"result": "late"}, id="during-write")427 await asyncio.sleep(0.01)428 429 monkeypatch.setattr(type(member), "write", original_write)430 replacement = asyncio.create_task(uds_harness.hub.call_frame("W:one", frame))431 await peer.wait_frames(2)432 await peer.send(REPLY_METHOD, "/old", {"result": "fresh"}, id="during-write")433 assert data_of(await replacement) == {"result": "fresh"}434 release.set()435 436 437 async def test_inbound_event_background_work_is_bounded(socket_dir, caplog):438 gate = asyncio.Event()439 440 async def held_event(member, frame):441 await gate.wait()442 443 harness = HubHarness(path=os.path.join(socket_dir, "bounded.sock"))444 harness.hub.on_event = held_event445 harness.hub.max_event_tasks = 1446 await harness.hub.start()447 try:448 peer = MemberPeer(harness.hub.address, "W:one")449 await peer.connect()450 await harness.wait_members(1)451 await peer.send(EVENT_METHOD, "/first")452 while len(harness.hub._event_tasks) != 1:453 await asyncio.sleep(0)454 await peer.send(EVENT_METHOD, "/dropped")455 await asyncio.sleep(0.02)456 assert len(harness.hub._event_tasks) == 1457 assert "event task limit 1 reached" in caplog.text458 gate.set()459 finally:460 await harness.hub.stop()461 462 463 async def test_post_reaches_one_member_only(uds_harness):464 one = MemberPeer(uds_harness.hub.address, "W:one")465 two = MemberPeer(uds_harness.hub.address, "W:two")466 await one.connect()467 await two.connect()468 await uds_harness.wait_members(2)469 470 frame_id = await uds_harness.hub.post("W:one", "/occupancy", {"users": 3})471 await one.wait_frames(1)472 assert one.received[0].id == frame_id473 assert one.received[0].method == EVENT_METHOD474 assert data_of(one.received[0]) == {"users": 3}475 assert two.received == []476 await one.close()477 await two.close()478 479 480 async def test_inbound_event_reaches_the_consumer(uds_harness):481 peer = MemberPeer(uds_harness.hub.address, "W:one")482 await peer.connect()483 await uds_harness.wait_members(1)484 485 await peer.send(EVENT_METHOD, "/op/drop_user", {"seq": 7})486 deadline = asyncio.get_running_loop().time() + 5.0487 while not uds_harness.events:488 assert asyncio.get_running_loop().time() < deadline, "no event reached the hub"489 await asyncio.sleep(0.01)490 name, frame = uds_harness.events[0]491 assert (name, frame.path, data_of(frame)) == ("W:one", "/op/drop_user", {"seq": 7})492 await peer.close()493 494 495 async def test_a_slow_event_consumer_does_not_delay_the_reply_behind_it(uds_harness):496 """Serving is a task: the member's receive loop stays free for the REPLY."""497 gate = asyncio.Event()498 served = []499 500 async def slow_on_event(member, frame):501 served.append(frame.path)502 await gate.wait()503 504 uds_harness.hub.on_event = slow_on_event505 peer = MemberPeer(uds_harness.hub.address, "W:one")506 peer.reply_result = {"ok": 1}507 await peer.connect()508 await uds_harness.wait_members(1)509 510 await peer.send(EVENT_METHOD, "/op/slow", {"seq": 1})511 deadline = asyncio.get_running_loop().time() + 5.0512 while not served:513 assert asyncio.get_running_loop().time() < deadline, "the consumer never ran"514 await asyncio.sleep(0.01)515 payload = await asyncio.wait_for(uds_harness.hub.call("W:one", "/op/ping"), timeout=5.0)516 assert payload["result"] == {"ok": 1}517 518 gate.set()519 await peer.close()520 521 522 async def test_inbound_call_is_an_unexpected_envelope(uds_harness, caplog):523 peer = MemberPeer(uds_harness.hub.address, "W:one")524 await peer.connect()525 await uds_harness.wait_members(1)526 527 with caplog.at_level(logging.WARNING, logger="genro_asgi.channel.hub"):528 await peer.send(CALL_METHOD, "/ask", {"q": 1})529 await asyncio.sleep(0.1)530 531 assert "Unknown envelope CALL from W:one" in caplog.text532 assert peer.received == []533 assert uds_harness.hub.resolve("W:one") is not None534 await peer.close()535 536 537 async def test_member_eof_sweeps_the_rubric(uds_harness):538 peer = MemberPeer(uds_harness.hub.address, "W:one")539 await peer.connect()540 await uds_harness.wait_members(1)541 542 await peer.close()543 await uds_harness.wait_lost(1)544 assert uds_harness.lost == ["W:one"]545 assert uds_harness.hub.resolve("W:one") is None546 547 548 async def test_deliberate_hub_stop_fires_no_channel_lost(socket_dir):549 harness = HubHarness(path=os.path.join(socket_dir, "hub.sock"))550 await harness.hub.start()551 peer = MemberPeer(harness.hub.address, "W:one")552 await peer.connect()553 await harness.wait_members(1)554 555 await harness.hub.stop()556 await asyncio.sleep(0.1)557 assert harness.lost == []558 await peer.close()559 560 561 async def test_protocol_violation_isolates_that_member(uds_harness):562 survivor = MemberPeer(uds_harness.hub.address, "W:good")563 await survivor.connect()564 await uds_harness.wait_members(1)565 566 reader, writer = await asyncio.open_unix_connection(uds_harness.hub.path)567 writer.write(568 control_frame(569 method=REGISTER_METHOD, path=REGISTER_PATH, data={"name": "W:bad", "pid": 1}570 ).encode()571 )572 await writer.drain()573 await uds_harness.wait_members(2)574 payload = b"NOTWSX-garbage"575 writer.write(len(payload).to_bytes(4, "big") + payload)576 await writer.drain()577 578 await uds_harness.wait_lost(1)579 assert uds_harness.lost == ["W:bad"]580 assert uds_harness.hub.resolve("W:bad") is None581 writer.close()582 583 survivor.reply_result = "alive"584 payload = await uds_harness.hub.call("W:good", "/ping", None, timeout=5.0)585 assert payload["result"] == "alive"586 await survivor.close()587 588 589 async def test_duplicate_name_refuses_the_new_connection(uds_harness):590 first = MemberPeer(uds_harness.hub.address, "W:one")591 await first.connect()592 await uds_harness.wait_members(1)593 registered = uds_harness.hub.resolve("W:one")594 595 second = MemberPeer(uds_harness.hub.address, "W:one")596 await second.connect()597 # The refusal is the closed stream: the newcomer's receive loop reads EOF.598 await asyncio.wait_for(second._task, timeout=5.0)599 600 assert uds_harness.hub.resolve("W:one") is registered601 assert uds_harness.joined == ["W:one"]602 assert uds_harness.lost == []603 first.reply_result = "still here"604 payload = await uds_harness.hub.call("W:one", "/ping", None, timeout=5.0)605 assert payload["result"] == "still here"606 await second.close()607 await first.close()608 609 610 async def test_connection_without_register_is_rejected(uds_harness):611 reader, writer = await asyncio.open_unix_connection(uds_harness.hub.path)612 writer.write(control_frame(method=EVENT_METHOD, path="/hello", data=None).encode())613 await writer.drain()614 assert await reader.read() == b""615 assert uds_harness.hub.members == {}616 writer.close()617 618 619 async def test_address_before_start_raises(socket_dir):620 hub = ChannelHub(path=os.path.join(socket_dir, "hub.sock"))621 assert not hub.started622 with pytest.raises(RuntimeError):623 hub.address624 await hub.stop()625 626 627 async def test_path_and_host_together_are_rejected():628 with pytest.raises(ValueError):629 ChannelHub(path="/tmp/x.sock", host="127.0.0.1")630 631 632 @pytest.mark.parametrize("pid", [{}, "invalid"])633 async def test_malformed_register_pid_closes_only_offending_socket(uds_harness, pid):634 survivor = MemberPeer(uds_harness.hub.address, "W:good")635 await survivor.connect()636 await uds_harness.wait_members(1)637 reader, writer = await asyncio.open_unix_connection(uds_harness.hub.path)638 try:639 writer.write(control_frame(method=REGISTER_METHOD, path=REGISTER_PATH,640 data={"name": "W:bad", "pid": pid}).encode())641 await writer.drain()642 assert await asyncio.wait_for(reader.read(), timeout=1) == b""643 assert uds_harness.hub.resolve("W:bad") is None644 survivor.reply_result = "alive"645 reply = await uds_harness.hub.call("W:good", "/ping", timeout=1)646 assert reply["result"] == "alive"647 finally:648 writer.close()649 await writer.wait_closed()650 await survivor.close()