Tyler
8bcb23c78d
- api.parse_837 / parse_835: pass request.app.state.event_bus into store.add() - conftest: autouse fixture wires a fresh EventBus onto app.state for every test, since TestClient does not invoke the FastAPI lifespan handler unless used as a context manager - test_pubsub: split get_event_bus coverage into a raises-when-missing test and a returns-attached-bus test, both save/restore app.state.event_bus around the assertion so the autouse fixture's bus is preserved Phase 2 complete: db init moved to lifespan, EventBus is the process-wide publish point, and the two ingest endpoints publish claim_written / remittance_written / activity_recorded events on every store.add().
135 lines
4.0 KiB
Python
135 lines
4.0 KiB
Python
"""Tests for the in-process EventBus.
|
|
|
|
These exercise the drop-oldest overflow strategy, per-kind fan-out, and
|
|
the non-blocking publish contract. They are async because the bus
|
|
exposes an async iterator for subscribers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from cyclone.pubsub import EventBus, get_event_bus
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_no_subscribers_is_noop():
|
|
bus = EventBus()
|
|
# No subscribers at all — must not raise.
|
|
await bus.publish("x", {"a": 1})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_to_n_subscribers_delivers_to_all():
|
|
bus = EventBus()
|
|
queues = [bus.subscribe(["claim"]) for _ in range(3)]
|
|
|
|
await bus.publish("claim", {"id": "C1"})
|
|
|
|
received = [await asyncio.wait_for(q.__anext__(), timeout=0.5) for q in queues]
|
|
for event in received:
|
|
assert event == {"id": "C1", "_kind": "claim"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscriber_only_receives_matching_kinds():
|
|
bus = EventBus()
|
|
queue = bus.subscribe(["a"])
|
|
|
|
await bus.publish("b", {"n": 1})
|
|
await bus.publish("a", {"n": 2})
|
|
|
|
event = await asyncio.wait_for(queue.__anext__(), timeout=0.5)
|
|
assert event == {"n": 2, "_kind": "a"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slow_subscriber_does_not_block_publish():
|
|
bus = EventBus(max_queue_size=2)
|
|
queue = bus.subscribe(["e"])
|
|
|
|
async def emit():
|
|
for i in range(3):
|
|
await bus.publish("e", {"i": i})
|
|
|
|
start = time.monotonic()
|
|
await emit()
|
|
elapsed = time.monotonic() - start
|
|
|
|
# Publishing must not block even though the subscriber queue is full
|
|
# after the second event; the drop-oldest policy keeps publishes non-blocking.
|
|
assert elapsed < 0.05, f"publish blocked for {elapsed:.3f}s"
|
|
|
|
# The slow subscriber retains only the last two events.
|
|
seen = []
|
|
for _ in range(2):
|
|
seen.append(await asyncio.wait_for(queue.__anext__(), timeout=0.5))
|
|
assert [e["i"] for e in seen] == [1, 2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_queue_overflow_drops_oldest():
|
|
bus = EventBus(max_queue_size=2)
|
|
queue = bus.subscribe(["e"])
|
|
|
|
for i in range(3):
|
|
await bus.publish("e", {"i": i})
|
|
|
|
seen = []
|
|
for _ in range(2):
|
|
seen.append(await asyncio.wait_for(queue.__anext__(), timeout=0.5))
|
|
assert [e["i"] for e in seen] == [1, 2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_concurrent_subscribers_each_get_all_events():
|
|
bus = EventBus()
|
|
queues = [bus.subscribe(["claim"]) for _ in range(2)]
|
|
|
|
for i in range(5):
|
|
await bus.publish("claim", {"i": i})
|
|
|
|
for q in queues:
|
|
seen = []
|
|
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; if the lifespan handler
|
|
hasn't set ``app.state.event_bus`` (or it's been cleared), the
|
|
AttributeError must surface so callers can fail loudly."""
|
|
from cyclone.api import app
|
|
|
|
# The autouse conftest fixture sets app.state.event_bus for every test
|
|
# so that endpoint code can call request.app.state.event_bus. We pop
|
|
# the key entirely here so __getattr__ raises AttributeError rather
|
|
# than returning None for an explicitly-None slot.
|
|
state_dict = app.state._state
|
|
saved = state_dict.pop("event_bus", None)
|
|
try:
|
|
with pytest.raises((AttributeError, RuntimeError)):
|
|
get_event_bus()
|
|
finally:
|
|
if saved is not None:
|
|
state_dict["event_bus"] = saved
|
|
|
|
|
|
def test_get_event_bus_returns_app_state_bus():
|
|
"""When the lifespan handler has initialised the bus, get_event_bus()
|
|
returns it. Validates the happy path under the same late-import."""
|
|
from cyclone.api import app
|
|
from cyclone.pubsub import EventBus
|
|
|
|
saved = getattr(app.state, "event_bus", None)
|
|
bus = EventBus()
|
|
app.state.event_bus = bus
|
|
try:
|
|
assert get_event_bus() is bus
|
|
finally:
|
|
app.state.event_bus = saved
|