Developer
8cdeadd6dd
PR 1 of 4 for the runtime service registry change. - Add Fernet encryption helper (services/secrets.py) with a required MANAGE_ENCRYPTION_KEY; validate it on startup. - Add closed integrations/ registry with Pydantic config + widget-config definitions for grafana, prometheus, jellyfin, nextcloud, and ssh_tasks. - Add services + service_task_runs tables and SettingsStore CRUD with cascade-delete (defensive until widgets carry service_id). - Add /api/services/types and /api/services/instances CRUD (encrypted secrets, secrets_set flags only; never plaintext). - Declare cryptography as a direct dependency. - Require MANAGE_ENCRYPTION_KEY in compose + .env.example + README. - Add 25 backend tests (registry, encryption, CRUD, cascade, task-run history). Verification: ruff clean; pytest 225 passed; frontend lint/build green.
361 lines
12 KiB
Python
361 lines
12 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 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_five_service_types():
|
|
assert set(SERVICE_DEFINITIONS) == {
|
|
"grafana",
|
|
"prometheus",
|
|
"jellyfin",
|
|
"nextcloud",
|
|
"ssh_tasks",
|
|
}
|
|
|
|
|
|
def test_definitions_declare_widget_kinds():
|
|
assert {wk.kind for wk in get_service_definition("grafana").widget_kinds} == {"link"}
|
|
assert {wk.kind for wk in get_service_definition("prometheus").widget_kinds} == {"metric"}
|
|
assert {wk.kind for wk in get_service_definition("jellyfin").widget_kinds} == {"activity"}
|
|
assert get_service_definition("nextcloud").widget_kinds == []
|
|
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 == {"grafana", "prometheus", "jellyfin", "nextcloud", "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"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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": ""}},
|
|
)
|
|
# Pydantic accepts empty string; 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
|
|
|
|
|
|
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"
|