"""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, normalize_prometheus_matrix, step_for_window, ) class TestStepForWindow: """SC-104: every preset must yield 100–300 points.""" @pytest.mark.parametrize("preset", sorted(WINDOW_PRESETS)) def test_presets_yield_in_band_point_counts(self, preset: str) -> None: window = WINDOW_PRESETS[preset] step = step_for_window(window) # Clamped minimum. assert step >= 15 point_count = window // step assert 100 <= point_count <= 300, f"{preset}: {point_count} points (step={step})" 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}, ]