convert Credential to django migration

awx/main/migrations/0004_rbac_migrations.py

@@ -12,5 +12,6 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
-        migrations.RunPython(rbac.migrate_organization, rbac.unmigrate_organization),
+        migrations.RunPython(rbac.migrate_organization),
+        migrations.RunPython(rbac.migrate_credential),
     ]

awx/main/migrations/_rbac.py

@@ -13,5 +13,15 @@ def migrate_organization(apps, schema_editor):
     return migrations
 
 
-def unmigrate_organization(apps, schema_editor):
-    pass
+def migrate_credential(apps, schema_editor):
+    migrations = defaultdict(list)
+    credential = apps.get_model('main', "Credential")
+    for cred in credential.objects.all():
+        if cred.user:
+            cred.owner_role.members.add(cred.user)
+            migrations[cred.name].append(cred.user)
+        elif cred.team:
+            cred.owner_role.parents.add(cred.team.admin_role)
+            cred.usage_role.parents.add(cred.team.member_role)
+            migrations[cred.name].append(cred.team)
+    return migrations

awx/main/models/credential.py

@@ -363,14 +363,6 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
                 update_fields.append('cloud')
         super(Credential, self).save(*args, **kwargs)
 
-    def migrate_to_rbac(self):
-        if self.user:
-            self.owner_role.members.add(self.user)
-            return [self.user]
-        elif self.team:
-            self.owner_role.parents.add(self.team.admin_role)
-            self.usage_role.parents.add(self.team.member_role)
-            return [self.team]
 
 def validate_ssh_private_key(data):
     """Validate that the given SSH private key or certificate is,

awx/main/tests/functional/test_rbac_credential.py

@@ -1,10 +1,16 @@
 import pytest
 
+from awx.main.migrations import _rbac as rbac
+from django.apps import apps
+
 @pytest.mark.django_db
 def test_credential_migration_user(credential, user, permissions):
     u = user('user', False)
     credential.user = u
-    migrated = credential.migrate_to_rbac()
+    credential.save()
+
+    migrated = rbac.migrate_credential(apps, None)
+
     assert len(migrated) == 1
     assert credential.accessible_by(u, permissions['admin'])
 
@@ -19,11 +25,13 @@ def test_credential_migration_team_member(credential, team, user, permissions):
     u = user('user', False)
     team.admin_role.members.add(u)
     credential.team = team
+    credential.save()
 
     # No permissions pre-migration
     assert not credential.accessible_by(u, permissions['admin'])
 
-    migrated = credential.migrate_to_rbac()
+    migrated = rbac.migrate_credential(apps, None)
+
     # Admin permissions post migration
     assert len(migrated) == 1
     assert credential.accessible_by(u, permissions['admin'])
@@ -33,12 +41,13 @@ def test_credential_migration_team_admin(credential, team, user, permissions):
     u = user('user', False)
     team.member_role.members.add(u)
     credential.team = team
+    credential.save()
 
     # No permissions pre-migration
     assert not credential.accessible_by(u, permissions['usage'])
 
     # Usage permissions post migration
-    migrated = credential.migrate_to_rbac()
+    migrated = rbac.migrate_credential(apps, None)
     assert len(migrated) == 1
     assert credential.accessible_by(u, permissions['usage'])