Developer
94bf830955
Two fixes for the Grafana chart widget: 1. Invisible lines: the CHART_COLORS used 'hsl(var(--chart-1))' but the CSS variable is named '--color-chart-1' and already contains a hex color (#4f8cff). The hsl() wrapper produced invalid CSS, making every stroke invisible. Fixed to var(--color-chart-1). 2. Multiple series collision: the backend labeled all Prometheus series with the value field name (often just 'Value'), so multiple time series collided on the same recharts dataKey and overwrote each other. Now extracts meaningful labels from the Grafana frame metadata: prefers displayName, then Prometheus metric labels (e.g. 'instance=server1:9100 mode=iowait'), then falls back to the field name. Duplicate labels get a numeric suffix for uniqueness. Multi-series queries now render correctly: each Prometheus time series gets its own colored line with a unique label in the legend/tooltip. 280 backend tests pass (+1 labels test); 127 frontend tests pass; ruff/ eslint clean.
727 lines
24 KiB
Python
727 lines
24 KiB
Python
"""Tests for the dashboard widget system: service-bound + built-in widgets."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
from fastapi.testclient import TestClient
|
|
|
|
from media_library_viewer_api.dependencies import get_settings_store
|
|
from media_library_viewer_api.main import app
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
from media_library_viewer_api.widgets.sources import (
|
|
AlertmanagerWidgetSource,
|
|
BackupsWidgetSource,
|
|
GrafanaWidgetSource,
|
|
JellyfinWidgetSource,
|
|
ServiceRecord,
|
|
StaticWidgetSource,
|
|
)
|
|
|
|
TEST_KEY = Fernet.generate_key().decode()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _encryption_key(monkeypatch):
|
|
monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", TEST_KEY)
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
"""FastAPI test client with a fresh settings store and auth disabled."""
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
app.dependency_overrides[get_settings_store] = lambda: store
|
|
auth_settings = SimpleNamespace(auth_enabled=False)
|
|
with patch("media_library_viewer_api.auth.get_settings", return_value=auth_settings):
|
|
yield TestClient(app)
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def _make_grafana_service(client, name="Production Grafana", **config_overrides):
|
|
config = {"base_url": "https://grafana.example.com"}
|
|
config.update(config_overrides)
|
|
return client.post(
|
|
"/api/services/instances",
|
|
json={"service_type": "grafana", "name": name, "config": config, "enabled": True},
|
|
).json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Built-in kinds + built-in widget CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_builtin_kinds(client):
|
|
response = client.get("/api/widgets/builtin")
|
|
assert response.status_code == 200
|
|
kinds = {item["kind"] for item in response.json()}
|
|
assert kinds == {"backups", "static"}
|
|
|
|
|
|
def test_create_and_read_static_widget(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"widget_kind": "static",
|
|
"title": "Note",
|
|
"config": {"text": "hello"},
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
created = response.json()
|
|
assert created["widget_kind"] == "static"
|
|
assert created["service_id"] is None
|
|
assert created["config"]["text"] == "hello"
|
|
|
|
listed = client.get("/api/widgets/instances").json()
|
|
assert len(listed) == 1
|
|
assert listed[0]["id"] == created["id"]
|
|
|
|
|
|
def test_create_backups_widget(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "backups", "title": "Backups", "config": {}},
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
|
|
def test_widget_filtering_by_service_id_and_scope(client):
|
|
"""Test ?service_id= and ?scope= query params on GET /api/widgets/instances."""
|
|
service = _make_grafana_service(client)
|
|
# Create a dashboard-scoped (built-in) widget + a service-scoped widget.
|
|
client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "static", "title": "Note", "config": {"text": "hi"}},
|
|
)
|
|
client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "Dash",
|
|
"config": {"dashboard_uid": "o"},
|
|
},
|
|
)
|
|
|
|
# No filter: both widgets.
|
|
all_widgets = client.get("/api/widgets/instances").json()
|
|
assert len(all_widgets) == 2
|
|
|
|
# Filter by service_id: only the service-scoped one.
|
|
by_service = client.get(f"/api/widgets/instances?service_id={service['id']}").json()
|
|
assert len(by_service) == 1
|
|
assert by_service[0]["service_id"] == service["id"]
|
|
|
|
# scope=dashboard: only the built-in (NULL service_id).
|
|
dashboard_scope = client.get("/api/widgets/instances?scope=dashboard").json()
|
|
assert len(dashboard_scope) == 1
|
|
assert dashboard_scope[0]["service_id"] is None
|
|
|
|
# scope=service: only the non-null service_id widget.
|
|
service_scope = client.get("/api/widgets/instances?scope=service").json()
|
|
assert len(service_scope) == 1
|
|
assert service_scope[0]["service_id"] == service["id"]
|
|
|
|
|
|
def test_unknown_builtin_kind_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "bogus", "title": "x", "config": {}},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_credential_key_in_config_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "static", "title": "x", "config": {"api_key": "leak"}},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Service-bound widget CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_service_bound_widget(client):
|
|
service = _make_grafana_service(client)
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "Dashboard",
|
|
"config": {"dashboard_uid": "overview"},
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
created = response.json()
|
|
assert created["service_id"] == service["id"]
|
|
assert created["widget_kind"] == "link"
|
|
|
|
|
|
def test_service_bound_widget_unknown_kind_rejected(client):
|
|
service = _make_grafana_service(client)
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "metric",
|
|
"title": "x",
|
|
"config": {},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_service_bound_widget_service_not_found_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": "missing",
|
|
"widget_kind": "link",
|
|
"title": "x",
|
|
"config": {"dashboard_uid": "u"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_service_bound_widget_invalid_config_rejected(client):
|
|
service = _make_grafana_service(client)
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "x",
|
|
"config": {"dashboard_uid": ""}, # empty still validates; use bad type
|
|
},
|
|
)
|
|
# Empty string passes Pydantic; force a real failure with a bad type.
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "x",
|
|
"config": {"dashboard_uid": 123},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_update_and_delete_widget(client):
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "static", "title": "Note", "config": {"text": "a"}},
|
|
).json()
|
|
updated = client.put(
|
|
f"/api/widgets/instances/{created['id']}",
|
|
json={"widget_kind": "static", "title": "Note2", "config": {"text": "b"}},
|
|
).json()
|
|
assert updated["title"] == "Note2"
|
|
|
|
assert client.delete(f"/api/widgets/instances/{created['id']}").status_code == 200
|
|
assert client.get("/api/widgets/instances").json() == []
|
|
|
|
|
|
def test_update_nonexistent_returns_404(client):
|
|
response = client.put(
|
|
"/api/widgets/instances/missing",
|
|
json={"widget_kind": "static", "title": "x", "config": {}},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_update_id_mismatch_returns_400(client):
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "static", "title": "x", "config": {}},
|
|
).json()
|
|
response = client.put(
|
|
f"/api/widgets/instances/{created['id']}",
|
|
json={"id": "other", "widget_kind": "static", "title": "x", "config": {}},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_fetch_static_widget_data(client):
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "static", "title": "Note", "config": {"text": "hello"}},
|
|
).json()
|
|
response = client.get(f"/api/widgets/instances/{created['id']}/data")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["data"]["text"] == "hello"
|
|
assert body["error"] is None
|
|
|
|
|
|
def test_fetch_backups_widget_data(client):
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={"widget_kind": "backups", "title": "Backups", "config": {}},
|
|
).json()
|
|
response = client.get(f"/api/widgets/instances/{created['id']}/data")
|
|
assert response.status_code == 200
|
|
assert "total_jobs" in response.json()["data"]
|
|
|
|
|
|
def test_fetch_grafana_link_widget_data(client):
|
|
service = _make_grafana_service(client)
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "Dashboard",
|
|
"config": {"dashboard_uid": "overview", "panel_id": 2},
|
|
},
|
|
).json()
|
|
response = client.get(f"/api/widgets/instances/{created['id']}/data")
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["url"] == "https://grafana.example.com/d/overview?viewPanel=2"
|
|
|
|
|
|
def test_fetch_widget_service_not_found(client):
|
|
service = _make_grafana_service(client)
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "x",
|
|
"config": {"dashboard_uid": "u"},
|
|
},
|
|
).json()
|
|
# Deleting the service cascade-deletes its widgets, so the widget is gone.
|
|
client.delete(f"/api/services/instances/{service['id']}")
|
|
assert client.get("/api/widgets/instances").json() == []
|
|
assert client.get(f"/api/widgets/instances/{created['id']}/data").status_code == 404
|
|
|
|
|
|
def test_fetch_widget_service_disabled(client):
|
|
service = _make_grafana_service(client)
|
|
created = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"service_id": service["id"],
|
|
"widget_kind": "link",
|
|
"title": "x",
|
|
"config": {"dashboard_uid": "u"},
|
|
},
|
|
).json()
|
|
client.put(
|
|
f"/api/services/instances/{service['id']}",
|
|
json={
|
|
"service_type": "grafana",
|
|
"name": service["name"],
|
|
"config": {"base_url": "https://grafana.example.com"},
|
|
"enabled": False,
|
|
},
|
|
)
|
|
response = client.get(f"/api/widgets/instances/{created['id']}/data")
|
|
assert response.status_code == 200
|
|
assert "disabled" in response.json()["error"]
|
|
|
|
|
|
def test_fetch_widget_not_found(client):
|
|
assert client.get("/api/widgets/instances/missing/data").status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter_builds_url():
|
|
adapter = GrafanaWidgetSource()
|
|
service = ServiceRecord(id="s", service_type="grafana", name="g", config={"base_url": "http://g:3000"})
|
|
result = await adapter.fetch(service, "link", {"dashboard_uid": "ov"})
|
|
assert result["url"] == "http://g:3000/d/ov"
|
|
result = await adapter.fetch(service, "link", {"dashboard_uid": "ov", "panel_id": 4})
|
|
assert result["url"] == "http://g:3000/d/ov?viewPanel=4"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter_missing_service():
|
|
adapter = GrafanaWidgetSource()
|
|
result = await adapter.fetch(None, "link", {"dashboard_uid": "ov"})
|
|
assert "error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alertmanager_adapter_summarizes_alerts():
|
|
adapter = AlertmanagerWidgetSource()
|
|
service = ServiceRecord(id="s", service_type="alertmanager", name="am", config={"base_url": "http://am:9093"})
|
|
payload = SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {
|
|
"status": "success",
|
|
"data": [
|
|
{
|
|
"labels": {"alertname": "DiskFull", "severity": "critical"},
|
|
"annotations": {"summary": "disk full"},
|
|
"startsAt": "2026-06-23T00:00:00Z",
|
|
"status": "firing",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
with patch("media_library_viewer_api.widgets.sources.requests.get", return_value=payload):
|
|
result = await adapter.fetch(service, "active_alerts", {})
|
|
assert result["total"] == 1
|
|
assert result["by_severity"]["critical"] == 1
|
|
assert result["alerts"][0]["name"] == "DiskFull"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alertmanager_adapter_applies_severity_filter():
|
|
adapter = AlertmanagerWidgetSource()
|
|
service = ServiceRecord(id="s", service_type="alertmanager", name="am", config={"base_url": "http://am:9093"})
|
|
payload = SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {
|
|
"status": "success",
|
|
"data": [
|
|
{"labels": {"alertname": "A", "severity": "critical"}, "annotations": {}, "status": "firing"},
|
|
{"labels": {"alertname": "B", "severity": "warning"}, "annotations": {}, "status": "firing"},
|
|
],
|
|
},
|
|
)
|
|
with patch("media_library_viewer_api.widgets.sources.requests.get", return_value=payload):
|
|
result = await adapter.fetch(service, "active_alerts", {"severity_filter": "critical"})
|
|
assert result["total"] == 1
|
|
assert result["alerts"][0]["name"] == "A"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alertmanager_adapter_sends_bearer_token():
|
|
adapter = AlertmanagerWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="alertmanager",
|
|
name="am",
|
|
config={"base_url": "http://am:9093"},
|
|
secrets={"api_key": "tok"},
|
|
)
|
|
payload = SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"status": "success", "data": []})
|
|
with patch("media_library_viewer_api.widgets.sources.requests.get", return_value=payload) as mock_get:
|
|
result = await adapter.fetch(service, "active_alerts", {})
|
|
assert result["total"] == 0
|
|
_, kwargs = mock_get.call_args
|
|
assert kwargs["headers"]["Authorization"] == "Bearer tok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alertmanager_adapter_missing_service():
|
|
adapter = AlertmanagerWidgetSource()
|
|
result = await adapter.fetch(None, "active_alerts", {})
|
|
assert "error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_static_adapter():
|
|
adapter = StaticWidgetSource()
|
|
result = await adapter.fetch(None, "static", {"text": "hi"})
|
|
assert result == {"text": "hi"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backups_adapter(client):
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
with patch("media_library_viewer_api.widgets.sources.get_settings_store", return_value=store):
|
|
adapter = BackupsWidgetSource()
|
|
result = await adapter.fetch(None, "backups", {})
|
|
assert "total_jobs" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssh_task_adapter_missing_service():
|
|
from media_library_viewer_api.widgets.sources import SshTaskWidgetSource
|
|
|
|
adapter = SshTaskWidgetSource()
|
|
result = await adapter.fetch(None, "task_output", {"task_id": "t1"})
|
|
assert "error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssh_task_adapter_records_history_on_run(client):
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
# Save a task and an ssh_tasks service instance.
|
|
task = store.upsert_task(
|
|
{
|
|
"name": "echo",
|
|
"task_type": "shell",
|
|
"content": "echo hi",
|
|
"enabled": True,
|
|
"default_service_id": "",
|
|
}
|
|
)
|
|
service = store.upsert_service(
|
|
{"service_type": "ssh_tasks", "name": "box", "config": {"host": "h", "username": "u"}, "enabled": True}
|
|
)
|
|
|
|
fake_result = SimpleNamespace(exit_status=0, stdout="hi\n", stderr="")
|
|
fake_client = SimpleNamespace(run=lambda *a, **k: fake_result)
|
|
|
|
from media_library_viewer_api.widgets.sources import SshTaskWidgetSource
|
|
|
|
adapter = SshTaskWidgetSource()
|
|
service_record = ServiceRecord(
|
|
id=service["id"], service_type="ssh_tasks", name="box", config={"host": "h", "username": "u"}
|
|
)
|
|
with (
|
|
patch("media_library_viewer_api.widgets.sources.get_settings_store", return_value=store),
|
|
patch("media_library_viewer_api.services.task_runner.build_ssh_client", return_value=fake_client),
|
|
):
|
|
result = await adapter.fetch(service_record, "task_output", {"task_id": task["id"]})
|
|
|
|
assert result["exit_status"] == 0
|
|
runs = store.list_service_task_runs(service_id=service["id"])
|
|
assert len(runs) == 1
|
|
assert runs[0]["status"] == "success"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New widget kind tests (jellyfin now_playing + grafana panel)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_jellyfin_definition_has_now_playing_widget():
|
|
from media_library_viewer_api.integrations.registry import get_service_definition
|
|
|
|
definition = get_service_definition("jellyfin")
|
|
kinds = {wk.kind for wk in definition.widget_kinds}
|
|
assert "now_playing" in kinds
|
|
assert "activity" in kinds
|
|
|
|
|
|
def test_grafana_definition_has_chart_widget():
|
|
from media_library_viewer_api.integrations.registry import get_service_definition
|
|
|
|
definition = get_service_definition("grafana")
|
|
kinds = {wk.kind for wk in definition.widget_kinds}
|
|
assert "chart" in kinds
|
|
assert "link" in kinds
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter_chart_queries_datasource():
|
|
"""Chart widget should POST to /api/ds/query and normalize the response."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
adapter = GrafanaWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="grafana",
|
|
name="g",
|
|
config={"base_url": "http://g:3000", "timeout_seconds": 5},
|
|
secrets={"api_key": "tok"},
|
|
)
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = {
|
|
"results": {
|
|
"A": {
|
|
"frames": [
|
|
{
|
|
"data": {"values": [[1000, 2000], [0.5, 0.8]]},
|
|
"schema": {"fields": [{"name": "Time"}, {"name": "cpu_usage"}]},
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=mock_resp):
|
|
result = await adapter.fetch(
|
|
service,
|
|
"chart",
|
|
{"query": "rate(cpu[5m])", "datasource_uid": "prometheus"},
|
|
)
|
|
|
|
assert "series" in result
|
|
assert len(result["series"]) == 1
|
|
assert result["series"][0]["label"] == "cpu_usage"
|
|
assert result["series"][0]["points"] == [
|
|
{"t": 1000, "v": 0.5},
|
|
{"t": 2000, "v": 0.8},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter_chart_extracts_prometheus_labels():
|
|
"""Multiple Prometheus series should get unique labels from frame metadata."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
adapter = GrafanaWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="grafana",
|
|
name="g",
|
|
config={"base_url": "http://g:3000", "timeout_seconds": 5},
|
|
secrets={"api_key": "tok"},
|
|
)
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = {
|
|
"results": {
|
|
"A": {
|
|
"frames": [
|
|
{
|
|
"data": {"values": [[1000], [0.5]]},
|
|
"schema": {
|
|
"fields": [
|
|
{"name": "Time"},
|
|
{
|
|
"name": "Value",
|
|
"labels": {
|
|
"instance": "server1:9100",
|
|
"mode": "iowait",
|
|
},
|
|
},
|
|
]
|
|
},
|
|
},
|
|
{
|
|
"data": {"values": [[1000], [0.3]]},
|
|
"schema": {
|
|
"fields": [
|
|
{"name": "Time"},
|
|
{
|
|
"name": "Value",
|
|
"labels": {
|
|
"instance": "server2:9100",
|
|
"mode": "iowait",
|
|
},
|
|
},
|
|
]
|
|
},
|
|
},
|
|
]
|
|
}
|
|
}
|
|
}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=mock_resp):
|
|
result = await adapter.fetch(service, "chart", {"query": "rate(cpu[5m])"})
|
|
|
|
assert len(result["series"]) == 2
|
|
assert result["series"][0]["label"] == "instance=server1:9100 mode=iowait"
|
|
assert result["series"][1]["label"] == "instance=server2:9100 mode=iowait"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter_chart_requires_api_key():
|
|
adapter = GrafanaWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="grafana",
|
|
name="g",
|
|
config={"base_url": "http://g:3000"},
|
|
)
|
|
result = await adapter.fetch(service, "chart", {"query": "up"})
|
|
assert "error" in result
|
|
assert "api_key" in result["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter_chart_handles_http_failure():
|
|
from unittest.mock import patch
|
|
|
|
import requests as req_mod
|
|
|
|
adapter = GrafanaWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="grafana",
|
|
name="g",
|
|
config={"base_url": "http://g:3000", "timeout_seconds": 2},
|
|
secrets={"api_key": "tok"},
|
|
)
|
|
|
|
with patch(
|
|
"media_library_viewer_api.widgets.sources.requests.post",
|
|
side_effect=req_mod.ConnectionError("refused"),
|
|
):
|
|
result = await adapter.fetch(service, "chart", {"query": "up"})
|
|
|
|
assert "error" in result
|
|
assert "failed" in result["error"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jellyfin_now_playing_filters_active_sessions():
|
|
"""now_playing should exclude idle (no NowPlayingItem) and paused sessions."""
|
|
adapter = JellyfinWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="jellyfin",
|
|
name="jf",
|
|
config={"base_url": "http://jf:8096"},
|
|
secrets={"api_key": "k"},
|
|
)
|
|
playing_session = {
|
|
"UserName": "alice",
|
|
"NowPlayingItem": {"Name": "Movie", "Type": "Movie"},
|
|
"PlayState": {"IsPaused": False},
|
|
"DeviceName": "Web",
|
|
}
|
|
paused_session = {
|
|
"UserName": "bob",
|
|
"NowPlayingItem": {"Name": "Show", "Type": "Episode"},
|
|
"PlayState": {"IsPaused": True},
|
|
"DeviceName": "TV",
|
|
}
|
|
idle_session = {
|
|
"UserName": "carol",
|
|
"PlayState": {"IsPaused": False},
|
|
"DeviceName": "Phone",
|
|
}
|
|
mock_client = SimpleNamespace(sessions=lambda: [playing_session, paused_session, idle_session])
|
|
with patch("media_library_viewer_api.widgets.sources.JellyfinClient", return_value=mock_client):
|
|
result = await adapter.fetch(service, "now_playing", {})
|
|
sessions = result["sessions"]
|
|
assert len(sessions) == 1
|
|
assert sessions[0]["user"] == "alice"
|
|
assert sessions[0]["state"] == "playing"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jellyfin_activity_shows_all_sessions():
|
|
"""activity (default) should include idle and paused sessions."""
|
|
adapter = JellyfinWidgetSource()
|
|
service = ServiceRecord(
|
|
id="s",
|
|
service_type="jellyfin",
|
|
name="jf",
|
|
config={"base_url": "http://jf:8096"},
|
|
secrets={"api_key": "k"},
|
|
)
|
|
mock_client = SimpleNamespace(
|
|
sessions=lambda: [
|
|
{"UserName": "alice", "NowPlayingItem": {"Name": "M"}, "PlayState": {"IsPaused": False}},
|
|
{"UserName": "bob", "PlayState": {"IsPaused": False}},
|
|
]
|
|
)
|
|
with patch("media_library_viewer_api.widgets.sources.JellyfinClient", return_value=mock_client):
|
|
result = await adapter.fetch(service, "activity", {})
|
|
assert len(result["sessions"]) == 2
|