test_metrics_registry.py

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Tests for Python MetricsRegistry bindings.

This test suite verifies that Python can create, introspect, and use Prometheus
metrics through the Dynamo MetricsRegistry interface.
"""

import pytest


async def get_metrics_runtime(runtime, endpoint_name):
    """Helper to create a unique metrics runtime for each test."""
    namespace = runtime.namespace("test_metrics_ns")
    component = namespace.component("test_metrics_comp")
    endpoint = component.endpoint(endpoint_name)
    return endpoint.metrics


pytestmark = pytest.mark.pre_merge


@pytest.mark.asyncio
@pytest.mark.forked
async def test_counter_introspection(runtime):
    """Test Counter metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_counter_introspection")

    counter = metrics_runtime.create_counter(
        "test_counter", "A test counter", [("env", "test")]  # constant labels
    )

    # Test name() method
    name = counter.name()
    assert isinstance(name, str)
    assert "test_counter" in name
    assert name.endswith("test_counter")

    # Test const_labels() method
    labels = counter.const_labels()
    assert isinstance(labels, dict)
    assert "env" in labels
    assert labels["env"] == "test"
    assert "dynamo_namespace" in labels
    assert labels["dynamo_namespace"] == "test_metrics_ns"


@pytest.mark.asyncio
@pytest.mark.forked
async def test_intcounter_introspection(runtime):
    """Test IntCounter metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_intcounter_introspection")
    counter = metrics_runtime.create_intcounter(
        "test_int_counter", "A test int counter", [("type", "integer")]
    )

    name = counter.name()
    assert isinstance(name, str)
    assert "test_int_counter" in name

    labels = counter.const_labels()
    assert isinstance(labels, dict)
    assert labels["type"] == "integer"


@pytest.mark.asyncio
@pytest.mark.forked
async def test_gauge_introspection(runtime):
    """Test Gauge metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_gauge_introspection")
    gauge = metrics_runtime.create_gauge(
        "test_gauge", "A test gauge", [("unit", "bytes")]
    )

    name = gauge.name()
    assert isinstance(name, str)
    assert "test_gauge" in name

    labels = gauge.const_labels()
    assert isinstance(labels, dict)
    assert labels["unit"] == "bytes"


@pytest.mark.asyncio
@pytest.mark.forked
async def test_intgauge_introspection(runtime):
    """Test IntGauge metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_intgauge_introspection")
    gauge = metrics_runtime.create_intgauge(
        "test_int_gauge", "A test int gauge", []  # no constant labels
    )

    name = gauge.name()
    assert isinstance(name, str)
    assert "test_int_gauge" in name

    labels = gauge.const_labels()
    assert isinstance(labels, dict)
    # Should still have hierarchy labels
    assert "dynamo_namespace" in labels


@pytest.mark.asyncio
@pytest.mark.forked
async def test_histogram_introspection(runtime):
    """Test Histogram metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_histogram_introspection")
    histogram = metrics_runtime.create_histogram(
        "test_histogram", "A test histogram", [("method", "POST")]
    )

    name = histogram.name()
    assert isinstance(name, str)
    assert "test_histogram" in name

    labels = histogram.const_labels()
    assert isinstance(labels, dict)
    assert labels["method"] == "POST"


@pytest.mark.asyncio
@pytest.mark.forked
async def test_countervec_introspection(runtime):
    """Test CounterVec metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_countervec_introspection")
    counter_vec = metrics_runtime.create_countervec(
        "test_counter_vec",
        "A test counter vec",
        ["worker_id", "status"],  # variable labels
        [("cluster", "prod")],  # constant labels
    )

    # Test name()
    name = counter_vec.name()
    assert isinstance(name, str)
    assert "test_counter_vec" in name

    # Test const_labels()
    const_labels = counter_vec.const_labels()
    assert isinstance(const_labels, dict)
    assert const_labels["cluster"] == "prod"
    assert "dynamo_namespace" in const_labels

    # Test variable_labels()
    var_labels = counter_vec.variable_labels()
    assert isinstance(var_labels, list)
    assert len(var_labels) == 2
    assert "worker_id" in var_labels
    assert "status" in var_labels


@pytest.mark.asyncio
@pytest.mark.forked
async def test_intcountervec_introspection(runtime):
    """Test IntCounterVec metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(
        runtime, "ep_intcountervec_introspection"
    )
    counter_vec = metrics_runtime.create_intcountervec(
        "test_int_counter_vec",
        "A test int counter vec",
        ["region", "zone"],
        [],  # no constant labels
    )

    name = counter_vec.name()
    assert "test_int_counter_vec" in name

    const_labels = counter_vec.const_labels()
    assert isinstance(const_labels, dict)

    var_labels = counter_vec.variable_labels()
    assert len(var_labels) == 2
    assert "region" in var_labels
    assert "zone" in var_labels


@pytest.mark.asyncio
@pytest.mark.forked
async def test_gaugevec_introspection(runtime):
    """Test GaugeVec metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_gaugevec_introspection")
    gauge_vec = metrics_runtime.create_gaugevec(
        "test_gauge_vec", "A test gauge vec", ["instance", "job"], [("env", "staging")]
    )

    name = gauge_vec.name()
    assert "test_gauge_vec" in name

    const_labels = gauge_vec.const_labels()
    assert const_labels["env"] == "staging"

    var_labels = gauge_vec.variable_labels()
    assert len(var_labels) == 2
    assert "instance" in var_labels
    assert "job" in var_labels


@pytest.mark.asyncio
@pytest.mark.forked
async def test_intgaugevec_introspection(runtime):
    """Test IntGaugeVec metric introspection methods."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_intgaugevec_introspection")
    gauge_vec = metrics_runtime.create_intgaugevec(
        "test_int_gauge_vec",
        "A test int gauge vec",
        ["device", "partition"],
        [("datacenter", "us-west")],
    )

    name = gauge_vec.name()
    assert "test_int_gauge_vec" in name

    const_labels = gauge_vec.const_labels()
    assert const_labels["datacenter"] == "us-west"

    var_labels = gauge_vec.variable_labels()
    assert len(var_labels) == 2
    assert "device" in var_labels
    assert "partition" in var_labels


@pytest.mark.asyncio
@pytest.mark.forked
async def test_metric_operations(runtime):
    """Test that metrics can be used after introspection."""
    metrics_runtime = await get_metrics_runtime(runtime, "ep_metric_operations")
    # Counter operations
    counter = metrics_runtime.create_intcounter("ops_counter", "Operations counter", [])
    counter.inc()
    counter.inc_by(5)
    assert counter.get() == 6

    # Gauge operations
    gauge = metrics_runtime.create_intgauge(
        "connections_gauge", "Connections gauge", []
    )
    gauge.set(10)
    assert gauge.get() == 10
    gauge.inc()
    assert gauge.get() == 11
    gauge.dec()
    assert gauge.get() == 10

    # Vec operations
    gauge_vec = metrics_runtime.create_intgaugevec(
        "worker_gauge_vec", "Worker gauge vec", ["worker_id"], []
    )
    gauge_vec.set(5, {"worker_id": "w1"})
    assert gauge_vec.get({"worker_id": "w1"}) == 5
    gauge_vec.inc({"worker_id": "w1"})
    assert gauge_vec.get({"worker_id": "w1"}) == 6


@pytest.mark.asyncio
@pytest.mark.forked
async def test_multiple_metrics_same_runtime(runtime):
    """Test creating multiple metrics in the same runtime."""
    metrics_runtime = await get_metrics_runtime(
        runtime, "ep_multiple_metrics_same_runtime"
    )
    counter1 = metrics_runtime.create_intcounter("counter1", "Counter 1", [])
    counter2 = metrics_runtime.create_intcounter("counter2", "Counter 2", [])
    gauge1 = metrics_runtime.create_gauge("gauge1", "Gauge 1", [])

    # All should have unique names
    names = {counter1.name(), counter2.name(), gauge1.name()}
    assert len(names) == 3

    # All should share the same hierarchy labels
    for metric in [counter1, counter2, gauge1]:
        labels = metric.const_labels()
        assert labels["dynamo_namespace"] == "test_metrics_ns"
        assert "dynamo_component" in labels  # Component name is test-specific
        assert labels["dynamo_endpoint"] == "ep_multiple_metrics_same_runtime"