manage/backend/tests/test_widgets.py

"""Tests for the dashboard widget system: service-bound + built-in widgets."""

from __future__ import annotations

import time
from types import SimpleNamespace
from typing import Any
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,
    JellyfinWidgetSource,
    ServiceRecord,
    StaticWidgetSource,
)

TEST_KEY = Fernet.generate_key().decode()
PROMQL_REQUIRED_ERROR = "promql is required"


@pytest.fixture(autouse=True)
def _encryption_key(monkeypatch):
    monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", TEST_KEY)
    yield


@pytest.fixture(autouse=True)
def _clear_qbittorrent_client_cache():
    """Reset the per-service qBittorrent client cache between tests.

    QbittorrentWidgetSource reuses one authenticated client per service
    (lru_cache) so the SID cookie persists across fetches. Without clearing,
    a mock client cached by one test would leak into later tests that patch
    QbittorrentClient differently.
    """
    from media_library_viewer_api.widgets.sources import _qbittorrent_client

    _qbittorrent_client.cache_clear()
    yield
    _qbittorrent_client.cache_clear()


@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_prometheus_service(client, name="Production Prometheus", **config_overrides):
    config = {"grafana_url": "https://grafana.example.com", "datasource_uid": "prometheus"}
    config.update(config_overrides)
    return client.post(
        "/api/services/instances",
        json={"service_type": "prometheus", "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_prometheus_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": "metric",
            "title": "Dash",
            "config": {"promql": "up"},
        },
    )

    # 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_prometheus_service(client)
    response = client.post(
        "/api/widgets/instances",
        json={
            "service_id": service["id"],
            "widget_kind": "metric",
            "title": "Metrics",
            "config": {"promql": "up"},
        },
    )
    assert response.status_code == 201
    created = response.json()
    assert created["service_id"] == service["id"]
    assert created["widget_kind"] == "metric"


def test_service_bound_widget_unknown_kind_rejected(client):
    service = _make_prometheus_service(client)
    response = client.post(
        "/api/widgets/instances",
        json={
            "service_id": service["id"],
            "widget_kind": "nonexistent_kind",
            "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": "metric",
            "title": "x",
            "config": {"promql": "up"},
        },
    )
    assert response.status_code == 422


def test_service_bound_widget_invalid_config_rejected(client):
    service = _make_prometheus_service(client)
    response = client.post(
        "/api/widgets/instances",
        json={
            "service_id": service["id"],
            "widget_kind": "metric",
            "title": "x",
            "config": {"promql": 123},  # bad type: promql must be a string
        },
    )
    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_widget_service_not_found(client):
    service = _make_prometheus_service(client)
    created = client.post(
        "/api/widgets/instances",
        json={
            "service_id": service["id"],
            "widget_kind": "metric",
            "title": "x",
            "config": {"promql": "up"},
        },
    ).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_prometheus_service(client)
    created = client.post(
        "/api/widgets/instances",
        json={
            "service_id": service["id"],
            "widget_kind": "metric",
            "title": "x",
            "config": {"promql": "up"},
        },
    ).json()
    client.put(
        f"/api/services/instances/{service['id']}",
        json={
            "service_type": "prometheus",
            "name": service["name"],
            "config": {"grafana_url": "https://grafana.example.com", "datasource_uid": "prometheus"},
            "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_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)
# ---------------------------------------------------------------------------


def test_jellyfin_definition_has_now_playing_widget():
    from media_library_viewer_api.integrations.registry import get_service_definition

    definition = get_service_definition("jellyfin")
    assert definition is not None
    kinds = {wk.kind for wk in definition.widget_kinds}
    assert "now_playing" in kinds
    assert "activity" in kinds


@pytest.mark.asyncio
async def test_prometheus_chart_adapter_runs_range_query():
    """GM-106: chart kind hits /api/ds/query and returns {series}."""
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus", "timeout_seconds": 5},
        secrets={"grafana_api_key": "key"},
    )
    payload = SimpleNamespace(
        raise_for_status=lambda: None,
        json=lambda: {
            "results": {
                "A": {
                    "frames": [
                        {
                            "data": {"values": [[100, 130], [1.0, 1.0]]},
                            "schema": {
                                "fields": [
                                    {"name": "Time"},
                                    {"name": "Value", "labels": {"__name__": "up", "instance": "h:9100"}},
                                ]
                            },
                        }
                    ]
                }
            }
        },
    )
    with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=payload) as mock_post:
        result = await adapter.fetch(service, "chart", {"promql": "up", "window": "1h"})

    call = mock_post.call_args
    assert call.args[0].endswith("/api/ds/query")
    body = call.kwargs["json"]
    assert body["queries"][0]["expr"] == "up"
    assert body["queries"][0]["datasource"]["uid"] == "prometheus"
    assert "series" in result
    assert result["series"][0]["label"] == "instance=h:9100"
    assert result["series"][0]["points"] == [{"t": 100, "v": 1.0}, {"t": 130, "v": 1.0}]


@pytest.mark.asyncio
async def test_prometheus_chart_adapter_requires_promql():
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
        secrets={"grafana_api_key": "key"},
    )
    result = await adapter.fetch(service, "chart", {"promql": ""})
    assert result == {"error": PROMQL_REQUIRED_ERROR}


@pytest.mark.asyncio
async def test_prometheus_chart_adapter_degrades_on_http_error():
    """GM-103: a connection error returns {error} rather than raising."""
    import requests as req_mod

    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={
            "grafana_url": "http://grafana:3000",
            "datasource_uid": "prometheus",
            "timeout_seconds": 2,
        },
        secrets={"grafana_api_key": "key"},
    )
    with patch(
        "media_library_viewer_api.widgets.sources.requests.post",
        side_effect=req_mod.ConnectionError("refused"),
    ):
        result = await adapter.fetch(service, "chart", {"promql": "up", "window": "1h"})
    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, request):
    """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)
    request.addfinalizer(lambda: app.dependency_overrides.pop(get_settings_store, None))
    return client, store


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 Prometheus service).
    store.upsert_service(
        {
            "service_type": "prometheus",
            "name": "Prometheus",
            "config": {"grafana_url": "https://grafana.example.com", "datasource_uid": "prometheus"},
            "secrets": {"api_key": "tok"},
            "enabled": True,
        },
    )
    service = store.list_services("prometheus")[0]
    widget = store.upsert_widget(
        {
            "service_id": service["id"],
            "widget_kind": "chart",
            "title": "CPU IOWait",
            "config": {"promql": "rate(cpu[5m])", "window": "1h"},
            "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": "prometheus",
            "name": "Prometheus",
            "config": {"grafana_url": "https://grafana.example.com", "datasource_uid": "prometheus"},
            "secrets": {"api_key": "tok"},
            "enabled": True,
        },
    )
    service = store.list_services("prometheus")[0]
    widget = store.upsert_widget(
        {
            "service_id": service["id"],
            "widget_kind": "chart",
            "title": "Memory",
            "config": {"promql": "mem", "window": "1h"},
            "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"]["promql"] == "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


# ---------------------------------------------------------------------------
# Prometheus gauge + mean adapter tests (GM-107..GM-108)
# ---------------------------------------------------------------------------


def _grafana_single_frame(values, labels=None, display_name=None):
    """Build a Grafana /api/ds/query frames response for a single series."""
    field: dict[str, Any] = {"name": "Value"}
    if labels:
        field["labels"] = labels
    if display_name:
        field["config"] = {"displayName": display_name}
    return {
        "results": {
            "A": {
                "frames": [
                    {
                        "data": {"values": values},
                        "schema": {"fields": [{"name": "Time"}, field]},
                    }
                ]
            }
        }
    }


@pytest.mark.asyncio
async def test_prometheus_gauge_adapter_returns_scalar():
    """GM-107: gauge kind hits /api/ds/query and returns {value, thresholds}."""
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus", "timeout_seconds": 5},
        secrets={"grafana_api_key": "key"},
    )
    payload = SimpleNamespace(
        raise_for_status=lambda: None,
        json=lambda: _grafana_single_frame([[100], [0.75]]),
    )
    with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=payload) as mock_post:
        result = await adapter.fetch(
            service,
            "gauge",
            {"promql": "cpu_usage", "warn_at": 0.8, "crit_at": 0.95, "unit": "%"},
        )
    call = mock_post.call_args
    assert call.args[0].endswith("/api/ds/query")
    assert call.kwargs["json"]["queries"][0]["expr"] == "cpu_usage"
    assert result["value"] == 0.75
    assert result["warn_at"] == 0.8
    assert result["crit_at"] == 0.95
    assert result["unit"] == "%"


@pytest.mark.asyncio
async def test_prometheus_gauge_adapter_rejects_multi_series():
    """GM-107: gauge must be scalar-only; multi-series returns error."""
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
        secrets={"grafana_api_key": "key"},
    )
    payload = SimpleNamespace(
        raise_for_status=lambda: None,
        json=lambda: {
            "results": {
                "A": {
                    "frames": [
                        {"data": {"values": [[100], [1.0]]}, "schema": {"fields": [{}, {"name": "A"}]}},
                        {"data": {"values": [[100], [2.0]]}, "schema": {"fields": [{}, {"name": "A"}]}},
                    ]
                }
            }
        },
    )
    with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=payload):
        result = await adapter.fetch(service, "gauge", {"promql": "up"})
    assert "error" in result
    assert "single-series" in result["error"].lower()


@pytest.mark.asyncio
async def test_prometheus_gauge_adapter_requires_promql():
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
        secrets={"grafana_api_key": "key"},
    )
    result = await adapter.fetch(service, "gauge", {"promql": ""})
    assert result == {"error": PROMQL_REQUIRED_ERROR}


@pytest.mark.asyncio
async def test_prometheus_mean_adapter_computes_average():
    """GM-108: mean kind averages non-null values over the window."""
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus", "timeout_seconds": 5},
        secrets={"grafana_api_key": "key"},
    )
    payload = SimpleNamespace(
        raise_for_status=lambda: None,
        json=lambda: _grafana_single_frame([[100, 130, 160], [1.0, 2.0, 3.0]]),
    )
    with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=payload):
        result = await adapter.fetch(service, "mean", {"promql": "cpu", "window": "1h"})
    assert result["value"] == 2.0
    assert result["unit"] is None


@pytest.mark.asyncio
async def test_prometheus_mean_adapter_rejects_multi_series():
    """GM-108: mean must be scalar-only; multi-series returns error."""
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
        secrets={"grafana_api_key": "key"},
    )
    payload = SimpleNamespace(
        raise_for_status=lambda: None,
        json=lambda: {
            "results": {
                "A": {
                    "frames": [
                        {"data": {"values": [[100], [1.0]]}, "schema": {"fields": [{}, {"name": "A"}]}},
                        {"data": {"values": [[100], [2.0]]}, "schema": {"fields": [{}, {"name": "A"}]}},
                    ]
                }
            }
        },
    )
    with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=payload):
        result = await adapter.fetch(service, "mean", {"promql": "up", "window": "1h"})
    assert "error" in result
    assert "single-series" in result["error"].lower()


@pytest.mark.asyncio
async def test_prometheus_mean_adapter_skips_nan_values():
    """GM-108: NaN values are excluded from the mean computation."""
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
        secrets={"grafana_api_key": "key"},
    )
    # Grafana frames shape with NaN — normalize_grafana_frames converts string "NaN" to None
    payload = SimpleNamespace(
        raise_for_status=lambda: None,
        json=lambda: {
            "results": {
                "A": {
                    "frames": [
                        {"data": {"values": [[100, 130, 160], [2.0, "NaN", 4.0]]}, "schema": {"fields": [{}, {}]}}
                    ]
                }
            }
        },
    )
    with patch("media_library_viewer_api.widgets.sources.requests.post", return_value=payload):
        result = await adapter.fetch(service, "mean", {"promql": "up", "window": "1h"})
    assert result["value"] == 3.0


@pytest.mark.asyncio
async def test_prometheus_mean_adapter_requires_promql():
    from media_library_viewer_api.widgets.sources import MetricSource

    adapter = MetricSource()
    service = ServiceRecord(
        id="s",
        service_type="prometheus",
        name="p",
        config={"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
        secrets={"grafana_api_key": "key"},
    )
    result = await adapter.fetch(service, "mean", {"promql": ""})
    assert result == {"error": PROMQL_REQUIRED_ERROR}


# ---------------------------------------------------------------------------
# qBittorrent widget source adapter
# ---------------------------------------------------------------------------


def _fake_qbit_maindata():
    """Return a mock maindata response (server_state + torrents dict)."""
    return {
        "server_state": {"dl_info_speed": 500000, "up_info_speed": 100000},
        "torrents": {
            "h1": {
                "name": "Movie.mkv",
                "state": "downloading",
                "size": 1000,
                "progress": 0.5,
                "dlspeed": 500,
                "upspeed": 10,
            },
            "h2": {
                "name": "Show.mkv",
                "state": "uploading",
                "size": 2000,
                "progress": 1.0,
                "dlspeed": 0,
                "upspeed": 100,
            },
            "h3": {
                "name": "Queued",
                "state": "queuedDL",
                "size": 3000,
                "progress": 0.0,
                "dlspeed": 0,
                "upspeed": 0,
            },
            "h4": {
                "name": "Paused",
                "state": "pausedDL",
                "size": 4000,
                "progress": 0.3,
                "dlspeed": 0,
                "upspeed": 0,
            },
            "h5": {
                "name": "Forced download",
                "state": "forcedDL",
                "size": 5000,
                "progress": 0.4,
                "dlspeed": 0,
                "upspeed": 0,
            },
            "h6": {
                "name": "Stalled upload",
                "state": "stalledUP",
                "size": 6000,
                "progress": 1.0,
                "dlspeed": 0,
                "upspeed": 0,
            },
        },
    }


@pytest.mark.asyncio
async def test_qbittorrent_totals_counts_all_torrents():
    """Totals kind returns count of all listed items + by_state breakdown."""
    from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource

    adapter = QbittorrentWidgetSource()
    service = ServiceRecord(
        id="svc-1",
        service_type="qbittorrent",
        name="qbit",
        config={"base_url": "http://qbit:8080", "timeout_seconds": 5},
        secrets={"username": "admin", "password": "pass"},
    )
    with patch("media_library_viewer_api.widgets.sources.QbittorrentClient") as mock_client:
        mock_client.return_value.maindata.return_value = _fake_qbit_maindata()
        result = await adapter.fetch(service, "totals", {})

    assert result["total"] == 6
    assert result["by_state"]["downloading"] == 1
    assert result["by_state"]["uploading"] == 1
    assert result["by_state"]["queuedDL"] == 1
    assert result["by_state"]["pausedDL"] == 1
    assert result["by_direction"] == {"downloading": 3, "uploading": 2}


@pytest.mark.asyncio
async def test_qbittorrent_active_filters_dl_ul_only():
    """Active kind returns all active download/upload states, including queued work."""
    from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource

    adapter = QbittorrentWidgetSource()
    service = ServiceRecord(
        id="svc-1",
        service_type="qbittorrent",
        name="qbit",
        config={"base_url": "http://qbit:8080", "timeout_seconds": 5},
        secrets={"username": "admin", "password": "pass"},
    )
    with patch("media_library_viewer_api.widgets.sources.QbittorrentClient") as mock_client:
        mock_client.return_value.maindata.return_value = _fake_qbit_maindata()
        result = await adapter.fetch(service, "active", {})

    active = result["torrents"]
    assert len(active) == 5
    names = [t["name"] for t in active]
    assert "Movie.mkv" in names
    assert "Show.mkv" in names
    assert "Forced download" in names
    assert "Stalled upload" in names
    assert "Queued" in names
    # Paused torrents remain excluded, but queued transfer work is visible.
    assert "Paused" not in names


@pytest.mark.asyncio
async def test_qbittorrent_speed_reads_samples_without_polling(tmp_path):
    """Speed kind reads persisted samples and never calls qBittorrent itself."""
    from media_library_viewer_api.services.qbittorrent_store import QBITTORRENT_CONCERN, QbittorrentSampleStore
    from media_library_viewer_api.services.service_data import ServiceDataHarness
    from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource

    # Isolated harness so we don't pollute the real DB
    harness = ServiceDataHarness(base_dir=str(tmp_path))
    harness.register(QBITTORRENT_CONCERN)
    harness.run_migrations()

    adapter = QbittorrentWidgetSource()
    service = ServiceRecord(
        id="svc-speed",
        service_type="qbittorrent",
        name="qbit",
        config={"base_url": "http://qbit:8080", "timeout_seconds": 5},
        secrets={"username": "admin", "password": "pass"},
    )
    with (
        patch("media_library_viewer_api.widgets.sources.QbittorrentClient") as mock_client,
        patch("media_library_viewer_api.widgets.sources.QbittorrentSampleStore") as mock_store_cls,
    ):
        mock_client.return_value.maindata.return_value = _fake_qbit_maindata()
        # Wire the mock store to a real isolated store and seed headless data.
        real_store = QbittorrentSampleStore(harness)
        real_store.append("svc-speed", round(time.time()), 500000, 1000)
        mock_store_cls.return_value = real_store
        result = await adapter.fetch(service, "speed", {})

    mock_client.assert_not_called()

    assert "series" in result
    labels = [s["label"] for s in result["series"]]
    assert labels == ["download", "upload"]
    # The scheduler-supplied sample should be present.
    dl_points = result["series"][0]["points"]
    assert len(dl_points) >= 1
    # timestamps multiplied by 1000 for JS epoch
    assert dl_points[-1]["v"] == 500000


@pytest.mark.asyncio
async def test_qbittorrent_adapter_missing_service():
    from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource

    adapter = QbittorrentWidgetSource()
    result = await adapter.fetch(None, "totals", {})
    assert "error" in result


@pytest.mark.asyncio
async def test_qbittorrent_adapter_missing_credentials():
    from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource

    adapter = QbittorrentWidgetSource()
    service = ServiceRecord(
        id="s",
        service_type="qbittorrent",
        name="qbit",
        config={"base_url": "http://qbit:8080"},
        secrets={"username": "", "password": ""},
    )
    result = await adapter.fetch(service, "totals", {})
    assert "error" in result


@pytest.mark.asyncio
async def test_qbittorrent_adapter_timeout():
    """A timeout returns {error} rather than raising."""
    from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource

    adapter = QbittorrentWidgetSource()
    service = ServiceRecord(
        id="s",
        service_type="qbittorrent",
        name="qbit",
        config={"base_url": "http://qbit:8080", "timeout_seconds": 1},
        secrets={"username": "admin", "password": "pass"},
    )
    with patch("media_library_viewer_api.widgets.sources.QbittorrentClient") as mock_client:
        import asyncio as _asyncio

        async def _slow(*a, **kw):
            await _asyncio.sleep(10)

        # Make to_thread hang so wait_for times out
        mock_client.return_value.maindata.side_effect = lambda: (_ for _ in ()).throw(TimeoutError())
        result = await adapter.fetch(service, "totals", {})
    assert "error" in result