Clear permissions for unused models before creating types

AlanCoding · AlanCoding · commit 59cba5f900e0 · 2025-09-25T11:48:48.000-04:00
Reduce tests some

consolidate migration logic

Improve logging and code cleanup
diff --git a/ansible_base/rbac/migrations/0005_remote_permissions_data.py b/ansible_base/rbac/migrations/0005_remote_permissions_data.py
@@ -1,62 +1,8 @@
 # Generated by Django 4.2.23 on 2025-06-30 12:48
 
-import logging
-
 from django.db import migrations
 
-
-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)
-
-
-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')
+from . import _utils
 
 
 class Migration(migrations.Migration):
@@ -66,6 +12,6 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
-        migrations.RunPython(create_types_if_needed, migrations.RunPython.noop),
-        migrations.RunPython(migrate_content_type, migrations.RunPython.noop),
+        migrations.RunPython(_utils.create_types_if_needed, migrations.RunPython.noop),
+        migrations.RunPython(_utils.migrate_content_type, migrations.RunPython.noop),
     ]
diff --git a/ansible_base/rbac/migrations/_utils.py b/ansible_base/rbac/migrations/_utils.py
@@ -1,5 +1,10 @@
+# Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+import logging
+
 from django.db import models
 
+logger = logging.getLogger(__name__)
+
 # This method has moved, and this is put here temporarily to make branch management easier
 from ansible_base.rbac.management import create_dab_permissions as create_custom_permissions  # noqa
 
@@ -45,3 +50,127 @@ 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):
+    """
+    Delete orphaned DABPermission objects for models no longer in the permission registry.
+
+    This is used during migrations to clean up permissions for any previously-registered model
+    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)
+
+    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
+                referencing_roles = role_definition_cls.objects.filter(permissions=permission)
+                if referencing_roles.exists():
+                    # Log warning for unregistered model still referenced by role definitions
+                    role_names = list(referencing_roles.values_list('name', flat=True))
+                    logger.warning(
+                        f'Permission {permission.codename} for unregistered model '
+                        f'{permission.content_type.app_label}.{permission.content_type.model} '
+                        f'is still referenced by role definitions: {role_names}'
+                    )
+                else:
+                    logger.info(f'Deleting orphaned permission {permission.codename} for unregistered model {model_key}')
+                    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()
+        logger.info(f'Deleted {deleted_count} orphaned DABPermission objects for unregistered models')
+
+    return deleted_count
+
+
+def migrate_content_type(apps, schema_editor):
+    """
+    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)
+    """
+    # Pre-check: Delete orphaned DABPermission objects before migration
+    cleanup_orphaned_permissions(apps)
+
+    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')
+
+
+def create_types_if_needed(apps, schema_editor):
+    """
+    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)
+    """
+    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)
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,53 @@ 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 (should be deleted)
+    orphaned_permission = DABPermission.objects.create(name='Can view fake model', content_type=dab_fake_ct, codename='view_fakemodel')
+
+    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()
+
+
+@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()
