145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
"""Tests for the in-memory decision-event broker (events.py).
|
|
|
|
Offline, no network. Drive the thread-safe queue broker directly: publish,
|
|
subscribe-with-replay, unsubscribe, and the interaction where a decision is
|
|
published with no subscriber (must not error and must stay in the ring).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import queue
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
import dispatcher
|
|
import events
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_broker():
|
|
events.clear()
|
|
yield
|
|
events.clear()
|
|
|
|
|
|
def _decision(decision_id: int, model: str = "m") -> dict:
|
|
return {"id": decision_id, "selected_model": model}
|
|
|
|
|
|
def test_publish_adds_to_ring_and_stays_for_replay():
|
|
events.publish_decision(_decision(1))
|
|
events.publish_decision(_decision(2))
|
|
assert [d["id"] for d in events.recent_decisions()] == [1, 2]
|
|
|
|
|
|
def test_subscribe_replays_ring_into_new_queue():
|
|
events.publish_decision(_decision(1))
|
|
events.publish_decision(_decision(2))
|
|
sub = events.subscribe(replay=True)
|
|
assert [d["id"] for d in _drain(sub)] == [1, 2]
|
|
|
|
|
|
def test_subscribe_without_replay_starts_empty():
|
|
events.publish_decision(_decision(1))
|
|
sub = events.subscribe(replay=False)
|
|
assert _drain(sub) == []
|
|
|
|
|
|
def test_publish_fans_out_to_live_subscribers():
|
|
sub = events.subscribe()
|
|
events.publish_decision(_decision(1))
|
|
assert [d["id"] for d in _drain(sub)] == [1]
|
|
|
|
|
|
def test_unsubscribe_stops_delivery():
|
|
sub = events.subscribe()
|
|
events.unsubscribe(sub)
|
|
events.publish_decision(_decision(1))
|
|
assert _drain(sub) == []
|
|
|
|
|
|
def test_publish_with_no_subscriber_keeps_ring_and_does_not_raise():
|
|
events.publish_decision(_decision(1))
|
|
assert [d["id"] for d in events.recent_decisions()] == [1]
|
|
|
|
|
|
def test_full_subscriber_is_dropped_not_blocked():
|
|
events.clear()
|
|
subscriber = events.subscribe(replay=False, max_queue=1)
|
|
events.publish_decision(_decision(1))
|
|
events.publish_decision(_decision(2)) # must not raise or block; triggers eviction
|
|
# Eviction removed the subscriber from fan-out and delivered a sentinel so
|
|
# the SSE generator can terminate; the ring still contains both decisions.
|
|
assert subscriber not in events._subscribers
|
|
assert len(events.recent_decisions()) == 2
|
|
|
|
|
|
def _drain(q: queue.Queue) -> list:
|
|
out = []
|
|
while True:
|
|
try:
|
|
item = q.get_nowait()
|
|
except queue.Empty:
|
|
return out
|
|
if item is events._EVICTED:
|
|
return out
|
|
out.append(item)
|
|
|
|
|
|
def test_concurrent_publish_subscribe_unsubscribe_raises_no_error():
|
|
errors: list[Exception] = []
|
|
|
|
def _publisher():
|
|
for i in range(200):
|
|
try:
|
|
events.publish_decision({"id": i, "round": "pub"})
|
|
except (RuntimeError, queue.Full) as exc:
|
|
errors.append(exc)
|
|
|
|
def _subscriptor():
|
|
for _ in range(10):
|
|
try:
|
|
sub = events.subscribe(replay=False)
|
|
events.unsubscribe(sub)
|
|
except RuntimeError as exc:
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=_publisher) for _ in range(4)] + [
|
|
threading.Thread(target=_subscriptor) for _ in range(4)
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert not errors, f"unexpected errors during concurrent access: {errors}"
|
|
|
|
|
|
def test_full_subscriber_eviction_terminates_decision_event_stream(monkeypatch):
|
|
"""When a subscriber is evicted (full queue), the _decision_event_stream
|
|
generator breaks out of its loop instead of blocking forever."""
|
|
events.clear()
|
|
# Use a 1-slot queue we control. Monkeypatch subscribe to return it so
|
|
# _decision_event_stream's live loop reads from our controlled queue.
|
|
controlled = events.subscribe(replay=False, max_queue=1)
|
|
monkeypatch.setattr(events, "subscribe", lambda *args, **kwargs: controlled)
|
|
monkeypatch.setattr(events, "unsubscribe", lambda q: None)
|
|
|
|
# Fill the controlled queue; publish_decision would evict on next insert.
|
|
controlled.put_nowait(_decision(1))
|
|
|
|
feed = dispatcher._decision_event_stream()
|
|
assert next(feed).startswith("retry:")
|
|
# The replay drain is empty (replay=False); the live loop drains decision(1)
|
|
# and now blocks on controlled.get(timeout=SSE_HEARTBEAT_SECONDS).
|
|
assert next(feed) == dispatcher._sse_data(_decision(1))
|
|
|
|
# Now the queue is empty. Push another decision to fill it, then publish a
|
|
# second one: the controlled subscriber is full, so publish_decision evicts
|
|
# it and places the sentinel.
|
|
controlled.put_nowait(_decision(2))
|
|
events.publish_decision(_decision(3))
|
|
# The generator should wake, see the sentinel, and terminate.
|
|
with pytest.raises(StopIteration):
|
|
next(feed)
|