Developer
275 lines
9.2 KiB
Python
275 lines
9.2 KiB
Python
"""Unit tests for the shared Prometheus range-query helpers (SC-101..SC-104)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from media_library_viewer_api.widgets.prometheus_range import (
|
||
WINDOW_PRESETS,
|
||
_dedup_label,
|
||
normalize_grafana_frames,
|
||
normalize_prometheus_matrix,
|
||
step_for_window,
|
||
)
|
||
|
||
|
||
class TestStepForWindow:
|
||
"""SC-104: presets preserve usable resolution without sub-15s steps."""
|
||
|
||
@pytest.mark.parametrize("preset", sorted(WINDOW_PRESETS))
|
||
def test_presets_yield_supported_point_counts(self, preset: str) -> None:
|
||
window = WINDOW_PRESETS[preset]
|
||
step = step_for_window(window)
|
||
# The Prometheus-safe 15-second floor limits the two short presets to
|
||
# 20 and 60 points; all longer windows stay in the 100–300 target band.
|
||
assert step >= 15
|
||
point_count = window // step
|
||
assert min(100, window // 15) <= point_count <= 300, f"{preset}: {point_count} points (step={step})"
|
||
|
||
def test_window_presets_cover_the_shared_chart_windows(self) -> None:
|
||
assert WINDOW_PRESETS == {
|
||
"5m": 300,
|
||
"15m": 900,
|
||
"30m": 1_800,
|
||
"1h": 3_600,
|
||
"3h": 10_800,
|
||
"6h": 21_600,
|
||
"12h": 43_200,
|
||
"24h": 86_400,
|
||
"2d": 172_800,
|
||
"7d": 604_800,
|
||
"14d": 1_209_600,
|
||
"30d": 2_592_000,
|
||
}
|
||
|
||
def test_floor_of_fifteen_seconds(self) -> None:
|
||
# A tiny window that would otherwise produce a sub-15s step is clamped.
|
||
assert step_for_window(60) == 15
|
||
|
||
def test_custom_target_points(self) -> None:
|
||
# Targeting 100 points for 1h yields step 36 (3600/100).
|
||
assert step_for_window(3_600, target_points=100) == 36
|
||
|
||
|
||
class TestNormalizePrometheusMatrix:
|
||
"""SC-102: label rule + null handling + dedup."""
|
||
|
||
def test_empty_matrix(self) -> None:
|
||
assert normalize_prometheus_matrix([]) == []
|
||
|
||
def test_drops_dunder_labels_and_joins(self) -> None:
|
||
result = [
|
||
{
|
||
"metric": {"__name__": "node_cpu_seconds_total", "instance": "host:9100", "mode": "idle"},
|
||
"values": [[1_700_000_000, "12.5"], [1_700_000_030, "13.0"]],
|
||
}
|
||
]
|
||
out = normalize_prometheus_matrix(result)
|
||
assert len(out) == 1
|
||
assert out[0]["label"] == "instance=host:9100 mode=idle"
|
||
assert out[0]["points"] == [
|
||
{"t": 1_700_000_000, "v": 12.5},
|
||
{"t": 1_700_000_030, "v": 13.0},
|
||
]
|
||
|
||
def test_falls_back_to_value_when_no_labels(self) -> None:
|
||
result = [{"metric": {}, "values": [[100, "1"]]}]
|
||
out = normalize_prometheus_matrix(result)
|
||
assert out[0]["label"] == "value"
|
||
|
||
def test_dedup_collisions_with_suffix(self) -> None:
|
||
# Two series with identical visible labels get a "(1)" suffix on the 2nd.
|
||
result = [
|
||
{"metric": {"job": "x"}, "values": [[1, "1"]]},
|
||
{"metric": {"job": "x"}, "values": [[1, "2"]]},
|
||
]
|
||
out = normalize_prometheus_matrix(result)
|
||
labels = [s["label"] for s in out]
|
||
assert labels == ["job=x", "job=x (1)"]
|
||
|
||
def test_non_numeric_sentinels_become_none(self) -> None:
|
||
result = [
|
||
{
|
||
"metric": {"job": "x"},
|
||
"values": [
|
||
[1, "NaN"],
|
||
[2, "+Inf"],
|
||
[3, "-Inf"],
|
||
[4, "3.5"],
|
||
],
|
||
}
|
||
]
|
||
out = normalize_prometheus_matrix(result)
|
||
assert out[0]["points"] == [
|
||
{"t": 1, "v": None},
|
||
{"t": 2, "v": None},
|
||
{"t": 3, "v": None},
|
||
{"t": 4, "v": 3.5},
|
||
]
|
||
|
||
def test_malformed_values_are_ignored_not_raised(self) -> None:
|
||
result = [
|
||
{
|
||
"metric": {"job": "x"},
|
||
"values": [
|
||
[1, "3.5"],
|
||
["not-a-ts", "9"], # unusable timestamp → dropped
|
||
[3, "junk-value"], # unparseable value → v: None
|
||
],
|
||
}
|
||
]
|
||
out = normalize_prometheus_matrix(result)
|
||
assert out[0]["points"] == [
|
||
{"t": 1, "v": 3.5},
|
||
{"t": 3, "v": None},
|
||
]
|
||
|
||
|
||
class TestDedupLabel:
|
||
"""The shared label-dedup helper used by both normalizers (GM-104)."""
|
||
|
||
def test_first_use_returns_label_unchanged(self) -> None:
|
||
seen: dict[str, int] = {}
|
||
assert _dedup_label("value", seen) == "value"
|
||
assert seen == {"value": 0}
|
||
|
||
def test_collision_appends_suffix(self) -> None:
|
||
seen: dict[str, int] = {}
|
||
assert _dedup_label("job=x", seen) == "job=x"
|
||
assert _dedup_label("job=x", seen) == "job=x (1)"
|
||
assert _dedup_label("job=x", seen) == "job=x (2)"
|
||
|
||
def test_different_labels_dont_collide(self) -> None:
|
||
seen: dict[str, int] = {}
|
||
assert _dedup_label("a", seen) == "a"
|
||
assert _dedup_label("b", seen) == "b"
|
||
|
||
|
||
class TestNormalizeGrafanaFrames:
|
||
"""GM-104: frames normalizer recovered from 65bae95 + shared dedup."""
|
||
|
||
def test_empty_response(self) -> None:
|
||
assert normalize_grafana_frames({"results": {}}) == []
|
||
assert normalize_grafana_frames({}) == []
|
||
|
||
def test_single_frame_with_values(self) -> None:
|
||
raw = {
|
||
"results": {
|
||
"A": {
|
||
"frames": [
|
||
{
|
||
"data": {"values": [[1000, 2000], [1.5, 2.5]]},
|
||
"schema": {"fields": [{"name": "Time"}, {"name": "Value"}]},
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
out = normalize_grafana_frames(raw)
|
||
assert len(out) == 1
|
||
assert out[0]["label"] == "Value"
|
||
assert out[0]["points"] == [
|
||
{"t": 1000, "v": 1.5},
|
||
{"t": 2000, "v": 2.5},
|
||
]
|
||
|
||
def test_display_name_takes_priority(self) -> None:
|
||
raw = {
|
||
"results": {
|
||
"A": {
|
||
"frames": [
|
||
{
|
||
"data": {"values": [[100, 200], [0.75, 0.80]]},
|
||
"schema": {
|
||
"fields": [
|
||
{"name": "Time"},
|
||
{
|
||
"name": "Value",
|
||
"labels": {"instance": "host:9100"},
|
||
"config": {"displayName": "CPU Usage"},
|
||
},
|
||
]
|
||
},
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
out = normalize_grafana_frames(raw)
|
||
assert out[0]["label"] == "CPU Usage"
|
||
|
||
def test_labels_fallback_when_no_display_name(self) -> None:
|
||
raw = {
|
||
"results": {
|
||
"A": {
|
||
"frames": [
|
||
{
|
||
"data": {"values": [[100], [1.0]]},
|
||
"schema": {
|
||
"fields": [
|
||
{"name": "Time"},
|
||
{
|
||
"name": "Value",
|
||
"labels": {"__name__": "up", "instance": "h:9100"},
|
||
},
|
||
]
|
||
},
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
out = normalize_grafana_frames(raw)
|
||
assert out[0]["label"] == "instance=h:9100"
|
||
|
||
def test_falls_back_to_value_when_no_metadata(self) -> None:
|
||
raw = {
|
||
"results": {
|
||
"A": {
|
||
"frames": [
|
||
{
|
||
"data": {"values": [[100], [1.0]]},
|
||
"schema": {"fields": [{"name": "Time"}, {}]},
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
out = normalize_grafana_frames(raw)
|
||
assert out[0]["label"] == "value"
|
||
|
||
def test_dedup_collisions(self) -> None:
|
||
raw = {
|
||
"results": {
|
||
"A": {
|
||
"frames": [
|
||
{
|
||
"data": {"values": [[100], [1.0]]},
|
||
"schema": {"fields": [{}, {"name": "Value"}]},
|
||
},
|
||
{
|
||
"data": {"values": [[100], [2.0]]},
|
||
"schema": {"fields": [{}, {"name": "Value"}]},
|
||
},
|
||
]
|
||
}
|
||
}
|
||
}
|
||
out = normalize_grafana_frames(raw)
|
||
labels = [s["label"] for s in out]
|
||
assert labels == ["Value", "Value (1)"]
|
||
|
||
def test_skips_frames_with_insufficient_values(self) -> None:
|
||
raw = {
|
||
"results": {
|
||
"A": {
|
||
"frames": [
|
||
{"data": {"values": [[100]]}, "schema": {"fields": []}},
|
||
{"data": {"values": [[100], [1.0]]}, "schema": {"fields": [{}, {}]}},
|
||
]
|
||
}
|
||
}
|
||
}
|
||
out = normalize_grafana_frames(raw)
|
||
assert len(out) == 1
|