""" Data validation and integrity checks for permission migration """ from django.db import models, connection from django.core.exceptions import ValidationError from users.models import CustomUser, RolePermission, DefaultRolePermission, UserPermission from users.enhanced_permissions_models import PagePermission, RolePermissionTemplate, UserPagePermission from typing import Dict, List, Tuple, Any import logging logger = logging.getLogger(__name__) class PermissionDataValidator: """ Validates data integrity for permission systems """ def __init__(self): self.issues = [] self.warnings = [] self.stats = {} def validate_legacy_permissions(self) -> Dict[str, Any]: """ Validate legacy permission system data integrity Returns: Dict containing validation results """ self.issues = [] self.warnings = [] self.stats = {} # Check for orphaned permissions self._check_legacy_orphaned_permissions() # Check for duplicate permissions self._check_legacy_duplicate_permissions() # Check for invalid role/module/action combinations self._check_legacy_invalid_combinations() # Check for expired permissions self._check_legacy_expired_permissions() # Check for permission consistency self._check_legacy_permission_consistency() # Collect statistics self._collect_legacy_statistics() return { 'system': 'legacy', 'issues': self.issues, 'warnings': self.warnings, 'stats': self.stats, 'is_valid': len(self.issues) == 0 } def validate_granular_permissions(self) -> Dict[str, Any]: """ Validate granular permission system data integrity Returns: Dict containing validation results """ self.issues = [] self.warnings = [] self.stats = {} # Check for orphaned permissions self._check_granular_orphaned_permissions() # Check for duplicate permissions self._check_granular_duplicate_permissions() # Check for invalid permission dependencies self._check_granular_permission_dependencies() # Check for expired permissions self._check_granular_expired_permissions() # Check for permission template consistency self._check_granular_template_consistency() # Check for critical permission assignments self._check_granular_critical_permissions() # Collect statistics self._collect_granular_statistics() return { 'system': 'granular', 'issues': self.issues, 'warnings': self.warnings, 'stats': self.stats, 'is_valid': len(self.issues) == 0 } def validate_migration_integrity(self) -> Dict[str, Any]: """ Validate that migration from legacy to granular was successful Returns: Dict containing validation results """ self.issues = [] self.warnings = [] self.stats = {} # Check that all legacy permissions have granular equivalents self._check_migration_completeness() # Check that permission mappings are correct self._check_migration_mappings() # Check that user permissions were migrated correctly self._check_migration_user_permissions() # Check that role templates were created correctly self._check_migration_role_templates() # Collect migration statistics self._collect_migration_statistics() return { 'system': 'migration', 'issues': self.issues, 'warnings': self.warnings, 'stats': self.stats, 'is_valid': len(self.issues) == 0 } def _check_legacy_orphaned_permissions(self): """Check for orphaned permissions in legacy system""" # Check for user permissions with non-existent users orphaned_user_perms = UserPermission.objects.filter(user__isnull=True).count() if orphaned_user_perms > 0: self.issues.append(f"Found {orphaned_user_perms} user permissions with null users") # Check for user permissions with non-existent granted_by users invalid_granted_by = UserPermission.objects.exclude( granted_by__isnull=True ).exclude( granted_by__in=CustomUser.objects.all() ).count() if invalid_granted_by > 0: self.warnings.append(f"Found {invalid_granted_by} user permissions with invalid granted_by references") def _check_legacy_duplicate_permissions(self): """Check for duplicate permissions in legacy system""" # Check for duplicate role permissions role_duplicates = RolePermission.objects.values( 'role', 'module', 'action' ).annotate( count=models.Count('id') ).filter(count__gt=1) if role_duplicates.exists(): self.issues.append(f"Found {role_duplicates.count()} duplicate role permissions") # Check for duplicate default role permissions default_duplicates = DefaultRolePermission.objects.values( 'role', 'module', 'action' ).annotate( count=models.Count('id') ).filter(count__gt=1) if default_duplicates.exists(): self.issues.append(f"Found {default_duplicates.count()} duplicate default role permissions") # Check for duplicate user permissions user_duplicates = UserPermission.objects.values( 'user', 'module', 'action' ).annotate( count=models.Count('id') ).filter(count__gt=1) if user_duplicates.exists(): self.issues.append(f"Found {user_duplicates.count()} duplicate user permissions") def _check_legacy_invalid_combinations(self): """Check for invalid role/module/action combinations""" # Get valid choices from model valid_roles = [choice[0] for choice in RolePermission.ROLE_CHOICES] valid_modules = [choice[0] for choice in RolePermission.MODULE_CHOICES] valid_actions = [choice[0] for choice in RolePermission.ACTION_CHOICES] # Check role permissions invalid_roles = RolePermission.objects.exclude(role__in=valid_roles).count() invalid_modules = RolePermission.objects.exclude(module__in=valid_modules).count() invalid_actions = RolePermission.objects.exclude(action__in=valid_actions).count() if invalid_roles > 0: self.issues.append(f"Found {invalid_roles} role permissions with invalid roles") if invalid_modules > 0: self.issues.append(f"Found {invalid_modules} role permissions with invalid modules") if invalid_actions > 0: self.issues.append(f"Found {invalid_actions} role permissions with invalid actions") def _check_legacy_expired_permissions(self): """Check for expired permissions that should be cleaned up""" from django.utils import timezone expired_count = UserPermission.objects.filter( expires_at__lt=timezone.now() ).count() if expired_count > 0: self.warnings.append(f"Found {expired_count} expired user permissions that could be cleaned up") def _check_legacy_permission_consistency(self): """Check for permission consistency issues""" # Check if users have permissions for roles they don't have inconsistent_perms = 0 for user in CustomUser.objects.all(): user_role_perms = RolePermission.objects.filter(role=user.role, is_allowed=True) user_custom_perms = UserPermission.objects.filter(user=user, is_allowed=False) # Count permissions that are allowed by role but denied by custom permission for custom_perm in user_custom_perms: if user_role_perms.filter( module=custom_perm.module, action=custom_perm.action ).exists(): inconsistent_perms += 1 if inconsistent_perms > 0: self.warnings.append(f"Found {inconsistent_perms} permissions that override role defaults") def _collect_legacy_statistics(self): """Collect statistics about legacy permission system""" self.stats = { 'role_permissions': RolePermission.objects.count(), 'default_role_permissions': DefaultRolePermission.objects.count(), 'user_permissions': UserPermission.objects.count(), 'active_users': CustomUser.objects.filter(is_active=True).count(), 'roles_with_permissions': RolePermission.objects.values('role').distinct().count(), 'modules_with_permissions': RolePermission.objects.values('module').distinct().count(), 'actions_with_permissions': RolePermission.objects.values('action').distinct().count(), } def _check_granular_orphaned_permissions(self): """Check for orphaned permissions in granular system""" # Check for role templates with non-existent page permissions orphaned_templates = RolePermissionTemplate.objects.filter( page_permission__isnull=True ).count() if orphaned_templates > 0: self.issues.append(f"Found {orphaned_templates} role templates with null page permissions") # Check for user permissions with non-existent page permissions orphaned_user_perms = UserPagePermission.objects.filter( page_permission__isnull=True ).count() if orphaned_user_perms > 0: self.issues.append(f"Found {orphaned_user_perms} user permissions with null page permissions") # Check for user permissions with non-existent users orphaned_users = UserPagePermission.objects.filter(user__isnull=True).count() if orphaned_users > 0: self.issues.append(f"Found {orphaned_users} user permissions with null users") def _check_granular_duplicate_permissions(self): """Check for duplicate permissions in granular system""" # Check for duplicate page permissions page_duplicates = PagePermission.objects.values( 'page_name', 'action_code' ).annotate( count=models.Count('id') ).filter(count__gt=1) if page_duplicates.exists(): self.issues.append(f"Found {page_duplicates.count()} duplicate page permissions") # Check for duplicate role templates template_duplicates = RolePermissionTemplate.objects.values( 'role', 'page_permission' ).annotate( count=models.Count('id') ).filter(count__gt=1) if template_duplicates.exists(): self.issues.append(f"Found {template_duplicates.count()} duplicate role templates") # Check for duplicate user permissions user_duplicates = UserPagePermission.objects.values( 'user', 'page_permission' ).annotate( count=models.Count('id') ).filter(count__gt=1) if user_duplicates.exists(): self.issues.append(f"Found {user_duplicates.count()} duplicate user page permissions") def _check_granular_permission_dependencies(self): """Check for invalid permission dependencies""" # Check for circular dependencies circular_deps = 0 for permission in PagePermission.objects.all(): if self._has_circular_dependency(permission): circular_deps += 1 if circular_deps > 0: self.issues.append(f"Found {circular_deps} permissions with circular dependencies") # Check for dependencies on non-existent permissions invalid_deps = 0 for permission in PagePermission.objects.all(): for required_perm in permission.required_permissions.all(): if not required_perm.is_active: invalid_deps += 1 if invalid_deps > 0: self.warnings.append(f"Found {invalid_deps} dependencies on inactive permissions") def _check_granular_expired_permissions(self): """Check for expired permissions in granular system""" from django.utils import timezone expired_count = UserPagePermission.objects.filter( expires_at__lt=timezone.now() ).count() if expired_count > 0: self.warnings.append(f"Found {expired_count} expired user page permissions") def _check_granular_template_consistency(self): """Check for role template consistency""" # Check if all roles have templates for critical permissions critical_permissions = PagePermission.objects.filter(is_critical=True) roles = [choice[0] for choice in RolePermissionTemplate.ROLE_CHOICES] missing_critical = 0 for role in roles: for critical_perm in critical_permissions: if not RolePermissionTemplate.objects.filter( role=role, page_permission=critical_perm ).exists(): missing_critical += 1 if missing_critical > 0: self.warnings.append(f"Found {missing_critical} missing critical permission templates") def _check_granular_critical_permissions(self): """Check critical permission assignments""" # Check if admin role has all critical permissions critical_permissions = PagePermission.objects.filter(is_critical=True) admin_templates = RolePermissionTemplate.objects.filter( role='admin', page_permission__in=critical_permissions, is_allowed=False ) if admin_templates.exists(): self.warnings.append(f"Admin role is denied {admin_templates.count()} critical permissions") def _collect_granular_statistics(self): """Collect statistics about granular permission system""" self.stats = { 'page_permissions': PagePermission.objects.count(), 'role_permission_templates': RolePermissionTemplate.objects.count(), 'user_page_permissions': UserPagePermission.objects.count(), 'active_page_permissions': PagePermission.objects.filter(is_active=True).count(), 'critical_permissions': PagePermission.objects.filter(is_critical=True).count(), 'pages_with_permissions': PagePermission.objects.values('page_name').distinct().count(), 'roles_with_templates': RolePermissionTemplate.objects.values('role').distinct().count(), 'users_with_custom_permissions': UserPagePermission.objects.values('user').distinct().count(), } def _check_migration_completeness(self): """Check that all legacy permissions have granular equivalents""" # Get unique module/action combinations from legacy system legacy_combinations = set() for perm in RolePermission.objects.all(): legacy_combinations.add((perm.module, perm.action)) for perm in DefaultRolePermission.objects.all(): legacy_combinations.add((perm.module, perm.action)) for perm in UserPermission.objects.all(): legacy_combinations.add((perm.module, perm.action)) # Check if granular equivalents exist missing_granular = 0 for module, action in legacy_combinations: action_code = f"{module}_{action}" if not PagePermission.objects.filter(action_code=action_code).exists(): missing_granular += 1 if missing_granular > 0: self.issues.append(f"Found {missing_granular} legacy permissions without granular equivalents") def _check_migration_mappings(self): """Check that permission mappings are correct""" # Check that page names are correctly mapped incorrect_mappings = 0 for page_perm in PagePermission.objects.all(): # Extract module from action_code if '_' in page_perm.action_code: module = page_perm.action_code.split('_')[0] expected_page = self._get_expected_page_name(module) if page_perm.page_name != expected_page: incorrect_mappings += 1 if incorrect_mappings > 0: self.warnings.append(f"Found {incorrect_mappings} potentially incorrect page name mappings") def _check_migration_user_permissions(self): """Check that user permissions were migrated correctly""" # Count legacy user permissions legacy_user_perms = UserPermission.objects.count() # Count granular user permissions granular_user_perms = UserPagePermission.objects.count() # They should be approximately equal (some might not migrate due to missing page permissions) if abs(legacy_user_perms - granular_user_perms) > (legacy_user_perms * 0.1): # 10% tolerance self.warnings.append( f"User permission count mismatch: {legacy_user_perms} legacy vs {granular_user_perms} granular" ) def _check_migration_role_templates(self): """Check that role templates were created correctly""" # Count legacy role permissions legacy_role_perms = DefaultRolePermission.objects.count() if legacy_role_perms == 0: legacy_role_perms = RolePermission.objects.count() # Count granular role templates granular_templates = RolePermissionTemplate.objects.count() # They should be approximately equal if abs(legacy_role_perms - granular_templates) > (legacy_role_perms * 0.1): # 10% tolerance self.warnings.append( f"Role template count mismatch: {legacy_role_perms} legacy vs {granular_templates} granular" ) def _collect_migration_statistics(self): """Collect migration statistics""" self.stats = { 'legacy_role_permissions': RolePermission.objects.count(), 'legacy_default_permissions': DefaultRolePermission.objects.count(), 'legacy_user_permissions': UserPermission.objects.count(), 'granular_page_permissions': PagePermission.objects.count(), 'granular_role_templates': RolePermissionTemplate.objects.count(), 'granular_user_permissions': UserPagePermission.objects.count(), 'migration_coverage': self._calculate_migration_coverage(), } def _has_circular_dependency(self, permission, visited=None): """Check if a permission has circular dependencies""" if visited is None: visited = set() if permission.id in visited: return True visited.add(permission.id) for required_perm in permission.required_permissions.all(): if self._has_circular_dependency(required_perm, visited.copy()): return True return False def _get_expected_page_name(self, module): """Get expected page name for a module""" page_mapping = { 'dashboard': 'dashboard', 'clients': 'clients', 'loans': 'loans', 'repayments': 'repayments', 'portfolio': 'portfolio', 'reports_statements': 'reports', 'documents': 'documents', 'customer_documents': 'documents', 'payment_receipts': 'receipts', 'notifications': 'notifications', 'settings': 'settings', 'branch_settings': 'settings', 'system_settings': 'settings', } return page_mapping.get(module, module.split('_')[0] if '_' in module else module) def _calculate_migration_coverage(self): """Calculate what percentage of legacy permissions were migrated""" legacy_total = ( RolePermission.objects.count() + DefaultRolePermission.objects.count() + UserPermission.objects.count() ) granular_total = ( RolePermissionTemplate.objects.count() + UserPagePermission.objects.count() ) if legacy_total == 0: return 100.0 return round((granular_total / legacy_total) * 100, 2) def validate_permission_system(system='both'): """ Convenience function to validate permission systems Args: system: 'legacy', 'granular', 'migration', or 'both' Returns: Dict containing validation results """ validator = PermissionDataValidator() results = {} if system in ['legacy', 'both']: results['legacy'] = validator.validate_legacy_permissions() if system in ['granular', 'both']: results['granular'] = validator.validate_granular_permissions() if system == 'migration': results['migration'] = validator.validate_migration_integrity() return results