237 lines
6.5 KiB
Python
237 lines
6.5 KiB
Python
"""Unit tests for metrics collection."""
|
|
|
|
import pytest
|
|
|
|
from src.server.utils.metrics import (
|
|
MetricsCollector,
|
|
MetricType,
|
|
TimerContext,
|
|
get_metrics_collector,
|
|
)
|
|
|
|
|
|
def test_metrics_collector_initialization():
|
|
"""Test metrics collector initialization."""
|
|
collector = MetricsCollector()
|
|
|
|
assert collector is not None
|
|
assert collector._metrics == {}
|
|
assert collector._download_stats["completed"] == 0
|
|
assert collector._download_stats["failed"] == 0
|
|
|
|
|
|
def test_increment_counter():
|
|
"""Test counter metric increment."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.increment_counter("test_counter", 1.0, help_text="Test counter")
|
|
collector.increment_counter("test_counter", 2.0)
|
|
|
|
assert "test_counter" in collector._metrics
|
|
assert collector._metrics["test_counter"].value == 3.0
|
|
assert collector._metrics["test_counter"].metric_type == MetricType.COUNTER
|
|
|
|
|
|
def test_set_gauge():
|
|
"""Test gauge metric."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.set_gauge("test_gauge", 42.0, help_text="Test gauge")
|
|
assert collector._metrics["test_gauge"].value == 42.0
|
|
|
|
collector.set_gauge("test_gauge", 100.0)
|
|
assert collector._metrics["test_gauge"].value == 100.0
|
|
|
|
|
|
def test_observe_histogram():
|
|
"""Test histogram observation."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.observe_histogram("request_duration", 0.5)
|
|
collector.observe_histogram("request_duration", 1.2)
|
|
collector.observe_histogram("request_duration", 0.8)
|
|
|
|
assert len(collector._request_timings["request_duration"]) == 3
|
|
assert 0.5 in collector._request_timings["request_duration"]
|
|
|
|
|
|
def test_start_and_end_timer():
|
|
"""Test timer functionality."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.start_timer("test_timer")
|
|
import time
|
|
|
|
time.sleep(0.01) # Sleep for 10ms
|
|
duration = collector.end_timer("test_timer", "test_duration")
|
|
|
|
assert duration >= 0.01
|
|
assert "test_duration" in collector._metrics
|
|
|
|
|
|
def test_record_download_success():
|
|
"""Test download success recording."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.record_download_success(1000000)
|
|
collector.record_download_success(2000000)
|
|
|
|
stats = collector.get_download_stats()
|
|
assert stats["completed"] == 2
|
|
assert stats["total_size_bytes"] == 3000000
|
|
assert stats["failed"] == 0
|
|
|
|
|
|
def test_record_download_failure():
|
|
"""Test download failure recording."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.record_download_failure()
|
|
collector.record_download_failure()
|
|
|
|
stats = collector.get_download_stats()
|
|
assert stats["failed"] == 2
|
|
assert stats["completed"] == 0
|
|
|
|
|
|
def test_get_request_statistics():
|
|
"""Test request statistics calculation."""
|
|
collector = MetricsCollector()
|
|
|
|
for val in [0.5, 1.0, 0.8, 0.6, 0.9]:
|
|
collector.observe_histogram("request_latency", val)
|
|
|
|
stats = collector.get_request_statistics("request_latency")
|
|
|
|
assert stats is not None
|
|
assert stats["count"] == 5
|
|
assert stats["mean"] == pytest.approx(0.76, abs=0.01)
|
|
assert stats["min"] == 0.5
|
|
assert stats["max"] == 1.0
|
|
|
|
|
|
def test_get_request_statistics_not_found():
|
|
"""Test request statistics for non-existent metric."""
|
|
collector = MetricsCollector()
|
|
|
|
stats = collector.get_request_statistics("non_existent")
|
|
assert stats is None
|
|
|
|
|
|
def test_export_prometheus_format():
|
|
"""Test Prometheus format export."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.increment_counter(
|
|
"requests_total", 10, help_text="Total requests"
|
|
)
|
|
collector.set_gauge("active_connections", 5)
|
|
|
|
prometheus_output = collector.export_prometheus_format()
|
|
|
|
assert "requests_total" in prometheus_output
|
|
assert "active_connections" in prometheus_output
|
|
assert "10" in prometheus_output
|
|
assert "5" in prometheus_output
|
|
|
|
|
|
def test_export_prometheus_with_labels():
|
|
"""Test Prometheus format with labels."""
|
|
collector = MetricsCollector()
|
|
|
|
labels = {"endpoint": "/api/anime", "method": "GET"}
|
|
collector.increment_counter("requests_total", labels=labels)
|
|
|
|
prometheus_output = collector.export_prometheus_format()
|
|
|
|
assert "endpoint" in prometheus_output
|
|
assert "method" in prometheus_output
|
|
assert "/api/anime" in prometheus_output
|
|
assert "GET" in prometheus_output
|
|
|
|
|
|
def test_export_json():
|
|
"""Test JSON export."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.increment_counter("test_counter", 5)
|
|
collector.set_gauge("test_gauge", 42)
|
|
collector.record_download_success(1000000)
|
|
|
|
json_export = collector.export_json()
|
|
|
|
assert "metrics" in json_export
|
|
assert "downloads" in json_export
|
|
assert "request_timings" in json_export
|
|
assert json_export["downloads"]["completed"] == 1
|
|
assert json_export["downloads"]["total_size_bytes"] == 1000000
|
|
|
|
|
|
def test_reset_metrics():
|
|
"""Test metrics reset."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.increment_counter("test_counter", 10)
|
|
collector.record_download_success(1000000)
|
|
|
|
assert len(collector._metrics) > 0
|
|
assert collector._download_stats["completed"] == 1
|
|
|
|
collector.reset_metrics()
|
|
|
|
assert len(collector._metrics) == 0
|
|
assert collector._download_stats["completed"] == 0
|
|
|
|
|
|
def test_get_all_metrics():
|
|
"""Test getting all metrics."""
|
|
collector = MetricsCollector()
|
|
|
|
collector.increment_counter("counter1", 5)
|
|
collector.set_gauge("gauge1", 10)
|
|
collector.increment_counter("counter2", 3)
|
|
|
|
all_metrics = collector.get_all_metrics()
|
|
|
|
assert len(all_metrics) == 3
|
|
assert "counter1" in all_metrics
|
|
assert "gauge1" in all_metrics
|
|
assert "counter2" in all_metrics
|
|
|
|
|
|
def test_get_metrics_collector_singleton():
|
|
"""Test singleton metrics collector."""
|
|
collector1 = get_metrics_collector()
|
|
collector2 = get_metrics_collector()
|
|
|
|
assert collector1 is collector2
|
|
assert isinstance(collector1, MetricsCollector)
|
|
|
|
|
|
def test_timer_context_manager():
|
|
"""Test timer context manager."""
|
|
collector = get_metrics_collector()
|
|
collector.reset_metrics()
|
|
|
|
import time
|
|
|
|
with TimerContext("operation_duration", "timer1"):
|
|
time.sleep(0.01)
|
|
|
|
stats = collector.get_request_statistics("operation_duration")
|
|
assert stats is not None
|
|
assert stats["count"] == 1
|
|
assert stats["max"] >= 0.01
|
|
|
|
|
|
def test_timer_context_with_labels():
|
|
"""Test timer context manager with labels."""
|
|
collector = get_metrics_collector()
|
|
collector.reset_metrics()
|
|
|
|
labels = {"endpoint": "/api/test"}
|
|
with TimerContext("endpoint_duration", labels=labels):
|
|
pass
|
|
|
|
assert "endpoint_duration" in collector._metrics
|