Tyler
da29188dd0
Adds three live-tail streaming endpoints that emit an NDJSON snapshot
then forward new event-bus events as they arrive, with a 15s idle
heartbeat (overridable via CYCLONE_TAIL_HEARTBEAT_S for tests).
Each endpoint:
1. yields a snapshot of existing rows as {"type":"item","data":<row>}
2. terminates the snapshot with {"type":"snapshot_end","data":{"count":N}}
3. subscribes to its event kind and forwards each new event as an
{"type":"item","data":<event>} line
4. emits a {"type":"heartbeat","data":{"ts":<iso>}} line every
CYCLONE_TAIL_HEARTBEAT_S seconds when idle
5. checks request.is_disconnected() before each yield and unsubscribes
from the bus on cleanup so a closed stream releases its queue
The shared tail loop lives in api._tail_events, which polls
bus.subscribe_raw()'s queue directly instead of using the bus's
async-iterator wrapper — wait_for on an async generator cancels the
inner future on timeout, which poisons subsequent __anext__ calls with
StopAsyncIteration. Queue.get() is idempotent under cancellation, so
heartbeats don't break the subscription.
EventBus gains an unsubscribe(queue, kinds) method (idempotent) so
the tail loop can release its queue in a try/finally. The disconnect
test asserts the subscriber list is empty after the body iterator is
closed, validating no queue leak per open stream.
Tests in test_api_stream_live.py: 8 tests covering snapshot shape,
post-snapshot publish, heartbeat timing, multi-item snapshots, and
client disconnect cleanup. Plus 2 tests in test_pubsub.py for the
new unsubscribe method.
157 lines
4.6 KiB
Python
157 lines
4.6 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
|
|
|
|
|
|
def test_unsubscribe_removes_queue_from_each_kind():
|
|
bus = EventBus()
|
|
q1, _ = bus.subscribe_raw(["a", "b"])
|
|
q2, _ = bus.subscribe_raw(["a"])
|
|
|
|
bus.unsubscribe(q1, ["a", "b"])
|
|
|
|
assert q1 not in bus._subscribers.get("a", [])
|
|
assert q1 not in bus._subscribers.get("b", [])
|
|
assert q2 in bus._subscribers["a"]
|
|
|
|
|
|
def test_unsubscribe_is_idempotent():
|
|
bus = EventBus()
|
|
q, _ = bus.subscribe_raw(["a"])
|
|
bus.unsubscribe(q, ["a"])
|
|
# Second call must not raise.
|
|
bus.unsubscribe(q, ["a"])
|
|
bus.unsubscribe(q, ["b"]) # wrong kind — also fine
|
|
assert "a" not in bus._subscribers
|