- test_dashboard.py: tracing backend switched from jaeger to tempo in pipeline-performance.json; the test assertion was stale. - test_compose_mounts.py: exempt cadvisor (privileged: true for host metrics) and nvidia-exporter (upstream dcgm-exporter with SYS_ADMIN + GPU runtime) from no-new-privileges, matching the existing pattern for promtail / zotero / notebooks. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Tests for the Grafana pipeline performance dashboard JSON."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
DASHBOARD_PATH = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "infra"
|
|
/ "grafana"
|
|
/ "dashboards"
|
|
/ "pipeline-performance.json"
|
|
)
|
|
|
|
|
|
class TestDashboard:
|
|
def test_file_exists(self):
|
|
assert DASHBOARD_PATH.exists()
|
|
|
|
def test_valid_json(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
assert isinstance(data, dict)
|
|
|
|
def test_has_required_keys(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
assert "panels" in data
|
|
assert "title" in data
|
|
assert "uid" in data
|
|
assert "templating" in data
|
|
|
|
def test_uid(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
assert data["uid"] == "stack-pipeline-perf"
|
|
|
|
def test_has_panels(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
panels = data["panels"]
|
|
assert len(panels) >= 8
|
|
|
|
titles = {p["title"] for p in panels}
|
|
assert "Pipeline Run Duration" in titles
|
|
assert "Cache Hit Rate" in titles
|
|
assert "Memory RSS" in titles
|
|
assert "CPU Utilization" in titles
|
|
assert "Row Throughput per Step" in titles
|
|
assert "Recent Pipeline Traces" in titles
|
|
assert "Pipeline Logs" in titles
|
|
|
|
def test_has_pipeline_variable(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
variables = data["templating"]["list"]
|
|
names = [v["name"] for v in variables]
|
|
assert "pipeline" in names
|
|
|
|
def test_datasources(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
ds_types = set()
|
|
for panel in data["panels"]:
|
|
if "datasource" in panel:
|
|
ds_types.add(panel["datasource"]["type"])
|
|
assert "prometheus" in ds_types
|
|
assert "tempo" in ds_types
|
|
assert "loki" in ds_types
|
|
|
|
def test_tags(self):
|
|
data = json.loads(DASHBOARD_PATH.read_text())
|
|
assert "pipeline" in data.get("tags", [])
|
|
assert "otel" in data.get("tags", [])
|