"""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)