feat: add Authentik access widgets

This commit is contained in:
Developer
2026-07-14 21:41:24 +00:00
parent 4562a9dfca
commit 17976eab80
21 changed files with 1082 additions and 214 deletions
+103 -2
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Generator
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -19,7 +20,7 @@ TEST_KEY = Fernet.generate_key().decode()
@pytest.fixture(autouse=True)
def _encryption_key(monkeypatch: pytest.MonkeyPatch) -> None:
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()
@@ -28,7 +29,7 @@ def _encryption_key(monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.fixture()
def store(tmp_path: Path) -> SettingsStore:
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
@@ -181,3 +182,103 @@ class TestAuthentikUsersEndpoint:
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)