admin felület fejlesztése garázs, előfizeztési csomagok
This commit is contained in:
Roo
186
backend/app/tests/test_rbac_service.py
Normal file
186
backend/app/tests/test_rbac_service.py
Normal file
@@ -0,0 +1,186 @@
|
||||
"""
|
||||
Test script for Scope-Based Admin RBAC Service.
|
||||
Verifies that the RBACService correctly handles scope-based access control.
|
||||
"""
|
||||
import asyncio
|
||||
import sys
|
||||
from sqlalchemy import select, text
|
||||
from app.database import AsyncSessionLocal
|
||||
from app.models.identity.identity import User, UserRole
|
||||
from app.models.marketplace.organization import Organization
|
||||
from app.services.rbac_service import rbac_service, AdminAction
|
||||
|
||||
|
||||
async def test_rbac_service():
|
||||
print("=" * 60)
|
||||
print("RBAC SERVICE TEST SUITE")
|
||||
print("=" * 60)
|
||||
|
||||
async with AsyncSessionLocal() as db:
|
||||
# 1. Check that scope_type and scope_value columns exist on User
|
||||
print("\n[TEST 1] Checking User model has scope_type/scope_value columns...")
|
||||
result = await db.execute(
|
||||
text("SELECT column_name FROM information_schema.columns WHERE table_schema = 'identity' AND table_name = 'users' AND column_name IN ('scope_type', 'scope_value')")
|
||||
)
|
||||
cols = result.scalars().all()
|
||||
assert "scope_type" in cols, "scope_type column missing!"
|
||||
assert "scope_value" in cols, "scope_value column missing!"
|
||||
print(f" ✅ Columns found: {cols}")
|
||||
|
||||
# 2. Check that region and segment columns exist on Organization
|
||||
print("\n[TEST 2] Checking Organization model has region/segment columns...")
|
||||
result = await db.execute(
|
||||
text("SELECT column_name FROM information_schema.columns WHERE table_schema = 'fleet' AND table_name = 'organizations' AND column_name IN ('region', 'segment')")
|
||||
)
|
||||
cols = result.scalars().all()
|
||||
assert "region" in cols, "region column missing!"
|
||||
assert "segment" in cols, "segment column missing!"
|
||||
print(f" ✅ Columns found: {cols}")
|
||||
|
||||
# 3. Test RBACService.get_permitted_actions for each role
|
||||
print("\n[TEST 3] Testing get_permitted_actions for each role...")
|
||||
for role in UserRole:
|
||||
actions = rbac_service.get_permitted_actions(role)
|
||||
print(f" {role.value}: {len(actions)} permitted actions")
|
||||
assert isinstance(actions, (set, list)), f"Actions for {role} should be a set or list"
|
||||
print(" ✅ All roles return valid action lists")
|
||||
|
||||
# 4. Test is_action_permitted
|
||||
print("\n[TEST 4] Testing is_action_permitted...")
|
||||
assert rbac_service.is_action_permitted(UserRole.SUPERADMIN, AdminAction.EDIT_DATA) is True
|
||||
assert rbac_service.is_action_permitted(UserRole.ADMIN, AdminAction.EDIT_DATA) is True
|
||||
assert rbac_service.is_action_permitted(UserRole.MODERATOR, AdminAction.EDIT_DATA) is False
|
||||
assert rbac_service.is_action_permitted(UserRole.SALES_REP, AdminAction.VIEW_DATA) is True
|
||||
assert rbac_service.is_action_permitted(UserRole.SERVICE_MGR, AdminAction.VIEW_DATA) is True
|
||||
print(" ✅ All action permission checks pass")
|
||||
|
||||
# 5. Test SUPERADMIN bypasses all scope checks
|
||||
print("\n[TEST 5] Testing SUPERADMIN scope bypass...")
|
||||
# Create a mock user object
|
||||
class MockUser:
|
||||
role = UserRole.SUPERADMIN
|
||||
scope_type = None
|
||||
scope_value = None
|
||||
id = 999999
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUser(), AdminAction.VIEW_DATA, target_org_id=1)
|
||||
print(" ✅ SUPERADMIN bypasses scope check - no PermissionError raised")
|
||||
except PermissionError as e:
|
||||
print(f" ❌ SUPERADMIN should bypass scope check but got: {e}")
|
||||
|
||||
# 6. Test that a regular user without scope gets PermissionError
|
||||
print("\n[TEST 6] Testing regular user without scope...")
|
||||
class MockUserNoScope:
|
||||
role = UserRole.ADMIN
|
||||
scope_type = None
|
||||
scope_value = None
|
||||
id = 999998
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUserNoScope(), AdminAction.VIEW_DATA, target_org_id=1)
|
||||
print(" ❌ Should have raised PermissionError for user without scope")
|
||||
except PermissionError as e:
|
||||
print(f" ✅ Correctly raised PermissionError: {e}")
|
||||
|
||||
# 7. Test GLOBAL scope
|
||||
print("\n[TEST 7] Testing GLOBAL scope...")
|
||||
class MockUserGlobal:
|
||||
role = UserRole.ADMIN
|
||||
scope_type = "GLOBAL"
|
||||
scope_value = None
|
||||
id = 999997
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUserGlobal(), AdminAction.VIEW_DATA, target_org_id=1)
|
||||
print(" ✅ GLOBAL scope allows access to any org")
|
||||
except PermissionError as e:
|
||||
print(f" ❌ GLOBAL scope should allow access but got: {e}")
|
||||
|
||||
# 8. Test get_accessible_org_ids for GLOBAL scope
|
||||
print("\n[TEST 8] Testing get_accessible_org_ids for GLOBAL scope...")
|
||||
org_ids = await rbac_service.get_accessible_org_ids(db, MockUserGlobal())
|
||||
assert isinstance(org_ids, list), "Should return a list"
|
||||
print(f" ✅ GLOBAL scope returns {len(org_ids)} accessible orgs")
|
||||
|
||||
# 9. Test REGION scope matching
|
||||
print("\n[TEST 9] Testing REGION scope matching...")
|
||||
# First, set an org's region to a known value for testing
|
||||
await db.execute(
|
||||
text("UPDATE fleet.organizations SET region = 'Test_Region' WHERE id = 1")
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
class MockUserRegion:
|
||||
role = UserRole.ADMIN
|
||||
scope_type = "REGION"
|
||||
scope_value = "Test_Region"
|
||||
id = 999996
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUserRegion(), AdminAction.VIEW_DATA, target_org_id=1)
|
||||
print(" ✅ REGION scope matches org with same region")
|
||||
except PermissionError as e:
|
||||
print(f" ❌ REGION scope should match but got: {e}")
|
||||
|
||||
# Test non-matching region
|
||||
class MockUserRegionWrong:
|
||||
role = UserRole.ADMIN
|
||||
scope_type = "REGION"
|
||||
scope_value = "Other_Region"
|
||||
id = 999995
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUserRegionWrong(), AdminAction.VIEW_DATA, target_org_id=1)
|
||||
print(" ❌ Should have raised PermissionError for wrong region")
|
||||
except PermissionError as e:
|
||||
print(f" ✅ Correctly denied access for wrong region: {e}")
|
||||
|
||||
# 10. Test SEGMENT scope matching
|
||||
print("\n[TEST 10] Testing SEGMENT scope matching...")
|
||||
await db.execute(
|
||||
text("UPDATE fleet.organizations SET segment = 'Test_Segment' WHERE id = 1")
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
class MockUserSegment:
|
||||
role = UserRole.ADMIN
|
||||
scope_type = "SEGMENT"
|
||||
scope_value = "Test_Segment"
|
||||
id = 999994
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUserSegment(), AdminAction.VIEW_DATA, target_org_id=1)
|
||||
print(" ✅ SEGMENT scope matches org with same segment")
|
||||
except PermissionError as e:
|
||||
print(f" ❌ SEGMENT scope should match but got: {e}")
|
||||
|
||||
# 11. Test action not permitted for role
|
||||
print("\n[TEST 11] Testing action not permitted for role...")
|
||||
class MockUserModerator:
|
||||
role = UserRole.MODERATOR
|
||||
scope_type = "GLOBAL"
|
||||
scope_value = None
|
||||
id = 999993
|
||||
|
||||
try:
|
||||
await rbac_service.check_admin_access(db, MockUserModerator(), AdminAction.DELETE_DATA, target_org_id=1)
|
||||
print(" ❌ Should have raised PermissionError - MODERATOR cannot DELETE_DATA")
|
||||
except PermissionError as e:
|
||||
print(f" ✅ Correctly denied DELETE_DATA for MODERATOR: {e}")
|
||||
|
||||
# Cleanup test data
|
||||
await db.execute(
|
||||
text("UPDATE fleet.organizations SET region = NULL, segment = NULL WHERE id = 1")
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("ALL TESTS PASSED ✅")
|
||||
print("=" * 60)
|
||||
return True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = asyncio.run(test_rbac_service())
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user