391 lines
14 KiB
Python
391 lines
14 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import SQLModel, create_engine
|
|
|
|
|
|
@pytest.fixture
|
|
def api_client(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "setup_test.db"
|
|
monkeypatch.setenv("APPOPS_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token")
|
|
for name in [
|
|
"PORTAINER_URL",
|
|
"PORTAINER_API_KEY",
|
|
"PORTAINER_ENDPOINT_ID",
|
|
"UNRAID_URL",
|
|
"UNRAID_API_KEY",
|
|
"AUTHENTIK_URL",
|
|
"AUTHENTIK_TOKEN",
|
|
"AUTHENTIK_OUTPOST_URL",
|
|
"RADARR_URL",
|
|
"RADARR_API_KEY",
|
|
"PROWLARR_URL",
|
|
"PROWLARR_API_KEY",
|
|
]:
|
|
monkeypatch.delenv(name, raising=False)
|
|
|
|
from backend.app.main import app
|
|
from backend.app.core.config import settings
|
|
from backend.app.db import session
|
|
from backend.app.db.init_db import init_db
|
|
|
|
for name in [
|
|
"PORTAINER_URL",
|
|
"PORTAINER_API_KEY",
|
|
"PORTAINER_ENDPOINT_ID",
|
|
"UNRAID_URL",
|
|
"UNRAID_API_KEY",
|
|
"AUTHENTIK_URL",
|
|
"AUTHENTIK_TOKEN",
|
|
"AUTHENTIK_OUTPOST_URL",
|
|
"RADARR_URL",
|
|
"RADARR_API_KEY",
|
|
"PROWLARR_URL",
|
|
"PROWLARR_API_KEY",
|
|
]:
|
|
settings.apis[name] = None
|
|
settings.verify_tls = False
|
|
|
|
engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False})
|
|
session.engine = engine
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
init_db()
|
|
|
|
with TestClient(app) as client:
|
|
yield client
|
|
|
|
|
|
def test_setup_status_initial_is_pending_and_secret_safe(api_client):
|
|
payload = api_client.get("/api/setup/status").json()
|
|
|
|
assert payload["firstRunCompleted"] is False
|
|
assert payload["summary"]["containersDiscovered"] == 0
|
|
assert payload["summary"]["appsManaged"] >= 0
|
|
assert payload["summary"]["connectorsMissingConfig"] >= 1
|
|
assert {step["id"]: step["status"] for step in payload["steps"]}["portainer"] == "pending"
|
|
assert {step["id"]: step["status"] for step in payload["steps"]}["discovery"] == "pending"
|
|
assert "test-voice-token" not in str(payload)
|
|
|
|
|
|
def test_portainer_config_save_never_returns_secret(api_client):
|
|
secret = "PORTAINER_SUPER_SECRET"
|
|
|
|
payload = api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={
|
|
"url": "http://198.18.10.150:9443",
|
|
"endpointId": "1",
|
|
"verifyTls": False,
|
|
"apiKey": secret,
|
|
},
|
|
).json()
|
|
|
|
assert payload["connectorId"] == "portainer"
|
|
assert payload["configured"] is True
|
|
assert payload["hasSecret"] is True
|
|
assert payload["missing"] == []
|
|
assert payload["safeConfig"] == {
|
|
"url": "http://198.18.10.150:9443",
|
|
"endpointId": "1",
|
|
"verifyTls": False,
|
|
}
|
|
assert secret not in str(payload)
|
|
all_settings = api_client.get("/api/settings/connectors").json()
|
|
assert secret not in str(all_settings)
|
|
assert next(item for item in all_settings if item["connectorId"] == "portainer")["hasSecret"] is True
|
|
|
|
|
|
def test_generic_authentik_setup_cannot_set_or_overwrite_forge_oidc_binding(api_client):
|
|
import json
|
|
|
|
from sqlmodel import Session
|
|
|
|
from backend.app.db import session as db_session
|
|
from backend.app.schemas.forge_oidc import FORGE_OIDC_BINDING_SECRET_REF
|
|
from backend.app.services.secret_store import get_secret, set_secret
|
|
from backend.app.services.setup_service import _apply_forge_oidc_binding_to_runtime
|
|
from backend.app.core.config import settings
|
|
|
|
binding = json.dumps(
|
|
{
|
|
"version": 1,
|
|
"providerId": "31",
|
|
"scopeMappingIds": {
|
|
"openid": "75a34fd6-cc81-40b8-a8c3-2745af9567ab",
|
|
"email": "312810bb-11da-47d8-97d4-f2155348eaa2",
|
|
"profile": "7aa8c5f8-5e8e-4a1b-9c11-3a4f0a731cff",
|
|
},
|
|
}
|
|
)
|
|
with Session(db_session.engine) as session:
|
|
set_secret(session, FORGE_OIDC_BINDING_SECRET_REF, binding)
|
|
|
|
rejected = api_client.patch(
|
|
"/api/settings/connectors/authentik",
|
|
json={"forgeOidcBinding": binding, "providerId": "999"},
|
|
)
|
|
assert rejected.status_code == 422
|
|
|
|
saved = api_client.patch(
|
|
"/api/settings/connectors/authentik",
|
|
json={"url": "https://auth.example-host.com", "secret": "GENERIC_AUTHENTIK_TOKEN"},
|
|
)
|
|
assert saved.status_code == 200
|
|
assert binding not in saved.text
|
|
assert "GENERIC_AUTHENTIK_TOKEN" not in saved.text
|
|
with Session(db_session.engine) as session:
|
|
assert get_secret(session, FORGE_OIDC_BINDING_SECRET_REF, touch=False) == binding
|
|
assert _apply_forge_oidc_binding_to_runtime(session) is True
|
|
assert settings.apis["AUTHENTIK_FORGE_OIDC_PROVIDER_ID"] == "31"
|
|
|
|
all_settings = api_client.get("/api/settings/connectors").json()
|
|
assert binding not in str(all_settings)
|
|
assert "GENERIC_AUTHENTIK_TOKEN" not in str(all_settings)
|
|
|
|
|
|
def test_portainer_config_save_trims_safe_fields(api_client):
|
|
payload = api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={
|
|
"url": " http://198.18.10.150:9200/ ",
|
|
"endpointId": " 1 ",
|
|
"verifyTls": False,
|
|
"apiKey": "PORTAINER_SUPER_SECRET",
|
|
},
|
|
).json()
|
|
|
|
assert payload["safeConfig"]["url"] == "http://198.18.10.150:9200"
|
|
assert payload["safeConfig"]["endpointId"] == "1"
|
|
|
|
|
|
def test_portainer_config_save_defaults_endpoint_id(api_client):
|
|
payload = api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={
|
|
"url": "http://198.18.10.150:9200",
|
|
"endpointId": "",
|
|
"verifyTls": False,
|
|
"apiKey": "PORTAINER_SUPER_SECRET",
|
|
},
|
|
).json()
|
|
|
|
assert payload["safeConfig"]["endpointId"] == "1"
|
|
assert "PORTAINER_ENDPOINT_ID" not in payload["missing"]
|
|
|
|
|
|
def test_setup_test_portainer_missing_config_and_redaction(api_client):
|
|
response = api_client.post("/api/setup/test/portainer")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["success"] is False
|
|
assert payload["status"] == "missing_config"
|
|
assert "PORTAINER_API_KEY" in payload["requiredFields"]
|
|
|
|
|
|
def test_setup_test_portainer_success_summary(api_client, monkeypatch):
|
|
secret = "PORTAINER_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={"url": "http://portainer.local", "endpointId": "1", "verifyTls": False, "apiKey": secret},
|
|
)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.portainer.PortainerConnector.test_connection", lambda self: {"status": "connected"}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.portainer.PortainerConnector.list_containers", lambda self: {"status": "ok", "containers": [{"id": "one"}]}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.portainer.PortainerConnector.list_stacks", lambda self: {"status": "ok", "stacks": [{"id": "stack"}]}, raising=True)
|
|
|
|
payload = api_client.post("/api/setup/test/portainer").json()
|
|
|
|
assert payload["success"] is True
|
|
assert payload["status"] == "connected"
|
|
assert payload["summary"]["containers"] == 1
|
|
assert payload["summary"]["stacks"] == 1
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_setup_test_portainer_endpoint_error_is_actionable(api_client, monkeypatch):
|
|
secret = "PORTAINER_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={"url": "http://portainer.local", "endpointId": "99", "verifyTls": False, "apiKey": secret},
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.portainer.PortainerConnector.test_connection",
|
|
lambda self: {"status": "endpoint_error", "reason": "Portainer API error: HTTP 404 endpoint not found"},
|
|
raising=True,
|
|
)
|
|
|
|
payload = api_client.post("/api/setup/test/portainer").json()
|
|
|
|
assert payload["success"] is False
|
|
assert payload["status"] == "endpoint_error"
|
|
assert payload["requiredFields"] == ["PORTAINER_ENDPOINT_ID"]
|
|
assert payload["safeConfig"]["endpointId"] == "99"
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_setup_test_portainer_container_list_error_marks_endpoint(api_client, monkeypatch):
|
|
secret = "PORTAINER_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={"url": "http://portainer.local", "endpointId": "1", "verifyTls": False, "apiKey": secret},
|
|
)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.portainer.PortainerConnector.test_connection", lambda self: {"status": "connected"}, raising=True)
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.portainer.PortainerConnector.list_containers",
|
|
lambda self: {"status": "error", "reason": "Portainer API error: HTTP 404 endpoint not found", "containers": []},
|
|
raising=True,
|
|
)
|
|
monkeypatch.setattr("backend.app.connectors.portainer.PortainerConnector.list_stacks", lambda self: {"status": "ok", "stacks": []}, raising=True)
|
|
|
|
payload = api_client.post("/api/setup/test/portainer").json()
|
|
|
|
assert payload["success"] is False
|
|
assert payload["status"] == "endpoint_error"
|
|
assert payload["summary"]["endpointReachable"] is False
|
|
assert payload["requiredFields"] == ["PORTAINER_ENDPOINT_ID"]
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_unraid_config_save_never_returns_secret(api_client):
|
|
secret = "UNRAID_SUPER_SECRET"
|
|
|
|
payload = api_client.patch(
|
|
"/api/settings/connectors/unraid",
|
|
json={"url": "http://unraid.local", "verifyTls": False, "apiKey": secret},
|
|
).json()
|
|
|
|
assert payload["connectorId"] == "unraid"
|
|
assert payload["configured"] is True
|
|
assert payload["hasSecret"] is True
|
|
assert payload["safeConfig"] == {"url": "http://unraid.local", "verifyTls": False}
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_setup_test_authentik_404_marks_url_error(api_client, monkeypatch):
|
|
secret = "AUTHENTIK_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/authentik",
|
|
json={"url": "https://auth.example.local/if/flow/default-authentication-flow", "verifyTls": False, "apiKey": secret},
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.authentik.AuthentikConnector.test_connection",
|
|
lambda self: {"status": "unreachable", "reason": "Authentik API error: HTTP 404 <!DOCTYPE html>"},
|
|
raising=True,
|
|
)
|
|
|
|
payload = api_client.post("/api/setup/test/connector/authentik").json()
|
|
|
|
assert payload["success"] is False
|
|
assert payload["status"] == "url_error"
|
|
assert payload["requiredFields"] == ["AUTHENTIK_URL"]
|
|
assert "root/base URL" in payload["message"]
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_setup_test_authentik_auth_failed_marks_token(api_client, monkeypatch):
|
|
secret = "AUTHENTIK_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/authentik",
|
|
json={"url": "https://auth.example.local", "verifyTls": False, "apiKey": secret},
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.authentik.AuthentikConnector.test_connection",
|
|
lambda self: {"status": "auth_failed", "reason": "auth_failed"},
|
|
raising=True,
|
|
)
|
|
|
|
payload = api_client.post("/api/setup/test/connector/authentik").json()
|
|
|
|
assert payload["success"] is False
|
|
assert payload["status"] == "auth_failed"
|
|
assert payload["requiredFields"] == ["AUTHENTIK_TOKEN"]
|
|
assert "token ongeldig of verlopen" in payload["message"]
|
|
assert "API Token" in payload["message"]
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_unraid_partial_config_update_preserves_url_and_secret(api_client):
|
|
secret = "UNRAID_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/unraid",
|
|
json={"url": "https://unraid.local", "verifyTls": True, "apiKey": secret},
|
|
)
|
|
|
|
payload = api_client.patch(
|
|
"/api/settings/connectors/unraid",
|
|
json={"verifyTls": False},
|
|
).json()
|
|
|
|
assert payload["configured"] is True
|
|
assert payload["hasSecret"] is True
|
|
assert payload["safeConfig"] == {"url": "https://unraid.local", "verifyTls": False}
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_setup_test_prowlarr_uses_connector_test(api_client, monkeypatch):
|
|
secret = "PROWLARR_SUPER_SECRET"
|
|
api_client.patch(
|
|
"/api/settings/connectors/prowlarr",
|
|
json={"url": "http://prowlarr.local:9696", "verifyTls": False, "apiKey": secret},
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.prowlarr.ProwlarrConnector.test_connection",
|
|
lambda self: {"status": "connected", "version": "1.2.3"},
|
|
raising=True,
|
|
)
|
|
|
|
payload = api_client.post("/api/setup/test/connector/prowlarr").json()
|
|
|
|
assert payload["success"] is True
|
|
assert payload["status"] == "connected"
|
|
assert payload["connected"] is True
|
|
assert payload["settings"]["safeConfig"]["url"] == "http://prowlarr.local:9696"
|
|
assert secret not in str(payload)
|
|
|
|
|
|
def test_setup_test_prowlarr_duplicate_radarr_url_is_actionable(api_client, monkeypatch):
|
|
prowlarr_secret = "PROWLARR_SUPER_SECRET"
|
|
radarr_secret = "RADARR_SUPER_SECRET"
|
|
shared_url = "http://media.local:7878"
|
|
api_client.patch(
|
|
"/api/settings/connectors/radarr",
|
|
json={"url": shared_url, "verifyTls": False, "apiKey": radarr_secret},
|
|
)
|
|
api_client.patch(
|
|
"/api/settings/connectors/prowlarr",
|
|
json={"url": shared_url, "verifyTls": False, "apiKey": prowlarr_secret},
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.prowlarr.ProwlarrConnector.test_connection",
|
|
lambda self: {"status": "error", "reason": "unauthorized"},
|
|
raising=True,
|
|
)
|
|
|
|
payload = api_client.post("/api/setup/test/connector/prowlarr").json()
|
|
|
|
assert payload["success"] is False
|
|
assert payload["status"] == "url_mismatch"
|
|
assert payload["requiredFields"] == ["PROWLARR_URL", "PROWLARR_API_KEY"]
|
|
assert "Radarr" in payload["message"]
|
|
assert "Prowlarr" in payload["message"]
|
|
assert prowlarr_secret not in str(payload)
|
|
assert radarr_secret not in str(payload)
|
|
|
|
|
|
def test_complete_setup_sets_first_run_completed(api_client):
|
|
payload = api_client.post("/api/setup/complete").json()
|
|
|
|
assert payload["success"] is True
|
|
assert payload["status"]["firstRunCompleted"] is True
|
|
assert payload["status"]["setupCompletedAt"]
|