sientia-dataops-model-manager/tests/test_metrics.py

"""Unit tests for model_manager.metrics module.

This module tests the Prometheus metrics configuration used for
monitoring and observability in the Sientia DataOps Model Manager.
"""

from unittest.mock import MagicMock, patch

import pytest


def test_app_up_metric_exists():
    """Test that APP_UP metric is properly defined."""
    from model_manager.metrics import APP_UP

    assert APP_UP is not None
    assert APP_UP._name == 'app_up'
    assert APP_UP._documentation == 'Indicates if the application is running (1) or shutting down (0)'


def test_app_up_metric_has_pod_id_label():
    """Test that APP_UP metric has pod_id label."""
    from model_manager.metrics import APP_UP

    assert 'pod_id' in APP_UP._labelnames


def test_app_up_metric_is_gauge():
    """Test that APP_UP is a Gauge metric."""
    from prometheus_client import Gauge

    from model_manager.metrics import APP_UP

    assert isinstance(APP_UP, Gauge)


def test_app_up_metric_can_be_set_to_one():
    """Test that APP_UP metric can be set to 1 (running)."""
    from model_manager.metrics import APP_UP

    # Set metric to 1 for a specific pod
    APP_UP.labels(pod_id='test-pod-1').set(1)

    # Verify the metric value
    metric_value = APP_UP.labels(pod_id='test-pod-1')._value._value
    assert metric_value == 1


def test_app_up_metric_can_be_set_to_zero():
    """Test that APP_UP metric can be set to 0 (shutting down)."""
    from model_manager.metrics import APP_UP

    # Set metric to 0 for a specific pod
    APP_UP.labels(pod_id='test-pod-2').set(0)

    # Verify the metric value
    metric_value = APP_UP.labels(pod_id='test-pod-2')._value._value
    assert metric_value == 0


def test_app_up_metric_multiple_pods():
    """Test that APP_UP metric can track multiple pods independently."""
    from model_manager.metrics import APP_UP

    # Set different values for different pods
    APP_UP.labels(pod_id='pod-1').set(1)
    APP_UP.labels(pod_id='pod-2').set(0)
    APP_UP.labels(pod_id='pod-3').set(1)

    # Verify each pod has correct value
    assert APP_UP.labels(pod_id='pod-1')._value._value == 1
    assert APP_UP.labels(pod_id='pod-2')._value._value == 0
    assert APP_UP.labels(pod_id='pod-3')._value._value == 1


def test_app_up_metric_default_value():
    """Test that APP_UP metric starts with no value set."""
    from model_manager.metrics import APP_UP

    # Create a new label that hasn't been used yet
    import uuid

    unique_pod = f'test-pod-{uuid.uuid4()}'

    # The metric should exist but not have a value until set
    metric = APP_UP.labels(pod_id=unique_pod)
    assert metric is not None


def test_metrics_module_imports():
    """Test that metrics module can be imported successfully."""
    import model_manager.metrics

    assert hasattr(model_manager.metrics, 'APP_UP')
    assert hasattr(model_manager.metrics, 'Gauge')


def test_metrics_module_docstring():
    """Test that metrics module has proper documentation."""
    import model_manager.metrics

    assert model_manager.metrics.__doc__ is not None
    assert 'Prometheus' in model_manager.metrics.__doc__
    assert 'metrics' in model_manager.metrics.__doc__


def test_app_up_metric_can_increment():
    """Test that APP_UP metric value can be incremented."""
    from model_manager.metrics import APP_UP

    pod_id = 'test-pod-increment'
    APP_UP.labels(pod_id=pod_id).set(0)

    # Increment the metric
    APP_UP.labels(pod_id=pod_id).inc()

    metric_value = APP_UP.labels(pod_id=pod_id)._value._value
    assert metric_value == 1


def test_app_up_metric_can_decrement():
    """Test that APP_UP metric value can be decremented."""
    from model_manager.metrics import APP_UP

    pod_id = 'test-pod-decrement'
    APP_UP.labels(pod_id=pod_id).set(1)

    # Decrement the metric
    APP_UP.labels(pod_id=pod_id).dec()

    metric_value = APP_UP.labels(pod_id=pod_id)._value._value
    assert metric_value == 0


def test_app_up_metric_set_to_timestamp():
    """Test that APP_UP metric can be set to current timestamp."""
    import time

    from model_manager.metrics import APP_UP

    pod_id = 'test-pod-timestamp'
    current_time = time.time()

    # Set to timestamp
    APP_UP.labels(pod_id=pod_id).set_to_current_time()

    metric_value = APP_UP.labels(pod_id=pod_id)._value._value

    # Should be close to current time
    assert abs(metric_value - current_time) < 2  # Within 2 seconds


def test_app_up_metric_label_validation():
    """Test that APP_UP metric validates label names."""
    from model_manager.metrics import APP_UP

    # Should work with valid label
    APP_UP.labels(pod_id='valid-pod-name').set(1)

    # Should work with empty string (though not recommended)
    APP_UP.labels(pod_id='').set(1)

    # Should work with special characters
    APP_UP.labels(pod_id='pod-123_test.example').set(1)


def test_module_exports():
    """Test that metrics module exports expected symbols."""
    import model_manager.metrics as metrics_module

    # Check that module has the expected exports
    module_contents = dir(metrics_module)

    assert 'APP_UP' in module_contents
    assert 'Gauge' in module_contents


def test_app_up_metric_thread_safety():
    """Test that APP_UP metric is thread-safe."""
    import threading

    from model_manager.metrics import APP_UP

    pod_id = 'test-pod-threading'
    APP_UP.labels(pod_id=pod_id).set(0)

    def increment_metric():
        for _ in range(100):
            APP_UP.labels(pod_id=pod_id).inc()

    # Create multiple threads that increment the metric
    threads = [threading.Thread(target=increment_metric) for _ in range(5)]

    for thread in threads:
        thread.start()

    for thread in threads:
        thread.join()

    # Should have incremented 500 times total
    metric_value = APP_UP.labels(pod_id=pod_id)._value._value
    assert metric_value == 500


def test_prometheus_client_gauge_import():
    """Test that Gauge is properly imported from prometheus_client."""
    from model_manager.metrics import Gauge
    from prometheus_client import Gauge as PrometheusGauge

    assert Gauge is PrometheusGauge