"""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"} assert {wk.kind for wk in get_service_definition("prometheus").widget_kinds} == {"metric"} 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"} 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"] # --------------------------------------------------------------------------- # 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") == []