(self, queue: str)
| 704 | m.mock.assert_called_once_with("hello") |
| 705 | |
| 706 | async def test_publisher_mock(self, queue: str) -> None: |
| 707 | router = self.router_class() |
| 708 | |
| 709 | publisher = router.publisher(queue + "resp") |
| 710 | |
| 711 | args, kwargs = self.get_subscriber_params(queue) |
| 712 | sub = router.subscriber(*args, **kwargs) |
| 713 | |
| 714 | @publisher |
| 715 | @sub |
| 716 | async def m() -> str: |
| 717 | return "response" |
| 718 | |
| 719 | async with self.patch_broker(router.broker) as rb: |
| 720 | await rb.publish("hello", queue) |
| 721 | publisher.mock.assert_called_with("response") |
| 722 | |
| 723 | async def test_include(self, queue: str) -> None: |
| 724 | router = self.router_class() |
nothing calls this directly
no test coverage detected