Method test_queue_obj

tests/brokers/rabbit/test_router.py:107–137

(
        self,
        queue: str,
    )

        mock.assert_called_once_with(name="john", id=2)

    async def test_queue_obj(
        self,
        queue: str,
    ) -> None:
        broker = self.get_broker()
        router = self.get_router(prefix="test/")

        r_queue = RabbitQueue(queue)

        event = asyncio.Event()

        @router.subscriber(r_queue)
        def subscriber(m) -> None:
            event.set()

        broker.include_router(router)

        async with broker:
            await broker.start()

            await asyncio.wait(
                (
                    asyncio.create_task(
                        broker.publish("hello", f"test/{r_queue.name}"),
                    ),
                    asyncio.create_task(event.wait()),
                ),
                timeout=3,
            )

            assert event.is_set()

    async def test_queue_obj_with_routing_key(
        self,