Developer
10fd4ead4a
PR 2 of 4 for the runtime service registry change. - dashboard_widgets gains service_id + widget_kind columns (legacy addon_id/widget_type kept but unused). - Source adapters take (service: ServiceRecord | None, widget_kind, config). SERVICE_ADAPTERS keyed by service_type; BUILTIN_ADAPTERS for backups/static. - Backups and static stay as service-less built-ins (service_id nullable), exposed via GET /api/widgets/builtin. - SSH task adapter resolves the task + instance, runs over SSH, and appends a service_task_runs history row on success/failure/timeout/error. - Retire widgets/registry.py; widget metadata now comes from the integrations registry + widgets/builtin. Remove /api/widgets/types and /api/widgets/sources. - Stop default widget seeding (fresh install = empty dashboard). - Rewrite widget tests around the service-bound + built-in model (26 tests). Backend-only breaking change; frontend is reconciled in Slice 3. Build/lint stay green; pytest 222 passed.
387 lines
13 KiB
Python
387 lines
13 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 (
|
|
BackupsWidgetSource,
|
|
GrafanaWidgetSource,
|
|
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_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_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_machine_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.widgets.sources._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"
|