Developer
65bae95e3c
Add gauge widget (recharts RadialBarChart with configurable threshold bands, scalar-only per SC-111) and mean widget (client-side average over range-query window, scalar-only per SC-114). Extract shared _instant_query helper from the metric path; _fetch_gauge and _fetch_mean dispatch in PrometheusWidgetSource.fetch(). Both new widget kinds declared in integrations/prometheus.py and frontend registry. Backend: 305 pytest pass, ruff clean. Frontend: 136 vitest pass, build+lint green.
521 lines
18 KiB
Python
521 lines
18 KiB
Python
"""Tests for the service registry: definitions, encryption, CRUD, cascade delete."""
|
|
|
|
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 pydantic import ValidationError
|
|
|
|
from media_library_viewer_api.dependencies import get_settings_store
|
|
from media_library_viewer_api.integrations.registry import (
|
|
SERVICE_DEFINITIONS,
|
|
get_service_definition,
|
|
get_widget_kind,
|
|
)
|
|
from media_library_viewer_api.main import app
|
|
from media_library_viewer_api.services.secrets import (
|
|
EncryptionKeyError,
|
|
decrypt_secrets,
|
|
decrypt_value,
|
|
encrypt_secrets,
|
|
encrypt_value,
|
|
get_encryption_key,
|
|
reset_encryption_key_cache,
|
|
)
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
|
|
TEST_KEY = Fernet.generate_key().decode()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _encryption_key(monkeypatch):
|
|
"""Provide a stable MANAGE_ENCRYPTION_KEY for every test."""
|
|
monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", TEST_KEY)
|
|
reset_encryption_key_cache()
|
|
yield
|
|
reset_encryption_key_cache()
|
|
|
|
|
|
@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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_registry_contains_eight_service_types():
|
|
assert set(SERVICE_DEFINITIONS) == {
|
|
"grafana",
|
|
"prometheus",
|
|
"alertmanager",
|
|
"jellyfin",
|
|
"nextcloud",
|
|
"ssh_tasks",
|
|
"backups",
|
|
"authentik",
|
|
}
|
|
|
|
|
|
def test_jellyseerr_absorbed_into_jellyfin():
|
|
"""Jellyseerr is no longer its own service type (absorbed into Jellyfin)."""
|
|
assert "jellyseerr" not in SERVICE_DEFINITIONS
|
|
jellyfin_config = get_service_definition("jellyfin").config_schema["properties"]
|
|
assert "jellyseerr_url" in jellyfin_config
|
|
assert "jellyseerr_api_key" in jellyfin_config
|
|
|
|
|
|
def test_backups_service_definition():
|
|
definition = get_service_definition("backups")
|
|
assert definition is not None
|
|
assert definition.secret_fields == []
|
|
assert {wk.kind for wk in definition.widget_kinds} == {"summary"}
|
|
schema = definition.config_schema
|
|
assert "ingestion_label" in schema["properties"]
|
|
|
|
|
|
def test_authentik_service_definition():
|
|
definition = get_service_definition("authentik")
|
|
assert definition is not None
|
|
assert {sf.key for sf in definition.secret_fields} == {"api_token"}
|
|
assert definition.secret_fields[0].required is True
|
|
assert definition.widget_kinds == []
|
|
schema = definition.config_schema
|
|
assert "base_url" in schema["properties"]
|
|
assert "timeout_seconds" in schema["properties"]
|
|
|
|
|
|
def test_definitions_declare_widget_kinds():
|
|
assert {wk.kind for wk in get_service_definition("grafana").widget_kinds} == {"link", "chart"}
|
|
assert {wk.kind for wk in get_service_definition("prometheus").widget_kinds} == {"metric", "chart", "gauge", "mean"}
|
|
assert {wk.kind for wk in get_service_definition("alertmanager").widget_kinds} == {"active_alerts"}
|
|
assert {wk.kind for wk in get_service_definition("jellyfin").widget_kinds} == {"activity", "now_playing"}
|
|
assert get_service_definition("nextcloud").widget_kinds == []
|
|
assert get_service_definition("authentik").widget_kinds == []
|
|
assert {wk.kind for wk in get_service_definition("backups").widget_kinds} == {"summary"}
|
|
assert {wk.kind for wk in get_service_definition("ssh_tasks").widget_kinds} == {"task_output"}
|
|
|
|
|
|
def test_widget_kind_lookup():
|
|
assert get_widget_kind("grafana", "link") is not None
|
|
assert get_widget_kind("grafana", "missing") is None
|
|
assert get_widget_kind("unknown", "link") is None
|
|
|
|
|
|
def test_service_config_schema_is_json_schema():
|
|
schema = get_service_definition("grafana").config_schema
|
|
assert schema["type"] == "object"
|
|
assert "base_url" in schema["properties"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Encryption
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_encrypt_decrypt_round_trip():
|
|
cipher = encrypt_value("hunter2")
|
|
assert cipher != "hunter2"
|
|
assert decrypt_value(cipher) == "hunter2"
|
|
|
|
|
|
def test_encrypt_decrypt_secrets_dict():
|
|
blob = encrypt_secrets({"api_key": "abc", "token": "xyz"})
|
|
assert decrypt_secrets(blob) == {"api_key": "abc", "token": "xyz"}
|
|
|
|
|
|
def test_missing_encryption_key_raises(monkeypatch):
|
|
monkeypatch.delenv("MANAGE_ENCRYPTION_KEY", raising=False)
|
|
reset_encryption_key_cache()
|
|
with pytest.raises(EncryptionKeyError):
|
|
get_encryption_key()
|
|
reset_encryption_key_cache()
|
|
|
|
|
|
def test_decrypt_with_wrong_key_raises(monkeypatch):
|
|
blob = encrypt_secrets({"api_key": "abc"})
|
|
monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", Fernet.generate_key().decode())
|
|
reset_encryption_key_cache()
|
|
with pytest.raises(EncryptionKeyError):
|
|
decrypt_secrets(blob)
|
|
reset_encryption_key_cache()
|
|
|
|
|
|
def test_invalid_ciphertext_raises():
|
|
with pytest.raises(EncryptionKeyError):
|
|
decrypt_value("not-a-real-token")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Service type metadata endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_service_types(client):
|
|
response = client.get("/api/services/types")
|
|
assert response.status_code == 200
|
|
types = {item["service_type"] for item in response.json()}
|
|
assert types == {
|
|
"alertmanager",
|
|
"authentik",
|
|
"backups",
|
|
"grafana",
|
|
"jellyfin",
|
|
"nextcloud",
|
|
"prometheus",
|
|
"ssh_tasks",
|
|
}
|
|
|
|
|
|
def test_service_type_includes_secret_and_widget_metadata(client):
|
|
response = client.get("/api/services/types")
|
|
grafana = next(item for item in response.json() if item["service_type"] == "grafana")
|
|
assert [sf["key"] for sf in grafana["secret_fields"]] == ["api_key"]
|
|
assert [wk["kind"] for wk in grafana["widget_kinds"]] == ["link", "chart"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _grafana_payload(**overrides):
|
|
payload = {
|
|
"service_type": "grafana",
|
|
"name": "Production Grafana",
|
|
"config": {"base_url": "https://grafana.example.com"},
|
|
"secrets": {"api_key": "secret-token"},
|
|
"enabled": True,
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def test_create_and_list_service(client):
|
|
response = client.post("/api/services/instances", json=_grafana_payload())
|
|
assert response.status_code == 201
|
|
created = response.json()
|
|
assert created["service_type"] == "grafana"
|
|
assert created["config"]["base_url"] == "https://grafana.example.com"
|
|
# Plaintext secrets are never returned.
|
|
assert "secrets" not in created
|
|
assert created["secrets_set"] == {"api_key": True}
|
|
|
|
response = client.get("/api/services/instances")
|
|
assert response.status_code == 200
|
|
assert len(response.json()) == 1
|
|
|
|
|
|
def test_list_instances_filters_by_type(client):
|
|
client.post("/api/services/instances", json=_grafana_payload())
|
|
client.post(
|
|
"/api/services/instances",
|
|
json={
|
|
"service_type": "prometheus",
|
|
"name": "Prom",
|
|
"config": {"base_url": "http://prometheus:9090"},
|
|
},
|
|
)
|
|
response = client.get("/api/services/instances?service_type=grafana")
|
|
assert response.status_code == 200
|
|
assert len(response.json()) == 1
|
|
assert response.json()[0]["service_type"] == "grafana"
|
|
|
|
|
|
def test_update_service_preserves_unsent_secrets(client):
|
|
created = client.post("/api/services/instances", json=_grafana_payload()).json()
|
|
# Update without sending secrets; the existing key should remain set.
|
|
updated = client.put(
|
|
f"/api/services/instances/{created['id']}",
|
|
json={
|
|
"service_type": "grafana",
|
|
"name": "Renamed Grafana",
|
|
"config": {"base_url": "https://grafana.example.com", "timeout_seconds": 10},
|
|
},
|
|
).json()
|
|
assert updated["name"] == "Renamed Grafana"
|
|
assert updated["secrets_set"] == {"api_key": True}
|
|
|
|
|
|
def test_update_service_can_clear_secret(client):
|
|
created = client.post("/api/services/instances", json=_grafana_payload()).json()
|
|
updated = client.put(
|
|
f"/api/services/instances/{created['id']}",
|
|
json={
|
|
"service_type": "grafana",
|
|
"name": "Production Grafana",
|
|
"config": {"base_url": "https://grafana.example.com"},
|
|
"secrets": {"api_key": ""},
|
|
},
|
|
).json()
|
|
assert updated["secrets_set"] == {"api_key": False}
|
|
|
|
|
|
def test_unknown_service_type_rejected(client):
|
|
response = client.post(
|
|
"/api/services/instances",
|
|
json={"service_type": "bogus", "name": "x", "config": {}},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_invalid_config_rejected(client):
|
|
response = client.post(
|
|
"/api/services/instances",
|
|
json={"service_type": "grafana", "name": "x", "config": {"base_url": ""}},
|
|
)
|
|
assert response.status_code == 422
|
|
# Force a real validation error via bad type.
|
|
response = client.post(
|
|
"/api/services/instances",
|
|
json={"service_type": "grafana", "name": "x", "config": {"timeout_seconds": "fast"}},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad_url", ["grafana.example.com", "localhost:3000", "//grafana.example.com", "ftp://grafana.example.com"]
|
|
)
|
|
def test_service_base_url_requires_http_schema(bad_url):
|
|
"""Every service base_url must include an http:// or https:// schema."""
|
|
model = get_service_definition("grafana").config_model
|
|
with pytest.raises(ValidationError):
|
|
model.model_validate({"base_url": bad_url, "timeout_seconds": 5})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"service_type", ["grafana", "prometheus", "alertmanager", "jellyfin", "authentik", "nextcloud"]
|
|
)
|
|
def test_service_base_url_accepts_absolute_urls(service_type):
|
|
model = get_service_definition(service_type).config_model
|
|
instance = model.model_validate({"base_url": "https://example.com"})
|
|
assert instance.base_url == "https://example.com"
|
|
|
|
|
|
def test_unknown_secret_field_rejected(client):
|
|
response = client.post(
|
|
"/api/services/instances",
|
|
json={
|
|
"service_type": "grafana",
|
|
"name": "x",
|
|
"config": {"base_url": "https://grafana.example.com"},
|
|
"secrets": {"password": "leak"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_credential_key_in_config_rejected(client):
|
|
response = client.post(
|
|
"/api/services/instances",
|
|
json={
|
|
"service_type": "grafana",
|
|
"name": "x",
|
|
"config": {"base_url": "https://grafana.example.com", "api_key": "leak"},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_update_nonexistent_returns_404(client):
|
|
response = client.put(
|
|
"/api/services/instances/missing",
|
|
json=_grafana_payload(id="missing"),
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_update_id_mismatch_returns_400(client):
|
|
created = client.post("/api/services/instances", json=_grafana_payload()).json()
|
|
response = client.put(
|
|
f"/api/services/instances/{created['id']}",
|
|
json=_grafana_payload(id="other-id"),
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_delete_service(client):
|
|
created = client.post("/api/services/instances", json=_grafana_payload()).json()
|
|
response = client.delete(f"/api/services/instances/{created['id']}")
|
|
assert response.status_code == 200
|
|
assert client.get("/api/services/instances").json() == []
|
|
|
|
|
|
def test_delete_nonexistent_returns_404(client):
|
|
assert client.delete("/api/services/instances/missing").status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cascade delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_delete_service_cascades_to_widgets(client, tmp_path):
|
|
"""Once widgets carry service_id (Slice 2), deleting a service removes them.
|
|
|
|
This test seeds a widget row directly with the column present to prove the
|
|
cascade path; the column is added defensively here so the test is meaningful
|
|
even before Slice 2 lands.
|
|
"""
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
service = store.upsert_service(
|
|
{"service_type": "grafana", "name": "Grafana", "config": {"base_url": "u"}, "enabled": True}
|
|
)
|
|
|
|
# Ensure the service_id column exists and seed a referencing widget.
|
|
with store.connect() as conn:
|
|
cols = {row[1] for row in conn.execute("PRAGMA table_info(dashboard_widgets)").fetchall()}
|
|
if "service_id" not in cols:
|
|
conn.execute("ALTER TABLE dashboard_widgets ADD COLUMN service_id TEXT")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO dashboard_widgets (id, addon_id, widget_type, title, config_json,
|
|
enabled, sort_order, created_at, updated_at, service_id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("w1", "grafana", "grafana.link", "Link", "{}", 1, 0, 1, 1, service["id"]),
|
|
)
|
|
|
|
store.delete_service(service["id"])
|
|
assert store.get_service(service["id"]) is None
|
|
with store.connect() as conn:
|
|
remaining = conn.execute(
|
|
"SELECT COUNT(*) FROM dashboard_widgets WHERE service_id = ?",
|
|
(service["id"],),
|
|
).fetchone()
|
|
assert int(remaining[0]) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Service task run history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_record_and_list_service_task_runs(client):
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
service = store.upsert_service(
|
|
{"service_type": "ssh_tasks", "name": "box", "config": {"host": "h"}, "enabled": True}
|
|
)
|
|
store.record_service_task_run(
|
|
{
|
|
"task_id": "t1",
|
|
"service_id": service["id"],
|
|
"status": "success",
|
|
"exit_status": 0,
|
|
"stdout_tail": "ok",
|
|
}
|
|
)
|
|
runs = store.list_service_task_runs(service_id=service["id"])
|
|
assert len(runs) == 1
|
|
assert runs[0]["status"] == "success"
|
|
assert runs[0]["stdout_tail"] == "ok"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jellyseerr → Jellyfin migration (Slice 1.4 / 1.5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_jellyseerr_migrates_into_single_jellyfin(tmp_path):
|
|
"""A standalone jellyseerr service merges into the only jellyfin instance."""
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
|
|
jellyfin = store.upsert_service(
|
|
{
|
|
"service_type": "jellyfin",
|
|
"name": "Main Jellyfin",
|
|
"config": {"base_url": "https://jellyfin.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_key": "jf-key"},
|
|
)
|
|
store.upsert_service(
|
|
{
|
|
"service_type": "jellyseerr",
|
|
"name": "Main Jellyseerr",
|
|
"config": {"base_url": "https://jellyseerr.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_key": "js-key"},
|
|
)
|
|
|
|
# Run migration via ensure_defaults (idempotent entry point).
|
|
store.ensure_defaults()
|
|
|
|
# Jellyseerr row is gone.
|
|
assert store.list_services("jellyseerr") == []
|
|
|
|
# Jellyfin config gained the absorbed fields.
|
|
migrated = store.get_service(jellyfin["id"])
|
|
assert migrated["config"]["jellyseerr_url"] == "https://jellyseerr.example.com"
|
|
assert migrated["config"]["jellyseerr_api_key"] == "js-key"
|
|
|
|
|
|
def test_jellyseerr_dropped_when_no_jellyfin(tmp_path):
|
|
"""An unpaired jellyseerr (no jellyfin) is dropped with a warning, no crash."""
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
|
|
store.upsert_service(
|
|
{
|
|
"service_type": "jellyseerr",
|
|
"name": "Orphan Jellyseerr",
|
|
"config": {"base_url": "https://jellyseerr.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_key": "js-key"},
|
|
)
|
|
|
|
store.ensure_defaults()
|
|
|
|
assert store.list_services("jellyseerr") == []
|
|
assert store.list_services("jellyfin") == []
|
|
|
|
|
|
def test_jellyseerr_migration_is_idempotent(tmp_path):
|
|
"""Running ensure_defaults twice does nothing the second time."""
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
|
|
store.upsert_service(
|
|
{
|
|
"service_type": "jellyfin",
|
|
"name": "JF",
|
|
"config": {"base_url": "https://jellyfin.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_key": "k"},
|
|
)
|
|
store.upsert_service(
|
|
{
|
|
"service_type": "jellyseerr",
|
|
"name": "JS",
|
|
"config": {"base_url": "https://jellyseerr.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_key": "k"},
|
|
)
|
|
|
|
store.ensure_defaults()
|
|
first_jellyfin = store.list_services("jellyfin")[0]
|
|
first_url = first_jellyfin["config"]["jellyseerr_url"]
|
|
|
|
store.ensure_defaults() # second run
|
|
second_jellyfin = store.list_services("jellyfin")[0]
|
|
assert second_jellyfin["config"]["jellyseerr_url"] == first_url
|
|
assert store.list_services("jellyseerr") == []
|