Bug fix for AAP-47771 data migration update (#16058)

* Bug fix for AAP-47771 this data migration updates existing CredentialType entries
in the database and changes the kind from github_app to github_app_lookup

* Combine migration 0203 into 0202

* Add test to ensure reconciliation issue has been resolved

awx/main/migrations/0202_squashed_deletions.py

@@ -1,34 +1,55 @@
 # Generated by Django 4.2.10 on 2024-09-16 10:22
-
 from django.db import migrations, models
-
 from awx.main.migrations._create_system_jobs import delete_clear_tokens_sjt
 
 
+# --- START of function merged from 0203_rename_github_app_kind.py ---
+def update_github_app_kind(apps, schema_editor):
+    """
+    Updates the 'kind' field for CredentialType records
+    from 'github_app' to 'github_app_lookup'.
+    This addresses a change in the entry point key for the GitHub App plugin.
+    """
+    CredentialType = apps.get_model('main', 'CredentialType')
+    db_alias = schema_editor.connection.alias
+    CredentialType.objects.using(db_alias).filter(kind='github_app').update(kind='github_app_lookup')
+
+
+# --- END of function merged from 0203_rename_github_app_kind.py ---
+
+
 class Migration(migrations.Migration):
     dependencies = [
         ('main', '0201_create_managed_creds'),
     ]
-
     operations = [
         migrations.DeleteModel(
             name='Profile',
         ),
         # Remove SSO app content
         # delete all sso application migrations
-        migrations.RunSQL("DELETE FROM django_migrations WHERE app = 'sso';"),
+        # Added reverse_sql=migrations.RunSQL.noop to make this reversible for tests
+        migrations.RunSQL("DELETE FROM django_migrations WHERE app = 'sso';", reverse_sql=migrations.RunSQL.noop),
         # delete all sso application content group permissions
+        # Added reverse_sql=migrations.RunSQL.noop to make this reversible for tests
         migrations.RunSQL(
             "DELETE FROM auth_group_permissions "
             "WHERE permission_id IN "
-            "(SELECT id FROM auth_permission WHERE content_type_id in (SELECT id FROM django_content_type WHERE app_label = 'sso'));"
+            "(SELECT id FROM auth_permission WHERE content_type_id in (SELECT id FROM django_content_type WHERE app_label = 'sso'));",
+            reverse_sql=migrations.RunSQL.noop,
         ),
         # delete all sso application content permissions
-        migrations.RunSQL("DELETE FROM auth_permission " "WHERE content_type_id IN (SELECT id FROM django_content_type WHERE app_label = 'sso');"),
+        # Added reverse_sql=migrations.RunSQL.noop to make this reversible for tests
+        migrations.RunSQL(
+            "DELETE FROM auth_permission " "WHERE content_type_id IN (SELECT id FROM django_content_type WHERE app_label = 'sso');",
+            reverse_sql=migrations.RunSQL.noop,
+        ),
         # delete sso application content type
-        migrations.RunSQL("DELETE FROM django_content_type WHERE app_label = 'sso';"),
+        # Added reverse_sql=migrations.RunSQL.noop to make this reversible for tests
+        migrations.RunSQL("DELETE FROM django_content_type WHERE app_label = 'sso';", reverse_sql=migrations.RunSQL.noop),
         # drop sso application created table
-        migrations.RunSQL("DROP TABLE IF EXISTS sso_userenterpriseauth;"),
+        # Added reverse_sql=migrations.RunSQL.noop to make this reversible for tests
+        migrations.RunSQL("DROP TABLE IF EXISTS sso_userenterpriseauth;", reverse_sql=migrations.RunSQL.noop),
         # Alter inventory source source field
         migrations.AlterField(
             model_name='inventorysource',
@@ -97,4 +118,7 @@ class Migration(migrations.Migration):
                 max_length=32,
             ),
         ),
+        # --- START of operations merged from 0203_rename_github_app_kind.py ---
+        migrations.RunPython(update_github_app_kind, migrations.RunPython.noop),
+        # --- END of operations merged from 0203_rename_github_app_kind.py ---
     ]

awx/main/tests/functional/test_migrations.py

@@ -2,13 +2,13 @@ import pytest
 
 from django_test_migrations.plan import all_migrations, nodes_to_tuples
 from django.utils.timezone import now
+from django.utils import timezone
 
 """
 Most tests that live in here can probably be deleted at some point. They are mainly
 for a developer. When AWX versions that users upgrade from falls out of support that
 is when migration tests can be deleted. This is also a good time to squash. Squashing
 will likely mess with the tests that live here.
-
 The smoke test should be kept in here. The smoke test ensures that our migrations
 continue to work when sqlite is the backing database (vs. the default DB of postgres).
 """
@@ -19,27 +19,22 @@ class TestMigrationSmoke:
     def test_happy_path(self, migrator):
         """
         This smoke test runs all the migrations.
-
         Example of how to use django-test-migration to invoke particular migration(s)
         while weaving in object creation and assertions.
-
         Note that this is more than just an example. It is a smoke test because it runs ALL
         the migrations. Our "normal" unit tests subvert the migrations running because it is slow.
         """
         migration_nodes = all_migrations('default')
         migration_tuples = nodes_to_tuples(migration_nodes)
         final_migration = migration_tuples[-1]
-
         migrator.apply_initial_migration(('main', None))
         # I just picked a newish migration at the time of writing this.
         # If someone from the future finds themselves here because the are squashing migrations
         # it is fine to change the 0180_... below to some other newish migration
         intermediate_state = migrator.apply_tested_migration(('main', '0180_add_hostmetric_fields'))
-
         Instance = intermediate_state.apps.get_model('main', 'Instance')
         # Create any old object in the database
         Instance.objects.create(hostname='foobar', node_type='control')
-
         final_state = migrator.apply_tested_migration(final_migration)
         Instance = final_state.apps.get_model('main', 'Instance')
         assert Instance.objects.filter(hostname='foobar').count() == 1
@@ -52,20 +47,16 @@ class TestMigrationSmoke:
         foo = Instance.objects.create(hostname='foo', node_type='execution', listener_port=1234)
         bar = Instance.objects.create(hostname='bar', node_type='execution', listener_port=None)
         bar.peers.add(foo)
-
         new_state = migrator.apply_tested_migration(
             ('main', '0189_inbound_hop_nodes'),
         )
         Instance = new_state.apps.get_model('main', 'Instance')
         ReceptorAddress = new_state.apps.get_model('main', 'ReceptorAddress')
-
         # We can now test how our migration worked, new field is there:
         assert ReceptorAddress.objects.filter(address='foo', port=1234).count() == 1
         assert not ReceptorAddress.objects.filter(address='bar').exists()
-
         bar = Instance.objects.get(hostname='bar')
         fooaddr = ReceptorAddress.objects.get(address='foo')
-
         bar_peers = bar.peers.all()
         assert len(bar_peers) == 1
         assert fooaddr in bar_peers
@@ -75,38 +66,31 @@ class TestMigrationSmoke:
         Organization = old_state.apps.get_model('main', 'Organization')
         Team = old_state.apps.get_model('main', 'Team')
         User = old_state.apps.get_model('auth', 'User')
-
         org = Organization.objects.create(name='arbitrary-org', created=now(), modified=now())
         user = User.objects.create(username='random-user')
         org.read_role.members.add(user)
         org.member_role.members.add(user)
-
         team = Team.objects.create(name='arbitrary-team', organization=org, created=now(), modified=now())
         team.member_role.members.add(user)
-
         new_state = migrator.apply_tested_migration(
             ('main', '0192_custom_roles'),
         )
-
         RoleUserAssignment = new_state.apps.get_model('dab_rbac', 'RoleUserAssignment')
         assert RoleUserAssignment.objects.filter(user=user.id, object_id=org.id).exists()
         assert RoleUserAssignment.objects.filter(user=user.id, role_definition__name='Controller Organization Member', object_id=org.id).exists()
         assert RoleUserAssignment.objects.filter(user=user.id, role_definition__name='Controller Team Member', object_id=team.id).exists()
-
         # Regression testing for bug that comes from current vs past models mismatch
         RoleDefinition = new_state.apps.get_model('dab_rbac', 'RoleDefinition')
         assert not RoleDefinition.objects.filter(name='Organization Organization Admin').exists()
         # Test special cases in managed role creation
         assert not RoleDefinition.objects.filter(name='Organization Team Admin').exists()
         assert not RoleDefinition.objects.filter(name='Organization InstanceGroup Admin').exists()
-
         # Test that a removed EE model permission has been deleted
         new_state = migrator.apply_tested_migration(
             ('main', '0195_EE_permissions'),
         )
         DABPermission = new_state.apps.get_model('dab_rbac', 'DABPermission')
         assert not DABPermission.objects.filter(codename='view_executionenvironment').exists()
-
         # Test create a Project with a duplicate name
         Organization = new_state.apps.get_model('main', 'Organization')
         Project = new_state.apps.get_model('main', 'Project')
@@ -115,13 +99,11 @@ class TestMigrationSmoke:
         for i in range(3):
             proj = Project.objects.create(name='duplicate-project-name', organization=org, created=now(), modified=now())
             proj_ids.append(proj.id)
-
         # The uniqueness rules will not apply to InventorySource
         Inventory = new_state.apps.get_model('main', 'Inventory')
         InventorySource = new_state.apps.get_model('main', 'InventorySource')
         inv = Inventory.objects.create(name='migration-test-inv', organization=org, created=now(), modified=now())
         InventorySource.objects.create(name='migration-test-src', source='file', inventory=inv, organization=org, created=now(), modified=now())
-
         new_state = migrator.apply_tested_migration(
             ('main', '0200_template_name_constraint'),
         )
@@ -132,7 +114,6 @@ class TestMigrationSmoke:
             else:
                 assert proj.name != 'duplicate-project-name'
                 assert proj.name.startswith('duplicate-project-name')
-
         # The inventory source had this field set to avoid the constrains
         InventorySource = new_state.apps.get_model('main', 'InventorySource')
         inv_src = InventorySource.objects.get(name='migration-test-src')
@@ -140,3 +121,54 @@ class TestMigrationSmoke:
         Project = new_state.apps.get_model('main', 'Project')
         for proj in Project.objects.all():
             assert proj.org_unique is True
+
+
+@pytest.mark.django_db
+class TestGithubAppBug:
+    """
+    Tests that `awx-manage createsuperuser` runs successfully after
+    the `github_app` CredentialType kind is updated to `github_app_lookup`
+    via the migration.
+    """
+
+    def test_after_github_app_kind_migration(self, migrator):
+        """
+        Verifies that `createsuperuser` does not raise a KeyError
+        after the 0202_squashed_deletions migration (which includes
+        the `update_github_app_kind` logic) is applied.
+        """
+        # 1. Apply migrations up to the point *before* the 0202_squashed_deletions migration.
+        # This simulates the state where the problematic CredentialType might exist.
+        # We use 0201_create_managed_creds as the direct predecessor.
+        old_state = migrator.apply_tested_migration(('main', '0201_create_managed_creds'))
+
+        # Get the CredentialType model from the historical state.
+        CredentialType = old_state.apps.get_model('main', 'CredentialType')
+
+        # Create a CredentialType with the old, problematic 'kind' value
+        CredentialType.objects.create(
+            name='Legacy GitHub App Credential',
+            kind='github_app',  # The old, problematic 'kind' value
+            namespace='github_app',  # The namespace that causes the KeyError in the registry lookup
+            managed=True,
+            created=timezone.now(),
+            modified=timezone.now(),
+        )
+
+        # Apply the migration that includes the fix (0202_squashed_deletions).
+        new_state = migrator.apply_tested_migration(('main', '0202_squashed_deletions'))
+
+        # Verify that the CredentialType with the old 'kind' no longer exists
+        # and the 'kind' has been updated to the new value.
+        CredentialType = new_state.apps.get_model('main', 'CredentialType')  # Get CredentialType model from the new state
+
+        # Assertion 1: The CredentialType with the old 'github_app' kind should no longer exist.
+        assert not CredentialType.objects.filter(
+            kind='github_app'
+        ).exists(), "CredentialType with old 'github_app' kind should no longer exist after migration."
+
+        # Assertion 2: The CredentialType should now exist with the new 'github_app_lookup' kind
+        # and retain its original name.
+        assert CredentialType.objects.filter(
+            kind='github_app_lookup', name='Legacy GitHub App Credential'
+        ).exists(), "CredentialType should be updated to 'github_app_lookup' and retain its name."