Developer
e757f4ba21
The credential tester (POST /api/services/test) used only body.secrets — the
values typed in the form. When editing an existing service the secret fields
are masked and intentionally left blank ("leave blank to keep current"), so the
test ran with empty credentials and failed auth even though the stored secret
was valid.
When body.id is set, look up the stored service, decrypt its secrets, and fall
back to the stored value for any known secret key that is absent or blank in
the input. The test still uses the freshly-typed config (so you can test an
edited URL) but authenticates with the effective credentials. New-service tests
(no id) are unchanged.
Test: editing a service and testing with empty secrets now authenticates with
the stored secret (asserts the stored key reaches the upstream request).
402/402 backend pass; ruff clean.
685 lines
25 KiB
Python
685 lines
25 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) == {
|
|
"prometheus",
|
|
"alertmanager",
|
|
"jellyfin",
|
|
"nextcloud",
|
|
"ssh_tasks",
|
|
"backups",
|
|
"authentik",
|
|
"qbittorrent",
|
|
}
|
|
|
|
|
|
def test_jellyseerr_absorbed_into_jellyfin():
|
|
"""Jellyseerr is no longer its own service type (absorbed into Jellyfin)."""
|
|
assert "jellyseerr" not in SERVICE_DEFINITIONS
|
|
jellyfin = get_service_definition("jellyfin")
|
|
jellyfin_config = jellyfin.config_schema["properties"]
|
|
assert "jellyseerr_url" in jellyfin_config
|
|
# jellyseerr_api_key moved from config to a secret field.
|
|
assert "jellyseerr_api_key" not in jellyfin_config
|
|
assert "jellyseerr_api_key" in {sf.key for sf in jellyfin.secret_fields}
|
|
|
|
|
|
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("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",
|
|
"stat",
|
|
"stats_overview",
|
|
}
|
|
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("prometheus", "metric") is not None
|
|
assert get_widget_kind("prometheus", "missing") is None
|
|
assert get_widget_kind("unknown", "metric") is None
|
|
|
|
|
|
def test_chart_widget_kinds_expose_unit_and_scale_options():
|
|
"""Graph widgets share unit/scale options so axes/tooltips can be scaled."""
|
|
units = ["none", "bytes", "bytes_per_sec", "bits_per_sec", "bits", "percent", "seconds"]
|
|
scales = ["auto", "k", "m", "g", "t"]
|
|
|
|
prom_chart = get_widget_kind("prometheus", "chart")
|
|
assert prom_chart is not None
|
|
prom_props = prom_chart.config_schema["properties"]
|
|
assert prom_props["unit"]["enum"] == units
|
|
assert prom_props["scale"]["enum"] == scales
|
|
|
|
qbit_speed = get_widget_kind("qbittorrent", "speed")
|
|
assert qbit_speed is not None
|
|
qbit_props = qbit_speed.config_schema["properties"]
|
|
assert qbit_props["unit"]["enum"] == units
|
|
assert qbit_props["scale"]["enum"] == scales
|
|
# qBittorrent speed data is bytes/sec by default.
|
|
assert qbit_speed.default_config["unit"] == "bytes_per_sec"
|
|
# totals/active are not graphs and stay option-less.
|
|
assert "unit" not in get_widget_kind("qbittorrent", "totals").config_schema["properties"]
|
|
assert "unit" not in get_widget_kind("qbittorrent", "active").config_schema["properties"]
|
|
|
|
|
|
def test_service_config_schema_is_json_schema():
|
|
schema = get_service_definition("prometheus").config_schema
|
|
assert schema["type"] == "object"
|
|
assert "grafana_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",
|
|
"jellyfin",
|
|
"nextcloud",
|
|
"prometheus",
|
|
"qbittorrent",
|
|
"ssh_tasks",
|
|
}
|
|
|
|
|
|
def test_service_type_includes_secret_and_widget_metadata(client):
|
|
response = client.get("/api/services/types")
|
|
prom = next(item for item in response.json() if item["service_type"] == "prometheus")
|
|
assert [sf["key"] for sf in prom["secret_fields"]] == ["grafana_api_key"]
|
|
assert set(wk["kind"] for wk in prom["widget_kinds"]) == {"metric", "chart", "gauge", "mean"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _prometheus_payload(**overrides):
|
|
payload = {
|
|
"service_type": "prometheus",
|
|
"name": "Production Prometheus",
|
|
"config": {"grafana_url": "https://grafana.example.com", "datasource_uid": "prometheus"},
|
|
"secrets": {"grafana_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=_prometheus_payload())
|
|
assert response.status_code == 201
|
|
created = response.json()
|
|
assert created["service_type"] == "prometheus"
|
|
assert created["config"]["grafana_url"] == "https://grafana.example.com"
|
|
# Plaintext secrets are never returned.
|
|
assert "secrets" not in created
|
|
assert created["secrets_set"] == {"grafana_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=_prometheus_payload())
|
|
client.post(
|
|
"/api/services/instances",
|
|
json={
|
|
"service_type": "alertmanager",
|
|
"name": "AM",
|
|
"config": {"base_url": "http://am:9093"},
|
|
},
|
|
)
|
|
response = client.get("/api/services/instances?service_type=prometheus")
|
|
assert response.status_code == 200
|
|
assert len(response.json()) == 1
|
|
assert response.json()[0]["service_type"] == "prometheus"
|
|
|
|
|
|
def test_update_service_preserves_unsent_secrets(client):
|
|
created = client.post("/api/services/instances", json=_prometheus_payload()).json()
|
|
# Update without sending secrets; the existing key should remain set.
|
|
updated = client.put(
|
|
f"/api/services/instances/{created['id']}",
|
|
json={
|
|
"service_type": "prometheus",
|
|
"name": "Renamed Prometheus",
|
|
"config": {"grafana_url": "https://grafana.example.com", "timeout_seconds": 10},
|
|
},
|
|
).json()
|
|
assert updated["name"] == "Renamed Prometheus"
|
|
assert updated["secrets_set"] == {"grafana_api_key": True}
|
|
|
|
|
|
def test_update_service_can_clear_secret(client):
|
|
created = client.post("/api/services/instances", json=_prometheus_payload()).json()
|
|
updated = client.put(
|
|
f"/api/services/instances/{created['id']}",
|
|
json={
|
|
"service_type": "prometheus",
|
|
"name": "Production Prometheus",
|
|
"config": {"grafana_url": "https://grafana.example.com"},
|
|
"secrets": {"grafana_api_key": ""},
|
|
},
|
|
).json()
|
|
assert updated["secrets_set"] == {"grafana_api_key": False}
|
|
|
|
|
|
def test_service_test_uses_stored_secrets_when_not_reentered(client):
|
|
"""Testing an existing service merges stored secrets for blank fields.
|
|
|
|
The editor leaves secret fields blank ("leave blank to keep current"); the
|
|
test must authenticate with the stored secret rather than failing on empty.
|
|
"""
|
|
|
|
created = client.post("/api/services/instances", json=_prometheus_payload()).json()
|
|
assert created["secrets_set"] == {"grafana_api_key": True}
|
|
|
|
payload = SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"results": {}})
|
|
with patch(
|
|
"media_library_viewer_api.integrations.prometheus.requests.post",
|
|
return_value=payload,
|
|
) as mock_post:
|
|
res = client.post(
|
|
"/api/services/test",
|
|
json={
|
|
"id": created["id"],
|
|
"service_type": "prometheus",
|
|
"name": "Production Prometheus",
|
|
"config": {"grafana_url": "https://grafana.example.com"},
|
|
"secrets": {}, # not re-entered
|
|
"enabled": True,
|
|
},
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json()["ok"] is True
|
|
# The stored grafana_api_key was used for the request (not empty).
|
|
headers = mock_post.call_args.kwargs["headers"]
|
|
assert headers["Authorization"] == "Bearer secret-token"
|
|
|
|
|
|
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": "prometheus", "name": "x", "config": {"grafana_url": ""}},
|
|
)
|
|
assert response.status_code == 422
|
|
# Force a real validation error via bad type.
|
|
response = client.post(
|
|
"/api/services/instances",
|
|
json={"service_type": "prometheus", "name": "x", "config": {"timeout_seconds": "fast"}},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad_url", ["prometheus.example.com", "localhost:3000", "//bad.example.com", "ftp://bad.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("prometheus").config_model
|
|
with pytest.raises(ValidationError):
|
|
model.model_validate({"grafana_url": bad_url, "timeout_seconds": 5})
|
|
|
|
|
|
@pytest.mark.parametrize("service_type", ["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": "prometheus",
|
|
"name": "x",
|
|
"config": {"grafana_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": "prometheus",
|
|
"name": "x",
|
|
"config": {"grafana_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=_prometheus_payload(id="missing"),
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_update_id_mismatch_returns_400(client):
|
|
created = client.post("/api/services/instances", json=_prometheus_payload()).json()
|
|
response = client.put(
|
|
f"/api/services/instances/{created['id']}",
|
|
json=_prometheus_payload(id="other-id"),
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_delete_service(client):
|
|
created = client.post("/api/services/instances", json=_prometheus_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": "prometheus", "name": "Prometheus", "config": {"grafana_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", "prometheus", "prometheus.metric", "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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Harness cascade-delete (Slice 4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cascade_delete_removes_harness_data_across_concerns(tmp_path, monkeypatch):
|
|
"""Deleting a service cascades to both qbittorrent samples and media items.
|
|
|
|
Proves end-to-end cascade across both harness-managed concerns, and that
|
|
deleting one service preserves another service's data (multi-instance).
|
|
"""
|
|
monkeypatch.setenv("BACKEND_CACHE_DIR", str(tmp_path))
|
|
|
|
from media_library_viewer_api.services.media_index_impl import MediaIndex
|
|
from media_library_viewer_api.services.qbittorrent_store import QbittorrentSampleStore
|
|
from media_library_viewer_api.services.service_data import (
|
|
get_service_data_harness,
|
|
reset_service_data_harness,
|
|
)
|
|
|
|
reset_service_data_harness()
|
|
harness = get_service_data_harness()
|
|
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
|
|
# --- qBit: create service, add samples, delete, verify gone ---
|
|
qbit_a = store.upsert_service(
|
|
{"service_type": "qbittorrent", "name": "qA", "config": {"base_url": "http://a"}, "enabled": True}
|
|
)
|
|
qbit_b = store.upsert_service(
|
|
{"service_type": "qbittorrent", "name": "qB", "config": {"base_url": "http://b"}, "enabled": True}
|
|
)
|
|
sample_store = QbittorrentSampleStore(harness)
|
|
sample_store.append(qbit_a["id"], ts=1000, dl_speed=500, up_speed=100)
|
|
sample_store.append(qbit_b["id"], ts=1000, dl_speed=200, up_speed=50)
|
|
|
|
assert len(sample_store.window(qbit_a["id"])) == 1
|
|
assert len(sample_store.window(qbit_b["id"])) == 1
|
|
|
|
store.delete_service(qbit_a["id"])
|
|
|
|
assert sample_store.window(qbit_a["id"]) == []
|
|
assert len(sample_store.window(qbit_b["id"])) == 1 # B survives
|
|
|
|
# --- MediaIndex: create services, add items, delete, verify scoped ---
|
|
jelly_a = store.upsert_service(
|
|
{"service_type": "jellyfin", "name": "jA", "config": {"base_url": "http://ja"}, "enabled": True}
|
|
)
|
|
jelly_b = store.upsert_service(
|
|
{"service_type": "jellyfin", "name": "jB", "config": {"base_url": "http://jb"}, "enabled": True}
|
|
)
|
|
index = MediaIndex(harness.db_path("media_index"))
|
|
index.init_schema()
|
|
index.replace_items([{"id": "m1", "title": "A1"}], service_id=jelly_a["id"])
|
|
index.replace_items([{"id": "m2", "title": "B1"}], service_id=jelly_b["id"])
|
|
|
|
rows_a, total_a = index.query(service_id=jelly_a["id"])
|
|
rows_b, total_b = index.query(service_id=jelly_b["id"])
|
|
assert total_a == 1 and total_b == 1
|
|
|
|
store.delete_service(jelly_a["id"])
|
|
|
|
rows_a_after, total_a_after = index.query(service_id=jelly_a["id"])
|
|
rows_b_after, total_b_after = index.query(service_id=jelly_b["id"])
|
|
assert total_a_after == 0 # deleted
|
|
assert total_b_after == 1 # survives
|
|
|
|
reset_service_data_harness()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 jellyseerr_url; the api key is now a secret.
|
|
migrated = store.get_service(jellyfin["id"])
|
|
assert migrated["config"]["jellyseerr_url"] == "https://jellyseerr.example.com"
|
|
assert "jellyseerr_api_key" not in migrated["config"]
|
|
assert decrypt_value(migrated["secrets"]["jellyseerr_api_key"]) == "js-key"
|
|
# The existing Jellyfin api_key is preserved (not double-encrypted).
|
|
assert decrypt_value(migrated["secrets"]["api_key"]) == "jf-key"
|
|
|
|
|
|
def test_jellyseerr_api_key_migrates_from_config_to_secret(tmp_path):
|
|
"""A pre-existing plaintext jellyseerr_api_key in config moves to a secret."""
|
|
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",
|
|
"jellyseerr_url": "https://jellyseerr.example.com",
|
|
"jellyseerr_api_key": "plaintext-key", # legacy plaintext in config
|
|
},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_key": "jf-key"},
|
|
)
|
|
|
|
store.ensure_defaults() # runs the config->secret migration
|
|
|
|
migrated = store.get_service(jellyfin["id"])
|
|
assert "jellyseerr_api_key" not in migrated["config"]
|
|
assert decrypt_value(migrated["secrets"]["jellyseerr_api_key"]) == "plaintext-key"
|
|
# Idempotent: a second run keeps it in secrets, doesn't wipe it.
|
|
store.ensure_defaults()
|
|
migrated = store.get_service(jellyfin["id"])
|
|
assert decrypt_value(migrated["secrets"]["jellyseerr_api_key"]) == "plaintext-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") == []
|