feat(pubsub): get_event_bus() late-import accessor

This commit is contained in:
Tyler
2026-06-20 15:35:02 -06:00
parent b2f5a16541
commit 25a76a515d
2 changed files with 23 additions and 1 deletions
+11 -1
View File
@@ -12,7 +12,7 @@ import time
import pytest
from cyclone.pubsub import EventBus
from cyclone.pubsub import EventBus, get_event_bus
@pytest.mark.asyncio
@@ -97,3 +97,13 @@ async def test_multiple_concurrent_subscribers_each_get_all_events():
for _ in range(5):
seen.append(await asyncio.wait_for(q.__anext__(), timeout=0.5))
assert [e["i"] for e in seen] == [0, 1, 2, 3, 4]
def test_get_event_bus_raises_when_app_state_uninitialized():
"""get_event_bus() late-imports cyclone.api; without a running app,
accessing app.state.event_bus raises AttributeError."""
# Don't import cyclone.api at module level — that would force a load
# and could mask the real failure. Just call the helper and expect
# the AttributeError to surface from app.state.
with pytest.raises((AttributeError, RuntimeError)):
get_event_bus()