sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement super admin and platform admin roles

Browse Source 
Add multi-platform admin authorization system with:
- AdminPlatform junction table for admin-platform assignments
- is_super_admin flag on User model for global admin access
- Platform selection flow for platform admins after login
- JWT token updates to include platform context
- New API endpoints for admin user management (super admin only)
- Auth dependencies for super admin and platform access checks

Includes comprehensive test coverage:
- Unit tests for AdminPlatform model and User admin methods
- Unit tests for AdminPlatformService operations
- Integration tests for admin users API endpoints

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-24 18:44:49 +01:00
parent 7e39bb0564
commit 53e05dd497
18 changed files with 2792 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
tests/conftest.py
View File

@@ -158,6 +158,7 @@ def cleanup():
# Import fixtures from fixture modules
pytest_plugins = [
"tests.fixtures.admin_platform_fixtures",
"tests.fixtures.auth_fixtures",
"tests.fixtures.marketplace_product_fixtures",
"tests.fixtures.vendor_fixtures",

150
tests/fixtures/admin_platform_fixtures.py vendored Normal file
View File

@@ -0,0 +1,150 @@
# tests/fixtures/admin_platform_fixtures.py
"""
Admin platform assignment test fixtures.
Provides fixtures for testing super admin and platform admin functionality.
"""
import uuid
import pytest
from models.database.admin_platform import AdminPlatform
from models.database.platform import Platform
from models.database.user import User
@pytest.fixture
def test_platform(db):
"""Create a test platform."""
unique_id = str(uuid.uuid4())[:8]
platform = Platform(
code=f"test_{unique_id}",
name=f"Test Platform {unique_id}",
description="A test platform for unit tests",
path_prefix=f"test{unique_id}",
is_active=True,
is_public=True,
default_language="en",
supported_languages=["en", "fr", "de"],
)
db.add(platform)
db.commit()
db.refresh(platform)
return platform
@pytest.fixture
def another_platform(db):
"""Create another test platform."""
unique_id = str(uuid.uuid4())[:8]
platform = Platform(
code=f"another_{unique_id}",
name=f"Another Platform {unique_id}",
description="Another test platform",
path_prefix=f"another{unique_id}",
is_active=True,
is_public=True,
default_language="fr",
supported_languages=["fr", "en"],
)
db.add(platform)
db.commit()
db.refresh(platform)
return platform
@pytest.fixture
def test_admin_platform_assignment(db, test_platform_admin, test_platform, test_super_admin):
"""Create an admin platform assignment."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
return assignment
@pytest.fixture
def platform_admin_with_platform(db, auth_manager, test_platform, test_super_admin):
"""Create a platform admin with an assigned platform."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("platformadminpass123")
# Create platform admin
admin = User(
email=f"padmin_{unique_id}@example.com",
username=f"padmin_{unique_id}",
hashed_password=hashed_password,
role="admin",
is_active=True,
is_super_admin=False,
)
db.add(admin)
db.flush()
# Assign to platform
assignment = AdminPlatform(
user_id=admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(admin)
return admin
@pytest.fixture
def platform_admin_with_platform_headers(client, platform_admin_with_platform, test_platform):
"""Get authentication headers for platform admin with platform context."""
# First login
response = client.post(
"/api/v1/admin/auth/login",
json={
"email_or_username": platform_admin_with_platform.username,
"password": "platformadminpass123",
},
)
assert response.status_code == 200, f"Platform admin login failed: {response.text}"
token = response.json()["access_token"]
# Select platform
response = client.post(
"/api/v1/admin/auth/select-platform",
params={"platform_id": test_platform.id},
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200, f"Platform selection failed: {response.text}"
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def platform_factory(db):
"""Factory fixture for creating platforms."""
def _create_platform(**kwargs):
unique_id = str(uuid.uuid4())[:8]
defaults = {
"code": f"factory_{unique_id}",
"name": f"Factory Platform {unique_id}",
"path_prefix": f"factory{unique_id}",
"is_active": True,
"is_public": True,
"default_language": "en",
"supported_languages": ["en"],
}
defaults.update(kwargs)
platform = Platform(**defaults)
db.add(platform)
db.commit()
db.refresh(platform)
return platform
return _create_platform

66
tests/fixtures/auth_fixtures.py vendored
View File

@@ -40,7 +40,7 @@ def test_user(db, auth_manager):
@pytest.fixture
def test_admin(db, auth_manager):
"""Create a test admin user with unique username."""
"""Create a test admin user with unique username (super admin by default)."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("adminpass123")
admin = User(
@@ -49,6 +49,7 @@ def test_admin(db, auth_manager):
hashed_password=hashed_password,
role="admin",
is_active=True,
is_super_admin=True, # Default to super admin for backward compatibility
)
db.add(admin)
db.commit()
@@ -56,6 +57,68 @@ def test_admin(db, auth_manager):
return admin
@pytest.fixture
def test_super_admin(db, auth_manager):
"""Create a test super admin user with unique username."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("superadminpass123")
admin = User(
email=f"superadmin_{unique_id}@example.com",
username=f"superadmin_{unique_id}",
hashed_password=hashed_password,
role="admin",
is_active=True,
is_super_admin=True,
)
db.add(admin)
db.commit()
db.refresh(admin)
return admin
@pytest.fixture
def test_platform_admin(db, auth_manager):
"""Create a test platform admin user (not super admin)."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("platformadminpass123")
admin = User(
email=f"platformadmin_{unique_id}@example.com",
username=f"platformadmin_{unique_id}",
hashed_password=hashed_password,
role="admin",
is_active=True,
is_super_admin=False, # Platform admin, not super admin
)
db.add(admin)
db.commit()
db.refresh(admin)
return admin
@pytest.fixture
def super_admin_headers(client, test_super_admin):
"""Get authentication headers for super admin user."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_super_admin.username, "password": "superadminpass123"},
)
assert response.status_code == 200, f"Super admin login failed: {response.text}"
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def platform_admin_headers(client, test_platform_admin):
"""Get authentication headers for platform admin user (no platform context yet)."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_platform_admin.username, "password": "platformadminpass123"},
)
assert response.status_code == 200, f"Platform admin login failed: {response.text}"
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def another_admin(db, auth_manager):
"""Create another test admin user for testing admin-to-admin interactions."""
@@ -67,6 +130,7 @@ def another_admin(db, auth_manager):
hashed_password=hashed_password,
role="admin",
is_active=True,
is_super_admin=True, # Super admin for backward compatibility
)
db.add(admin)
db.commit()

381
tests/integration/api/v1/admin/test_admin_users.py Normal file
View File

@@ -0,0 +1,381 @@
# tests/integration/api/v1/admin/test_admin_users.py
"""
Integration tests for admin user management API endpoints.
Tests the /api/v1/admin/admin-users/* endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminUsersListAPI:
"""Test listing admin users."""
def test_list_admin_users_as_super_admin(
self, client, super_admin_headers, test_platform_admin
):
"""Test listing admin users as super admin."""
response = client.get(
"/api/v1/admin/admin-users",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "admins" in data
assert "total" in data
assert data["total"] >= 1
def test_list_admin_users_as_platform_admin_forbidden(
self, client, platform_admin_headers
):
"""Test that platform admin cannot list admin users."""
response = client.get(
"/api/v1/admin/admin-users",
headers=platform_admin_headers,
)
assert response.status_code == 403 # Super admin required
def test_list_admin_users_excludes_super_admins(
self, client, super_admin_headers, test_platform_admin
):
"""Test listing admin users excluding super admins."""
response = client.get(
"/api/v1/admin/admin-users",
params={"include_super_admins": False},
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
# All returned admins should not be super admins
for admin in data["admins"]:
assert admin["is_super_admin"] is False
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminUsersCreateAPI:
"""Test creating platform admins."""
def test_create_platform_admin_success(
self, client, super_admin_headers, test_platform, another_platform
):
"""Test creating a new platform admin."""
response = client.post(
"/api/v1/admin/admin-users",
json={
"email": "new_platform_admin@example.com",
"username": "new_platform_admin",
"password": "securepass123",
"first_name": "New",
"last_name": "Admin",
"platform_ids": [test_platform.id, another_platform.id],
},
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "new_platform_admin@example.com"
assert data["username"] == "new_platform_admin"
assert data["is_super_admin"] is False
assert len(data["platform_assignments"]) == 2
def test_create_platform_admin_duplicate_email(
self, client, super_admin_headers, test_platform, test_platform_admin
):
"""Test creating platform admin with duplicate email fails."""
response = client.post(
"/api/v1/admin/admin-users",
json={
"email": test_platform_admin.email,
"username": "unique_username",
"password": "securepass123",
"platform_ids": [test_platform.id],
},
headers=super_admin_headers,
)
assert response.status_code == 422 # Validation error
def test_create_platform_admin_as_platform_admin_forbidden(
self, client, platform_admin_headers, test_platform
):
"""Test that platform admin cannot create other admins."""
response = client.post(
"/api/v1/admin/admin-users",
json={
"email": "another@example.com",
"username": "another",
"password": "securepass123",
"platform_ids": [test_platform.id],
},
headers=platform_admin_headers,
)
assert response.status_code == 403 # Forbidden for non-super admin
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminUsersPlatformAssignmentAPI:
"""Test admin platform assignment endpoints."""
def test_assign_admin_to_platform(
self, client, super_admin_headers, test_platform_admin, test_platform
):
"""Test assigning an admin to a platform."""
response = client.post(
f"/api/v1/admin/admin-users/{test_platform_admin.id}/platforms/{test_platform.id}",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "message" in data
assert data["user_id"] == test_platform_admin.id
assert data["platform_id"] == test_platform.id
def test_remove_admin_from_platform(
self, client, super_admin_headers, db, test_platform_admin, test_platform, test_super_admin
):
"""Test removing an admin from a platform."""
from models.database.admin_platform import AdminPlatform
# First create an assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
response = client.delete(
f"/api/v1/admin/admin-users/{test_platform_admin.id}/platforms/{test_platform.id}",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "message" in data
def test_get_admin_platforms(
self, client, super_admin_headers, db, test_platform_admin, test_platform, test_super_admin
):
"""Test getting platforms for an admin."""
from models.database.admin_platform import AdminPlatform
# Create assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
response = client.get(
f"/api/v1/admin/admin-users/{test_platform_admin.id}/platforms",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "platforms" in data
assert len(data["platforms"]) == 1
assert data["platforms"][0]["id"] == test_platform.id
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminUsersSuperAdminToggleAPI:
"""Test super admin promotion/demotion."""
def test_promote_to_super_admin(
self, client, super_admin_headers, test_platform_admin
):
"""Test promoting a platform admin to super admin."""
response = client.put(
f"/api/v1/admin/admin-users/{test_platform_admin.id}/super-admin",
json={"is_super_admin": True},
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["is_super_admin"] is True
def test_demote_from_super_admin(
self, client, super_admin_headers, db, auth_manager
):
"""Test demoting a super admin to platform admin."""
from models.database.user import User
# Create another super admin to demote
another_super = User(
email="demote_test@example.com",
username="demote_test",
hashed_password=auth_manager.hash_password("pass"),
role="admin",
is_active=True,
is_super_admin=True,
)
db.add(another_super)
db.commit()
response = client.put(
f"/api/v1/admin/admin-users/{another_super.id}/super-admin",
json={"is_super_admin": False},
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["is_super_admin"] is False
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminAuthPlatformSelectionAPI:
"""Test platform selection for platform admins."""
def test_get_accessible_platforms_super_admin(
self, client, super_admin_headers, test_platform
):
"""Test getting accessible platforms as super admin."""
response = client.get(
"/api/v1/admin/auth/accessible-platforms",
headers=super_admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["is_super_admin"] is True
assert "platforms" in data
assert data["requires_platform_selection"] is False
def test_get_accessible_platforms_platform_admin(
self, client, db, test_platform_admin, test_platform, test_super_admin, auth_manager
):
"""Test getting accessible platforms as platform admin."""
from models.database.admin_platform import AdminPlatform
# Create assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
# Login as platform admin
response = client.post(
"/api/v1/admin/auth/login",
json={
"email_or_username": test_platform_admin.username,
"password": "platformadminpass123",
},
)
assert response.status_code == 200
token = response.json()["access_token"]
# Get accessible platforms
response = client.get(
"/api/v1/admin/auth/accessible-platforms",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
data = response.json()
assert data["is_super_admin"] is False
assert len(data["platforms"]) == 1
assert data["platforms"][0]["id"] == test_platform.id
assert data["requires_platform_selection"] is True
def test_select_platform_success(
self, client, db, test_platform_admin, test_platform, test_super_admin
):
"""Test selecting a platform as platform admin."""
from models.database.admin_platform import AdminPlatform
# Create assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
# Login
response = client.post(
"/api/v1/admin/auth/login",
json={
"email_or_username": test_platform_admin.username,
"password": "platformadminpass123",
},
)
token = response.json()["access_token"]
# Select platform
response = client.post(
"/api/v1/admin/auth/select-platform",
params={"platform_id": test_platform.id},
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
# New token should be different
assert data["access_token"] != token
def test_select_platform_not_assigned(
self, client, db, test_platform_admin, test_platform
):
"""Test selecting a platform that admin is not assigned to."""
# Login (no platform assignment exists)
response = client.post(
"/api/v1/admin/auth/login",
json={
"email_or_username": test_platform_admin.username,
"password": "platformadminpass123",
},
)
token = response.json()["access_token"]
# Try to select platform
response = client.post(
"/api/v1/admin/auth/select-platform",
params={"platform_id": test_platform.id},
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 403 # Access denied
def test_select_platform_as_super_admin_fails(
self, client, super_admin_headers, test_platform
):
"""Test that super admin cannot select platform (they don't need to)."""
response = client.post(
"/api/v1/admin/auth/select-platform",
params={"platform_id": test_platform.id},
headers=super_admin_headers,
)
assert response.status_code == 401 # Super admins don't need this

271
tests/unit/models/database/test_admin_platform.py Normal file
View File

@@ -0,0 +1,271 @@
# tests/unit/models/database/test_admin_platform.py
"""
Unit tests for AdminPlatform model.
Tests the admin-platform junction table model and its relationships.
"""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.admin_platform import AdminPlatform
@pytest.mark.unit
@pytest.mark.database
@pytest.mark.admin
class TestAdminPlatformModel:
"""Test AdminPlatform model creation and constraints."""
def test_create_admin_platform_assignment(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test creating an admin platform assignment."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
assert assignment.id is not None
assert assignment.user_id == test_platform_admin.id
assert assignment.platform_id == test_platform.id
assert assignment.is_active is True
assert assignment.assigned_by_user_id == test_super_admin.id
assert assignment.assigned_at is not None
def test_admin_platform_unique_constraint(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test that an admin can only be assigned to a platform once."""
# Create first assignment
assignment1 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment1)
db.commit()
# Try to create duplicate assignment
assignment2 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment2)
with pytest.raises(IntegrityError):
db.commit()
def test_admin_platform_cascade_delete_user(
self, db, auth_manager, test_platform, test_super_admin
):
"""Test that deleting user cascades to admin platform assignments."""
from models.database.user import User
# Create a temporary admin
temp_admin = User(
email="temp_admin@example.com",
username="temp_admin",
hashed_password=auth_manager.hash_password("temppass"),
role="admin",
is_active=True,
is_super_admin=False,
)
db.add(temp_admin)
db.flush()
# Create assignment
assignment = AdminPlatform(
user_id=temp_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
assignment_id = assignment.id
# Delete user - should cascade to assignment
db.delete(temp_admin)
db.commit()
# Verify assignment is gone
remaining = db.query(AdminPlatform).filter(AdminPlatform.id == assignment_id).first()
assert remaining is None
def test_admin_platform_relationships(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test AdminPlatform relationships are loaded correctly."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
# Test relationships
assert assignment.user is not None
assert assignment.user.id == test_platform_admin.id
assert assignment.platform is not None
assert assignment.platform.id == test_platform.id
assert assignment.assigned_by is not None
assert assignment.assigned_by.id == test_super_admin.id
def test_admin_platform_properties(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test AdminPlatform computed properties."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
# Test properties
assert assignment.platform_code == test_platform.code
assert assignment.platform_name == test_platform.name
def test_admin_platform_repr(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test AdminPlatform string representation."""
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(assignment)
repr_str = repr(assignment)
assert "AdminPlatform" in repr_str
assert str(test_platform_admin.id) in repr_str
assert str(test_platform.id) in repr_str
@pytest.mark.unit
@pytest.mark.database
@pytest.mark.admin
class TestUserAdminMethods:
"""Test User model admin-related methods."""
def test_is_super_admin_user_true(self, db, test_super_admin):
"""Test is_super_admin_user property for super admin."""
assert test_super_admin.is_super_admin_user is True
def test_is_super_admin_user_false_for_platform_admin(self, db, test_platform_admin):
"""Test is_super_admin_user property for platform admin."""
assert test_platform_admin.is_super_admin_user is False
def test_is_platform_admin_true(self, db, test_platform_admin):
"""Test is_platform_admin property for platform admin."""
assert test_platform_admin.is_platform_admin is True
def test_is_platform_admin_false_for_super_admin(self, db, test_super_admin):
"""Test is_platform_admin property for super admin."""
assert test_super_admin.is_platform_admin is False
def test_can_access_platform_super_admin(self, db, test_super_admin, test_platform):
"""Test that super admin can access any platform."""
assert test_super_admin.can_access_platform(test_platform.id) is True
def test_can_access_platform_assigned(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test that platform admin can access assigned platform."""
# Create assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(test_platform_admin)
assert test_platform_admin.can_access_platform(test_platform.id) is True
def test_can_access_platform_not_assigned(
self, db, test_platform_admin, test_platform
):
"""Test that platform admin cannot access unassigned platform."""
# No assignment created
assert test_platform_admin.can_access_platform(test_platform.id) is False
def test_can_access_platform_inactive_assignment(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test that platform admin cannot access platform with inactive assignment."""
# Create inactive assignment
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=False, # Inactive
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
db.refresh(test_platform_admin)
assert test_platform_admin.can_access_platform(test_platform.id) is False
def test_get_accessible_platform_ids_super_admin(self, db, test_super_admin):
"""Test get_accessible_platform_ids returns None for super admin."""
result = test_super_admin.get_accessible_platform_ids()
assert result is None # None means all platforms
def test_get_accessible_platform_ids_platform_admin(
self, db, test_platform_admin, test_platform, another_platform, test_super_admin
):
"""Test get_accessible_platform_ids returns correct list for platform admin."""
# Create assignments for both platforms
assignment1 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
assignment2 = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=another_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add_all([assignment1, assignment2])
db.commit()
db.refresh(test_platform_admin)
result = test_platform_admin.get_accessible_platform_ids()
assert len(result) == 2
assert test_platform.id in result
assert another_platform.id in result
def test_get_accessible_platform_ids_no_assignments(self, db, test_platform_admin):
"""Test get_accessible_platform_ids returns empty list when no assignments."""
result = test_platform_admin.get_accessible_platform_ids()
assert result == []
def test_get_accessible_platform_ids_vendor_user(self, db, test_vendor_user):
"""Test get_accessible_platform_ids returns empty list for non-admin."""
result = test_vendor_user.get_accessible_platform_ids()
assert result == []

463
tests/unit/services/test_admin_platform_service.py Normal file
View File

@@ -0,0 +1,463 @@
# tests/unit/services/test_admin_platform_service.py
"""
Unit tests for AdminPlatformService.
Tests the admin platform assignment service operations.
"""
import pytest
from app.exceptions import AdminOperationException, CannotModifySelfException, ValidationException
from app.services.admin_platform_service import AdminPlatformService
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceAssign:
"""Test AdminPlatformService.assign_admin_to_platform."""
def test_assign_admin_to_platform_success(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test successfully assigning an admin to a platform."""
service = AdminPlatformService()
assignment = service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert assignment is not None
assert assignment.user_id == test_platform_admin.id
assert assignment.platform_id == test_platform.id
assert assignment.is_active is True
assert assignment.assigned_by_user_id == test_super_admin.id
def test_assign_admin_user_not_found(self, db, test_platform, test_super_admin):
"""Test assigning non-existent user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=99999,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "User not found" in str(exc.value)
def test_assign_admin_not_admin_role(
self, db, test_vendor_user, test_platform, test_super_admin
):
"""Test assigning non-admin user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_vendor_user.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "must be an admin" in str(exc.value)
def test_assign_super_admin_raises_error(
self, db, test_super_admin, test_platform
):
"""Test assigning super admin raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_super_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "Super admins don't need platform assignments" in str(exc.value)
def test_assign_platform_not_found(
self, db, test_platform_admin, test_super_admin
):
"""Test assigning to non-existent platform raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=99999,
assigned_by_user_id=test_super_admin.id,
)
assert "Platform not found" in str(exc.value)
def test_assign_admin_already_assigned(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test assigning already assigned admin raises error."""
service = AdminPlatformService()
# First assignment
service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
db.commit()
# Try to assign again
with pytest.raises(AdminOperationException) as exc:
service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert "already assigned" in str(exc.value)
def test_reactivate_inactive_assignment(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test reactivating an inactive assignment."""
from models.database.admin_platform import AdminPlatform
service = AdminPlatformService()
# Create inactive assignment directly
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=False,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
# Assign again - should reactivate
result = service.assign_admin_to_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
assigned_by_user_id=test_super_admin.id,
)
assert result.is_active is True
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceRemove:
"""Test AdminPlatformService.remove_admin_from_platform."""
def test_remove_admin_from_platform_success(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test successfully removing an admin from a platform."""
from models.database.admin_platform import AdminPlatform
service = AdminPlatformService()
# Create assignment first
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
# Remove
service.remove_admin_from_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
removed_by_user_id=test_super_admin.id,
)
db.commit()
db.refresh(assignment)
assert assignment.is_active is False
def test_remove_admin_not_assigned(
self, db, test_platform_admin, test_platform, test_super_admin
):
"""Test removing non-existent assignment raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.remove_admin_from_platform(
db=db,
admin_user_id=test_platform_admin.id,
platform_id=test_platform.id,
removed_by_user_id=test_super_admin.id,
)
assert "not assigned" in str(exc.value)
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceQueries:
"""Test AdminPlatformService query methods."""
def test_get_platforms_for_admin(
self, db, test_platform_admin, test_platform, another_platform, test_super_admin
):
"""Test getting platforms for an admin."""
from models.database.admin_platform import AdminPlatform
service = AdminPlatformService()
# Create assignments
for platform in [test_platform, another_platform]:
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
platforms = service.get_platforms_for_admin(db, test_platform_admin.id)
assert len(platforms) == 2
platform_ids = [p.id for p in platforms]
assert test_platform.id in platform_ids
assert another_platform.id in platform_ids
def test_get_platforms_for_admin_no_assignments(self, db, test_platform_admin):
"""Test getting platforms when no assignments exist."""
service = AdminPlatformService()
platforms = service.get_platforms_for_admin(db, test_platform_admin.id)
assert platforms == []
def test_get_admins_for_platform(
self, db, test_platform_admin, test_platform, test_super_admin, auth_manager
):
"""Test getting admins for a platform."""
from models.database.admin_platform import AdminPlatform
from models.database.user import User
service = AdminPlatformService()
# Create another platform admin
another_admin = User(
email="another_padmin@example.com",
username="another_padmin",
hashed_password=auth_manager.hash_password("pass"),
role="admin",
is_active=True,
is_super_admin=False,
)
db.add(another_admin)
db.flush()
# Create assignments for both admins
for admin in [test_platform_admin, another_admin]:
assignment = AdminPlatform(
user_id=admin.id,
platform_id=test_platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
admins = service.get_admins_for_platform(db, test_platform.id)
assert len(admins) == 2
admin_ids = [a.id for a in admins]
assert test_platform_admin.id in admin_ids
assert another_admin.id in admin_ids
def test_get_admin_assignments(
self, db, test_platform_admin, test_platform, another_platform, test_super_admin
):
"""Test getting admin assignments with platform details."""
from models.database.admin_platform import AdminPlatform
service = AdminPlatformService()
# Create assignments
for platform in [test_platform, another_platform]:
assignment = AdminPlatform(
user_id=test_platform_admin.id,
platform_id=platform.id,
is_active=True,
assigned_by_user_id=test_super_admin.id,
)
db.add(assignment)
db.commit()
assignments = service.get_admin_assignments(db, test_platform_admin.id)
assert len(assignments) == 2
# Verify platform relationship is loaded
for assignment in assignments:
assert assignment.platform is not None
assert assignment.platform.code is not None
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceSuperAdmin:
"""Test AdminPlatformService super admin operations."""
def test_toggle_super_admin_promote(
self, db, test_platform_admin, test_super_admin
):
"""Test promoting admin to super admin."""
service = AdminPlatformService()
result = service.toggle_super_admin(
db=db,
user_id=test_platform_admin.id,
is_super_admin=True,
current_admin_id=test_super_admin.id,
)
db.commit()
assert result.is_super_admin is True
def test_toggle_super_admin_demote(
self, db, test_super_admin, auth_manager
):
"""Test demoting super admin to platform admin."""
from models.database.user import User
service = AdminPlatformService()
# Create another super admin to demote
another_super = User(
email="another_super@example.com",
username="another_super",
hashed_password=auth_manager.hash_password("pass"),
role="admin",
is_active=True,
is_super_admin=True,
)
db.add(another_super)
db.commit()
result = service.toggle_super_admin(
db=db,
user_id=another_super.id,
is_super_admin=False,
current_admin_id=test_super_admin.id,
)
db.commit()
assert result.is_super_admin is False
def test_toggle_super_admin_cannot_demote_self(self, db, test_super_admin):
"""Test that super admin cannot demote themselves."""
service = AdminPlatformService()
with pytest.raises(CannotModifySelfException):
service.toggle_super_admin(
db=db,
user_id=test_super_admin.id,
is_super_admin=False,
current_admin_id=test_super_admin.id,
)
def test_toggle_super_admin_user_not_found(self, db, test_super_admin):
"""Test toggling non-existent user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.toggle_super_admin(
db=db,
user_id=99999,
is_super_admin=True,
current_admin_id=test_super_admin.id,
)
assert "User not found" in str(exc.value)
def test_toggle_super_admin_not_admin(
self, db, test_vendor_user, test_super_admin
):
"""Test toggling non-admin user raises error."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.toggle_super_admin(
db=db,
user_id=test_vendor_user.id,
is_super_admin=True,
current_admin_id=test_super_admin.id,
)
assert "must be an admin" in str(exc.value)
@pytest.mark.unit
@pytest.mark.admin
class TestAdminPlatformServiceCreatePlatformAdmin:
"""Test AdminPlatformService.create_platform_admin."""
def test_create_platform_admin_success(
self, db, test_platform, another_platform, test_super_admin
):
"""Test creating a new platform admin with assignments."""
service = AdminPlatformService()
user, assignments = service.create_platform_admin(
db=db,
email="new_padmin@example.com",
username="new_padmin",
password="securepass123",
platform_ids=[test_platform.id, another_platform.id],
created_by_user_id=test_super_admin.id,
first_name="New",
last_name="Admin",
)
db.commit()
assert user is not None
assert user.email == "new_padmin@example.com"
assert user.username == "new_padmin"
assert user.role == "admin"
assert user.is_super_admin is False
assert user.first_name == "New"
assert user.last_name == "Admin"
assert len(assignments) == 2
def test_create_platform_admin_duplicate_email(
self, db, test_platform, test_super_admin, test_platform_admin
):
"""Test creating platform admin with duplicate email fails."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.create_platform_admin(
db=db,
email=test_platform_admin.email, # Duplicate
username="unique_username",
password="securepass123",
platform_ids=[test_platform.id],
created_by_user_id=test_super_admin.id,
)
assert "Email already exists" in str(exc.value)
def test_create_platform_admin_duplicate_username(
self, db, test_platform, test_super_admin, test_platform_admin
):
"""Test creating platform admin with duplicate username fails."""
service = AdminPlatformService()
with pytest.raises(ValidationException) as exc:
service.create_platform_admin(
db=db,
email="unique@example.com",
username=test_platform_admin.username, # Duplicate
password="securepass123",
platform_ids=[test_platform.id],
created_by_user_id=test_super_admin.id,
)
assert "Username already exists" in str(exc.value)