Developer
29650ca512
Move to openspec/changes/archive/2026-07-09-per-instance-hook-scoping/ (R100 renames preserved). 9 artifacts. Canonical openspec/specs/ service-instance-scoping/ remains. Resolves multi-instance wrong-data bug (hooks now scope by instance.id; instance switcher re-scopes). Carry-overs: fetchBackupDashboard untouched (design decision 5); subquery scoping for runs/alerts (schema asymmetry).
152 lines
6.4 KiB
Python
152 lines
6.4 KiB
Python
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from media_library_viewer_api.main import app
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
|
|
|
|
def test_dashboard_backups():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "test_settings.sqlite"
|
|
store = SettingsStore(db_path)
|
|
store.init_schema()
|
|
|
|
# Monkey-patch the global store for this test
|
|
import media_library_viewer_api.auth as auth_module
|
|
from media_library_viewer_api.services import settings_store
|
|
|
|
original_store = settings_store._store
|
|
settings_store._store = store
|
|
auth_module._API_KEY = None
|
|
|
|
try:
|
|
client = TestClient(app)
|
|
response = client.get("/api/dashboard/backups")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_jobs" in data
|
|
assert "success_rate_24h" in data
|
|
finally:
|
|
settings_store._store = original_store
|
|
auth_module._API_KEY = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-instance service scoping (PI-110, PI-111, PI-119)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
"""Fresh SettingsStore with schema initialized."""
|
|
db_path = tmp_path / "test_settings.sqlite"
|
|
s = SettingsStore(db_path)
|
|
s.init_schema()
|
|
return s
|
|
|
|
|
|
class TestBackupServiceScoping:
|
|
"""Verify service_id filtering on list_backup_jobs/runs/alerts."""
|
|
|
|
def test_list_backup_jobs_filtered_by_service(self, store: SettingsStore):
|
|
store.upsert_backup_job({"name": "job-a", "service_id": "svc-a"})
|
|
store.upsert_backup_job({"name": "job-b", "service_id": "svc-b"})
|
|
assert len(store.list_backup_jobs(service_id="svc-a")) == 1
|
|
assert len(store.list_backup_jobs(service_id="svc-b")) == 1
|
|
|
|
def test_list_backup_jobs_unfiltered_returns_all(self, store: SettingsStore):
|
|
store.upsert_backup_job({"name": "job-a", "service_id": "svc-a"})
|
|
store.upsert_backup_job({"name": "job-b", "service_id": "svc-b"})
|
|
assert len(store.list_backup_jobs()) == 2
|
|
assert len(store.list_backup_jobs(service_id="")) == 2
|
|
|
|
def test_list_backup_runs_filtered_by_service(self, store: SettingsStore):
|
|
job_a = store.upsert_backup_job({"name": "job-a", "service_id": "svc-a"})
|
|
job_b = store.upsert_backup_job({"name": "job-b", "service_id": "svc-b"})
|
|
store.create_backup_run({"job_id": job_a["id"], "started_at": 1700000000, "status": "success"})
|
|
store.create_backup_run({"job_id": job_b["id"], "started_at": 1700000000, "status": "success"})
|
|
assert len(store.list_backup_runs(service_id="svc-a")) == 1
|
|
assert len(store.list_backup_runs(service_id="svc-b")) == 1
|
|
|
|
def test_list_backup_alerts_filtered_by_service(self, store: SettingsStore):
|
|
job_a = store.upsert_backup_job({"name": "job-a", "service_id": "svc-a"})
|
|
job_b = store.upsert_backup_job({"name": "job-b", "service_id": "svc-b"})
|
|
store.create_backup_run({"job_id": job_a["id"], "started_at": 1700000000, "status": "success"})
|
|
store.create_backup_run({"job_id": job_b["id"], "started_at": 1700000000, "status": "success"})
|
|
store.create_backup_alert({"job_id": job_a["id"], "alert_type": "test", "severity": "warning"})
|
|
store.create_backup_alert({"job_id": job_b["id"], "alert_type": "test", "severity": "warning"})
|
|
assert len(store.list_backup_alerts(service_id="svc-a")) == 1
|
|
assert len(store.list_backup_alerts(service_id="svc-b")) == 1
|
|
|
|
def test_list_backup_runs_unfiltered_returns_all(self, store: SettingsStore):
|
|
job_a = store.upsert_backup_job({"name": "job-a", "service_id": "svc-a"})
|
|
job_b = store.upsert_backup_job({"name": "job-b", "service_id": "svc-b"})
|
|
store.create_backup_run({"job_id": job_a["id"], "started_at": 1700000000, "status": "success"})
|
|
store.create_backup_run({"job_id": job_b["id"], "started_at": 1700000000, "status": "success"})
|
|
assert len(store.list_backup_runs()) == 2
|
|
|
|
def test_endpoint_threads_service_id_to_store(self, store: SettingsStore):
|
|
"""GET /api/backups/jobs?service_id=svc-a filters via the endpoint."""
|
|
import media_library_viewer_api.auth as auth_module
|
|
from media_library_viewer_api.services import settings_store
|
|
|
|
store.upsert_backup_job({"name": "job-a", "service_id": "svc-a"})
|
|
store.upsert_backup_job({"name": "job-b", "service_id": "svc-b"})
|
|
|
|
original_store = settings_store._store
|
|
settings_store._store = store
|
|
auth_module._API_KEY = None
|
|
|
|
try:
|
|
client = TestClient(app)
|
|
resp = client.get("/api/backups/jobs", params={"service_id": "svc-a"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "job-a"
|
|
finally:
|
|
settings_store._store = original_store
|
|
auth_module._API_KEY = None
|
|
|
|
|
|
def test_post_backup_report():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "test_settings.sqlite"
|
|
store = SettingsStore(db_path)
|
|
store.init_schema()
|
|
|
|
# Monkey-patch the global store for this test
|
|
import media_library_viewer_api.auth as auth_module
|
|
from media_library_viewer_api.auth import get_api_key
|
|
from media_library_viewer_api.services import settings_store
|
|
|
|
original_store = settings_store._store
|
|
settings_store._store = store
|
|
auth_module._API_KEY = None
|
|
|
|
try:
|
|
client = TestClient(app)
|
|
api_key = get_api_key()
|
|
|
|
response = client.post(
|
|
"/api/backups/report",
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
json={
|
|
"name": "test-backup",
|
|
"started_at": "2026-05-11T02:00:00Z",
|
|
"status": "success",
|
|
"ended_at": "2026-05-11T02:15:00Z",
|
|
"duration_ms": 900000,
|
|
"bytes_transferred": 1024,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "success"
|
|
finally:
|
|
settings_store._store = original_store
|
|
auth_module._API_KEY = None
|