litellm/tests/unit/integrations/test_prometheus_labels.py

"""
Unit tests for prometheus metric labels configuration
"""

from typing import Final, Literal

import pytest

import litellm
from litellm.types.integrations.prometheus import (
    PrometheusMetricLabels,
    UserAPIKeyLabelNames,
)


def _clear_prometheus_registry() -> None:
    from prometheus_client import REGISTRY

    for collector in list(REGISTRY._collector_to_names.keys()):
        try:
            REGISTRY.unregister(collector)
        except Exception:
            pass


def _collected_samples(metric_name: str):
    from prometheus_client import REGISTRY

    return [
        sample
        for metric in REGISTRY.collect()
        for sample in metric.samples
        if sample.name == metric_name
    ]


def test_user_email_in_required_metrics():
    """
    Test that user_email label is present in all the metrics that should have it:
    - litellm_proxy_total_requests_metric (already had it)
    - litellm_proxy_failed_requests_metric (added)
    - litellm_input_tokens_metric (added)
    - litellm_output_tokens_metric (added)
    - litellm_requests_metric (already had it)
    - litellm_spend_metric (added)
    """
    user_email_label = UserAPIKeyLabelNames.USER_EMAIL.value

    # Metrics that should have user_email
    metrics_with_user_email = [
        "litellm_proxy_total_requests_metric",
        "litellm_proxy_failed_requests_metric",
        "litellm_input_tokens_metric",
        "litellm_output_tokens_metric",
        "litellm_requests_metric",
        "litellm_spend_metric",
    ]

    for metric_name in metrics_with_user_email:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            user_email_label in labels
        ), f"Metric {metric_name} should contain user_email label"
        print(f"✅ {metric_name} contains user_email label")


def test_model_id_in_extended_metric_set():
    """
    Test that model_id label is present in all the metrics that should have it
    """
    model_id_label = UserAPIKeyLabelNames.MODEL_ID.value

    # Metrics that should have model_id
    metrics_with_model_id = [
        "litellm_proxy_total_requests_metric",
        "litellm_proxy_failed_requests_metric",
        "litellm_input_tokens_metric",
        "litellm_output_tokens_metric",
        "litellm_requests_metric",
        "litellm_spend_metric",
        "litellm_llm_api_latency_metric",
        "litellm_remaining_requests_metric",
        "litellm_deployment_successful_fallbacks",
        "litellm_cache_hits_metric",
        "litellm_cache_misses_metric",
        "litellm_remaining_api_key_requests_for_model",
        "litellm_remaining_api_key_tokens_for_model",
        "litellm_llm_api_failed_requests_metric",
    ]

    for metric_name in metrics_with_model_id:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            model_id_label in labels
        ), f"Metric {metric_name} should contain model_id label"
        print(f"✅ {metric_name} contains model_id label")


def test_api_provider_in_spend_and_requests_metrics():
    """
    Test that api_provider label is present in spend and requests metrics
    so users can build spend-by-provider and request-count-by-provider dashboards.
    """
    api_provider_label = UserAPIKeyLabelNames.API_PROVIDER.value

    for metric_name in ["litellm_spend_metric", "litellm_requests_metric"]:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            api_provider_label in labels
        ), f"Metric {metric_name} should contain api_provider label"
        print(f"✅ {metric_name} contains api_provider label")


def test_api_provider_in_token_latency_and_request_metrics():
    """
    Regression test for LIT-4178.

    These metrics are all emitted from the same call site (async_log_success_event
    / async_post_call_failure_hook) as litellm_spend_metric and
    litellm_requests_metric, which already carry api_provider. They were the odd
    ones out with no way to break down tokens, latency, request counts or cache
    hits by upstream provider, even though the provider is available on the
    payload as custom_llm_provider.
    """
    api_provider_label = UserAPIKeyLabelNames.API_PROVIDER.value

    metrics_with_api_provider = [
        "litellm_llm_api_latency_metric",
        "litellm_llm_api_time_to_first_token_metric",
        "litellm_request_total_latency_metric",
        "litellm_request_queue_time_seconds",
        "litellm_proxy_total_requests_metric",
        "litellm_proxy_failed_requests_metric",
        "litellm_input_tokens_metric",
        "litellm_total_tokens_metric",
        "litellm_output_tokens_metric",
        "litellm_cache_hits_metric",
        "litellm_cache_misses_metric",
        # The remaining cache metrics share _cache_metric_labels, so they pick up
        # api_provider from the same change. Assert them explicitly so the shared
        # list can't silently drop the label from them.
        "litellm_cached_tokens_metric",
        "litellm_provider_cache_read_input_tokens_metric",
        "litellm_provider_cache_creation_input_tokens_metric",
    ]

    for metric_name in metrics_with_api_provider:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            api_provider_label in labels
        ), f"Metric {metric_name} should contain api_provider label"


def test_api_provider_value_flows_through_label_factory():
    """
    The label being in the allow-list is necessary but not sufficient: the
    factory must also carry the value from the enum through to the emitted
    label. This would fail if the label were dropped from the metric's list or
    if the value plumbing regressed, which the allow-list assertion above cannot
    catch on its own.
    """
    from unittest.mock import MagicMock

    from litellm.integrations.prometheus import (
        PrometheusLogger,
        UserAPIKeyLabelValues,
        prometheus_label_factory,
    )

    prometheus_logger = MagicMock()
    prometheus_logger._cached_metric_labels = {}
    prometheus_logger.label_filters = {}
    prometheus_logger.get_labels_for_metric = (
        PrometheusLogger.get_labels_for_metric.__get__(prometheus_logger)
    )

    enum_values = UserAPIKeyLabelValues(
        api_provider="anthropic",
        litellm_model_name="claude-sonnet-4",
        requested_model="claude",
        status_code="200",
    )

    for metric_name in [
        "litellm_input_tokens_metric",
        "litellm_total_tokens_metric",
        "litellm_output_tokens_metric",
        "litellm_llm_api_latency_metric",
        "litellm_request_total_latency_metric",
        "litellm_proxy_total_requests_metric",
        "litellm_cache_hits_metric",
    ]:
        labels = prometheus_label_factory(
            supported_enum_labels=prometheus_logger.get_labels_for_metric(
                metric_name=metric_name
            ),
            enum_values=enum_values,
        )
        assert (
            labels.get("api_provider") == "anthropic"
        ), f"{metric_name} should emit api_provider=anthropic, got {labels.get('api_provider')!r}"


def test_extract_api_provider_from_request_data_failure_path():
    """
    On the client-side failure path the provider is not always known. Prefer the
    resolved custom_llm_provider on litellm_params, fall back to a partial
    standard_logging_object (e.g. a stream that broke mid-flight), then infer it
    from the requested model name, and return None only when nothing maps so the
    label emits empty rather than a guess.
    """
    from litellm.integrations.prometheus import PrometheusLogger

    extract = PrometheusLogger._extract_api_provider_from_request_data

    assert extract({"litellm_params": {"custom_llm_provider": "bedrock"}}) == "bedrock"
    assert (
        extract({"standard_logging_object": {"custom_llm_provider": "vertex_ai"}})
        == "vertex_ai"
    )
    # litellm_params wins over standard_logging_object when both are present
    assert (
        extract(
            {
                "litellm_params": {"custom_llm_provider": "openai"},
                "standard_logging_object": {"custom_llm_provider": "azure"},
            }
        )
        == "openai"
    )
    # Fallback: infer provider from the requested model name when the proxy's
    # failure request_data carries only the client-supplied model. This is the
    # common client-side failure case (e.g. an invalid param rejected with 400).
    assert extract({"model": "gpt-4o-mini"}) == "openai"
    assert extract({"litellm_params": {"model": "anthropic/claude-haiku-4-5"}}) == "anthropic"
    assert extract({}) is None
    # Unmappable model name -> None, not a guess
    assert extract({"model": "some-unknown-model-xyz"}) is None


_PASS_THROUGH_CALL_TYPE: Final = "pass_through_endpoint"


@pytest.mark.parametrize(
    "request_data, expected_provider",
    [
        ({"model": "gpt-4o-mini", "call_type": "acompletion", "custom_llm_provider": "anthropic"}, "openai"),
        ({"model": "gpt-4o-mini", "litellm_params": {"custom_llm_provider": "humanloop"}}, "humanloop"),
        ({"model": "gemini-3.8-flash", "call_type": _PASS_THROUGH_CALL_TYPE}, "vertex_ai"),
        (
            {"model": "gemini-3.8-flash", "call_type": _PASS_THROUGH_CALL_TYPE, "custom_llm_provider": "gemini"},
            "gemini",
        ),
        (
            {
                "call_type": _PASS_THROUGH_CALL_TYPE,
                "custom_llm_provider": "gemini",
                "standard_logging_object": {"custom_llm_provider": "vertex_ai"},
            },
            "vertex_ai",
        ),
    ],
)
def test_extract_api_provider_reads_the_request_declared_provider_only_on_pass_through_failures(
    request_data: dict[str, object], expected_provider: str
):
    from litellm.integrations.prometheus import PrometheusLogger

    assert PrometheusLogger._extract_api_provider_from_request_data(request_data) == expected_provider


@pytest.mark.parametrize(
    "declared_provider",
    ["azure", "bedrock", "openrouter", "client-supplied-value", "", {"nested": "object"}, 42],
)
def test_extract_api_provider_ignores_pass_through_declared_values_outside_gemini_and_vertex(
    declared_provider: object,
):
    from litellm.integrations.prometheus import PrometheusLogger

    extract = PrometheusLogger._extract_api_provider_from_request_data
    declared = {"call_type": _PASS_THROUGH_CALL_TYPE, "custom_llm_provider": declared_provider}

    assert extract({"model": "gpt-4o-mini", **declared}) == "openai"
    assert extract({"model": "no-such-model", **declared}) is None


def test_extract_api_provider_swallows_unknown_model_but_logs_unexpected_errors():
    """
    get_llm_provider raises BadRequestError for a model that maps to no
    provider; that is the expected miss and must resolve to None quietly. Any
    other exception is unexpected (e.g. a real bug) and must not vanish
    silently, nor break metric emission: it is logged and still returns None.
    """
    from unittest.mock import patch

    import litellm
    from litellm.integrations.prometheus import PrometheusLogger

    extract = PrometheusLogger._extract_api_provider_from_request_data

    # Expected miss: BadRequestError -> None, no log noise
    with patch.object(
        litellm,
        "get_llm_provider",
        side_effect=litellm.exceptions.BadRequestError(
            message="no provider", model="x", llm_provider="y"
        ),
    ):
        with patch("litellm.integrations.prometheus.verbose_logger") as mock_logger:
            assert extract({"model": "x"}) is None
            mock_logger.debug.assert_not_called()

    # Unexpected error: must be logged and still return None (never raised)
    with patch.object(litellm, "get_llm_provider", side_effect=RuntimeError("boom")):
        with patch("litellm.integrations.prometheus.verbose_logger") as mock_logger:
            assert extract({"model": "x"}) is None
            mock_logger.debug.assert_called_once()


def test_user_email_label_exists():
    """Test that the USER_EMAIL label is properly defined"""
    assert UserAPIKeyLabelNames.USER_EMAIL.value == "user_email"


def test_prometheus_metric_labels_structure():
    """Test that all required prometheus metrics have proper label structure"""
    from typing import get_args

    from litellm.types.integrations.prometheus import DEFINED_PROMETHEUS_METRICS

    # Test a few key metrics to ensure they have proper label structure
    test_metrics = [
        "litellm_proxy_total_requests_metric",
        "litellm_proxy_failed_requests_metric",
        "litellm_input_tokens_metric",
        "litellm_output_tokens_metric",
        "litellm_spend_metric",
    ]

    for metric_name in test_metrics:
        # Check metric is in DEFINED_PROMETHEUS_METRICS
        assert metric_name in get_args(
            DEFINED_PROMETHEUS_METRICS
        ), f"{metric_name} should be in DEFINED_PROMETHEUS_METRICS"

        # Check labels can be retrieved
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert isinstance(labels, list), f"Labels for {metric_name} should be a list"
        assert len(labels) > 0, f"Labels for {metric_name} should not be empty"

        # Check user_email is in the labels
        assert "user_email" in labels, f"{metric_name} should have user_email label"

        print(f"✅ {metric_name} has proper label structure with user_email")


def test_model_id_in_required_metrics():
    """
    Test that model_id label is present in all the metrics that should have it:
    - litellm_proxy_total_requests_metric
    - litellm_proxy_failed_requests_metric
    - litellm_request_total_latency_metric
    - litellm_llm_api_time_to_first_token_metric
    """
    model_id_label = UserAPIKeyLabelNames.MODEL_ID.value

    # Metrics that should have model_id
    metrics_with_model_id = [
        "litellm_proxy_total_requests_metric",
        "litellm_proxy_failed_requests_metric",
        "litellm_request_total_latency_metric",
        "litellm_llm_api_time_to_first_token_metric",
    ]

    for metric_name in metrics_with_model_id:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            model_id_label in labels
        ), f"Metric {metric_name} should contain model_id label"
        print(f"✅ {metric_name} contains model_id label")


def test_requested_model_in_spend_and_requests_metrics():
    """
    Regression test for LIT-3796.

    litellm_spend_metric and litellm_requests_metric must expose the
    requested_model label so spend and request counts can be grouped by the
    model alias the caller asked for, not just the backend deployment that
    served the request. The sibling token metrics (input/output/total)
    already carry this label; spend and requests were the odd ones out, even
    though both are emitted side-by-side from the same call site.
    """
    requested_model_label = UserAPIKeyLabelNames.REQUESTED_MODEL.value

    metrics_with_requested_model = [
        "litellm_spend_metric",
        "litellm_requests_metric",
        "litellm_input_tokens_metric",
        "litellm_output_tokens_metric",
        "litellm_total_tokens_metric",
    ]

    for metric_name in metrics_with_requested_model:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert requested_model_label in labels, (
            f"Metric {metric_name} should contain requested_model label"
        )


def test_route_normalization_for_responses_api():
    """
    Test that route normalization prevents high cardinality in Prometheus metrics
    for the /v1/responses/{response_id} endpoint.

    Issue: https://github.com/BerriAI/litellm/issues/XXXX
    Each unique response ID was creating a separate metric line, causing the
    /metrics endpoint to grow to ~30MB and take ~40 seconds to respond.

    Fix: Routes are normalized to collapse dynamic IDs into placeholders.
    """
    from litellm.proxy.auth.auth_utils import normalize_request_route

    # Test responses API routes
    responses_routes = [
        ("/v1/responses/1234567890", "/v1/responses/{response_id}"),
        ("/v1/responses/9876543210", "/v1/responses/{response_id}"),
        ("/v1/responses/abcdefghij", "/v1/responses/{response_id}"),
        ("/v1/responses/resp_abc123", "/v1/responses/{response_id}"),
        ("/v1/responses/litellm_poll_xyz", "/v1/responses/{response_id}"),
    ]

    for original, expected in responses_routes:
        normalized = normalize_request_route(original)
        assert (
            normalized == expected
        ), f"Failed: {original} -> {normalized} (expected {expected})"

    # Verify cardinality reduction
    unique_normalized = set(
        normalize_request_route(route) for route, _ in responses_routes
    )
    assert (
        len(unique_normalized) == 1
    ), f"Expected 1 unique normalized route, got {len(unique_normalized)}: {unique_normalized}"

    print(
        f"✅ Responses API routes: {len(responses_routes)} different IDs normalized to 1 metric label"
    )


def test_route_normalization_for_sub_routes():
    """Test that sub-routes like /cancel and /input_items are normalized correctly"""
    from litellm.proxy.auth.auth_utils import normalize_request_route

    sub_routes = [
        ("/v1/responses/id1/cancel", "/v1/responses/{response_id}/cancel"),
        ("/v1/responses/id2/cancel", "/v1/responses/{response_id}/cancel"),
        ("/v1/responses/id3/input_items", "/v1/responses/{response_id}/input_items"),
        (
            "/openai/v1/responses/id4/input_items",
            "/openai/v1/responses/{response_id}/input_items",
        ),
    ]

    for original, expected in sub_routes:
        normalized = normalize_request_route(original)
        assert (
            normalized == expected
        ), f"Failed: {original} -> {normalized} (expected {expected})"

    print("✅ Sub-routes normalized correctly")


def test_route_normalization_preserves_static_routes():
    """Test that static routes are not affected by normalization"""
    from litellm.proxy.auth.auth_utils import normalize_request_route

    static_routes = [
        "/chat/completions",
        "/v1/chat/completions",
        "/v1/embeddings",
        "/health",
        "/metrics",
        "/v1/models",
        "/v1/responses",  # List endpoint without ID
    ]

    for route in static_routes:
        normalized = normalize_request_route(route)
        assert (
            normalized == route
        ), f"Static route should not be modified: {route} -> {normalized}"

    print(f"✅ {len(static_routes)} static routes preserved")


def test_route_normalization_other_dynamic_apis():
    """Test normalization for other OpenAI-compatible APIs with dynamic IDs"""
    from litellm.proxy.auth.auth_utils import normalize_request_route

    test_cases = [
        # Threads API
        ("/v1/threads/thread_123", "/v1/threads/{thread_id}"),
        ("/v1/threads/thread_abc/messages", "/v1/threads/{thread_id}/messages"),
        (
            "/v1/threads/thread_abc/runs/run_123",
            "/v1/threads/{thread_id}/runs/{run_id}",
        ),
        # Vector Stores API
        ("/v1/vector_stores/vs_123", "/v1/vector_stores/{vector_store_id}"),
        ("/v1/vector_stores/vs_123/files", "/v1/vector_stores/{vector_store_id}/files"),
        # Assistants API
        ("/v1/assistants/asst_123", "/v1/assistants/{assistant_id}"),
        # Files API
        ("/v1/files/file_123", "/v1/files/{file_id}"),
        ("/v1/files/file_123/content", "/v1/files/{file_id}/content"),
        # Batches API
        ("/v1/batches/batch_123", "/v1/batches/{batch_id}"),
        ("/v1/batches/batch_123/cancel", "/v1/batches/{batch_id}/cancel"),
    ]

    for original, expected in test_cases:
        normalized = normalize_request_route(original)
        assert (
            normalized == expected
        ), f"Failed: {original} -> {normalized} (expected {expected})"

    print(f"✅ {len(test_cases)} other API routes normalized correctly")


def test_prometheus_metrics_use_normalized_routes():
    """
    Test that Prometheus metrics use the normalized route in labels
    to prevent high cardinality.
    """
    from unittest.mock import MagicMock

    from litellm.integrations.prometheus import (
        PrometheusLogger,
        UserAPIKeyLabelValues,
        prometheus_label_factory,
    )

    # Create a mock PrometheusLogger
    prometheus_logger = MagicMock()
    # ``get_labels_for_metric`` reads ``_cached_metric_labels`` and
    # ``label_filters`` off ``self``; default MagicMock attribute access
    # returns Mocks that masquerade as a populated cache, so seed real
    # containers before binding the real method.
    prometheus_logger._cached_metric_labels = {}
    prometheus_logger.label_filters = {}
    prometheus_logger.get_labels_for_metric = (
        PrometheusLogger.get_labels_for_metric.__get__(prometheus_logger)
    )

    # Test with a normalized route
    enum_values = UserAPIKeyLabelValues(
        route="/v1/responses/{response_id}",  # Normalized route
        status_code="200",
        requested_model="gpt-4",
    )

    labels = prometheus_label_factory(
        supported_enum_labels=prometheus_logger.get_labels_for_metric(
            metric_name="litellm_proxy_total_requests_metric"
        ),
        enum_values=enum_values,
    )

    # Verify the route is normalized in labels
    assert (
        labels["route"] == "/v1/responses/{response_id}"
    ), f"Expected normalized route in labels, got: {labels.get('route')}"

    print("✅ Prometheus metrics use normalized routes in labels")


def test_prometheus_label_value_sanitization():
    """
    Test that Prometheus label values are sanitized to prevent breaking
    the Prometheus text format.

    Issue: Unicode Line Separator (U+2028) in label values (e.g. from a
    malformed model name) breaks the Prometheus exposition format, causing
    scrapers like Datadog to fail parsing the entire /metrics endpoint.
    """
    from litellm.integrations.prometheus import (
        PrometheusLogger,
        UserAPIKeyLabelValues,
        prometheus_label_factory,
    )
    from unittest.mock import MagicMock

    prometheus_logger = MagicMock()
    prometheus_logger._cached_metric_labels = {}
    prometheus_logger.label_filters = {}
    prometheus_logger.get_labels_for_metric = (
        PrometheusLogger.get_labels_for_metric.__get__(prometheus_logger)
    )

    # Simulate a model name with U+2028 (Unicode Line Separator) appended
    # and an api_key_alias with newlines and quotes
    enum_values = UserAPIKeyLabelValues(
        requested_model="claude-haiku-4-5-20251001\u2028",
        api_key_alias='My Key "test"\nwith newline',
        route="/v1/chat/completions",
        status_code="400",
    )

    labels = prometheus_label_factory(
        supported_enum_labels=prometheus_logger.get_labels_for_metric(
            metric_name="litellm_proxy_total_requests_metric"
        ),
        enum_values=enum_values,
    )

    # U+2028 must be stripped
    assert (
        "\u2028" not in labels["requested_model"]
    ), f"U+2028 should be removed from label value, got: {repr(labels['requested_model'])}"
    assert labels["requested_model"] == "claude-haiku-4-5-20251001"

    # Newlines must be replaced with spaces, quotes must be escaped
    assert "\n" not in labels["api_key_alias"]
    assert labels["api_key_alias"] == 'My Key \\"test\\" with newline'

    print("✅ Prometheus label values are properly sanitized")


def test_prometheus_label_value_sanitization_unicode_paragraph_separator():
    """Test that U+2029 (Paragraph Separator) is also stripped."""
    from litellm.types.integrations.prometheus import sanitize_prometheus_label_value

    result = sanitize_prometheus_label_value("model\u2029name")
    assert result == "modelname"
    assert "\u2029" not in result

    print("✅ U+2029 Paragraph Separator is stripped")


def test_prometheus_label_value_sanitization_none():
    """Test that None values pass through unchanged."""
    from litellm.types.integrations.prometheus import sanitize_prometheus_label_value

    assert sanitize_prometheus_label_value(None) is None

    print("✅ None values pass through unchanged")


def test_prometheus_label_value_sanitization_non_string_types():
    """Test that non-string values (int, bool, etc.) are coerced to str."""
    from litellm.types.integrations.prometheus import sanitize_prometheus_label_value

    assert sanitize_prometheus_label_value(200) == "200"
    assert sanitize_prometheus_label_value(True) == "True"
    assert sanitize_prometheus_label_value(3.14) == "3.14"

    print("✅ Non-string values are coerced to str")


@pytest.mark.asyncio
@pytest.mark.parametrize("metadata_bucket", ("metadata", "litellm_metadata"))
@pytest.mark.parametrize("populate_other_bucket", (False, True))
@pytest.mark.parametrize("original_model_group", (None, "", 123, "requested-source"))
@pytest.mark.parametrize("stream", (False, True))
async def test_success_hook_preserves_requested_model_and_deployment_labels(
    metadata_bucket: Literal["metadata", "litellm_metadata"],
    populate_other_bucket: bool,
    original_model_group: str | int | None,
    stream: bool,
) -> None:
    """
    End-to-end emit wiring for the success path.

    The label-list and factory tests prove the label exists and that the factory
    carries a value handed to it, but neither drives the real logger, so deleting
    the production api_provider=standard_logging_payload["custom_llm_provider"]
    assignment in async_log_success_event would still pass them. This drives
    async_log_success_event with a payload whose provider is openai and asserts
    the collected litellm_total_tokens_metric sample actually carries
    api_provider="openai"; it fails if that assignment is removed.
    """
    import datetime

    from litellm.integrations.prometheus import PrometheusLogger

    payload: Final = {
        "id": "t",
        "call_type": "completion",
        "response_cost": 0.001,
        "status": "success",
        "stream": stream,
        "total_tokens": 30,
        "prompt_tokens": 20,
        "completion_tokens": 10,
        "startTime": 1.0,
        "endTime": 2.0,
        "completionStartTime": 1.5,
        "model": "gpt-4o-mini",
        "model_id": "model-123",
        "model_group": "gpt-4o-mini",
        "api_base": "https://api.openai.com",
        "custom_llm_provider": "openai",
        "request_tags": [],
        "end_user": None,
        "cache_hit": False,
        "metadata": {
            "user_api_key_hash": "h",
            "user_api_key_alias": "a",
            "user_api_key_team_id": "t",
            "user_api_key_team_alias": "ta",
            "user_api_key_user_id": "u",
            "user_api_key_user_email": "e@x.com",
            "user_api_key_org_id": None,
            "user_api_key_org_alias": None,
            "requester_metadata": None,
            "user_api_key_end_user_id": None,
        },
        "hidden_params": {"litellm_overhead_time_ms": None, "additional_headers": None},
    }

    other_bucket: Final = "litellm_metadata" if metadata_bucket == "metadata" else "metadata"
    litellm_params: Final = {
        metadata_bucket: {
            "request_tag": "router",
            **({"original_model_group": original_model_group} if original_model_group is not None else {}),
        },
        **({other_bucket: {"request_tag": "client"}} if populate_other_bucket else {}),
    }
    _clear_prometheus_registry()
    try:
        logger: Final = PrometheusLogger()
        now: Final = datetime.datetime.fromtimestamp(1, tz=datetime.timezone.utc)
        await logger.async_log_success_event(
            {
                "model": "gpt-4o-mini",
                "litellm_params": litellm_params,
                "standard_logging_object": payload,
            },
            None,
            now,
            now,
        )
        request_samples: Final = _collected_samples("litellm_proxy_total_requests_metric_total")
        assert len(request_samples) == 1
        assert request_samples[0].value == 1
        expected_requested_model: Final = (
            original_model_group
            if isinstance(original_model_group, str) and original_model_group
            else payload["model_group"]
        )
        assert request_samples[0].labels["requested_model"] == expected_requested_model
        assert request_samples[0].labels["model_id"] == payload["model_id"]
        assert request_samples[0].labels["api_provider"] == payload["custom_llm_provider"]
        assert request_samples[0].labels["status_code"] == "200"
        samples: Final = _collected_samples("litellm_total_tokens_metric_total")
        assert samples, "expected litellm_total_tokens_metric to be emitted"
        assert all(s.labels.get("api_provider") == "openai" for s in samples), (
            "collected token metric must carry api_provider=openai, got "
            f"{[s.labels.get('api_provider') for s in samples]}"
        )
        assert all(sample.labels["requested_model"] == payload["model_group"] for sample in samples)
        spend_samples: Final = _collected_samples("litellm_spend_metric_total")
        assert len(spend_samples) == 1
        assert spend_samples[0].value == payload["response_cost"]
        assert spend_samples[0].labels["requested_model"] == payload["model_group"]
        assert spend_samples[0].labels["model"] == payload["model"]
    finally:
        _clear_prometheus_registry()


@pytest.mark.asyncio
async def test_failure_hook_emits_api_provider_value_on_failed_requests_metric():
    """
    End-to-end emit wiring for the failure path.

    async_post_call_failure_hook on a request whose model resolves to openai must
    emit litellm_proxy_failed_requests_metric with api_provider="openai". This
    fails if the api_provider assignment in the failure hook is removed, which the
    helper-only test cannot catch.
    """
    from litellm.integrations.prometheus import PrometheusLogger
    from litellm.proxy._types import UserAPIKeyAuth

    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        await logger.async_post_call_failure_hook(
            request_data={"model": "gpt-4o-mini", "metadata": {}},
            original_exception=Exception("boom"),
            user_api_key_dict=UserAPIKeyAuth(token="tok"),
        )
        samples = _collected_samples("litellm_proxy_failed_requests_metric_total")
        assert samples, "expected litellm_proxy_failed_requests_metric to be emitted"
        assert any(s.labels.get("api_provider") == "openai" for s in samples), (
            "collected failed-requests metric must carry api_provider=openai, got "
            f"{[s.labels.get('api_provider') for s in samples]}"
        )
    finally:
        _clear_prometheus_registry()


@pytest.mark.asyncio
async def test_failure_hook_labels_a_cli_session_with_the_per_user_alias_not_the_login_token():
    from litellm.integrations.prometheus import PrometheusLogger
    from litellm.proxy._types import UserAPIKeyAuth

    _clear_prometheus_registry()
    try:
        await PrometheusLogger().async_post_call_failure_hook(
            request_data={"model": "gpt-4o-mini", "metadata": {}},
            original_exception=Exception("boom"),
            user_api_key_dict=UserAPIKeyAuth(
                api_key="cli-session-Qm7xJ2kP9sLw4vT1nR8yAa",
                user_id="alice",
                key_alias="cli-session-alice",
                is_session_token=True,
            ),
        )
        hashed_keys = {s.labels.get("hashed_api_key") for s in _collected_samples("litellm_proxy_failed_requests_metric_total")}
        assert hashed_keys == {"cli-session-alice"}, hashed_keys
    finally:
        _clear_prometheus_registry()


async def _failed_requests_api_provider_labels(
    request_data: dict[str, object],
    original_exception: Exception,
) -> list[str]:
    from litellm.integrations.prometheus import PrometheusLogger
    from litellm.proxy._types import UserAPIKeyAuth

    _clear_prometheus_registry()
    try:
        await PrometheusLogger().async_post_call_failure_hook(
            request_data=request_data,
            original_exception=original_exception,
            user_api_key_dict=UserAPIKeyAuth(token="tok"),
        )
        return [
            s.labels.get("api_provider")
            for s in _collected_samples("litellm_proxy_failed_requests_metric_total")
        ]
    finally:
        _clear_prometheus_registry()


@pytest.mark.asyncio
async def test_failure_hook_emits_api_provider_from_pre_call_rate_limit_error_for_router_alias():
    """
    Pre-call limiters reject before a deployment lands on request_data and a
    router alias cannot be inferred from its name, so the provider the limiter
    resolved onto the exception is the only source for the label.
    """
    from litellm.exceptions import RateLimitType
    from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError

    err = ProxyRateLimitError(
        detail={"error": "rpm exceeded"},
        rate_limit_type=RateLimitType.REQUESTS,
        model="openai/gpt-5.4-mini",
        llm_provider="openai",
    )

    assert await _failed_requests_api_provider_labels(
        {"model": "team-chat-model", "metadata": {}}, err
    ) == ["openai"]


@pytest.mark.asyncio
async def test_failure_hook_leaves_api_provider_unset_when_rate_limiter_could_not_resolve_provider():
    from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError

    err = ProxyRateLimitError(detail={"error": "rpm exceeded"}, model="unknown-alias")

    assert await _failed_requests_api_provider_labels(
        {"model": "unknown-alias", "metadata": {}}, err
    ) == ["None"]


@pytest.mark.asyncio
async def test_failure_hook_prefers_request_data_provider_over_exception_provider():
    from litellm.exceptions import RateLimitError

    err = RateLimitError(message="upstream 429", llm_provider="openai", model="gpt-4o")

    assert await _failed_requests_api_provider_labels(
        {
            "model": "gpt-4o",
            "metadata": {},
            "litellm_params": {"custom_llm_provider": "azure"},
        },
        err,
    ) == ["azure"]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "call_type, custom_llm_provider, expected_requested_model",
    [
        (_PASS_THROUGH_CALL_TYPE, "vertex_ai", "gcp/google/gemini-2.5-flash"),
        ("acompletion", "vertex_ai", "other"),
        (_PASS_THROUGH_CALL_TYPE, "bedrock", "other"),
    ],
)
async def test_failure_hook_labels_a_pass_through_failure_with_the_router_model_the_client_requested(
    call_type: str, custom_llm_provider: str, expected_requested_model: str
):
    from unittest.mock import patch

    from litellm.integrations.prometheus import PrometheusLogger
    from litellm.proxy._types import UserAPIKeyAuth

    router = litellm.Router(
        model_list=[
            {
                "model_name": "gcp/google/gemini-2.5-flash",
                "litellm_params": {
                    "model": "vertex_ai/gemini-2.5-flash",
                    "vertex_project": "p",
                    "vertex_location": "us-central1",
                },
            }
        ]
    )
    _clear_prometheus_registry()
    try:
        with patch("litellm.proxy.proxy_server.llm_router", router, create=True):  # test-quality-ok: production reads proxy_server.llm_router lazily, no injection seam
            await PrometheusLogger().async_post_call_failure_hook(
                request_data={
                    "model": "gemini-2.5-flash",
                    "call_type": call_type,
                    "custom_llm_provider": custom_llm_provider,
                    "metadata": {},
                    "litellm_params": {"metadata": {"model_group": "gcp/google/gemini-2.5-flash"}},
                },
                original_exception=Exception("upstream 400"),
                user_api_key_dict=UserAPIKeyAuth(token="tok"),
            )
        assert {
            s.labels.get("requested_model") for s in _collected_samples("litellm_proxy_failed_requests_metric_total")
        } == {expected_requested_model}
    finally:
        _clear_prometheus_registry()


def test_model_group_in_deployment_metrics():
    """
    Test that model_group label is present on the deployment-scoped metrics
    needed to build model-group dashboards (request counts, success/failure
    counts, tpm/rpm limits). These metrics previously only carried
    requested_model, litellm_model_name and model_id, none of which identify
    the model_group a pooled deployment belongs to.
    """
    model_group_label = UserAPIKeyLabelNames.MODEL_GROUP.value

    metrics_with_model_group = [
        "litellm_deployment_total_requests",
        "litellm_deployment_success_responses",
        "litellm_deployment_failure_responses",
        "litellm_deployment_tpm_limit",
        "litellm_deployment_rpm_limit",
    ]

    for metric_name in metrics_with_model_group:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            model_group_label in labels
        ), f"Metric {metric_name} should contain model_group label"
        print(f"✅ {metric_name} contains model_group label")


def test_model_group_value_flows_through_deployment_metrics_label_factory():
    """
    The label being in the allow-list is necessary but not sufficient: the
    factory must also carry the value from the enum through to the emitted
    label. This would fail if the label were dropped from a metric's list or
    if the value plumbing regressed, which the allow-list assertion above
    cannot catch on its own.
    """
    from unittest.mock import MagicMock

    from litellm.integrations.prometheus import (
        PrometheusLogger,
        UserAPIKeyLabelValues,
        prometheus_label_factory,
    )

    prometheus_logger = MagicMock()
    prometheus_logger._cached_metric_labels = {}
    prometheus_logger.label_filters = {}
    prometheus_logger.get_labels_for_metric = (
        PrometheusLogger.get_labels_for_metric.__get__(prometheus_logger)
    )

    enum_values = UserAPIKeyLabelValues(
        model_group="example-model-group",
        litellm_model_name="gpt-4o-mini",
        requested_model="example-model-group",
        status_code="200",
    )

    for metric_name in [
        "litellm_deployment_total_requests",
        "litellm_deployment_success_responses",
        "litellm_deployment_failure_responses",
        "litellm_deployment_tpm_limit",
        "litellm_deployment_rpm_limit",
    ]:
        labels = prometheus_label_factory(
            supported_enum_labels=prometheus_logger.get_labels_for_metric(
                metric_name=metric_name
            ),
            enum_values=enum_values,
        )
        assert (
            labels.get("model_group") == "example-model-group"
        ), f"{metric_name} should emit model_group=example-model-group, got {labels.get('model_group')!r}"


def test_deployment_failure_metrics_emit_model_group_from_standard_logging_payload():
    """
    End-to-end emit wiring for the failure path.

    The label-list and factory tests above prove the label exists and that
    the factory carries a value handed to it, but neither drives the real
    set_llm_deployment_failure_metrics code path, so deleting the production
    model_group=model_group assignment there would still pass them. This
    calls it directly with a standard_logging_object carrying model_group and
    asserts the real litellm_deployment_failure_responses / _total_requests
    Counter series actually carry it.
    """
    from litellm.integrations.prometheus import PrometheusLogger

    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        logger.set_llm_deployment_failure_metrics(
            request_kwargs={
                "model": "gpt-4o-mini",
                "litellm_params": {"metadata": {}},
                "standard_logging_object": {
                    "model_group": "example-model-group",
                    "model_id": "model-123",
                    "api_base": "https://api.openai.com",
                    "request_tags": [],
                },
                "exception": Exception("boom"),
            }
        )

        for metric in (
            logger.litellm_deployment_failure_responses,
            logger.litellm_deployment_total_requests,
        ):
            index = metric._labelnames.index("model_group")
            values = {sample_key[index] for sample_key in metric._metrics}
            assert values == {"example-model-group"}, (
                f"expected model_group=example-model-group on {metric._name}, got {values}"
            )
    finally:
        _clear_prometheus_registry()


def _routed(deployment: object) -> dict[str, object]:
    return {"deployment": deployment, "model_info": {"id": "model-123"}}


@pytest.mark.parametrize(
    ("litellm_params", "logged_provider", "expected"),
    [
        pytest.param(
            {"custom_llm_provider": "bedrock", "metadata": _routed("azure/gpt-4o")},
            "vertex_ai",
            "bedrock",
            id="resolved-provider-wins",
        ),
        pytest.param(
            {"custom_llm_provider": None, "metadata": _routed("anthropic/claude-haiku-4-5")},
            "openai",
            "openai",
            id="provider-on-the-payload-wins",
        ),
        pytest.param({"metadata": _routed("azure/limited-azure")}, None, "azure", id="deployment-prefix"),
        pytest.param(
            {"litellm_metadata": _routed("anthropic/claude-haiku-4-5"), "metadata": {}},
            None,
            "anthropic",
            id="litellm-metadata-deployment",
        ),
        pytest.param(
            {"litellm_metadata": _routed("anthropic/claude-haiku-4-5"), "metadata": _routed("bedrock/converse/model")},
            None,
            "anthropic",
            id="litellm-metadata-deployment-wins",
        ),
        pytest.param(
            {"litellm_metadata": {"deployment": "bedrock/forged"}, "metadata": _routed("anthropic/claude-haiku-4-5")},
            None,
            "anthropic",
            id="routed-bucket-wins-over-a-bare-deployment",
        ),
        pytest.param({"metadata": _routed("abliteration/model")}, None, "abliteration", id="json-configured-provider"),
        pytest.param({"metadata": _routed("langfuse/my-prompt")}, None, "", id="prompt-management-deployment"),
        pytest.param({"metadata": _routed("gpt-4o")}, None, "", id="unprefixed-deployment"),
        pytest.param({"metadata": _routed("meta-llama/Llama-3")}, None, "", id="prefix-is-not-a-provider"),
        pytest.param({"metadata": _routed(5)}, None, "", id="non-string-deployment"),
        pytest.param({"metadata": {}}, None, "", id="nothing-known"),
    ],
)
def test_deployment_failure_metrics_label_the_selected_deployment_provider(litellm_params, logged_provider, expected):
    """
    A failure on a selected deployment labels api_provider with that
    deployment's provider, never the string "None" and never a guess from the
    requested model name (gpt-4o-mini here would read openai).
    """
    from litellm.integrations.prometheus import PrometheusLogger

    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        logger.set_llm_deployment_failure_metrics(
            request_kwargs={
                "model": "gpt-4o-mini",
                "litellm_params": litellm_params,
                "standard_logging_object": {
                    "model_group": "gpt-4o-mini",
                    "model_id": "model-123",
                    "api_base": "https://example.openai.azure.com",
                    "custom_llm_provider": logged_provider,
                    "request_tags": [],
                },
                "exception": Exception("boom"),
            }
        )

        for metric in (
            logger.litellm_deployment_failure_responses,
            logger.litellm_deployment_total_requests,
        ):
            index = metric._labelnames.index("api_provider")
            values = {sample_key[index] for sample_key in metric._metrics}
            assert values == {expected}, f"expected api_provider={expected!r} on {metric._name}, got {values}"
    finally:
        _clear_prometheus_registry()


def test_deployment_failure_metrics_label_a_custom_provider_deployment(monkeypatch):
    from litellm.integrations.prometheus import PrometheusLogger

    monkeypatch.setattr(litellm, "custom_provider_map", [{"provider": "in-house", "custom_handler": object()}])
    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        logger.set_llm_deployment_failure_metrics(
            request_kwargs={
                "model": "gpt-4o-mini",
                "litellm_params": {"metadata": _routed("in-house/model")},
                "standard_logging_object": {
                    "model_group": "gpt-4o-mini",
                    "model_id": "model-123",
                    "api_base": "https://in-house.example",
                    "request_tags": [],
                },
                "exception": Exception("boom"),
            }
        )

        index = logger.litellm_deployment_failure_responses._labelnames.index("api_provider")
        values = {sample_key[index] for sample_key in logger.litellm_deployment_failure_responses._metrics}
        assert values == {"in-house"}
    finally:
        _clear_prometheus_registry()


def test_deployment_tpm_rpm_limit_metrics_emit_model_group_from_enum_values():
    """
    End-to-end emit wiring for the tpm/rpm limit gauges.

    _set_deployment_tpm_rpm_limit_metrics used to build its own
    UserAPIKeyLabelValues with no model_group parameter at all, dropping the
    value even though its only caller (set_llm_deployment_success_metrics)
    already had it on enum_values. This drives set_llm_deployment_success_metrics
    directly with a deployment that has tpm/rpm configured and asserts the real
    litellm_deployment_tpm_limit / litellm_deployment_rpm_limit Gauge series
    carry model_group; it fails if that plumbing is removed.
    """
    import datetime

    from litellm.integrations.prometheus import PrometheusLogger, UserAPIKeyLabelValues

    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        now = datetime.datetime.now()
        enum_values = UserAPIKeyLabelValues(
            model_group="example-model-group",
            litellm_model_name="gpt-4o-mini",
            requested_model="example-model-group",
            status_code="200",
        )
        logger.set_llm_deployment_success_metrics(
            request_kwargs={
                "model": "gpt-4o-mini",
                "litellm_params": {"metadata": {"model_info": {"id": "model-123", "tpm": 1000, "rpm": 10}}},
                "standard_logging_object": {
                    "model_group": "example-model-group",
                    "model_id": "model-123",
                    "api_base": "https://api.openai.com",
                    "hidden_params": {"additional_headers": None, "litellm_overhead_time_ms": None},
                },
            },
            start_time=now,
            end_time=now,
            enum_values=enum_values,
        )

        for metric in (logger.litellm_deployment_tpm_limit, logger.litellm_deployment_rpm_limit):
            index = metric._labelnames.index("model_group")
            values = {sample_key[index] for sample_key in metric._metrics}
            assert values == {"example-model-group"}, (
                f"expected model_group=example-model-group on {metric._name}, got {values}"
            )
    finally:
        _clear_prometheus_registry()


def test_model_group_in_latency_metrics():
    """
    Test that model_group label is present on the end-to-end / per-call
    latency metrics needed to build model-group latency dashboards. These
    metrics previously only carried requested_model, litellm_model_name and
    model_id, none of which identify the model_group a pooled deployment
    belongs to -- only the proxy-overhead-only latency metrics
    (litellm_overhead_latency_metric and friends) carried model_group.
    """
    model_group_label = UserAPIKeyLabelNames.MODEL_GROUP.value

    metrics_with_model_group = [
        "litellm_llm_api_latency_metric",
        "litellm_llm_api_time_to_first_token_metric",
        "litellm_request_total_latency_metric",
        "litellm_deployment_latency_per_output_token",
    ]

    for metric_name in metrics_with_model_group:
        labels = PrometheusMetricLabels.get_labels(metric_name)
        assert (
            model_group_label in labels
        ), f"Metric {metric_name} should contain model_group label"
        print(f"✅ {metric_name} contains model_group label")


def test_model_group_value_flows_through_latency_metrics_label_factory():
    """
    The label being in the allow-list is necessary but not sufficient: the
    factory must also carry the value from the enum through to the emitted
    label. This would fail if the label were dropped from a metric's list or
    if the value plumbing regressed, which the allow-list assertion above
    cannot catch on its own.
    """
    from unittest.mock import MagicMock

    from litellm.integrations.prometheus import (
        PrometheusLogger,
        UserAPIKeyLabelValues,
        prometheus_label_factory,
    )

    prometheus_logger = MagicMock()
    prometheus_logger._cached_metric_labels = {}
    prometheus_logger.label_filters = {}
    prometheus_logger.get_labels_for_metric = (
        PrometheusLogger.get_labels_for_metric.__get__(prometheus_logger)
    )

    enum_values = UserAPIKeyLabelValues(
        model_group="example-model-group",
        litellm_model_name="gpt-4o-mini",
        requested_model="example-model-group",
        status_code="200",
    )

    for metric_name in [
        "litellm_llm_api_latency_metric",
        "litellm_llm_api_time_to_first_token_metric",
        "litellm_request_total_latency_metric",
        "litellm_deployment_latency_per_output_token",
    ]:
        labels = prometheus_label_factory(
            supported_enum_labels=prometheus_logger.get_labels_for_metric(
                metric_name=metric_name
            ),
            enum_values=enum_values,
        )
        assert (
            labels.get("model_group") == "example-model-group"
        ), f"{metric_name} should emit model_group=example-model-group, got {labels.get('model_group')!r}"


def test_latency_metrics_emit_model_group_from_set_latency_metrics():
    """
    End-to-end emit wiring for _set_latency_metrics.

    The label-list and factory tests above prove the label exists and that
    the factory carries a value handed to it, but neither drives the real
    _set_latency_metrics code path, so deleting the production model_group
    plumbing there would still pass them. This calls it directly with a
    streaming request (so the time-to-first-token branch also fires) and
    asserts the real litellm_llm_api_latency_metric,
    litellm_llm_api_time_to_first_token_metric and
    litellm_request_total_latency_metric Histogram series actually carry it.
    """
    import datetime

    from litellm.integrations.prometheus import PrometheusLogger, UserAPIKeyLabelValues

    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        start_time = datetime.datetime(2024, 1, 1, 0, 0, 0)
        api_call_start_time = datetime.datetime(2024, 1, 1, 0, 0, 1)
        completion_start_time = datetime.datetime(2024, 1, 1, 0, 0, 2)
        end_time = datetime.datetime(2024, 1, 1, 0, 0, 3)

        enum_values = UserAPIKeyLabelValues(
            model_group="example-model-group",
            litellm_model_name="gpt-4o-mini",
            requested_model="example-model-group",
            status_code="200",
        )

        logger._set_latency_metrics(
            kwargs={
                "start_time": start_time,
                "end_time": end_time,
                "api_call_start_time": api_call_start_time,
                "completion_start_time": completion_start_time,
                "stream": True,
                "litellm_params": {"metadata": {}},
            },
            model="gpt-4o-mini",
            user_api_key=None,
            user_api_key_alias=None,
            user_api_team=None,
            user_api_team_alias=None,
            enum_values=enum_values,
        )

        for metric in (
            logger.litellm_llm_api_latency_metric,
            logger.litellm_llm_api_time_to_first_token_metric,
            logger.litellm_request_total_latency_metric,
        ):
            index = metric._labelnames.index("model_group")
            values = {sample_key[index] for sample_key in metric._metrics}
            assert values == {"example-model-group"}, (
                f"expected model_group=example-model-group on {metric._name}, got {values}"
            )
    finally:
        _clear_prometheus_registry()


def test_deployment_latency_per_output_token_emits_model_group_from_enum_values():
    """
    End-to-end emit wiring for litellm_deployment_latency_per_output_token.

    Drives set_llm_deployment_success_metrics directly (its only caller) with
    output_tokens > 0 so the latency-per-token branch fires, and asserts the
    real Histogram series carries model_group; fails if that label-list
    addition or the enum_values plumbing is removed.
    """
    import datetime

    from litellm.integrations.prometheus import PrometheusLogger, UserAPIKeyLabelValues

    _clear_prometheus_registry()
    try:
        logger = PrometheusLogger()
        start_time = datetime.datetime(2024, 1, 1, 0, 0, 0)
        end_time = datetime.datetime(2024, 1, 1, 0, 0, 2)
        enum_values = UserAPIKeyLabelValues(
            model_group="example-model-group",
            litellm_model_name="gpt-4o-mini",
            requested_model="example-model-group",
            status_code="200",
        )
        logger.set_llm_deployment_success_metrics(
            request_kwargs={
                "model": "gpt-4o-mini",
                "litellm_params": {"metadata": {"model_info": {"id": "model-123"}}},
                "standard_logging_object": {
                    "model_group": "example-model-group",
                    "model_id": "model-123",
                    "api_base": "https://api.openai.com",
                    "hidden_params": {"additional_headers": None, "litellm_overhead_time_ms": None},
                },
            },
            start_time=start_time,
            end_time=end_time,
            enum_values=enum_values,
            output_tokens=10.0,
        )

        metric = logger.litellm_deployment_latency_per_output_token
        index = metric._labelnames.index("model_group")
        values = {sample_key[index] for sample_key in metric._metrics}
        assert values == {"example-model-group"}, (
            f"expected model_group=example-model-group on {metric._name}, got {values}"
        )
    finally:
        _clear_prometheus_registry()


if __name__ == "__main__":
    test_user_email_in_required_metrics()
    test_user_email_label_exists()
    test_prometheus_metric_labels_structure()
    test_route_normalization_for_responses_api()
    test_route_normalization_for_sub_routes()
    test_route_normalization_preserves_static_routes()
    test_route_normalization_other_dynamic_apis()
    test_prometheus_metrics_use_normalized_routes()
    test_prometheus_label_value_sanitization()
    test_prometheus_label_value_sanitization_unicode_paragraph_separator()
    test_prometheus_label_value_sanitization_none()
    test_prometheus_label_value_sanitization_non_string_types()
    print("\n✅ All prometheus label tests passed!")