blob: 3fd72a74eef0ca9d6091fbc82e2ff15b8c7cd608 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
"""Tests for Prometheus metrics utilities."""
from prometheus_client import CollectorRegistry
def make_metrics(service_name="test-service", registry=None):
from shared.metrics import ServiceMetrics
return ServiceMetrics(service_name, registry=registry)
def test_metrics_creates_with_prefix():
"""ServiceMetrics stores sanitized service name prefix."""
registry = CollectorRegistry()
m = make_metrics("my-service", registry=registry)
assert m.service_name == "my_service"
def test_errors_total_increment():
"""errors_total counter can be incremented with labels."""
registry = CollectorRegistry()
m = make_metrics("test-svc", registry=registry)
m.errors_total.labels(service="test_svc", error_type="timeout").inc()
assert m.errors_total.labels(service="test_svc", error_type="timeout")._value.get() == 1.0
def test_events_processed_increment():
"""events_processed counter can be incremented with labels."""
registry = CollectorRegistry()
m = make_metrics("test-svc", registry=registry)
m.events_processed.labels(service="test_svc", event_type="candle").inc(5)
assert m.events_processed.labels(service="test_svc", event_type="candle")._value.get() == 5.0
def test_processing_seconds_observe():
"""processing_seconds histogram can observe values."""
registry = CollectorRegistry()
m = make_metrics("test-svc", registry=registry)
m.processing_seconds.labels(service="test_svc").observe(0.5)
m.processing_seconds.labels(service="test_svc").observe(1.5)
# Sum should be 2.0
assert m.processing_seconds.labels(service="test_svc")._sum.get() == 2.0
def test_service_up_gauge():
"""service_up gauge can be set."""
registry = CollectorRegistry()
m = make_metrics("test-svc", registry=registry)
m.service_up.labels(service="test_svc").set(1)
assert m.service_up.labels(service="test_svc")._value.get() == 1.0
m.service_up.labels(service="test_svc").set(0)
assert m.service_up.labels(service="test_svc")._value.get() == 0.0
|
