manage/backend/tests/test_widgets.py

"""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"