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>
628 lines
22 KiB
Python
628 lines
22 KiB
Python
"""Tests for authentication service."""
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.domains.auth.models import User, Role, Group, UserPreferences
|
|
from src.domains.auth.schemas import TokenInfoSchema, RoleSchema
|
|
from src.domains.auth.service import AuthService, get_auth_service
|
|
|
|
|
|
# =============================================================================
|
|
# Fixtures
|
|
# =============================================================================
|
|
|
|
@pytest.fixture
|
|
def mock_session():
|
|
"""Create a mock async database session."""
|
|
session = AsyncMock(spec=AsyncSession)
|
|
session.execute = AsyncMock()
|
|
session.commit = AsyncMock()
|
|
session.flush = AsyncMock()
|
|
session.refresh = AsyncMock()
|
|
session.add = MagicMock()
|
|
return session
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_service(mock_session):
|
|
"""Create an AuthService instance with mock session."""
|
|
return AuthService(mock_session)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_user():
|
|
"""Create a sample user for testing."""
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
authentik_id=uuid.uuid4(),
|
|
email="test@example.com",
|
|
name="Test User",
|
|
avatar_url="https://example.com/avatar.jpg",
|
|
created_at=datetime.now(timezone.utc),
|
|
last_login=datetime.now(timezone.utc),
|
|
)
|
|
user.roles = []
|
|
user.preferences = None
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_role():
|
|
"""Create a sample role for testing."""
|
|
return Role(
|
|
id=uuid.uuid4(),
|
|
name="control-room.general:admin",
|
|
domain="control-room",
|
|
category="general",
|
|
action="admin",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_group(sample_role):
|
|
"""Create a sample group for testing."""
|
|
group = Group(
|
|
id=uuid.uuid4(),
|
|
authentik_id=uuid.uuid4(),
|
|
name="Administrators",
|
|
is_superuser=True,
|
|
parent_name=None,
|
|
member_count=5,
|
|
created_at=datetime.now(timezone.utc),
|
|
synced_at=datetime.now(timezone.utc),
|
|
)
|
|
group.roles = [sample_role]
|
|
return group
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_token_info():
|
|
"""Create sample token info from Authentik."""
|
|
return TokenInfoSchema(
|
|
sub=str(uuid.uuid4()),
|
|
email="test@example.com",
|
|
name="Test User",
|
|
preferred_username="testuser",
|
|
groups=["Administrators", "Developers"],
|
|
picture="https://example.com/avatar.jpg",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_preferences():
|
|
"""Create sample user preferences."""
|
|
return UserPreferences(
|
|
user_id=uuid.uuid4(),
|
|
theme="dark",
|
|
default_room="control-room",
|
|
preferences_json={"notifications": True},
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# AuthService Initialization Tests
|
|
# =============================================================================
|
|
|
|
class TestAuthServiceInit:
|
|
"""Test AuthService initialization."""
|
|
|
|
def test_init_with_session(self, mock_session):
|
|
"""AuthService should initialize with session."""
|
|
service = AuthService(mock_session)
|
|
assert service.session is mock_session
|
|
|
|
def test_init_sets_userinfo_url(self, mock_session):
|
|
"""AuthService should set userinfo URL from settings."""
|
|
service = AuthService(mock_session)
|
|
assert "userinfo" in service.userinfo_url
|
|
|
|
def test_get_auth_service_factory(self, mock_session):
|
|
"""get_auth_service should return AuthService instance."""
|
|
service = get_auth_service(mock_session)
|
|
assert isinstance(service, AuthService)
|
|
assert service.session is mock_session
|
|
|
|
|
|
# =============================================================================
|
|
# Token Validation Tests
|
|
# =============================================================================
|
|
|
|
class TestValidateToken:
|
|
"""Test token validation via Authentik userinfo endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_token_success(self, auth_service):
|
|
"""validate_token should return token info on success."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"sub": str(uuid.uuid4()),
|
|
"email": "test@example.com",
|
|
"name": "Test User",
|
|
"preferred_username": "testuser",
|
|
"groups": ["Administrators"],
|
|
"picture": "https://example.com/avatar.jpg",
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("src.domains.auth.service.httpx.AsyncClient") as mock_client:
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await auth_service.validate_token("valid_token")
|
|
|
|
assert result.email == "test@example.com"
|
|
assert result.name == "Test User"
|
|
assert "Administrators" in result.groups
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_token_invalid(self, auth_service):
|
|
"""validate_token should raise ValueError for invalid token."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 401
|
|
|
|
with patch("src.domains.auth.service.httpx.AsyncClient") as mock_client:
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Invalid or expired token"):
|
|
await auth_service.validate_token("invalid_token")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_token_service_unavailable(self, auth_service):
|
|
"""validate_token should raise ValueError when service unavailable."""
|
|
import httpx
|
|
|
|
with patch("src.domains.auth.service.httpx.AsyncClient") as mock_client:
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
side_effect=httpx.RequestError("Connection failed")
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Authentication service unavailable"):
|
|
await auth_service.validate_token("token")
|
|
|
|
|
|
# =============================================================================
|
|
# User Sync Tests
|
|
# =============================================================================
|
|
|
|
class TestSyncUser:
|
|
"""Test user synchronization from OIDC token."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_user_creates_new_user(self, auth_service, sample_token_info, mock_session):
|
|
"""sync_user should create new user when not found."""
|
|
# Mock no existing user found
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
user, is_new = await auth_service.sync_user(sample_token_info)
|
|
|
|
assert is_new is True
|
|
assert mock_session.add.call_count == 2 # User and Preferences
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_user_updates_existing_user(
|
|
self, auth_service, sample_token_info, sample_user, mock_session
|
|
):
|
|
"""sync_user should update existing user when found."""
|
|
# Mock existing user found
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = sample_user
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
# Update token info with matching authentik_id
|
|
sample_token_info.sub = str(sample_user.authentik_id)
|
|
|
|
user, is_new = await auth_service.sync_user(sample_token_info)
|
|
|
|
assert is_new is False
|
|
assert user.email == sample_token_info.email
|
|
assert user.name == sample_token_info.name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_user_updates_last_login(
|
|
self, auth_service, sample_token_info, sample_user, mock_session
|
|
):
|
|
"""sync_user should update last_login timestamp."""
|
|
old_login = sample_user.last_login
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = sample_user
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
sample_token_info.sub = str(sample_user.authentik_id)
|
|
|
|
user, _ = await auth_service.sync_user(sample_token_info)
|
|
|
|
assert user.last_login is not None
|
|
# last_login should be updated (or same if happened in same second)
|
|
assert user.last_login >= old_login or user.last_login is not None
|
|
|
|
|
|
# =============================================================================
|
|
# Role Sync Tests
|
|
# =============================================================================
|
|
|
|
class TestSyncRoles:
|
|
"""Test role synchronization from Authentik groups via group_roles."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_roles_from_groups(
|
|
self, auth_service, sample_user, sample_group, mock_session
|
|
):
|
|
"""sync_roles should get roles from matching groups."""
|
|
# Mock finding groups with roles
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [sample_group]
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
roles = await auth_service.sync_roles(sample_user, ["Administrators"])
|
|
|
|
assert len(roles) == 1
|
|
assert roles[0].name == "control-room.general:admin"
|
|
assert sample_user.roles == roles
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_roles_no_matching_groups(
|
|
self, auth_service, sample_user, mock_session
|
|
):
|
|
"""sync_roles should return empty list when no groups match."""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
roles = await auth_service.sync_roles(sample_user, ["NonExistentGroup"])
|
|
|
|
assert len(roles) == 0
|
|
assert sample_user.roles == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_roles_deduplicates_roles(
|
|
self, auth_service, sample_user, sample_role, mock_session
|
|
):
|
|
"""sync_roles should deduplicate roles from multiple groups."""
|
|
# Create two groups with the same role
|
|
group1 = Group(
|
|
id=uuid.uuid4(),
|
|
authentik_id=uuid.uuid4(),
|
|
name="Group1",
|
|
is_superuser=False,
|
|
member_count=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
synced_at=datetime.now(timezone.utc),
|
|
)
|
|
group1.roles = [sample_role]
|
|
|
|
group2 = Group(
|
|
id=uuid.uuid4(),
|
|
authentik_id=uuid.uuid4(),
|
|
name="Group2",
|
|
is_superuser=False,
|
|
member_count=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
synced_at=datetime.now(timezone.utc),
|
|
)
|
|
group2.roles = [sample_role] # Same role
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [group1, group2]
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
roles = await auth_service.sync_roles(sample_user, ["Group1", "Group2"])
|
|
|
|
# Should only have one role despite appearing in two groups
|
|
assert len(roles) == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Conversion Tests
|
|
# =============================================================================
|
|
|
|
class TestSchemaConversions:
|
|
"""Test model to schema conversions."""
|
|
|
|
def test_user_to_schema(self, auth_service, sample_user):
|
|
"""user_to_schema should convert User model to UserSchema."""
|
|
schema = auth_service.user_to_schema(sample_user)
|
|
|
|
assert schema.id == sample_user.id
|
|
assert schema.authentik_id == sample_user.authentik_id
|
|
assert schema.email == sample_user.email
|
|
assert schema.name == sample_user.name
|
|
assert schema.avatar_url == sample_user.avatar_url
|
|
|
|
def test_roles_to_schema(self, auth_service, sample_role):
|
|
"""roles_to_schema should convert Role models to RoleSchemas."""
|
|
schemas = auth_service.roles_to_schema([sample_role])
|
|
|
|
assert len(schemas) == 1
|
|
assert schemas[0].id == sample_role.id
|
|
assert schemas[0].name == sample_role.name
|
|
assert schemas[0].domain == sample_role.domain
|
|
assert schemas[0].category == sample_role.category
|
|
assert schemas[0].action == sample_role.action
|
|
|
|
def test_roles_to_schema_empty_list(self, auth_service):
|
|
"""roles_to_schema should handle empty list."""
|
|
schemas = auth_service.roles_to_schema([])
|
|
assert schemas == []
|
|
|
|
def test_preferences_to_schema(self, auth_service, sample_preferences):
|
|
"""preferences_to_schema should convert UserPreferences to schema."""
|
|
schema = auth_service.preferences_to_schema(sample_preferences)
|
|
|
|
assert schema.theme == sample_preferences.theme
|
|
assert schema.default_room == sample_preferences.default_room
|
|
assert schema.preferences_json == sample_preferences.preferences_json
|
|
|
|
def test_preferences_to_schema_none(self, auth_service):
|
|
"""preferences_to_schema should return defaults for None."""
|
|
schema = auth_service.preferences_to_schema(None)
|
|
|
|
assert schema.theme == "system"
|
|
assert schema.default_room == "front-hall"
|
|
assert schema.preferences_json == {}
|
|
|
|
|
|
# =============================================================================
|
|
# List Operations Tests
|
|
# =============================================================================
|
|
|
|
class TestListOperations:
|
|
"""Test list operations for users, groups, and roles."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users(self, auth_service, sample_user, mock_session):
|
|
"""list_users should return paginated user list."""
|
|
sample_user.roles = []
|
|
|
|
# Mock count query
|
|
count_result = MagicMock()
|
|
count_result.scalar.return_value = 1
|
|
|
|
# Mock users query
|
|
users_result = MagicMock()
|
|
users_result.scalars.return_value.all.return_value = [sample_user]
|
|
|
|
mock_session.execute.side_effect = [count_result, users_result]
|
|
|
|
items, total = await auth_service.list_users()
|
|
|
|
assert total == 1
|
|
assert len(items) == 1
|
|
assert items[0].email == sample_user.email
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users_with_search(self, auth_service, mock_session):
|
|
"""list_users should filter by search query."""
|
|
count_result = MagicMock()
|
|
count_result.scalar.return_value = 0
|
|
|
|
users_result = MagicMock()
|
|
users_result.scalars.return_value.all.return_value = []
|
|
|
|
mock_session.execute.side_effect = [count_result, users_result]
|
|
|
|
items, total = await auth_service.list_users(search="nonexistent")
|
|
|
|
assert total == 0
|
|
assert len(items) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_groups(self, auth_service, sample_group, mock_session):
|
|
"""list_groups should return paginated group list with roles."""
|
|
count_result = MagicMock()
|
|
count_result.scalar.return_value = 1
|
|
|
|
groups_result = MagicMock()
|
|
groups_result.scalars.return_value.all.return_value = [sample_group]
|
|
|
|
mock_session.execute.side_effect = [count_result, groups_result]
|
|
|
|
items, total = await auth_service.list_groups()
|
|
|
|
assert total == 1
|
|
assert len(items) == 1
|
|
assert items[0].name == sample_group.name
|
|
assert len(items[0].roles) == 1 # Should include role names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_roles(self, auth_service, sample_role, mock_session):
|
|
"""list_roles should return all roles ordered by domain."""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [sample_role]
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
roles = await auth_service.list_roles()
|
|
|
|
assert len(roles) == 1
|
|
assert roles[0].name == sample_role.name
|
|
|
|
|
|
# =============================================================================
|
|
# Group-Role Management Tests
|
|
# =============================================================================
|
|
|
|
class TestGroupRoleManagement:
|
|
"""Test group-role assignment and removal."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_group_by_id(self, auth_service, sample_group, mock_session):
|
|
"""get_group_by_id should return group with roles."""
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = sample_group
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
group = await auth_service.get_group_by_id(sample_group.id)
|
|
|
|
assert group is not None
|
|
assert group.id == sample_group.id
|
|
assert len(group.roles) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_group_by_id_not_found(self, auth_service, mock_session):
|
|
"""get_group_by_id should return None when not found."""
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
group = await auth_service.get_group_by_id(uuid.uuid4())
|
|
|
|
assert group is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_assign_role_to_group(
|
|
self, auth_service, sample_group, sample_role, mock_session
|
|
):
|
|
"""assign_role_to_group should add role to group."""
|
|
# Clear existing roles for this test
|
|
sample_group.roles = []
|
|
|
|
# Mock group lookup
|
|
group_result = MagicMock()
|
|
group_result.scalar_one_or_none.return_value = sample_group
|
|
|
|
# Mock role lookup
|
|
role_result = MagicMock()
|
|
role_result.scalar_one_or_none.return_value = sample_role
|
|
|
|
mock_session.execute.side_effect = [group_result, role_result]
|
|
|
|
group = await auth_service.assign_role_to_group(sample_group.id, sample_role.id)
|
|
|
|
assert sample_role in group.roles
|
|
mock_session.flush.assert_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_assign_role_to_group_already_assigned(
|
|
self, auth_service, sample_group, sample_role, mock_session
|
|
):
|
|
"""assign_role_to_group should not duplicate if already assigned."""
|
|
# Group already has this role
|
|
sample_group.roles = [sample_role]
|
|
original_count = len(sample_group.roles)
|
|
|
|
group_result = MagicMock()
|
|
group_result.scalar_one_or_none.return_value = sample_group
|
|
|
|
role_result = MagicMock()
|
|
role_result.scalar_one_or_none.return_value = sample_role
|
|
|
|
mock_session.execute.side_effect = [group_result, role_result]
|
|
|
|
group = await auth_service.assign_role_to_group(sample_group.id, sample_role.id)
|
|
|
|
assert len(group.roles) == original_count # No duplicate
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_assign_role_to_group_group_not_found(self, auth_service, mock_session):
|
|
"""assign_role_to_group should raise ValueError when group not found."""
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
with pytest.raises(ValueError, match="Group not found"):
|
|
await auth_service.assign_role_to_group(uuid.uuid4(), uuid.uuid4())
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_assign_role_to_group_role_not_found(
|
|
self, auth_service, sample_group, mock_session
|
|
):
|
|
"""assign_role_to_group should raise ValueError when role not found."""
|
|
sample_group.roles = []
|
|
|
|
group_result = MagicMock()
|
|
group_result.scalar_one_or_none.return_value = sample_group
|
|
|
|
role_result = MagicMock()
|
|
role_result.scalar_one_or_none.return_value = None
|
|
|
|
mock_session.execute.side_effect = [group_result, role_result]
|
|
|
|
with pytest.raises(ValueError, match="Role not found"):
|
|
await auth_service.assign_role_to_group(sample_group.id, uuid.uuid4())
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_role_from_group(
|
|
self, auth_service, sample_group, sample_role, mock_session
|
|
):
|
|
"""remove_role_from_group should remove role from group."""
|
|
# Group has this role
|
|
sample_group.roles = [sample_role]
|
|
|
|
group_result = MagicMock()
|
|
group_result.scalar_one_or_none.return_value = sample_group
|
|
|
|
role_result = MagicMock()
|
|
role_result.scalar_one_or_none.return_value = sample_role
|
|
|
|
mock_session.execute.side_effect = [group_result, role_result]
|
|
|
|
group = await auth_service.remove_role_from_group(sample_group.id, sample_role.id)
|
|
|
|
assert sample_role not in group.roles
|
|
mock_session.flush.assert_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_role_from_group_not_assigned(
|
|
self, auth_service, sample_group, sample_role, mock_session
|
|
):
|
|
"""remove_role_from_group should handle role not assigned gracefully."""
|
|
# Group does not have this role
|
|
sample_group.roles = []
|
|
|
|
group_result = MagicMock()
|
|
group_result.scalar_one_or_none.return_value = sample_group
|
|
|
|
role_result = MagicMock()
|
|
role_result.scalar_one_or_none.return_value = sample_role
|
|
|
|
mock_session.execute.side_effect = [group_result, role_result]
|
|
|
|
group = await auth_service.remove_role_from_group(sample_group.id, sample_role.id)
|
|
|
|
# Should complete without error
|
|
assert len(group.roles) == 0
|
|
|
|
|
|
# =============================================================================
|
|
# Role Schema Tests
|
|
# =============================================================================
|
|
|
|
class TestRoleSchema:
|
|
"""Test RoleSchema validation."""
|
|
|
|
def test_role_schema_creation(self):
|
|
"""RoleSchema should be creatable with valid data."""
|
|
schema = RoleSchema(
|
|
id=uuid.uuid4(),
|
|
name="control-room.general:admin",
|
|
domain="control-room",
|
|
category="general",
|
|
action="admin",
|
|
)
|
|
|
|
assert schema.name == "control-room.general:admin"
|
|
assert schema.domain == "control-room"
|
|
assert schema.category == "general"
|
|
assert schema.action == "admin"
|
|
|
|
def test_role_schema_category_default(self):
|
|
"""RoleSchema should default category to 'general'."""
|
|
schema = RoleSchema(
|
|
id=uuid.uuid4(),
|
|
name="media.general:viewer",
|
|
domain="media",
|
|
action="viewer",
|
|
)
|
|
|
|
assert schema.category == "general"
|