Developer
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
"""Tests for observability metrics helpers."""
|
|
|
|
import time
|
|
|
|
from prometheus_client import REGISTRY
|
|
|
|
from media_library_viewer_api.observability import record_backup_run
|
|
|
|
|
|
def _metric_samples(metric_name):
|
|
return {
|
|
tuple(s.labels.values()): s.value
|
|
for family in REGISTRY.collect()
|
|
for s in family.samples
|
|
if s.name == metric_name
|
|
}
|
|
|
|
|
|
def test_record_backup_run_increments_counter():
|
|
record_backup_run("job-a", "success", success=True)
|
|
record_backup_run("job-a", "failure")
|
|
record_backup_run("job-b", "success", success=True)
|
|
|
|
samples = _metric_samples("manage_backup_runs_total")
|
|
assert samples[("job-a", "success")] >= 1.0
|
|
assert samples[("job-a", "failure")] >= 1.0
|
|
assert samples[("job-b", "success")] >= 1.0
|
|
|
|
|
|
def test_record_backup_run_sets_last_success_timestamp():
|
|
before = time.time()
|
|
record_backup_run("job-ts", "success", success=True)
|
|
after = time.time()
|
|
|
|
samples = _metric_samples("manage_backup_runs_last_success_timestamp")
|
|
value = samples[("job-ts",)]
|
|
assert before <= value <= after
|