tests/core/test_event_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 """Tests for EventHub (core 1e Phase 3): in-memory per-session event fan-out.16 17 Real objects, no mocks: a live ``EventHub`` and real ``asyncio.Queue``s. The18 tests are ``async def`` so subscriptions bind to a running event loop, matching19 the hub's runtime (one server event loop).20 """21 22 from __future__ import annotations23 24 from genro_asgi.tasks import EventHub25 from genro_asgi.tasks.hub import QUEUE_MAXSIZE26 27 28 class TestSubscribePublish:29     """A subscriber receives every event published to its session."""30 31     async def test_publish_reaches_subscriber(self) -> None:32         hub = EventHub()33         queue = hub.subscribe("s1")34         hub.publish("s1", {"type": "progress", "value": 1})35         assert (await queue.get()) == {"type": "progress", "value": 1}36 37     async def test_two_subscribers_same_session_both_receive(self) -> None:38         hub = EventHub()39         q1 = hub.subscribe("s1")40         q2 = hub.subscribe("s1")41         hub.publish("s1", {"n": 7})42         assert (await q1.get()) == {"n": 7}43         assert (await q2.get()) == {"n": 7}44 45     async def test_publish_isolated_per_session(self) -> None:46         hub = EventHub()47         q1 = hub.subscribe("s1")48         hub.subscribe("s2")49         hub.publish("s1", {"only": "s1"})50         assert (await q1.get()) == {"only": "s1"}51         assert hub.subscribe("s2").empty()      # a fresh s2 queue saw nothing52 53 54 class TestNoSubscriber:55     """Publishing to a session nobody watches is a silent no-op."""56 57     async def test_publish_without_subscriber_is_noop(self) -> None:58         hub = EventHub()59         hub.publish("ghost", {"lost": True})        # must not raise60         queue = hub.subscribe("ghost")61         assert queue.empty()                         # the earlier event is not replayed62 63 64 class TestBackpressure:65     """A full queue drops the OLDEST event (progress is a snapshot)."""66 67     async def test_full_queue_drops_oldest(self) -> None:68         hub = EventHub()69         queue = hub.subscribe("s1")70         for i in range(QUEUE_MAXSIZE + 3):71             hub.publish("s1", {"i": i})72         assert queue.qsize() == QUEUE_MAXSIZE73         first = await queue.get()74         assert first == {"i": 3}                     # 0,1,2 dropped as oldest75 76 77 class TestUnsubscribe:78     """Unsubscribing removes the queue and cleans up the empty session."""79 80     async def test_unsubscribe_stops_delivery(self) -> None:81         hub = EventHub()82         queue = hub.subscribe("s1")83         hub.unsubscribe("s1", queue)84         hub.publish("s1", {"after": "unsub"})85         assert queue.empty()86 87     async def test_unsubscribe_unknown_session_is_noop(self) -> None:88         hub = EventHub()89         queue = hub.subscribe("s1")90         hub.unsubscribe("nope", queue)              # must not raise91 92     async def test_last_unsubscribe_drops_the_session(self) -> None:93         hub = EventHub()94         queue = hub.subscribe("s1")95         hub.unsubscribe("s1", queue)96         # a fresh subscribe rebuilds a distinct empty queue for the session97         again = hub.subscribe("s1")98         assert again is not queue99         assert again.empty()