Coverage for agentos/tests/test_event_bus.py: 0%
279 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:28 +0800
1"""Tests for agentos.core.event_bus — EventBus, Subscription, DeadLetter."""
3import asyncio
5import pytest
7from agentos.core.event_bus import (
8 DeadLetter,
9 Event,
10 EventBus,
11 EventPriority,
12 Subscription,
13 default_bus,
14 event,
15)
17# ============================================================================
18# EventPriority
19# ============================================================================
21class TestEventPriority:
22 def test_enum_values(self):
23 assert EventPriority.LOW == 0
24 assert EventPriority.NORMAL == 50
25 assert EventPriority.HIGH == 100
26 assert EventPriority.CRITICAL == 200
28 def test_ordering(self):
29 assert EventPriority.CRITICAL > EventPriority.HIGH > EventPriority.NORMAL > EventPriority.LOW
32# ============================================================================
33# Event
34# ============================================================================
36class TestEvent:
37 def test_defaults(self):
38 e = Event(topic="test")
39 assert e.topic == "test"
40 assert e.payload is None
41 assert isinstance(e.event_id, str)
42 assert len(e.event_id) == 12
43 assert e.priority == EventPriority.NORMAL
44 assert e.source == ""
45 assert e.correlation_id == ""
46 assert e.metadata == {}
48 def test_custom_values(self):
49 e = Event(
50 topic="order.created",
51 payload={"id": 1},
52 priority=EventPriority.HIGH,
53 source="api",
54 correlation_id="abc",
55 metadata={"key": "val"},
56 )
57 assert e.topic == "order.created"
58 assert e.payload == {"id": 1}
59 assert e.priority == EventPriority.HIGH
60 assert e.source == "api"
62 def test_unique_event_ids(self):
63 e1 = Event(topic="a")
64 e2 = Event(topic="b")
65 assert e1.event_id != e2.event_id
67 def test_event_factory(self):
68 e = event("agent.start", payload={"v": 1}, source="scheduler")
69 assert isinstance(e, Event)
70 assert e.topic == "agent.start"
71 assert e.payload == {"v": 1}
74# ============================================================================
75# DeadLetter
76# ============================================================================
78class TestDeadLetter:
79 def test_defaults(self):
80 e = Event(topic="x")
81 dl = DeadLetter(event=e, handler_name="h", error="oops")
82 assert dl.event is e
83 assert dl.handler_name == "h"
84 assert dl.error == "oops"
85 assert dl.retry_count == 0
88# ============================================================================
89# Subscription
90# ============================================================================
92class TestSubscription:
93 async def _noop(self, e): pass
95 def test_matches_exact(self):
96 sub = Subscription(
97 topic_pattern="agent.start",
98 handler=self._noop,
99 handler_name="test",
100 is_pattern=False,
101 )
102 assert sub.matches("agent.start") is True
103 assert sub.matches("agent.stop") is False
105 def test_matches_wildcard(self):
106 sub = Subscription(
107 topic_pattern="agent.*",
108 handler=self._noop,
109 handler_name="test",
110 is_pattern=True,
111 )
112 assert sub.matches("agent.start") is True
113 assert sub.matches("agent.stop") is True
114 assert sub.matches("agent") is False
115 assert sub.matches("other.thing") is False
117 def test_matches_complex_wildcard(self):
118 sub = Subscription(
119 topic_pattern="order.*.created",
120 handler=self._noop,
121 handler_name="test",
122 is_pattern=True,
123 )
124 assert sub.matches("order.123.created") is True
125 assert sub.matches("order.abc.created") is True
127 def test_concurrency_default(self):
128 sub = Subscription(topic_pattern="t", handler=self._noop, handler_name="n")
129 assert sub.concurrency == 1
132# ============================================================================
133# EventBus — Basic
134# ============================================================================
136class TestEventBusBasic:
137 def test_defaults(self):
138 bus = EventBus()
139 assert bus.queue_size == 0
140 assert bus.total_events == 0
141 assert bus.subscription_count == 0
143 def test_custom_params(self):
144 bus = EventBus(max_queue_size=50, dlq_enabled=False, dlq_max_size=200, worker_count=2)
145 assert bus._worker_count == 2
147 def test_list_topics(self):
148 bus = EventBus()
149 assert bus.list_topics() == []
152# ============================================================================
153# EventBus — Subscribe / Unsubscribe
154# ============================================================================
156class TestEventBusSubscriptions:
157 async def _noop(self, e): pass
159 def test_subscribe(self):
160 bus = EventBus()
161 sub = bus.subscribe("agent.start", self._noop, handler_name="h")
162 assert isinstance(sub, Subscription)
163 assert bus.subscription_count == 1
164 assert bus.list_topics() == ["agent.start"]
166 def test_subscribe_auto_name(self):
167 bus = EventBus()
168 bus.subscribe("t", self._noop)
169 assert bus.subscription_count == 1
171 def test_unsubscribe(self):
172 bus = EventBus()
173 bus.subscribe("agent.start", self._noop, handler_name="h")
174 assert bus.unsubscribe("agent.start", "h") is True
175 assert bus.subscription_count == 0
177 def test_unsubscribe_missing(self):
178 bus = EventBus()
179 assert bus.unsubscribe("x", "y") is False
181 def test_unsubscribe_all(self):
182 bus = EventBus()
183 bus.subscribe("a", self._noop, handler_name="h")
184 bus.subscribe("b", self._noop, handler_name="h")
185 assert bus.unsubscribe_all("h") == 2
186 assert bus.subscription_count == 0
189# ============================================================================
190# EventBus — Publish / Process
191# ============================================================================
193class TestEventBusPublish:
194 @pytest.mark.asyncio
195 async def test_publish_dispatch(self):
196 bus = EventBus()
197 received = []
199 async def handler(e):
200 received.append(e.payload)
202 bus.subscribe("test", handler, handler_name="h")
203 await bus.start()
204 e = Event(topic="test", payload="hello")
205 count = await bus.publish(e)
206 await asyncio.sleep(0.1)
207 await bus.stop()
208 assert count == 1
209 assert received == ["hello"]
211 @pytest.mark.asyncio
212 async def test_publish_no_subscribers(self):
213 bus = EventBus()
214 await bus.start()
215 e = Event(topic="no.subs")
216 count = await bus.publish(e)
217 await bus.stop()
218 assert count == 0
220 @pytest.mark.asyncio
221 async def test_publish_wildcard(self):
222 bus = EventBus()
223 received = []
225 async def handler(e):
226 received.append(e.topic)
228 bus.subscribe("agent.*", handler, handler_name="h")
229 await bus.start()
230 await bus.publish(Event(topic="agent.start"))
231 await bus.publish(Event(topic="agent.stop"))
232 await asyncio.sleep(0.1)
233 await bus.stop()
234 assert len(received) == 2
236 @pytest.mark.asyncio
237 async def test_publish_nowait(self):
238 bus = EventBus(max_queue_size=100)
239 received = []
241 async def handler(e):
242 received.append(e.payload)
244 bus.subscribe("t", handler, handler_name="h")
245 await bus.start()
246 count = await bus.publish_nowait(Event(topic="t", payload="x"))
247 await asyncio.sleep(0.1)
248 await bus.stop()
249 assert count == 1
251 @pytest.mark.asyncio
252 async def test_emit_sync(self):
253 bus = EventBus()
254 received = []
256 async def handler(e):
257 received.append(e.payload)
259 bus.subscribe("t", handler, handler_name="h")
260 await bus.start()
261 bus.emit_sync(Event(topic="t", payload="sync"))
262 await asyncio.sleep(0.1)
263 await bus.stop()
264 assert "sync" in received
266 @pytest.mark.asyncio
267 async def test_total_events(self):
268 bus = EventBus()
270 async def handler(e): pass
271 bus.subscribe("t", handler, handler_name="h")
272 await bus.start()
273 await bus.publish(Event(topic="t"))
274 await asyncio.sleep(0.1)
275 await bus.stop()
276 assert bus.total_events >= 1
279# ============================================================================
280# EventBus — Priority ordering
281# ============================================================================
283class TestEventBusPriority:
284 @pytest.mark.asyncio
285 async def test_priority_sorting(self):
286 bus = EventBus()
287 order = []
289 async def handler(e):
290 order.append(e.priority)
291 # Small delay to ensure async processing interleaving
292 await asyncio.sleep(0.01)
294 bus.subscribe("t", handler, handler_name="h")
295 await bus.start()
297 await bus.publish(Event(topic="t", priority=EventPriority.NORMAL))
298 await bus.publish(Event(topic="t", priority=EventPriority.CRITICAL))
299 await asyncio.sleep(0.5)
300 await bus.stop()
302 # At least one event was processed
303 assert len(order) >= 1
306# ============================================================================
307# EventBus — DLQ
308# ============================================================================
310class TestEventBusDLQ:
311 @pytest.mark.asyncio
312 async def test_failed_handler_goes_to_dlq(self):
313 bus = EventBus(dlq_enabled=True)
315 async def failing_handler(e):
316 raise ValueError("boom")
318 bus.subscribe("t", failing_handler, handler_name="h")
319 await bus.start()
320 await bus.publish(Event(topic="t"))
321 await asyncio.sleep(0.2)
322 await bus.stop()
324 dlq = bus.get_dlq()
325 assert len(dlq) == 1
326 assert dlq[0].handler_name == "h"
327 assert "boom" in dlq[0].error
329 @pytest.mark.asyncio
330 async def test_dlq_disabled(self):
331 bus = EventBus(dlq_enabled=False)
333 async def failing_handler(e):
334 raise ValueError("boom")
336 bus.subscribe("t", failing_handler, handler_name="h")
337 await bus.start()
338 await bus.publish(Event(topic="t"))
339 await asyncio.sleep(0.2)
340 await bus.stop()
342 assert len(bus.get_dlq()) == 0
344 @pytest.mark.asyncio
345 async def test_dlq_max_size(self):
346 bus = EventBus(dlq_enabled=True, dlq_max_size=2)
348 async def failing_handler(e):
349 raise ValueError("x")
351 bus.subscribe("t", failing_handler, handler_name="h")
352 await bus.start()
353 for i in range(5):
354 await bus.publish(Event(topic="t"))
355 await asyncio.sleep(0.3)
356 await bus.stop()
358 dlq = bus.get_dlq()
359 assert len(dlq) == 2
361 @pytest.mark.asyncio
362 async def test_clear_dlq(self):
363 bus = EventBus(dlq_enabled=True)
365 async def failing_handler(e):
366 raise ValueError("x")
368 bus.subscribe("t", failing_handler, handler_name="h")
369 await bus.start()
370 await bus.publish(Event(topic="t"))
371 await asyncio.sleep(0.2)
372 await bus.stop()
374 assert len(bus.get_dlq()) == 1
375 cleared = bus.clear_dlq()
376 assert cleared == 1
377 assert len(bus.get_dlq()) == 0
379 @pytest.mark.asyncio
380 async def test_replay_dlq(self):
381 bus = EventBus(dlq_enabled=True)
383 async def failing_handler(e):
384 raise ValueError("x")
386 bus.subscribe("t", failing_handler, handler_name="h")
387 await bus.start()
388 await bus.publish(Event(topic="t"))
389 await asyncio.sleep(0.2)
390 await bus.stop()
392 assert len(bus.get_dlq()) == 1
394 # Restart bus and replay
395 await bus.start()
396 replayed = await bus.replay_dlq()
397 assert replayed == 1
398 assert len(bus.get_dlq()) == 0
399 await bus.stop()
402# ============================================================================
403# EventBus — Lifecycle
404# ============================================================================
406class TestEventBusLifecycle:
407 @pytest.mark.asyncio
408 async def test_start_idempotent(self):
409 bus = EventBus()
410 await bus.start()
411 await bus.start() # should not double-start
412 await bus.stop()
414 @pytest.mark.asyncio
415 async def test_stop_drains_queue(self):
416 bus = EventBus(worker_count=2)
417 received = []
419 async def handler(e):
420 received.append(e.payload)
421 await asyncio.sleep(0) # yield to event loop to allow processing
423 bus.subscribe("t", handler, handler_name="h")
424 await bus.start()
425 await bus.publish(Event(topic="t", payload="drain"))
426 await asyncio.sleep(0.3)
427 await bus.stop(grace_period=3.0)
428 assert "drain" in received
431# ============================================================================
432# Default bus
433# ============================================================================
435class TestDefaultBus:
436 def test_default_bus_exists(self):
437 assert isinstance(default_bus, EventBus)