Clear permissions for unused models before creating types

AlanCoding · AlanCoding · commit 059b42cd2297 · 2025-09-25T10:51:54.000-04:00
diff --git a/ansible_base/rbac/migrations/0005_remote_permissions_data.py b/ansible_base/rbac/migrations/0005_remote_permissions_data.py
@@ -4,59 +4,18 @@
 
 from django.db import migrations
 
+from . import _utils
+
 
 logger = logging.getLogger(__name__)
 
 
 def create_types_if_needed(apps, schema_editor):
-    """Before we can migrate to the new DABContentType, entries in that table must be created.
-
-    This method runs what is ordinarily the post_migrate logic, but in the migration case here.
-    Only needed in the upgrade case, otherwise better to run at true post-migrate.
-    """
-    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-    rd_cls = apps.get_model('dab_rbac', 'RoleDefinition')
-    if permission_cls.objects.exists() or rd_cls.objects.exists():
-        logger.info('Running DABContentType creation script as part of 0005 migration')
-        from ansible_base.rbac.management.create_types import create_DAB_contenttypes
-
-        create_DAB_contenttypes(apps=apps)
+    return _utils.create_types_if_needed(apps, schema_editor, logger)
 
 
 def migrate_content_type(apps, schema_editor):
-    ct_cls = apps.get_model('dab_rbac', 'DABContentType')
-    ct_cls.objects.clear_cache()
-    for model_name in ('dabpermission', 'objectrole', 'roledefinition', 'roleuserassignment', 'roleteamassignment'):
-        cls = apps.get_model('dab_rbac', model_name)
-        update_ct = 0
-        for obj in cls.objects.all():
-            old_ct = obj.content_type
-            if old_ct:
-                try:
-                    # NOTE: could give duplicate normally, but that is impossible in migration path
-                    obj.new_content_type = ct_cls.objects.get_by_natural_key(old_ct.app_label, old_ct.model)
-                except Exception as e:
-                    raise RuntimeError(
-                        f"Failed to get new content type for a {model_name} pk={obj.pk}, obj={obj.__dict__}"
-                    ) from e
-                obj.save()
-                update_ct += 1
-        if update_ct:
-            logger.info(f'Updated content_type reference to new model for {model_name} for {update_ct} entries')
-    for model_name in ('roleevaluation', 'roleevaluationuuid'):
-        cls = apps.get_model('dab_rbac', model_name)
-        cls.objects.all().delete()
-
-    # DABPermission model had api_slug added in last migration
-    # if records existed before this point, it needs to be filled in
-    mod_ct = 0
-    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-    for permission in permission_cls.objects.all():
-        permission.api_slug = f'{permission.new_content_type.service}.{permission.codename}'
-        permission.save()
-        mod_ct += 1
-    if mod_ct:
-        logger.info(f'Set new field DABPermission.api_slug for {mod_ct} existing permissions')
+    return _utils.migrate_content_type(apps, schema_editor, logger)
 
 
 class Migration(migrations.Migration):
diff --git a/ansible_base/rbac/migrations/_utils.py b/ansible_base/rbac/migrations/_utils.py
@@ -1,3 +1,4 @@
+# Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
 from django.db import models
 
 # This method has moved, and this is put here temporarily to make branch management easier
@@ -45,3 +46,123 @@ def give_permissions(apps, rd, users=(), teams=(), object_id=None, content_type_
                 for team_id in teams
             ]
         RoleTeamAssignment.objects.bulk_create(team_assignments, ignore_conflicts=True)
+
+
+def cleanup_orphaned_permissions(apps, logger=None):
+    """
+    Delete orphaned DABPermission objects for models no longer in the permission registry.
+
+    This is used during migrations to clean up permissions that were auto-created for models
+    that are no longer tracked by RBAC, but only if they are not referenced by any RoleDefinition.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        logger: Optional logger for reporting deleted permissions
+
+    Returns:
+        int: Number of permissions deleted
+    """
+    # Get model classes from apps registry
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    role_definition_cls = apps.get_model('dab_rbac', 'RoleDefinition')
+
+    # Get permission registry to check which models are registered
+    from ansible_base.rbac import permission_registry
+    registered_model_keys = set()
+    for model in permission_registry._registry:
+        registered_model_keys.add((model._meta.app_label, model._meta.model_name))
+
+    # Find orphaned permissions
+    orphaned_permissions = []
+    for permission in permission_cls.objects.all():
+        if permission.content_type:
+            model_key = (permission.content_type.app_label, permission.content_type.model)
+            if model_key not in registered_model_keys:
+                # Check if this permission is referenced by any RoleDefinition
+                is_referenced = role_definition_cls.objects.filter(permissions=permission).exists()
+                if not is_referenced:
+                    orphaned_permissions.append(permission)
+
+    # Delete orphaned permissions
+    deleted_count = 0
+    if orphaned_permissions:
+        deleted_count = len(orphaned_permissions)
+        permission_cls.objects.filter(id__in=[p.id for p in orphaned_permissions]).delete()
+        if logger:
+            logger.info(f'Deleted {deleted_count} orphaned DABPermission objects for unregistered models')
+
+    return deleted_count
+
+
+def migrate_content_type(apps, schema_editor, logger=None):
+    """
+    Migrate content type references from Django ContentType to DABContentType.
+
+    This function handles the migration of content type references across all RBAC models
+    from the old Django ContentType to the new DABContentType system.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        schema_editor: Django schema editor (unused but required for migration signature)
+        logger: Optional logger for reporting migration progress
+    """
+    ct_cls = apps.get_model('dab_rbac', 'DABContentType')
+    ct_cls.objects.clear_cache()
+
+    # Pre-check: Delete orphaned DABPermission objects before migration
+    cleanup_orphaned_permissions(apps, logger)
+
+    for model_name in ('dabpermission', 'objectrole', 'roledefinition', 'roleuserassignment', 'roleteamassignment'):
+        cls = apps.get_model('dab_rbac', model_name)
+        update_ct = 0
+        for obj in cls.objects.all():
+            old_ct = obj.content_type
+            if old_ct:
+                try:
+                    # NOTE: could give duplicate normally, but that is impossible in migration path
+                    obj.new_content_type = ct_cls.objects.get_by_natural_key(old_ct.app_label, old_ct.model)
+                except Exception as e:
+                    raise RuntimeError(
+                        f"Failed to get new content type for a {model_name} pk={obj.pk}, obj={obj.__dict__}"
+                    ) from e
+                obj.save()
+                update_ct += 1
+        if update_ct and logger:
+            logger.info(f'Updated content_type reference to new model for {model_name} for {update_ct} entries')
+    for model_name in ('roleevaluation', 'roleevaluationuuid'):
+        cls = apps.get_model('dab_rbac', model_name)
+        cls.objects.all().delete()
+
+    # DABPermission model had api_slug added in last migration
+    # if records existed before this point, it needs to be filled in
+    mod_ct = 0
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    for permission in permission_cls.objects.all():
+        permission.api_slug = f'{permission.new_content_type.service}.{permission.codename}'
+        permission.save()
+        mod_ct += 1
+    if mod_ct and logger:
+        logger.info(f'Set new field DABPermission.api_slug for {mod_ct} existing permissions')
+
+
+def create_types_if_needed(apps, schema_editor, logger=None):
+    """
+    Create DABContentType entries if needed before migration.
+
+    Before we can migrate to the new DABContentType, entries in that table must be created.
+    This method runs what is ordinarily the post_migrate logic, but in the migration case here.
+    Only needed in the upgrade case, otherwise better to run at true post-migrate.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        schema_editor: Django schema editor (unused but required for migration signature)
+        logger: Optional logger for reporting creation progress
+    """
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    rd_cls = apps.get_model('dab_rbac', 'RoleDefinition')
+    if permission_cls.objects.exists() or rd_cls.objects.exists():
+        if logger:
+            logger.info('Running DABContentType creation script as part of 0005 migration')
+        from ansible_base.rbac.management.create_types import create_DAB_contenttypes
+
+        create_DAB_contenttypes(apps=apps)
diff --git a/test_app/tests/rbac/test_migrations.py b/test_app/tests/rbac/test_migrations.py
@@ -1,8 +1,8 @@
 import pytest
 from django.apps import apps
 
-from ansible_base.rbac.migrations._utils import give_permissions
-from ansible_base.rbac.models import DABContentType, DABPermission, RoleTeamAssignment, RoleUserAssignment
+from ansible_base.rbac.migrations._utils import cleanup_orphaned_permissions, give_permissions
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ansible_base.rbac.permission_registry import permission_registry
 from test_app.models import Team, User
 
@@ -33,3 +33,117 @@ def test_give_permissions_by_id(organization, inventory, inv_rd):
 def test_permission_migration():
     "These are expected to be created via a post_migrate signal just like auth.Permission"
     assert len(DABPermission.objects.order_by('content_type').values_list('content_type').distinct()) == len(permission_registry.all_registered_models)
+
+
+@pytest.mark.django_db
+def test_cleanup_orphaned_permissions_deletes_unregistered_unreferenced():
+    """Test that cleanup_orphaned_permissions deletes permissions for unregistered models that are not referenced by any RoleDefinition.
+
+    Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+    """
+    # Create a DABContentType for a model not in the permission registry
+    dab_fake_ct = DABContentType.objects.create(app_label='fake_app', model='fakemodel', service='local')
+
+    # Create a permission for this fake model
+    orphaned_permission = DABPermission.objects.create(name='Can view fake model', content_type=dab_fake_ct, codename='view_fakemodel')
+
+    # For registered model, let's use an existing DABContentType from the test setup
+    # Get any existing DABContentType that represents a registered model
+    existing_registered_ct = DABContentType.objects.filter(app_label='test_app').first()
+
+    if not existing_registered_ct:
+        # Create one if none exist
+        existing_registered_ct = DABContentType.objects.create(app_label='test_app', model='organization', service='shared')
+
+    valid_permission = DABPermission.objects.create(name='Can view registered model', content_type=existing_registered_ct, codename='view_registered')
+
+    initial_count = DABPermission.objects.count()
+
+    # Run cleanup
+    deleted_count = cleanup_orphaned_permissions(apps)
+
+    # Verify the orphaned permission was deleted
+    assert deleted_count == 1
+    assert DABPermission.objects.count() == initial_count - 1
+    assert not DABPermission.objects.filter(id=orphaned_permission.id).exists()
+    assert DABPermission.objects.filter(id=valid_permission.id).exists()
+
+
+@pytest.mark.django_db
+def test_cleanup_orphaned_permissions_preserves_referenced():
+    """Test that cleanup_orphaned_permissions preserves permissions for unregistered models that are referenced by RoleDefinitions.
+
+    Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+    """
+    # Create a DABContentType for a model not in the permission registry
+    dab_fake_ct = DABContentType.objects.create(app_label='fake_app', model='fakemodel', service='local')
+
+    # Create a permission for this fake model
+    referenced_permission = DABPermission.objects.create(name='Can view fake model', content_type=dab_fake_ct, codename='view_fakemodel')
+
+    # Create a RoleDefinition that references this permission
+    role_def = RoleDefinition.objects.create(name='Fake Role', content_type=dab_fake_ct)
+    role_def.permissions.add(referenced_permission)
+
+    initial_count = DABPermission.objects.count()
+
+    # Run cleanup
+    deleted_count = cleanup_orphaned_permissions(apps)
+
+    # Verify the referenced permission was NOT deleted
+    assert deleted_count == 0
+    assert DABPermission.objects.count() == initial_count
+    assert DABPermission.objects.filter(id=referenced_permission.id).exists()
+
+
+@pytest.mark.django_db
+def test_cleanup_orphaned_permissions_preserves_registered():
+    """Test that cleanup_orphaned_permissions preserves all permissions for registered models.
+
+    Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+    """
+    initial_count = DABPermission.objects.count()
+
+    # Run cleanup - should not delete any existing permissions since they're for registered models
+    deleted_count = cleanup_orphaned_permissions(apps)
+
+    # Verify no permissions were deleted
+    assert deleted_count == 0
+    assert DABPermission.objects.count() == initial_count
+
+
+@pytest.mark.django_db
+def test_cleanup_orphaned_permissions_mixed_scenario():
+    """Test cleanup with a mix of registered, unregistered referenced, and unregistered unreferenced permissions.
+
+    Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+    """
+    # Create DABContentTypes for models not in the permission registry
+    dab_fake_ct1 = DABContentType.objects.create(app_label='fake_app', model='orphanedmodel', service='local')
+    dab_fake_ct2 = DABContentType.objects.create(app_label='fake_app', model='referencedmodel', service='local')
+
+    # Create orphaned permission (should be deleted)
+    orphaned_permission = DABPermission.objects.create(name='Can view orphaned model', content_type=dab_fake_ct1, codename='view_orphanedmodel')
+
+    # Create referenced permission (should be preserved)
+    referenced_permission = DABPermission.objects.create(name='Can view referenced model', content_type=dab_fake_ct2, codename='view_referencedmodel')
+
+    # Create a RoleDefinition that references the second permission
+    role_def = RoleDefinition.objects.create(name='Referenced Role', content_type=dab_fake_ct2)
+    role_def.permissions.add(referenced_permission)
+
+    # Get count of valid permissions (for registered models)
+    valid_perms_count = DABPermission.objects.exclude(id__in=[orphaned_permission.id, referenced_permission.id]).count()
+
+    initial_count = DABPermission.objects.count()
+
+    # Run cleanup
+    deleted_count = cleanup_orphaned_permissions(apps)
+
+    # Verify only the orphaned permission was deleted
+    assert deleted_count == 1
+    assert DABPermission.objects.count() == initial_count - 1
+    assert not DABPermission.objects.filter(id=orphaned_permission.id).exists()
+    assert DABPermission.objects.filter(id=referenced_permission.id).exists()
+    # All other permissions should still exist
+    assert DABPermission.objects.exclude(id=referenced_permission.id).count() == valid_perms_count
