Method test_publisher_object

tests/brokers/base/publish.py:321–356

(
        self,
        queue: str,
        mock: MagicMock,
    )

    @pytest.mark.asyncio()
    async def test_publisher_object(
        self,
        queue: str,
        mock: MagicMock,
    ) -> None:
        event = asyncio.Event()

        pub_broker = self.get_broker(apply_types=True)

        publisher = pub_broker.publisher(queue + "resp")

        args, kwargs = self.get_subscriber_params(queue)

        @publisher
        @pub_broker.subscriber(*args, **kwargs)
        async def m() -> str:
            return ""

        args, kwargs = self.get_subscriber_params(queue + "resp")

        @pub_broker.subscriber(*args, **kwargs)
        async def resp(msg) -> None:
            event.set()
            mock(msg)

        async with self.patch_broker(pub_broker) as br:
            await br.start()
            await asyncio.wait(
                (
                    asyncio.create_task(br.publish("", queue)),
                    asyncio.create_task(event.wait()),
                ),
                timeout=self.timeout,
            )

        mock.assert_called_once_with("")

    @pytest.mark.asyncio()
    async def test_publish_manual(