From 87c64eac2e6e30b47eba50ab25386f16326f770e Mon Sep 17 00:00:00 2001 From: Kou-Kinoshita Date: Thu, 30 Oct 2025 09:47:23 -0300 Subject: [PATCH] SIENTIAPDE-1241: Metrics test --- tests/test_metrics.py | 211 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 tests/test_metrics.py diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..2e92813 --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,211 @@ +"""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 +