Coverage for tests / brokers / base / middlewares.py: 99%

1import asyncio (empty)

2from unittest.mock import MagicMock, call (empty)

3 

4import pytest (empty)

5 

6from faststream import Context (empty)

7from faststream._internal.basic_types import DecodedMessage (empty)

8from faststream.exceptions import SkipMessage (empty)

9from faststream.middlewares import BaseMiddleware, ExceptionMiddleware (empty)

10 

11from .basic import BaseTestcaseConfig (empty)

12 

13 

14@pytest.mark.asyncio() (empty)

15class MiddlewaresOrderTestcase(BaseTestcaseConfig): (empty)

16 async def test_broker_middleware_order(self, queue: str, mock: MagicMock) -> None: (empty)

17 class InnerMiddleware(BaseMiddleware): (empty)

18 async def __aenter__(self) -> None: (empty)

19 mock.enter_inner() (empty)

20 mock.enter("inner") (empty)

21 

22 async def __aexit__(self, *args) -> None: (empty)

23 mock.exit_inner() (empty)

24 mock.exit("inner") (empty)

25 

26 async def consume_scope(self, call_next, msg) -> None: (empty)

27 mock.consume_inner() (empty)

28 mock.sub("inner") (empty)

29 return await call_next(msg) (empty)

30 

31 async def publish_scope(self, call_next, cmd) -> None: (empty)

32 mock.publish_inner() (empty)

33 mock.pub("inner") (empty)

34 return await call_next(cmd) (empty)

35 

36 class OuterMiddleware(BaseMiddleware): (empty)

37 async def __aenter__(self) -> None: (empty)

38 mock.enter_outer() (empty)

39 mock.enter("outer") (empty)

40 

41 async def __aexit__(self, *args) -> None: (empty)

42 mock.exit_outer() (empty)

43 mock.exit("outer") (empty)

44 

45 async def consume_scope(self, call_next, msg) -> None: (empty)

46 mock.consume_outer() (empty)

47 mock.sub("outer") (empty)

48 return await call_next(msg) (empty)

49 

50 async def publish_scope(self, call_next, cmd) -> None: (empty)

51 mock.publish_outer() (empty)

52 mock.pub("outer") (empty)

53 return await call_next(cmd) (empty)

54 

55 broker = self.get_broker(middlewares=[OuterMiddleware, InnerMiddleware]) (empty)

56 

57 args, kwargs = self.get_subscriber_params(queue) (empty)

58 

59 @broker.subscriber(*args, **kwargs) (empty)

60 async def handler(msg): (empty)

61 pass (empty)

62 

63 async with self.patch_broker(broker) as br: (empty)

64 await br.publish(None, queue) (empty)

65 

66 mock.consume_inner.assert_called_once() (empty)

67 mock.consume_outer.assert_called_once() (empty)

68 mock.publish_inner.assert_called_once() (empty)

69 mock.publish_outer.assert_called_once() (empty)

70 mock.enter_inner.assert_called_once() (empty)

71 mock.enter_outer.assert_called_once() (empty)

72 mock.exit_inner.assert_called_once() (empty)

73 mock.exit_outer.assert_called_once() (empty)

74 

75 assert [c.args[0] for c in mock.sub.call_args_list] == ["outer", "inner"] (empty)

76 assert [c.args[0] for c in mock.pub.call_args_list] == ["outer", "inner"] (empty)

77 assert [c.args[0] for c in mock.enter.call_args_list] == ["outer", "inner"] (empty)

78 assert [c.args[0] for c in mock.exit.call_args_list] == ["inner", "outer"] (empty)

79 

80 async def test_publisher_middleware_order( (empty)

81 self, 

82 queue: str, 

83 mock: MagicMock, 

84 ) -> None: 

85 class InnerMiddleware(BaseMiddleware): (empty)

86 async def publish_scope(self, call_next, cmd): (empty)

87 mock.publish_inner() (empty)

88 mock("inner") (empty)

89 return await call_next(cmd) (empty)

90 

91 class MiddleMiddleware(BaseMiddleware): (empty)

92 async def publish_scope(self, call_next, cmd): (empty)

93 mock.publish_middle() (empty)

94 mock("middle") (empty)

95 return await call_next(cmd) (empty)

96 

97 class OuterMiddleware(BaseMiddleware): (empty)

98 async def publish_scope(self, call_next, cmd): (empty)

99 mock.publish_outer() (empty)

100 mock("outer") (empty)

101 return await call_next(cmd) (empty)

102 

103 broker = self.get_broker( (empty)

104 middlewares=[OuterMiddleware, MiddleMiddleware, InnerMiddleware], 

105 ) 

106 publisher = broker.publisher(queue) (empty)

107 

108 args, kwargs = self.get_subscriber_params(queue) (empty)

109 

110 @broker.subscriber(*args, **kwargs) (empty)

111 async def handler(msg): (empty)

112 pass (empty)

113 

114 async with self.patch_broker(broker): (empty)

115 await publisher.publish(None, queue) (empty)

116 

117 mock.publish_inner.assert_called_once() (empty)

118 mock.publish_middle.assert_called_once() (empty)

119 mock.publish_outer.assert_called_once() (empty)

120 

121 assert [c.args[0] for c in mock.call_args_list] == ["outer", "middle", "inner"] (empty)

122 

123 async def test_publisher_with_router_middleware_order( (empty)

124 self, 

125 queue: str, 

126 mock: MagicMock, 

127 ) -> None: 

128 class InnerMiddleware(BaseMiddleware): (empty)

129 async def publish_scope(self, call_next, cmd): (empty)

130 mock.publish_inner() (empty)

131 mock("inner") (empty)

132 return await call_next(cmd) (empty)

133 

134 class MiddleMiddleware(BaseMiddleware): (empty)

135 async def publish_scope(self, call_next, cmd): (empty)

136 mock.publish_middle() (empty)

137 mock("middle") (empty)

138 return await call_next(cmd) (empty)

139 

140 class OuterMiddleware(BaseMiddleware): (empty)

141 async def publish_scope(self, call_next, cmd): (empty)

142 mock.publish_outer() (empty)

143 mock("outer") (empty)

144 return await call_next(cmd) (empty)

145 

146 broker = self.get_broker(middlewares=[OuterMiddleware]) (empty)

147 router = self.get_router(middlewares=[MiddleMiddleware]) (empty)

148 router2 = self.get_router(middlewares=[InnerMiddleware]) (empty)

149 

150 publisher = router2.publisher(queue) (empty)

151 

152 args, kwargs = self.get_subscriber_params(queue) (empty)

153 

154 @router2.subscriber(*args, **kwargs) (empty)

155 async def handler(msg): (empty)

156 pass (empty)

157 

158 router.include_router(router2) (empty)

159 broker.include_router(router) (empty)

160 

161 async with self.patch_broker(broker): (empty)

162 await publisher.publish(None, queue) (empty)

163 

164 mock.publish_inner.assert_called_once() (empty)

165 mock.publish_middle.assert_called_once() (empty)

166 mock.publish_outer.assert_called_once() (empty)

167 

168 assert [c.args[0] for c in mock.call_args_list] == ["outer", "middle", "inner"] (empty)

169 

170 async def test_consume_middleware_order(self, queue: str, mock: MagicMock) -> None: (empty)

171 class InnerMiddleware(BaseMiddleware): (empty)

172 async def consume_scope(self, call_next, cmd): (empty)

173 mock.consume_inner() (empty)

174 mock("inner") (empty)

175 return await call_next(cmd) (empty)

176 

177 class MiddleMiddleware(BaseMiddleware): (empty)

178 async def consume_scope(self, call_next, cmd): (empty)

179 mock.consume_middle() (empty)

180 mock("middle") (empty)

181 return await call_next(cmd) (empty)

182 

183 class OuterMiddleware(BaseMiddleware): (empty)

184 async def consume_scope(self, call_next, cmd): (empty)

185 mock.consume_outer() (empty)

186 mock("outer") (empty)

187 return await call_next(cmd) (empty)

188 

189 broker = self.get_broker( (empty)

190 middlewares=[OuterMiddleware, MiddleMiddleware, InnerMiddleware], 

191 ) 

192 

193 args, kwargs = self.get_subscriber_params(queue) (empty)

194 

195 @broker.subscriber(*args, **kwargs) (empty)

196 async def handler(msg): (empty)

197 pass (empty)

198 

199 async with self.patch_broker(broker) as br: (empty)

200 await br.publish(None, queue) (empty)

201 

202 mock.consume_inner.assert_called_once() (empty)

203 mock.consume_middle.assert_called_once() (empty)

204 mock.consume_outer.assert_called_once() (empty)

205 

206 assert [c.args[0] for c in mock.call_args_list] == ["outer", "middle", "inner"] (empty)

207 

208 async def test_consume_with_router_middleware_order( (empty)

209 self, 

210 queue: str, 

211 mock: MagicMock, 

212 ) -> None: 

213 class InnerMiddleware(BaseMiddleware): (empty)

214 async def consume_scope(self, call_next, cmd): (empty)

215 mock("inner") (empty)

216 return await call_next(cmd) (empty)

217 

218 class MiddleMiddleware(BaseMiddleware): (empty)

219 async def consume_scope(self, call_next, cmd): (empty)

220 mock("middle") (empty)

221 return await call_next(cmd) (empty)

222 

223 class OuterMiddleware(BaseMiddleware): (empty)

224 async def consume_scope(self, call_next, cmd): (empty)

225 mock("outer") (empty)

226 return await call_next(cmd) (empty)

227 

228 broker = self.get_broker(middlewares=[OuterMiddleware]) (empty)

229 router = self.get_router(middlewares=[MiddleMiddleware]) (empty)

230 router2 = self.get_router(middlewares=[InnerMiddleware]) (empty)

231 

232 args, kwargs = self.get_subscriber_params(queue) (empty)

233 

234 @router2.subscriber(*args, **kwargs) (empty)

235 async def handler(msg): (empty)

236 pass (empty)

237 

238 router.include_router(router2) (empty)

239 broker.include_router(router) (empty)

240 async with self.patch_broker(broker) as br: (empty)

241 await br.publish(None, queue) (empty)

242 

243 call_order = [c.args[0] for c in mock.call_args_list] (empty)

244 assert call_order == ["outer", "middle", "inner"], call_order (empty)

245 

246 

247@pytest.mark.asyncio() (empty)

248class LocalMiddlewareTestcase(BaseTestcaseConfig): (empty)

249 async def test_subscriber_middleware( (empty)

250 self, 

251 queue: str, 

252 mock: MagicMock, 

253 ) -> None: 

254 event = asyncio.Event() (empty)

255 

256 class TapMiddleware(BaseMiddleware): (empty)

257 async def consume_scope(self, call_next, msg): (empty)

258 mock.start(await msg.decode()) (empty)

259 result = await call_next(msg) (empty)

260 mock.end() (empty)

261 event.set() (empty)

262 return result (empty)

263 

264 broker = self.get_broker(middlewares=(TapMiddleware,)) (empty)

265 

266 args, kwargs = self.get_subscriber_params(queue) (empty)

267 

268 @broker.subscriber(*args, **kwargs) (empty)

269 async def handler(m) -> str: (empty)

270 mock.inner(m) (empty)

271 return "end" (empty)

272 

273 async with self.patch_broker(broker) as br: (empty)

274 await br.start() (empty)

275 await asyncio.wait( (empty)

276 ( 

277 asyncio.create_task(br.publish("start", queue)), 

278 asyncio.create_task(event.wait()), 

279 ), 

280 timeout=self.timeout, 

281 ) 

282 

283 mock.start.assert_called_once_with("start") (empty)

284 mock.inner.assert_called_once_with("start") (empty)

285 

286 assert event.is_set() (empty)

287 mock.end.assert_called_once() (empty)

288 

289 async def test_error_traceback( (empty)

290 self, 

291 queue: str, 

292 mock: MagicMock, 

293 event: asyncio.Event, 

294 ) -> None: 

295 class ErrorTraceMiddleware(BaseMiddleware): (empty)

296 async def consume_scope(self, call_next, msg): (empty)

297 try: (empty)

298 return await call_next(msg) (empty)

299 except Exception as e: (empty)

300 mock(isinstance(e, ValueError)) (empty)

301 raise (empty)

302 

303 broker = self.get_broker(middlewares=(ErrorTraceMiddleware,)) (empty)

304 

305 args, kwargs = self.get_subscriber_params(queue) (empty)

306 

307 @broker.subscriber(*args, **kwargs) (empty)

308 async def handler2(m): (empty)

309 event.set() (empty)

310 raise ValueError (empty)

311 

312 async with self.patch_broker(broker) as br: (empty)

313 await br.start() (empty)

314 

315 await asyncio.wait( (empty)

316 ( 

317 asyncio.create_task(br.publish("", queue)), 

318 asyncio.create_task(event.wait()), 

319 ), 

320 timeout=self.timeout, 

321 ) 

322 

323 assert event.is_set() (empty)

324 mock.assert_called_once_with(True) (empty)

325 

326 

327@pytest.mark.asyncio() (empty)

328class MiddlewareTestcase(LocalMiddlewareTestcase): (empty)

329 async def test_global_middleware( (empty)

330 self, 

331 queue: str, 

332 mock: MagicMock, 

333 event: asyncio.Event, 

334 ) -> None: 

335 class mid(BaseMiddleware): # noqa: N801 (empty)

336 async def on_receive(self): (empty)

337 mock.start(self.msg) (empty)

338 return await super().on_receive() (empty)

339 

340 async def after_processed(self, exc_type, exc_val, exc_tb): (empty)

341 mock.end() (empty)

342 return await super().after_processed(exc_type, exc_val, exc_tb) (empty)

343 

344 broker = self.get_broker( (empty)

345 middlewares=(mid,), 

346 ) 

347 

348 args, kwargs = self.get_subscriber_params(queue) (empty)

349 

350 @broker.subscriber(*args, **kwargs) (empty)

351 async def handler(m) -> str: (empty)

352 event.set() (empty)

353 return "" (empty)

354 

355 async with self.patch_broker(broker) as br: (empty)

356 await br.start() (empty)

357 await asyncio.wait( (empty)

358 ( 

359 asyncio.create_task(br.publish("", queue)), 

360 asyncio.create_task(event.wait()), 

361 ), 

362 timeout=self.timeout, 

363 ) 

364 assert event.is_set() (empty)

365 

366 mock.start.assert_called_once() (empty)

367 mock.end.assert_called_once() (empty)

368 

369 async def test_add_global_middleware( (empty)

370 self, 

371 queue: str, 

372 mock: MagicMock, 

373 event: asyncio.Event, 

374 ) -> None: 

375 class mid(BaseMiddleware): # noqa: N801 (empty)

376 async def on_receive(self): (empty)

377 mock.start(self.msg) (empty)

378 return await super().on_receive() (empty)

379 

380 async def after_processed(self, exc_type, exc_val, exc_tb): (empty)

381 mock.end() (empty)

382 return await super().after_processed(exc_type, exc_val, exc_tb) (empty)

383 

384 broker = self.get_broker() (empty)

385 

386 # already registered subscriber 

387 args, kwargs = self.get_subscriber_params(queue) (empty)

388 

389 @broker.subscriber(*args, **kwargs) (empty)

390 async def handler(m) -> str: (empty)

391 event.set() (empty)

392 return "" (empty)

393 

394 # should affect to already registered and a new subscriber both 

395 broker.add_middleware(mid) (empty)

396 

397 event2 = asyncio.Event() (empty)

398 

399 # new subscriber 

400 args2, kwargs2 = self.get_subscriber_params(queue + "1") (empty)

401 

402 @broker.subscriber(*args2, **kwargs2) (empty)

403 async def handler2(m) -> str: (empty)

404 event2.set() (empty)

405 return "" (empty)

406 

407 async with self.patch_broker(broker) as br: (empty)

408 await br.start() (empty)

409 await asyncio.wait( (empty)

410 ( 

411 asyncio.create_task(br.publish("", queue)), 

412 asyncio.create_task(br.publish("", f"{queue}1")), 

413 asyncio.create_task(event.wait()), 

414 asyncio.create_task(event2.wait()), 

415 ), 

416 timeout=self.timeout, 

417 ) 

418 

419 assert event.is_set() (empty)

420 assert mock.start.call_count == 2, mock.start.call_count (empty)

421 assert mock.end.call_count == 2, mock.end.call_count (empty)

422 

423 async def test_patch_publish( (empty)

424 self, 

425 queue: str, 

426 mock: MagicMock, 

427 event: asyncio.Event, 

428 ) -> None: 

429 class Mid(BaseMiddleware): (empty)

430 async def publish_scope(self, call_next, cmd): (empty)

431 cmd.body *= 2 (empty)

432 return await call_next(cmd) (empty)

433 

434 broker = self.get_broker(middlewares=(Mid,)) (empty)

435 

436 args, kwargs = self.get_subscriber_params(queue) (empty)

437 

438 @broker.subscriber(*args, **kwargs) (empty)

439 async def handler(m): (empty)

440 return m (empty)

441 

442 args2, kwargs2 = self.get_subscriber_params(queue + "r") (empty)

443 

444 @broker.subscriber(*args2, **kwargs2) (empty)

445 async def handler_resp(m) -> None: (empty)

446 mock(m) (empty)

447 event.set() (empty)

448 

449 async with self.patch_broker(broker) as br: (empty)

450 await br.start() (empty)

451 

452 await asyncio.wait( (empty)

453 ( 

454 asyncio.create_task(br.publish("r", queue, reply_to=queue + "r")), 

455 asyncio.create_task(event.wait()), 

456 ), 

457 timeout=self.timeout, 

458 ) 

459 

460 assert event.is_set() (empty)

461 mock.assert_called_once_with("rrrr") (empty)

462 

463 async def test_global_publisher_middleware( (empty)

464 self, 

465 queue: str, 

466 mock: MagicMock, 

467 event: asyncio.Event, 

468 ) -> None: 

469 class Mid(BaseMiddleware): (empty)

470 async def publish_scope(self, call_next, cmd): (empty)

471 cmd.body *= 2 (empty)

472 mock.enter(cmd.body) (empty)

473 try: (empty)

474 return await call_next(cmd) (empty)

475 finally: 

476 mock.end() (empty)

477 if mock.end.call_count > 2: (empty)

478 event.set() (empty)

479 

480 broker = self.get_broker(middlewares=(Mid,)) (empty)

481 

482 args, kwargs = self.get_subscriber_params(queue) (empty)

483 

484 @broker.subscriber(*args, **kwargs) (empty)

485 @broker.publisher(queue + "1") (empty)

486 @broker.publisher(queue + "2") (empty)

487 async def handler(m): (empty)

488 mock.inner(m) (empty)

489 return m (empty)

490 

491 async with self.patch_broker(broker) as br: (empty)

492 await br.start() (empty)

493 await asyncio.wait( (empty)

494 ( 

495 asyncio.create_task(br.publish("1", queue)), 

496 asyncio.create_task(event.wait()), 

497 ), 

498 timeout=self.timeout, 

499 ) 

500 

501 assert event.is_set() (empty)

502 mock.inner.assert_called_once_with("11") (empty)

503 assert mock.enter.call_count == 3 (empty)

504 mock.enter.assert_called_with("1111") (empty)

505 assert mock.end.call_count == 3 (empty)

506 

507 

508@pytest.mark.asyncio() (empty)

509class ExceptionMiddlewareTestcase(BaseTestcaseConfig): (empty)

510 async def test_exception_middleware_default_msg( (empty)

511 self, 

512 queue: str, 

513 mock: MagicMock, 

514 event: asyncio.Event, 

515 ) -> None: 

516 mid = ExceptionMiddleware() (empty)

517 

518 @mid.add_handler(ValueError, publish=True) (empty)

519 async def value_error_handler(exc) -> str: (empty)

520 return "value" (empty)

521 

522 broker = self.get_broker(apply_types=True, middlewares=(mid,)) (empty)

523 

524 args, kwargs = self.get_subscriber_params(queue) (empty)

525 

526 @broker.subscriber(*args, **kwargs) (empty)

527 @broker.publisher(queue + "1") (empty)

528 async def subscriber1(m): (empty)

529 raise ValueError (empty)

530 

531 args, kwargs = self.get_subscriber_params(queue + "1") (empty)

532 

533 @broker.subscriber(*args, **kwargs) (empty)

534 async def subscriber2(msg=Context("message")) -> None: (empty)

535 mock(await msg.decode()) (empty)

536 event.set() (empty)

537 

538 async with self.patch_broker(broker) as br: (empty)

539 await br.start() (empty)

540 await asyncio.wait( (empty)

541 ( 

542 asyncio.create_task(br.publish("", queue)), 

543 asyncio.create_task(event.wait()), 

544 ), 

545 timeout=self.timeout, 

546 ) 

547 

548 assert event.is_set() (empty)

549 assert mock.call_count == 1 (empty)

550 mock.assert_called_once_with("value") (empty)

551 

552 async def test_exception_middleware_skip_msg( (empty)

553 self, 

554 queue: str, 

555 mock: MagicMock, 

556 event: asyncio.Event, 

557 ) -> None: 

558 mid = ExceptionMiddleware() (empty)

559 

560 @mid.add_handler(ValueError, publish=True) (empty)

561 async def value_error_handler(exc): (empty)

562 event.set() (empty)

563 raise SkipMessage (empty)

564 

565 broker = self.get_broker(middlewares=(mid,)) (empty)

566 args, kwargs = self.get_subscriber_params(queue) (empty)

567 

568 @broker.subscriber(*args, **kwargs) (empty)

569 @broker.publisher(queue + "1") (empty)

570 async def subscriber1(m): (empty)

571 raise ValueError (empty)

572 

573 args2, kwargs2 = self.get_subscriber_params(queue + "1") (empty)

574 

575 @broker.subscriber(*args2, **kwargs2) (empty)

576 async def subscriber2(msg=Context("message")) -> None: (empty)

577 mock(await msg.decode()) 

578 

579 async with self.patch_broker(broker) as br: (empty)

580 await br.start() (empty)

581 await asyncio.wait( (empty)

582 ( 

583 asyncio.create_task(br.publish("", queue)), 

584 asyncio.create_task(event.wait()), 

585 ), 

586 timeout=self.timeout, 

587 ) 

588 

589 assert event.is_set() (empty)

590 assert mock.call_count == 0 (empty)

591 

592 async def test_exception_middleware_do_not_catch_skip_msg( (empty)

593 self, 

594 queue: str, 

595 mock: MagicMock, 

596 event: asyncio.Event, 

597 ) -> None: 

598 mid = ExceptionMiddleware() (empty)

599 

600 @mid.add_handler(Exception) (empty)

601 async def value_error_handler(exc) -> None: (empty)

602 mock() 

603 

604 broker = self.get_broker(middlewares=(mid,)) (empty)

605 args, kwargs = self.get_subscriber_params(queue) (empty)

606 

607 @broker.subscriber(*args, **kwargs) (empty)

608 async def subscriber(m): (empty)

609 event.set() (empty)

610 raise SkipMessage (empty)

611 

612 async with self.patch_broker(broker) as br: (empty)

613 await br.start() (empty)

614 await asyncio.wait( (empty)

615 ( 

616 asyncio.create_task(br.publish("", queue)), 

617 asyncio.create_task(event.wait()), 

618 ), 

619 timeout=self.timeout, 

620 ) 

621 await asyncio.sleep(0.001) (empty)

622 

623 assert event.is_set() (empty)

624 assert mock.call_count == 0 (empty)

625 

626 async def test_exception_middleware_reraise( (empty)

627 self, 

628 queue: str, 

629 mock: MagicMock, 

630 event: asyncio.Event, 

631 ) -> None: 

632 mid = ExceptionMiddleware() (empty)

633 

634 @mid.add_handler(ValueError, publish=True) (empty)

635 async def value_error_handler(exc): (empty)

636 event.set() (empty)

637 raise exc (empty)

638 

639 broker = self.get_broker(middlewares=(mid,)) (empty)

640 args, kwargs = self.get_subscriber_params(queue) (empty)

641 

642 @broker.subscriber(*args, **kwargs) (empty)

643 @broker.publisher(queue + "1") (empty)

644 async def subscriber1(m): (empty)

645 raise ValueError (empty)

646 

647 args2, kwargs2 = self.get_subscriber_params(queue + "1") (empty)

648 

649 @broker.subscriber(*args2, **kwargs2) (empty)

650 async def subscriber2(msg=Context("message")) -> None: (empty)

651 mock(await msg.decode()) 

652 

653 async with self.patch_broker(broker) as br: (empty)

654 await br.start() (empty)

655 await asyncio.wait( (empty)

656 ( 

657 asyncio.create_task(br.publish("", queue)), 

658 asyncio.create_task(event.wait()), 

659 ), 

660 timeout=self.timeout, 

661 ) 

662 

663 assert event.is_set() (empty)

664 assert mock.call_count == 0 (empty)

665 

666 async def test_exception_middleware_different_handler( (empty)

667 self, 

668 queue: str, 

669 mock: MagicMock, 

670 event: asyncio.Event, 

671 ) -> None: 

672 mid = ExceptionMiddleware() (empty)

673 

674 @mid.add_handler(ZeroDivisionError, publish=True) (empty)

675 async def zero_error_handler(exc) -> str: (empty)

676 return "zero" (empty)

677 

678 @mid.add_handler(ValueError, publish=True) (empty)

679 async def value_error_handler(exc) -> str: (empty)

680 return "value" (empty)

681 

682 broker = self.get_broker(apply_types=True, middlewares=(mid,)) (empty)

683 args, kwargs = self.get_subscriber_params(queue) (empty)

684 

685 publisher = broker.publisher(queue + "2") (empty)

686 

687 @broker.subscriber(*args, **kwargs) (empty)

688 @publisher (empty)

689 async def subscriber1(m): (empty)

690 raise ZeroDivisionError (empty)

691 

692 args2, kwargs2 = self.get_subscriber_params(queue + "1") (empty)

693 

694 @broker.subscriber(*args2, **kwargs2) (empty)

695 @publisher (empty)

696 async def subscriber2(m): (empty)

697 raise ValueError (empty)

698 

699 args3, kwargs3 = self.get_subscriber_params(queue + "2") (empty)

700 

701 @broker.subscriber(*args3, **kwargs3) (empty)

702 async def subscriber3(msg=Context("message")) -> None: (empty)

703 mock(await msg.decode()) (empty)

704 if mock.call_count > 1: (empty)

705 event.set() (empty)

706 

707 async with self.patch_broker(broker) as br: (empty)

708 await br.start() (empty)

709 await asyncio.wait( (empty)

710 ( 

711 asyncio.create_task(br.publish("", queue)), 

712 asyncio.create_task(br.publish("", queue + "1")), 

713 asyncio.create_task(event.wait()), 

714 ), 

715 timeout=self.timeout, 

716 ) 

717