Method test_rpc_request

tests/prometheus/basic.py:288–325

(
        self,
        queue: str,
    )

class LocalRPCPrometheusTestcase:
    @pytest.mark.asyncio()
    async def test_rpc_request(
        self,
        queue: str,
    ) -> None:
        event = asyncio.Event()
        registry = CollectorRegistry()

        middleware = self.get_middleware(registry=registry)

        broker = self.get_broker(apply_types=True, middlewares=(middleware,))

        message = None

        @broker.subscriber(queue)
        async def handle(m=Context("message")):
            event.set()

            nonlocal message
            message = m

            return ""

        async with self.patch_broker(broker) as br:
            await br.start()

            await asyncio.wait_for(
                br.request("", queue),
                timeout=3,
            )

        assert event.is_set()

        self.assert_metrics(
            registry=registry,
            message=message,
            exception_class=None,
            custom_labels={},
        )


class LocalMetricsSettingsProviderTestcase: