Developer
1e636fdbe2
Three reusable-widget follow-up fixes:
1. Reference sort_order independently reorderable. Reordering a
referenced widget now updates the widget_references.sort_order (per-
dashboard), not the shared widget instance sort_order. New backend
update_widget_reference method + PUT /api/widgets/references/{id}
endpoint. Frontend moveInstance checks _ref_id to choose the right
mutation (updateRef for references, saveWidget for owned).
2. Detach preserves service_id. detach_widget_reference now copies the
original widget's service_id into the clone, so service-bound widgets
(Grafana chart, Jellyfin activity) continue to render after detach.
3. Named dashboards support widget references. NamedDashboardPage
fetches useWidgetReferences('named:<slug>') and renders them via
WidgetInstanceCard alongside pinned links. 'Edit widgets' button
opens WidgetConfigDialog with dashboardScope='named:<slug>'.
Also: removed useMemo on combinedWidgets in WidgetConfigDialog to fix
a react-hooks/preserve-manual-memoization lint error (the React Compiler
ESLint plugin couldn't verify the spread+sort memoization).
283 backend tests pass (+1 update_reference test); 128 frontend tests
pass; ruff clean; 0 lint errors.
927 lines
31 KiB
Python
927 lines
31 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Widget references (live-link widgets across dashboards)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def widget_ref_client(monkeypatch):
|
|
"""TestClient with an isolated SettingsStore + encryption key."""
|
|
monkeypatch.setenv(
|
|
"MANAGE_ENCRYPTION_KEY",
|
|
Fernet.generate_key().decode(),
|
|
)
|
|
from media_library_viewer_api.services.secrets import reset_encryption_key_cache
|
|
|
|
reset_encryption_key_cache()
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
store = SettingsStore(str(Path(tempfile.mkdtemp()) / "test.db"))
|
|
store.ensure_defaults()
|
|
|
|
def get_store_override():
|
|
return store
|
|
|
|
app.dependency_overrides[get_settings_store] = get_store_override
|
|
client = TestClient(app)
|
|
yield client, store
|
|
app.dependency_overrides.pop(get_settings_store, None)
|
|
|
|
|
|
def test_widget_reference_lifecycle(widget_ref_client):
|
|
"""Create a widget, reference it on 'main', verify it appears, delete reference."""
|
|
client, store = widget_ref_client
|
|
|
|
# Create a service-bound widget (simulating one on a Grafana Overview).
|
|
store.upsert_service(
|
|
{
|
|
"service_type": "grafana",
|
|
"name": "Grafana",
|
|
"config": {"base_url": "https://grafana.example.com"},
|
|
"secrets": {"api_key": "tok"},
|
|
"enabled": True,
|
|
},
|
|
)
|
|
service = store.list_services("grafana")[0]
|
|
widget = store.upsert_widget(
|
|
{
|
|
"service_id": service["id"],
|
|
"widget_kind": "chart",
|
|
"title": "CPU IOWait",
|
|
"config": {"query": "rate(cpu[5m])", "datasource_uid": "prometheus"},
|
|
"enabled": True,
|
|
"sort_order": 0,
|
|
}
|
|
)
|
|
|
|
# Reference it on "main" dashboard.
|
|
resp = client.post(
|
|
"/api/widgets/references",
|
|
json={
|
|
"dashboard_scope": "main",
|
|
"widget_id": widget["id"],
|
|
"sort_order": 5,
|
|
},
|
|
)
|
|
assert resp.status_code == 201
|
|
ref = resp.json()
|
|
assert ref["dashboard_scope"] == "main"
|
|
assert ref["widget_id"] == widget["id"]
|
|
ref_id = ref["id"]
|
|
|
|
# List references for "main" — should include our widget.
|
|
resp = client.get("/api/widgets/references", params={"dashboard_scope": "main"})
|
|
assert resp.status_code == 200
|
|
refs = resp.json()
|
|
assert len(refs) == 1
|
|
assert refs[0]["widget"]["title"] == "CPU IOWait"
|
|
|
|
# Delete the reference.
|
|
resp = client.delete(f"/api/widgets/references/{ref_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "deleted"
|
|
|
|
# Reference is gone, original widget still exists.
|
|
resp = client.get("/api/widgets/references", params={"dashboard_scope": "main"})
|
|
assert len(resp.json()) == 0
|
|
assert store.get_widget(widget["id"]) is not None
|
|
|
|
|
|
def test_widget_reference_detach(widget_ref_client):
|
|
"""Detach clones the widget into a standalone instance and removes the reference."""
|
|
client, store = widget_ref_client
|
|
|
|
store.upsert_service(
|
|
{
|
|
"service_type": "grafana",
|
|
"name": "Grafana",
|
|
"config": {"base_url": "https://grafana.example.com"},
|
|
"secrets": {"api_key": "tok"},
|
|
"enabled": True,
|
|
},
|
|
)
|
|
service = store.list_services("grafana")[0]
|
|
widget = store.upsert_widget(
|
|
{
|
|
"service_id": service["id"],
|
|
"widget_kind": "chart",
|
|
"title": "Memory",
|
|
"config": {"query": "mem", "datasource_uid": "prometheus"},
|
|
"enabled": True,
|
|
"sort_order": 0,
|
|
}
|
|
)
|
|
|
|
# Reference on "main".
|
|
resp = client.post(
|
|
"/api/widgets/references",
|
|
json={
|
|
"dashboard_scope": "main",
|
|
"widget_id": widget["id"],
|
|
},
|
|
)
|
|
ref_id = resp.json()["id"]
|
|
|
|
# Detach.
|
|
resp = client.post(f"/api/widgets/references/{ref_id}/detach")
|
|
assert resp.status_code == 200
|
|
cloned = resp.json()
|
|
assert cloned["title"] == "Memory"
|
|
assert cloned["widget_kind"] == "chart"
|
|
assert cloned["service_id"] == service["id"] # Fix 2: preserves service binding
|
|
assert cloned["config"]["query"] == "mem"
|
|
assert cloned["id"] != widget["id"] # new independent widget
|
|
|
|
# Reference is gone.
|
|
refs = client.get("/api/widgets/references", params={"dashboard_scope": "main"}).json()
|
|
assert len(refs) == 0
|
|
# Original still exists.
|
|
assert store.get_widget(widget["id"]) is not None
|
|
|
|
|
|
def test_widget_reference_update_sort_order(widget_ref_client):
|
|
"""PUT /references/{id} updates only the reference's sort_order (Fix 1)."""
|
|
client, store = widget_ref_client
|
|
|
|
widget_a = store.upsert_widget(
|
|
{
|
|
"service_id": None,
|
|
"widget_kind": "static",
|
|
"title": "A",
|
|
"config": {"text": "a"},
|
|
"enabled": True,
|
|
"sort_order": 0,
|
|
}
|
|
)
|
|
widget_b = store.upsert_widget(
|
|
{
|
|
"service_id": None,
|
|
"widget_kind": "static",
|
|
"title": "B",
|
|
"config": {"text": "b"},
|
|
"enabled": True,
|
|
"sort_order": 1,
|
|
}
|
|
)
|
|
|
|
# Two references on the same dashboard scope.
|
|
resp = client.post(
|
|
"/api/widgets/references",
|
|
json={
|
|
"dashboard_scope": "named:test",
|
|
"widget_id": widget_a["id"],
|
|
"sort_order": 0,
|
|
},
|
|
)
|
|
ref_a = resp.json()
|
|
resp = client.post(
|
|
"/api/widgets/references",
|
|
json={
|
|
"dashboard_scope": "named:test",
|
|
"widget_id": widget_b["id"],
|
|
"sort_order": 1,
|
|
},
|
|
)
|
|
ref_b = resp.json()
|
|
|
|
# Swap sort orders via PUT (per-dashboard reorder).
|
|
resp = client.put(f"/api/widgets/references/{ref_a['id']}?sort_order=1")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["sort_order"] == 1
|
|
|
|
resp = client.put(f"/api/widgets/references/{ref_b['id']}?sort_order=0")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["sort_order"] == 0
|
|
|
|
# Widget instances themselves are unchanged.
|
|
assert store.get_widget(widget_a["id"])["sort_order"] == 0
|
|
assert store.get_widget(widget_b["id"])["sort_order"] == 1
|