Jeroen SchweitzerandClaude Opus 4.5
7752cd9d23
Architecture changes: - Permission format: domain.category:action (e.g., control-room.general:admin) - Decoupled groups from roles via group_roles mapping table - Groups are organizational (synced from Authentik) - Roles are permissions (admin-managed via API) New features: - require_permission() and require_any_permission() dependency factories - Action hierarchy: admin > editor > user > viewer - Global admin override (admin.general:admin grants all) - Group-role management endpoints (assign/remove roles) - GET /auth/roles endpoint to list all roles Database changes: - Added category column to roles table (default: general) - Removed authentik_group column (decoupled) - Added group_roles association table - Added user_groups association table - Migration updates role names to domain.general:action format Tests: - 67 new tests for auth service and controller - Covers token validation, user sync, role sync - Covers group-role assignment/removal - Covers schema conversions and permission system 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
518 lines
19 KiB
Python
518 lines
19 KiB
Python
"""Tests for authentication controller endpoints."""
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.main import app
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
"""Create a test client."""
|
|
return TestClient(app)
|
|
|
|
|
|
# =============================================================================
|
|
# OpenAPI Spec Tests
|
|
# =============================================================================
|
|
|
|
class TestAuthOpenAPISpec:
|
|
"""Test that auth endpoints are documented in OpenAPI spec."""
|
|
|
|
def test_auth_sync_in_openapi(self, client):
|
|
"""Auth sync endpoint should be in OpenAPI spec."""
|
|
response = client.get("/openapi.json")
|
|
assert response.status_code == 200
|
|
spec = response.json()
|
|
assert "/auth/sync" in spec["paths"]
|
|
assert "post" in spec["paths"]["/auth/sync"]
|
|
|
|
def test_auth_users_in_openapi(self, client):
|
|
"""Auth users endpoint should be in OpenAPI spec."""
|
|
response = client.get("/openapi.json")
|
|
spec = response.json()
|
|
assert "/auth/users" in spec["paths"]
|
|
assert "get" in spec["paths"]["/auth/users"]
|
|
|
|
def test_auth_groups_in_openapi(self, client):
|
|
"""Auth groups endpoint should be in OpenAPI spec."""
|
|
response = client.get("/openapi.json")
|
|
spec = response.json()
|
|
assert "/auth/groups" in spec["paths"]
|
|
assert "get" in spec["paths"]["/auth/groups"]
|
|
|
|
def test_auth_roles_in_openapi(self, client):
|
|
"""Auth roles endpoint should be in OpenAPI spec."""
|
|
response = client.get("/openapi.json")
|
|
spec = response.json()
|
|
assert "/auth/roles" in spec["paths"]
|
|
assert "get" in spec["paths"]["/auth/roles"]
|
|
|
|
def test_group_role_assignment_in_openapi(self, client):
|
|
"""Group role assignment endpoints should be in OpenAPI spec."""
|
|
response = client.get("/openapi.json")
|
|
spec = response.json()
|
|
path = "/auth/groups/{group_id}/roles/{role_id}"
|
|
assert path in spec["paths"]
|
|
assert "post" in spec["paths"][path] # Assign
|
|
assert "delete" in spec["paths"][path] # Remove
|
|
|
|
def test_sync_from_authentik_endpoints(self, client):
|
|
"""Sync from Authentik endpoints should be in OpenAPI spec."""
|
|
response = client.get("/openapi.json")
|
|
spec = response.json()
|
|
assert "/auth/users/sync-from-authentik" in spec["paths"]
|
|
assert "/auth/groups/sync-from-authentik" in spec["paths"]
|
|
|
|
|
|
# =============================================================================
|
|
# Controller Module Tests
|
|
# =============================================================================
|
|
|
|
class TestAuthControllerModule:
|
|
"""Test auth controller module imports and configuration."""
|
|
|
|
def test_controller_imports(self):
|
|
"""Auth controller should be importable."""
|
|
from src.domains.auth.controller import AuthController, auth_controller
|
|
assert AuthController is not None
|
|
assert auth_controller is not None
|
|
|
|
def test_controller_has_correct_prefix(self):
|
|
"""Auth controller should have /auth prefix."""
|
|
from src.domains.auth.controller import auth_controller
|
|
assert auth_controller.prefix == "/auth"
|
|
|
|
def test_controller_has_correct_tags(self):
|
|
"""Auth controller should have Authentication tag."""
|
|
from src.domains.auth.controller import auth_controller
|
|
assert "Authentication" in auth_controller.tags
|
|
|
|
|
|
# =============================================================================
|
|
# Sync Endpoint Tests
|
|
# =============================================================================
|
|
|
|
class TestSyncEndpoint:
|
|
"""Test POST /auth/sync endpoint."""
|
|
|
|
def test_sync_requires_access_token(self, client):
|
|
"""Sync should require access_token in body."""
|
|
response = client.post("/auth/sync", json={})
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
def test_sync_with_invalid_token(self, client):
|
|
"""Sync should return 401 for invalid token."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.validate_token = AsyncMock(
|
|
side_effect=ValueError("Invalid or expired token")
|
|
)
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.post(
|
|
"/auth/sync",
|
|
json={"access_token": "invalid_token"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
# =============================================================================
|
|
# Users Endpoint Tests
|
|
# =============================================================================
|
|
|
|
class TestUsersEndpoint:
|
|
"""Test GET /auth/users endpoint."""
|
|
|
|
def test_list_users_returns_list(self, client):
|
|
"""List users should return paginated response."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_users = AsyncMock(return_value=([], 0))
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.get("/auth/users")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
|
|
def test_list_users_with_search(self, client):
|
|
"""List users should accept search parameter."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_users = AsyncMock(return_value=([], 0))
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.get("/auth/users?search=test")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_list_users_with_pagination(self, client):
|
|
"""List users should accept pagination parameters."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_users = AsyncMock(return_value=([], 0))
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.get("/auth/users?offset=10&limit=20")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_list_users_limit_validation(self, client):
|
|
"""List users should reject limit > 100."""
|
|
response = client.get("/auth/users?limit=200")
|
|
assert response.status_code == 422
|
|
|
|
|
|
# =============================================================================
|
|
# Groups Endpoint Tests
|
|
# =============================================================================
|
|
|
|
class TestGroupsEndpoint:
|
|
"""Test GET /auth/groups endpoint."""
|
|
|
|
def test_list_groups_returns_list(self, client):
|
|
"""List groups should return paginated response."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_groups = AsyncMock(return_value=([], 0))
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.get("/auth/groups")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
|
|
def test_list_groups_with_search(self, client):
|
|
"""List groups should accept search parameter."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_groups = AsyncMock(return_value=([], 0))
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.get("/auth/groups?search=admin")
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
# =============================================================================
|
|
# Roles Endpoint Tests
|
|
# =============================================================================
|
|
|
|
class TestRolesEndpoint:
|
|
"""Test GET /auth/roles endpoint."""
|
|
|
|
def test_list_roles_returns_list(self, client):
|
|
"""List roles should return all roles."""
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_roles = AsyncMock(return_value=[])
|
|
mock_instance.roles_to_schema = MagicMock(return_value=[])
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.get("/auth/roles")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
|
|
|
|
# =============================================================================
|
|
# Group-Role Assignment Endpoint Tests
|
|
# =============================================================================
|
|
|
|
class TestGroupRoleAssignmentEndpoints:
|
|
"""Test group-role assignment and removal endpoints."""
|
|
|
|
def test_assign_role_to_group_success(self, client):
|
|
"""POST /auth/groups/{id}/roles/{id} should assign role."""
|
|
group_id = str(uuid.uuid4())
|
|
role_id = str(uuid.uuid4())
|
|
|
|
mock_group = MagicMock()
|
|
mock_group.id = uuid.UUID(group_id)
|
|
mock_group.name = "Test Group"
|
|
mock_group.roles = []
|
|
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.assign_role_to_group = AsyncMock(return_value=mock_group)
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.post(f"/auth/groups/{group_id}/roles/{role_id}")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["group_id"] == group_id
|
|
assert data["group_name"] == "Test Group"
|
|
assert "roles" in data
|
|
|
|
def test_assign_role_to_group_not_found(self, client):
|
|
"""POST should return 404 when group not found."""
|
|
group_id = str(uuid.uuid4())
|
|
role_id = str(uuid.uuid4())
|
|
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.assign_role_to_group = AsyncMock(
|
|
side_effect=ValueError("Group not found")
|
|
)
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.post(f"/auth/groups/{group_id}/roles/{role_id}")
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_remove_role_from_group_success(self, client):
|
|
"""DELETE /auth/groups/{id}/roles/{id} should remove role."""
|
|
group_id = str(uuid.uuid4())
|
|
role_id = str(uuid.uuid4())
|
|
|
|
mock_group = MagicMock()
|
|
mock_group.id = uuid.UUID(group_id)
|
|
mock_group.name = "Test Group"
|
|
mock_group.roles = []
|
|
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.remove_role_from_group = AsyncMock(return_value=mock_group)
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.delete(f"/auth/groups/{group_id}/roles/{role_id}")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["group_id"] == group_id
|
|
|
|
def test_remove_role_from_group_not_found(self, client):
|
|
"""DELETE should return 404 when role not found."""
|
|
group_id = str(uuid.uuid4())
|
|
role_id = str(uuid.uuid4())
|
|
|
|
with patch("src.domains.auth.controller.AuthService") as MockService:
|
|
mock_instance = MagicMock()
|
|
mock_instance.remove_role_from_group = AsyncMock(
|
|
side_effect=ValueError("Role not found")
|
|
)
|
|
MockService.return_value = mock_instance
|
|
|
|
response = client.delete(f"/auth/groups/{group_id}/roles/{role_id}")
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_assign_role_invalid_uuid(self, client):
|
|
"""POST should return 422 for invalid UUID."""
|
|
response = client.post("/auth/groups/not-a-uuid/roles/also-not-uuid")
|
|
assert response.status_code == 422
|
|
|
|
|
|
# =============================================================================
|
|
# Me Endpoint Tests
|
|
# =============================================================================
|
|
|
|
class TestMeEndpoint:
|
|
"""Test GET /auth/me endpoint."""
|
|
|
|
def test_me_not_implemented(self, client):
|
|
"""Me endpoint should return 501 (not implemented yet)."""
|
|
response = client.get("/auth/me")
|
|
assert response.status_code == 501
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Tests
|
|
# =============================================================================
|
|
|
|
class TestAuthSchemas:
|
|
"""Test auth schema imports and structure."""
|
|
|
|
def test_all_schemas_importable(self):
|
|
"""All auth schemas should be importable."""
|
|
from src.domains.auth.schemas import (
|
|
AuthSyncRequest,
|
|
AuthSyncResponse,
|
|
RoleSchema,
|
|
UserSchema,
|
|
UserPreferencesSchema,
|
|
TokenInfoSchema,
|
|
UserListItemSchema,
|
|
UsersListResponse,
|
|
BulkSyncResultSchema,
|
|
GroupListItemSchema,
|
|
GroupsListResponse,
|
|
RolesListResponse,
|
|
GroupRoleAssignmentResponse,
|
|
)
|
|
|
|
assert AuthSyncRequest is not None
|
|
assert AuthSyncResponse is not None
|
|
assert RoleSchema is not None
|
|
assert UserSchema is not None
|
|
assert UserPreferencesSchema is not None
|
|
assert TokenInfoSchema is not None
|
|
assert UserListItemSchema is not None
|
|
assert UsersListResponse is not None
|
|
assert BulkSyncResultSchema is not None
|
|
assert GroupListItemSchema is not None
|
|
assert GroupsListResponse is not None
|
|
assert RolesListResponse is not None
|
|
assert GroupRoleAssignmentResponse is not None
|
|
|
|
def test_role_schema_includes_id(self):
|
|
"""RoleSchema should include id field."""
|
|
from src.domains.auth.schemas import RoleSchema
|
|
|
|
schema = RoleSchema(
|
|
id=uuid.uuid4(),
|
|
name="test.general:admin",
|
|
domain="test",
|
|
category="general",
|
|
action="admin",
|
|
)
|
|
assert schema.id is not None
|
|
|
|
def test_role_schema_includes_category(self):
|
|
"""RoleSchema should include category field."""
|
|
from src.domains.auth.schemas import RoleSchema
|
|
|
|
schema = RoleSchema(
|
|
id=uuid.uuid4(),
|
|
name="test.specific:viewer",
|
|
domain="test",
|
|
category="specific",
|
|
action="viewer",
|
|
)
|
|
assert schema.category == "specific"
|
|
|
|
def test_group_list_item_includes_roles(self):
|
|
"""GroupListItemSchema should include roles list."""
|
|
from src.domains.auth.schemas import GroupListItemSchema
|
|
|
|
schema = GroupListItemSchema(
|
|
id=uuid.uuid4(),
|
|
authentik_id=uuid.uuid4(),
|
|
name="Test Group",
|
|
is_superuser=False,
|
|
parent_name=None,
|
|
member_count=5,
|
|
synced_at=datetime.now(timezone.utc),
|
|
roles=["admin.general:admin", "media.general:viewer"],
|
|
)
|
|
assert len(schema.roles) == 2
|
|
|
|
|
|
# =============================================================================
|
|
# Model Tests
|
|
# =============================================================================
|
|
|
|
class TestAuthModels:
|
|
"""Test auth model imports."""
|
|
|
|
def test_all_models_importable(self):
|
|
"""All auth models should be importable."""
|
|
from src.domains.auth.models import (
|
|
User,
|
|
Role,
|
|
UserRole,
|
|
Group,
|
|
UserPreferences,
|
|
ApiKey,
|
|
user_groups,
|
|
group_roles,
|
|
)
|
|
|
|
assert User is not None
|
|
assert Role is not None
|
|
assert UserRole is not None
|
|
assert Group is not None
|
|
assert UserPreferences is not None
|
|
assert ApiKey is not None
|
|
assert user_groups is not None
|
|
assert group_roles is not None
|
|
|
|
def test_role_model_has_category(self):
|
|
"""Role model should have category attribute."""
|
|
from src.domains.auth.models import Role
|
|
|
|
role = Role(
|
|
name="test.general:admin",
|
|
domain="test",
|
|
category="general",
|
|
action="admin",
|
|
)
|
|
assert role.category == "general"
|
|
|
|
def test_group_has_roles_relationship(self):
|
|
"""Group model should have roles relationship."""
|
|
from src.domains.auth.models import Group
|
|
|
|
assert hasattr(Group, "roles")
|
|
|
|
|
|
# =============================================================================
|
|
# Permission System Tests (from oidc.py)
|
|
# =============================================================================
|
|
|
|
class TestPermissionSystem:
|
|
"""Test permission checking functions."""
|
|
|
|
def test_permission_constants_defined(self):
|
|
"""Permission constants should be defined."""
|
|
from src.domains.auth.oidc import (
|
|
ACTION_HIERARCHY,
|
|
VALID_DOMAINS,
|
|
DEFAULT_CATEGORY,
|
|
)
|
|
|
|
assert "viewer" in ACTION_HIERARCHY
|
|
assert "user" in ACTION_HIERARCHY
|
|
assert "editor" in ACTION_HIERARCHY
|
|
assert "admin" in ACTION_HIERARCHY
|
|
|
|
assert "control-room" in VALID_DOMAINS
|
|
assert "media" in VALID_DOMAINS
|
|
assert "admin" in VALID_DOMAINS
|
|
|
|
assert DEFAULT_CATEGORY == "general"
|
|
|
|
def test_action_hierarchy_ordering(self):
|
|
"""Action hierarchy should be ordered correctly."""
|
|
from src.domains.auth.oidc import ACTION_HIERARCHY
|
|
|
|
assert ACTION_HIERARCHY["viewer"] < ACTION_HIERARCHY["user"]
|
|
assert ACTION_HIERARCHY["user"] < ACTION_HIERARCHY["editor"]
|
|
assert ACTION_HIERARCHY["editor"] < ACTION_HIERARCHY["admin"]
|
|
|
|
def test_require_permission_importable(self):
|
|
"""require_permission should be importable."""
|
|
from src.domains.auth.oidc import require_permission, require_any_permission
|
|
|
|
assert callable(require_permission)
|
|
assert callable(require_any_permission)
|
|
|
|
def test_require_permission_returns_dependency(self):
|
|
"""require_permission should return a callable dependency."""
|
|
from src.domains.auth.oidc import require_permission
|
|
|
|
dependency = require_permission("control-room", "admin")
|
|
assert callable(dependency)
|
|
|
|
def test_require_any_permission_returns_dependency(self):
|
|
"""require_any_permission should return a callable dependency."""
|
|
from src.domains.auth.oidc import require_any_permission
|
|
|
|
dependency = require_any_permission(
|
|
("control-room", "admin"),
|
|
("media", "editor"),
|
|
)
|
|
assert callable(dependency)
|