Developer
1cd8e926de
PR 2 of 4 for configurable dashboard widgets.
- Add grafana_url and prometheus_url settings (config.py + compose/env).
- Create WidgetSource protocol and adapters for jellyfin, backups, grafana,
prometheus, ssh_task, and static sources.
- Add GET /api/widgets/instances/{id}/data endpoint.
- Extract shared dashboard helpers into domain/dashboard.py so widgets and
the dashboard router reuse the same logic.
- Add adapter and data-endpoint tests.
- Update apply-progress.md.
Verification: ruff clean; backend pytest 200 passed; frontend lint/build green.
476 lines
13 KiB
Python
476 lines
13 KiB
Python
"""Tests for the dashboard widget backend: registry, CRUD, validation, seeding, adapters."""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
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 (
|
|
SOURCE_REGISTRY,
|
|
GrafanaWidgetSource,
|
|
SshTaskWidgetSource,
|
|
StaticWidgetSource,
|
|
)
|
|
|
|
|
|
@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 test_widget_sources(client):
|
|
response = client.get("/api/widgets/sources")
|
|
assert response.status_code == 200
|
|
assert set(response.json()) == {
|
|
"jellyfin",
|
|
"backups",
|
|
"grafana",
|
|
"prometheus",
|
|
"ssh_task",
|
|
"static",
|
|
}
|
|
|
|
|
|
def test_widget_types(client):
|
|
response = client.get("/api/widgets/types")
|
|
assert response.status_code == 200
|
|
types = {item["widget_type"] for item in response.json()}
|
|
assert types == {
|
|
"jellyfin",
|
|
"backups",
|
|
"grafana-link",
|
|
"prometheus-metric",
|
|
"ssh-task",
|
|
"static",
|
|
}
|
|
|
|
|
|
def test_create_and_read_widget(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Note",
|
|
"config": {"text": "hello"},
|
|
"enabled": True,
|
|
"sort_order": 5,
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
widget = response.json()
|
|
assert widget["title"] == "Note"
|
|
assert widget["config"] == {"text": "hello"}
|
|
assert widget["enabled"] is True
|
|
assert widget["sort_order"] == 5
|
|
widget_id = widget["id"]
|
|
|
|
response = client.get("/api/widgets/instances")
|
|
assert response.status_code == 200
|
|
assert any(w["id"] == widget_id for w in response.json())
|
|
|
|
|
|
def test_update_widget(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Note",
|
|
"config": {"text": "hello"},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.put(
|
|
f"/api/widgets/instances/{widget_id}",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Updated",
|
|
"config": {"text": "world"},
|
|
"enabled": False,
|
|
"sort_order": 10,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["title"] == "Updated"
|
|
assert data["config"] == {"text": "world"}
|
|
assert data["enabled"] is False
|
|
assert data["sort_order"] == 10
|
|
|
|
|
|
def test_delete_widget(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "To delete",
|
|
"config": {"text": "bye"},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.delete(f"/api/widgets/instances/{widget_id}")
|
|
assert response.status_code == 200
|
|
|
|
response = client.get("/api/widgets/instances")
|
|
assert not any(w["id"] == widget_id for w in response.json())
|
|
|
|
|
|
def test_unknown_widget_type_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "unknown",
|
|
"title": "Bad",
|
|
"config": {},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_addon_id_mismatch_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "grafana",
|
|
"widget_type": "static",
|
|
"title": "Bad",
|
|
"config": {"text": "x"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_credential_key_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Bad",
|
|
"config": {"api_key": "secret123"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_update_nonexistent_widget(client):
|
|
response = client.put(
|
|
"/api/widgets/instances/does-not-exist",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Bad",
|
|
"config": {"text": "x"},
|
|
},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_delete_nonexistent_widget(client):
|
|
response = client.delete("/api/widgets/instances/does-not-exist")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_default_widgets_seeded(client):
|
|
response = client.get("/api/widgets/instances")
|
|
assert response.status_code == 200
|
|
widgets = response.json()
|
|
types = [w["widget_type"] for w in widgets]
|
|
assert "jellyfin" in types
|
|
assert "backups" in types
|
|
|
|
|
|
def test_no_reseed_when_widgets_exist(tmp_path):
|
|
db_path = tmp_path / "settings.sqlite"
|
|
store = SettingsStore(db_path)
|
|
store.ensure_defaults()
|
|
widgets = store.list_widgets()
|
|
assert len(widgets) == 2
|
|
|
|
store.delete_widget(widgets[0]["id"])
|
|
store.ensure_defaults()
|
|
|
|
remaining = store.list_widgets()
|
|
assert len(remaining) == 1
|
|
|
|
|
|
def test_update_id_mismatch_returns_400(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Note",
|
|
"config": {"text": "hello"},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.put(
|
|
f"/api/widgets/instances/{widget_id}",
|
|
json={
|
|
"id": "different-id",
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Updated",
|
|
"config": {"text": "world"},
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_empty_title_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "",
|
|
"config": {"text": "hello"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_config_type_error_rejected(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "grafana",
|
|
"widget_type": "grafana-link",
|
|
"title": "Grafana",
|
|
"config": {"panel_id": "not-an-integer"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_list_instances_respects_sort_order(client):
|
|
response = client.get("/api/widgets/instances")
|
|
assert response.status_code == 200
|
|
widgets = response.json()
|
|
orders = [w["sort_order"] for w in widgets]
|
|
assert orders == sorted(orders)
|
|
|
|
|
|
def test_enabled_round_trip(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Toggle",
|
|
"config": {"text": "x"},
|
|
"enabled": False,
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.put(
|
|
f"/api/widgets/instances/{widget_id}",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Toggle",
|
|
"config": {"text": "x"},
|
|
"enabled": True,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["enabled"] is True
|
|
|
|
|
|
def test_fetch_static_widget_data(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Note",
|
|
"config": {"text": "hello world"},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.get(f"/api/widgets/instances/{widget_id}/data")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["widget_id"] == widget_id
|
|
assert data["widget_type"] == "static"
|
|
assert data["data"] == {"text": "hello world"}
|
|
assert data["error"] is None
|
|
assert isinstance(data["fetched_at"], int)
|
|
|
|
|
|
def test_fetch_grafana_widget_data(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "grafana",
|
|
"widget_type": "grafana-link",
|
|
"title": "Grafana",
|
|
"config": {"dashboard_uid": "overview", "panel_id": 3},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.get(f"/api/widgets/instances/{widget_id}/data")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["widget_type"] == "grafana-link"
|
|
assert data["data"]["url"] == "http://grafana:3000/d/overview?viewPanel=3"
|
|
|
|
|
|
def test_fetch_prometheus_widget_data(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "prometheus",
|
|
"widget_type": "prometheus-metric",
|
|
"title": "CPU",
|
|
"config": {"promql": '100 - avg(irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100'},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
fake_payload = {"data": {"resultType": "scalar", "result": [1718900000, "42.5"]}}
|
|
with patch("media_library_viewer_api.widgets.sources.requests.get") as mock_get:
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.return_value = None
|
|
mock_response.json.return_value = fake_payload
|
|
mock_get.return_value = mock_response
|
|
|
|
response = client.get(f"/api/widgets/instances/{widget_id}/data")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["widget_type"] == "prometheus-metric"
|
|
assert data["data"]["result"]["resultType"] == "scalar"
|
|
|
|
|
|
def test_fetch_jellyfin_widget_data_error(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "jellyfin",
|
|
"title": "Activity",
|
|
"config": {"machine_id": ""},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
response = client.get(f"/api/widgets/instances/{widget_id}/data")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["widget_type"] == "jellyfin"
|
|
assert data["data"] is None
|
|
assert data["error"] is not None
|
|
assert "Jellyfin" in data["error"] or "machine" in data["error"].lower()
|
|
|
|
|
|
def test_fetch_widget_data_not_found(client):
|
|
response = client.get("/api/widgets/instances/does-not-exist/data")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_fetch_widget_data_unhandled_exception_returns_500(client):
|
|
response = client.post(
|
|
"/api/widgets/instances",
|
|
json={
|
|
"addon_id": "core",
|
|
"widget_type": "static",
|
|
"title": "Note",
|
|
"config": {"text": "x"},
|
|
},
|
|
)
|
|
widget_id = response.json()["id"]
|
|
|
|
class _ExplodingAdapter:
|
|
source_type = "static"
|
|
|
|
async def fetch(self, config):
|
|
raise RuntimeError("boom")
|
|
|
|
with patch("media_library_viewer_api.routers.widgets.get_source_adapter", return_value=_ExplodingAdapter()):
|
|
response = client.get(f"/api/widgets/instances/{widget_id}/data")
|
|
|
|
assert response.status_code == 500
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_static_adapter():
|
|
adapter = StaticWidgetSource()
|
|
result = await adapter.fetch({"text": "hello"})
|
|
assert result == {"text": "hello"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grafana_adapter():
|
|
adapter = GrafanaWidgetSource()
|
|
result = await adapter.fetch({"dashboard_uid": "overview", "panel_id": 2})
|
|
assert result["url"] == "http://grafana:3000/d/overview?viewPanel=2"
|
|
|
|
result = await adapter.fetch({"dashboard_uid": "overview"})
|
|
assert result["url"] == "http://grafana:3000/d/overview"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssh_task_adapter_timeout(tmp_path):
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
|
|
# Create a local machine and a simple shell task.
|
|
machine = store.list_machines()[0]
|
|
task = store.upsert_task(
|
|
{
|
|
"name": "slow-task",
|
|
"task_type": "shell",
|
|
"content": "echo hello",
|
|
"enabled": True,
|
|
"default_machine_id": machine["id"],
|
|
}
|
|
)
|
|
|
|
adapter = SshTaskWidgetSource()
|
|
with patch(
|
|
"media_library_viewer_api.widgets.sources.get_settings_store",
|
|
return_value=store,
|
|
), patch(
|
|
"media_library_viewer_api.widgets.sources.asyncio.wait_for",
|
|
side_effect=asyncio.TimeoutError,
|
|
):
|
|
result = await adapter.fetch({"task_id": task["id"]})
|
|
|
|
assert "error" in result
|
|
assert "timed out" in result["error"].lower()
|
|
|
|
|
|
def test_source_registry_closed():
|
|
assert set(SOURCE_REGISTRY.keys()) == {
|
|
"jellyfin",
|
|
"backups",
|
|
"grafana",
|
|
"prometheus",
|
|
"ssh_task",
|
|
"static",
|
|
}
|