Method assert_metrics

tests/prometheus/basic.py:138–283

(
        self,
        *,
        registry: CollectorRegistry,
        message: Any,
        exception_class: type[Exception] | None,
        custom_labels: dict[str, str],
    )

        )

    def assert_metrics(
        self,
        *,
        registry: CollectorRegistry,
        message: Any,
        exception_class: type[Exception] | None,
        custom_labels: dict[str, str],
    ) -> None:
        settings_provider = self.get_settings_provider()
        consume_attrs = settings_provider.get_consume_attrs_from_message(message)

        received_messages_metric = get_received_messages_metric(
            metrics_prefix="faststream",
            app_name="faststream",
            broker=settings_provider.messaging_system,
            queue=consume_attrs["destination_name"],
            messages_amount=consume_attrs["messages_count"],
            custom_labels=custom_labels,
        )

        received_messages_size_bytes_metric = get_received_messages_size_bytes_metric(
            metrics_prefix="faststream",
            app_name="faststream",
            broker=settings_provider.messaging_system,
            queue=consume_attrs["destination_name"],
            buckets=(
                2.0**4,
                2.0**6,
                2.0**8,
                2.0**10,
                2.0**12,
                2.0**14,
                2.0**16,
                2.0**18,
                2.0**20,
                2.0**22,
                2.0**24,
                float("inf"),
            ),
            size=consume_attrs["message_size"],
            messages_amount=1,
            custom_labels=custom_labels,
        )

        received_messages_in_process_metric = get_received_messages_in_process_metric(
            metrics_prefix="faststream",
            app_name="faststream",
            broker=settings_provider.messaging_system,
            queue=consume_attrs["destination_name"],
            messages_amount=0,
            custom_labels=custom_labels,
        )

        received_processed_messages_duration_seconds_metric = (
            get_received_processed_messages_duration_seconds_metric(
                metrics_prefix="faststream",
                app_name="faststream",
                broker=settings_provider.messaging_system,