Coverage for agentos/tests/test_event_bus.py: 0%

1"""Tests for agentos.core.event_bus — EventBus, Subscription, DeadLetter.""" 

2 

3import asyncio 

4import pytest 

5from agentos.core.event_bus import ( 

6 DeadLetter, 

7 Event, 

8 EventBus, 

9 EventPriority, 

10 Subscription, 

11 default_bus, 

12 event, 

13) 

14 

15 

16# ============================================================================ 

17# EventPriority 

18# ============================================================================ 

19 

20class TestEventPriority: 

21 def test_enum_values(self): 

22 assert EventPriority.LOW == 0 

23 assert EventPriority.NORMAL == 50 

24 assert EventPriority.HIGH == 100 

25 assert EventPriority.CRITICAL == 200 

26 

27 def test_ordering(self): 

28 assert EventPriority.CRITICAL > EventPriority.HIGH > EventPriority.NORMAL > EventPriority.LOW 

29 

30 

31# ============================================================================ 

32# Event 

33# ============================================================================ 

34 

35class TestEvent: 

36 def test_defaults(self): 

37 e = Event(topic="test") 

38 assert e.topic == "test" 

39 assert e.payload is None 

40 assert isinstance(e.event_id, str) 

41 assert len(e.event_id) == 12 

42 assert e.priority == EventPriority.NORMAL 

43 assert e.source == "" 

44 assert e.correlation_id == "" 

45 assert e.metadata == {} 

46 

47 def test_custom_values(self): 

48 e = Event( 

49 topic="order.created", 

50 payload={"id": 1}, 

51 priority=EventPriority.HIGH, 

52 source="api", 

53 correlation_id="abc", 

54 metadata={"key": "val"}, 

55 ) 

56 assert e.topic == "order.created" 

57 assert e.payload == {"id": 1} 

58 assert e.priority == EventPriority.HIGH 

59 assert e.source == "api" 

60 

61 def test_unique_event_ids(self): 

62 e1 = Event(topic="a") 

63 e2 = Event(topic="b") 

64 assert e1.event_id != e2.event_id 

65 

66 def test_event_factory(self): 

67 e = event("agent.start", payload={"v": 1}, source="scheduler") 

68 assert isinstance(e, Event) 

69 assert e.topic == "agent.start" 

70 assert e.payload == {"v": 1} 

71 

72 

73# ============================================================================ 

74# DeadLetter 

75# ============================================================================ 

76 

77class TestDeadLetter: 

78 def test_defaults(self): 

79 e = Event(topic="x") 

80 dl = DeadLetter(event=e, handler_name="h", error="oops") 

81 assert dl.event is e 

82 assert dl.handler_name == "h" 

83 assert dl.error == "oops" 

84 assert dl.retry_count == 0 

85 

86 

87# ============================================================================ 

88# Subscription 

89# ============================================================================ 

90 

91class TestSubscription: 

92 async def _noop(self, e): pass 

93 

94 def test_matches_exact(self): 

95 sub = Subscription( 

96 topic_pattern="agent.start", 

97 handler=self._noop, 

98 handler_name="test", 

99 is_pattern=False, 

100 ) 

101 assert sub.matches("agent.start") is True 

102 assert sub.matches("agent.stop") is False 

103 

104 def test_matches_wildcard(self): 

105 sub = Subscription( 

106 topic_pattern="agent.*", 

107 handler=self._noop, 

108 handler_name="test", 

109 is_pattern=True, 

110 ) 

111 assert sub.matches("agent.start") is True 

112 assert sub.matches("agent.stop") is True 

113 assert sub.matches("agent") is False 

114 assert sub.matches("other.thing") is False 

115 

116 def test_matches_complex_wildcard(self): 

117 sub = Subscription( 

118 topic_pattern="order.*.created", 

119 handler=self._noop, 

120 handler_name="test", 

121 is_pattern=True, 

122 ) 

123 assert sub.matches("order.123.created") is True 

124 assert sub.matches("order.abc.created") is True 

125 

126 def test_concurrency_default(self): 

127 sub = Subscription(topic_pattern="t", handler=self._noop, handler_name="n") 

128 assert sub.concurrency == 1 

129 

130 

131# ============================================================================ 

132# EventBus — Basic 

133# ============================================================================ 

134 

135class TestEventBusBasic: 

136 def test_defaults(self): 

137 bus = EventBus() 

138 assert bus.queue_size == 0 

139 assert bus.total_events == 0 

140 assert bus.subscription_count == 0 

141 

142 def test_custom_params(self): 

143 bus = EventBus(max_queue_size=50, dlq_enabled=False, dlq_max_size=200, worker_count=2) 

144 assert bus._worker_count == 2 

145 

146 def test_list_topics(self): 

147 bus = EventBus() 

148 assert bus.list_topics() == [] 

149 

150 

151# ============================================================================ 

152# EventBus — Subscribe / Unsubscribe 

153# ============================================================================ 

154 

155class TestEventBusSubscriptions: 

156 async def _noop(self, e): pass 

157 

158 def test_subscribe(self): 

159 bus = EventBus() 

160 sub = bus.subscribe("agent.start", self._noop, handler_name="h") 

161 assert isinstance(sub, Subscription) 

162 assert bus.subscription_count == 1 

163 assert bus.list_topics() == ["agent.start"] 

164 

165 def test_subscribe_auto_name(self): 

166 bus = EventBus() 

167 bus.subscribe("t", self._noop) 

168 assert bus.subscription_count == 1 

169 

170 def test_unsubscribe(self): 

171 bus = EventBus() 

172 bus.subscribe("agent.start", self._noop, handler_name="h") 

173 assert bus.unsubscribe("agent.start", "h") is True 

174 assert bus.subscription_count == 0 

175 

176 def test_unsubscribe_missing(self): 

177 bus = EventBus() 

178 assert bus.unsubscribe("x", "y") is False 

179 

180 def test_unsubscribe_all(self): 

181 bus = EventBus() 

182 bus.subscribe("a", self._noop, handler_name="h") 

183 bus.subscribe("b", self._noop, handler_name="h") 

184 assert bus.unsubscribe_all("h") == 2 

185 assert bus.subscription_count == 0 

186 

187 

188# ============================================================================ 

189# EventBus — Publish / Process 

190# ============================================================================ 

191 

192class TestEventBusPublish: 

193 @pytest.mark.asyncio 

194 async def test_publish_dispatch(self): 

195 bus = EventBus() 

196 received = [] 

197 

198 async def handler(e): 

199 received.append(e.payload) 

200 

201 bus.subscribe("test", handler, handler_name="h") 

202 await bus.start() 

203 e = Event(topic="test", payload="hello") 

204 count = await bus.publish(e) 

205 await asyncio.sleep(0.1) 

206 await bus.stop() 

207 assert count == 1 

208 assert received == ["hello"] 

209 

210 @pytest.mark.asyncio 

211 async def test_publish_no_subscribers(self): 

212 bus = EventBus() 

213 await bus.start() 

214 e = Event(topic="no.subs") 

215 count = await bus.publish(e) 

216 await bus.stop() 

217 assert count == 0 

218 

219 @pytest.mark.asyncio 

220 async def test_publish_wildcard(self): 

221 bus = EventBus() 

222 received = [] 

223 

224 async def handler(e): 

225 received.append(e.topic) 

226 

227 bus.subscribe("agent.*", handler, handler_name="h") 

228 await bus.start() 

229 await bus.publish(Event(topic="agent.start")) 

230 await bus.publish(Event(topic="agent.stop")) 

231 await asyncio.sleep(0.1) 

232 await bus.stop() 

233 assert len(received) == 2 

234 

235 @pytest.mark.asyncio 

236 async def test_publish_nowait(self): 

237 bus = EventBus(max_queue_size=100) 

238 received = [] 

239 

240 async def handler(e): 

241 received.append(e.payload) 

242 

243 bus.subscribe("t", handler, handler_name="h") 

244 await bus.start() 

245 count = await bus.publish_nowait(Event(topic="t", payload="x")) 

246 await asyncio.sleep(0.1) 

247 await bus.stop() 

248 assert count == 1 

249 

250 @pytest.mark.asyncio 

251 async def test_emit_sync(self): 

252 bus = EventBus() 

253 received = [] 

254 

255 async def handler(e): 

256 received.append(e.payload) 

257 

258 bus.subscribe("t", handler, handler_name="h") 

259 await bus.start() 

260 bus.emit_sync(Event(topic="t", payload="sync")) 

261 await asyncio.sleep(0.1) 

262 await bus.stop() 

263 assert "sync" in received 

264 

265 @pytest.mark.asyncio 

266 async def test_total_events(self): 

267 bus = EventBus() 

268 

269 async def handler(e): pass 

270 bus.subscribe("t", handler, handler_name="h") 

271 await bus.start() 

272 await bus.publish(Event(topic="t")) 

273 await asyncio.sleep(0.1) 

274 await bus.stop() 

275 assert bus.total_events >= 1 

276 

277 

278# ============================================================================ 

279# EventBus — Priority ordering 

280# ============================================================================ 

281 

282class TestEventBusPriority: 

283 @pytest.mark.asyncio 

284 async def test_priority_sorting(self): 

285 bus = EventBus() 

286 order = [] 

287 

288 async def handler(e): 

289 order.append(e.priority) 

290 # Small delay to ensure async processing interleaving 

291 await asyncio.sleep(0.01) 

292 

293 bus.subscribe("t", handler, handler_name="h") 

294 await bus.start() 

295 

296 await bus.publish(Event(topic="t", priority=EventPriority.NORMAL)) 

297 await bus.publish(Event(topic="t", priority=EventPriority.CRITICAL)) 

298 await asyncio.sleep(0.5) 

299 await bus.stop() 

300 

301 # At least one event was processed 

302 assert len(order) >= 1 

303 

304 

305# ============================================================================ 

306# EventBus — DLQ 

307# ============================================================================ 

308 

309class TestEventBusDLQ: 

310 @pytest.mark.asyncio 

311 async def test_failed_handler_goes_to_dlq(self): 

312 bus = EventBus(dlq_enabled=True) 

313 

314 async def failing_handler(e): 

315 raise ValueError("boom") 

316 

317 bus.subscribe("t", failing_handler, handler_name="h") 

318 await bus.start() 

319 await bus.publish(Event(topic="t")) 

320 await asyncio.sleep(0.2) 

321 await bus.stop() 

322 

323 dlq = bus.get_dlq() 

324 assert len(dlq) == 1 

325 assert dlq[0].handler_name == "h" 

326 assert "boom" in dlq[0].error 

327 

328 @pytest.mark.asyncio 

329 async def test_dlq_disabled(self): 

330 bus = EventBus(dlq_enabled=False) 

331 

332 async def failing_handler(e): 

333 raise ValueError("boom") 

334 

335 bus.subscribe("t", failing_handler, handler_name="h") 

336 await bus.start() 

337 await bus.publish(Event(topic="t")) 

338 await asyncio.sleep(0.2) 

339 await bus.stop() 

340 

341 assert len(bus.get_dlq()) == 0 

342 

343 @pytest.mark.asyncio 

344 async def test_dlq_max_size(self): 

345 bus = EventBus(dlq_enabled=True, dlq_max_size=2) 

346 

347 async def failing_handler(e): 

348 raise ValueError("x") 

349 

350 bus.subscribe("t", failing_handler, handler_name="h") 

351 await bus.start() 

352 for i in range(5): 

353 await bus.publish(Event(topic="t")) 

354 await asyncio.sleep(0.3) 

355 await bus.stop() 

356 

357 dlq = bus.get_dlq() 

358 assert len(dlq) == 2 

359 

360 @pytest.mark.asyncio 

361 async def test_clear_dlq(self): 

362 bus = EventBus(dlq_enabled=True) 

363 

364 async def failing_handler(e): 

365 raise ValueError("x") 

366 

367 bus.subscribe("t", failing_handler, handler_name="h") 

368 await bus.start() 

369 await bus.publish(Event(topic="t")) 

370 await asyncio.sleep(0.2) 

371 await bus.stop() 

372 

373 assert len(bus.get_dlq()) == 1 

374 cleared = bus.clear_dlq() 

375 assert cleared == 1 

376 assert len(bus.get_dlq()) == 0 

377 

378 @pytest.mark.asyncio 

379 async def test_replay_dlq(self): 

380 bus = EventBus(dlq_enabled=True) 

381 

382 async def failing_handler(e): 

383 raise ValueError("x") 

384 

385 bus.subscribe("t", failing_handler, handler_name="h") 

386 await bus.start() 

387 await bus.publish(Event(topic="t")) 

388 await asyncio.sleep(0.2) 

389 await bus.stop() 

390 

391 assert len(bus.get_dlq()) == 1 

392 

393 # Restart bus and replay 

394 await bus.start() 

395 replayed = await bus.replay_dlq() 

396 assert replayed == 1 

397 assert len(bus.get_dlq()) == 0 

398 await bus.stop() 

399 

400 

401# ============================================================================ 

402# EventBus — Lifecycle 

403# ============================================================================ 

404 

405class TestEventBusLifecycle: 

406 @pytest.mark.asyncio 

407 async def test_start_idempotent(self): 

408 bus = EventBus() 

409 await bus.start() 

410 await bus.start() # should not double-start 

411 await bus.stop() 

412 

413 @pytest.mark.asyncio 

414 async def test_stop_drains_queue(self): 

415 bus = EventBus(worker_count=2) 

416 received = [] 

417 

418 async def handler(e): 

419 received.append(e.payload) 

420 await asyncio.sleep(0) # yield to event loop to allow processing 

421 

422 bus.subscribe("t", handler, handler_name="h") 

423 await bus.start() 

424 await bus.publish(Event(topic="t", payload="drain")) 

425 await asyncio.sleep(0.3) 

426 await bus.stop(grace_period=3.0) 

427 assert "drain" in received 

428 

429 

430# ============================================================================ 

431# Default bus 

432# ============================================================================ 

433 

434class TestDefaultBus: 

435 def test_default_bus_exists(self): 

436 assert isinstance(default_bus, EventBus)