Developer
285 lines
12 KiB
Python
285 lines
12 KiB
Python
"""Tests for AuthentikClient and the directory endpoint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Generator
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
from fastapi.testclient import TestClient
|
|
|
|
from media_library_viewer_api.clients.authentik import AuthentikClient
|
|
from media_library_viewer_api.dependencies import get_settings_store
|
|
from media_library_viewer_api.main import app
|
|
from media_library_viewer_api.services.secrets import 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: pytest.MonkeyPatch) -> Generator[None, None, None]:
|
|
"""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 store(tmp_path: Path) -> Generator[SettingsStore, None, None]:
|
|
s = SettingsStore(tmp_path / "settings.sqlite")
|
|
s.ensure_defaults()
|
|
app.dependency_overrides[get_settings_store] = lambda: s
|
|
yield s
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Client unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAuthentikClient:
|
|
def test_base_url_normalizes_trailing_slash(self) -> None:
|
|
c = AuthentikClient(base_url="https://auth.example.com/", api_token="t")
|
|
assert c.base_url == "https://auth.example.com"
|
|
|
|
def test_base_url_strips_api_v3_suffix(self) -> None:
|
|
c = AuthentikClient(base_url="https://auth.example.com/api/v3", api_token="t")
|
|
assert c.base_url == "https://auth.example.com"
|
|
|
|
def test_bearer_header_is_set(self) -> None:
|
|
c = AuthentikClient(base_url="https://auth.example.com", api_token="tok")
|
|
assert c.session.headers["Authorization"] == "Bearer tok"
|
|
|
|
def test_empty_base_url_raises(self) -> None:
|
|
with pytest.raises(ValueError):
|
|
AuthentikClient(base_url="", api_token="t")
|
|
|
|
def test_empty_api_token_raises(self) -> None:
|
|
with pytest.raises(ValueError):
|
|
AuthentikClient(base_url="https://auth.example.com", api_token="")
|
|
|
|
@patch.object(AuthentikClient, "get")
|
|
def test_users_normalizes_pagination(self, mock_get: MagicMock) -> None:
|
|
mock_get.return_value = {
|
|
"pagination": {"count": 42, "next": 2, "previous": 0, "current": 1},
|
|
"results": [
|
|
{"pk": 1, "username": "alice", "email": "alice@example.com"},
|
|
{"pk": 2, "username": "bob", "email": "bob@example.com"},
|
|
],
|
|
}
|
|
client = AuthentikClient(base_url="https://auth.example.com", api_token="t")
|
|
result = client.users(search="ali", page=1, page_size=2)
|
|
assert result["total"] == 42
|
|
assert result["page"] == 1
|
|
assert result["page_size"] == 2
|
|
assert len(result["items"]) == 2
|
|
assert result["items"][0]["username"] == "alice"
|
|
|
|
@patch.object(AuthentikClient, "get")
|
|
def test_users_handles_empty_results(self, mock_get: MagicMock) -> None:
|
|
mock_get.return_value = {"pagination": {"count": 0}, "results": []}
|
|
client = AuthentikClient(base_url="https://auth.example.com", api_token="t")
|
|
result = client.users()
|
|
assert result["items"] == []
|
|
assert result["total"] == 0
|
|
|
|
@patch.object(AuthentikClient, "get")
|
|
def test_users_handles_non_dict_payload(self, mock_get: MagicMock) -> None:
|
|
mock_get.return_value = []
|
|
client = AuthentikClient(base_url="https://auth.example.com", api_token="t")
|
|
result = client.users()
|
|
assert result["items"] == []
|
|
assert result["total"] == 0
|
|
|
|
@patch("media_library_viewer_api.clients.authentik.requests.Session")
|
|
def test_get_sends_correct_url_and_params(self, mock_session_cls: MagicMock) -> None:
|
|
mock_session = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"results": []}
|
|
mock_response.raise_for_status.return_value = None
|
|
mock_session.get.return_value = mock_response
|
|
|
|
c = AuthentikClient(base_url="https://auth.example.com", api_token="t")
|
|
c.get("/core/users/", search="x", page=2)
|
|
|
|
call_args = mock_session.get.call_args
|
|
assert call_args.kwargs["params"] == {"search": "x", "page": 2}
|
|
assert call_args.args[0] == "https://auth.example.com/api/v3/core/users/"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Endpoint integration tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAuthentikUsersEndpoint:
|
|
def test_not_configured_returns_empty_with_error(self, store: SettingsStore) -> None:
|
|
client = TestClient(app)
|
|
response = client.get("/api/services/authentik/nonexistent/users")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["items"] == []
|
|
assert data["total"] == 0
|
|
assert "error" in data
|
|
|
|
@patch("media_library_viewer_api.routers.authentik_users.AuthentikClient")
|
|
def test_success_returns_users(self, mock_client_cls: MagicMock, store: SettingsStore) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.users.return_value = {
|
|
"items": [{"pk": 1, "username": "alice"}],
|
|
"total": 1,
|
|
"page": 1,
|
|
"page_size": 50,
|
|
}
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
created = store.upsert_service(
|
|
{
|
|
"service_type": "authentik",
|
|
"name": "Main",
|
|
"config": {"base_url": "https://auth.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_token": "secret-token"},
|
|
)
|
|
service_id = created["id"]
|
|
|
|
client = TestClient(app)
|
|
response = client.get(f"/api/services/authentik/{service_id}/users?search=ali")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 1
|
|
assert data["items"][0]["username"] == "alice"
|
|
assert data["total"] == 1
|
|
assert "error" not in data
|
|
|
|
@patch("media_library_viewer_api.routers.authentik_users.AuthentikClient")
|
|
def test_unreachable_returns_error(self, mock_client_cls: MagicMock, store: SettingsStore) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.users.side_effect = ConnectionError("refused")
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
created = store.upsert_service(
|
|
{
|
|
"service_type": "authentik",
|
|
"name": "Main",
|
|
"config": {"base_url": "https://auth.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_token": "secret-token"},
|
|
)
|
|
service_id = created["id"]
|
|
|
|
client = TestClient(app)
|
|
response = client.get(f"/api/services/authentik/{service_id}/users")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["items"] == []
|
|
assert "error" in data
|
|
|
|
|
|
class TestAuthentikAccessMetadata:
|
|
@patch.object(AuthentikClient, "get")
|
|
def test_groups_and_applications_paginate_and_whitelist_fields(self, mock_get: MagicMock) -> None:
|
|
def payload(path: str, **params: object) -> dict[str, object]:
|
|
if path == "/core/groups/":
|
|
if params["page"] == 1:
|
|
return {"pagination": {"count": 2}, "results": [{"pk": 1, "name": "Admins"}]}
|
|
return {"pagination": {"count": 2}, "results": [{"id": "g2", "display_name": "Readers"}]}
|
|
return {
|
|
"pagination": {"count": 1},
|
|
"results": [
|
|
{
|
|
"pk": 3,
|
|
"name": "Portal",
|
|
"slug": "portal",
|
|
"meta_launch_url": "https://portal.example.com",
|
|
"provider": {"client_secret": "must-not-leak"},
|
|
"policy_engine_mode": "any",
|
|
}
|
|
],
|
|
}
|
|
|
|
mock_get.side_effect = payload
|
|
auth = AuthentikClient(base_url="https://auth.example.com", api_token="t")
|
|
assert auth.groups(limit=2)["items"] == [{"id": "1", "name": "Admins"}, {"id": "g2", "name": "Readers"}]
|
|
application = auth.applications(limit=1)["items"][0]
|
|
assert application == {
|
|
"id": "3",
|
|
"name": "Portal",
|
|
"slug": "portal",
|
|
"launch_url": "https://portal.example.com",
|
|
}
|
|
assert "provider" not in application
|
|
|
|
@patch.object(AuthentikClient, "get")
|
|
def test_access_summary_uses_group_references_without_user_detail_calls(self, mock_get: MagicMock) -> None:
|
|
def payload(path: str, **params: object) -> dict[str, object]:
|
|
if path == "/core/users/":
|
|
return {
|
|
"pagination": {"count": 1},
|
|
"results": [
|
|
{
|
|
"pk": 7,
|
|
"username": "alice",
|
|
"name": "Alice",
|
|
"groups": [1, {"id": "missing"}],
|
|
"is_superuser": True,
|
|
"is_staff": False,
|
|
}
|
|
],
|
|
}
|
|
assert path == "/core/groups/"
|
|
return {"pagination": {"count": 1}, "results": [{"pk": 1, "name": "Admins"}]}
|
|
|
|
mock_get.side_effect = payload
|
|
result = AuthentikClient(base_url="https://auth.example.com", api_token="t").access_summaries()
|
|
assert result["items"][0]["groups"] == [
|
|
{"id": "1", "name": "Admins", "known": True},
|
|
{"id": "missing", "name": "Unknown group (missing)", "known": False},
|
|
]
|
|
assert bool(result["items"][0]["is_superuser"])
|
|
assert all(call.args[0] in {"/core/users/", "/core/groups/"} for call in mock_get.call_args_list)
|
|
|
|
|
|
class TestAuthentikAccessEndpoints:
|
|
def test_not_configured_access_collections_return_empty_envelopes(self, store: SettingsStore) -> None:
|
|
client = TestClient(app)
|
|
for path in ("access-summary", "groups", "applications"):
|
|
response = client.get(f"/api/services/authentik/missing/{path}")
|
|
assert response.status_code == 200
|
|
assert response.json()["items"] == []
|
|
assert response.json()["error"] == "Authentik service not configured"
|
|
|
|
@patch("media_library_viewer_api.routers.authentik_users.AuthentikClient")
|
|
def test_access_summary_endpoint_returns_normalized_data(
|
|
self, mock_client_cls: MagicMock, store: SettingsStore
|
|
) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.access_summaries.return_value = {
|
|
"items": [{"id": "1", "groups": []}],
|
|
"total": 1,
|
|
"page": 1,
|
|
"page_size": 25,
|
|
}
|
|
mock_client_cls.return_value = mock_client
|
|
service = store.upsert_service(
|
|
{
|
|
"service_type": "authentik",
|
|
"name": "Main",
|
|
"config": {"base_url": "https://auth.example.com"},
|
|
"enabled": True,
|
|
},
|
|
secret_values={"api_token": "secret-token"},
|
|
)
|
|
response = TestClient(app).get(f"/api/services/authentik/{service['id']}/access-summary?page_size=25")
|
|
assert response.status_code == 200
|
|
assert response.json()["items"] == [{"id": "1", "groups": []}]
|
|
mock_client.access_summaries.assert_called_once_with(search=None, page=1, page_size=25)
|