fix: skip already converted credentials (ansible#825)

src/aap_eda/core/management/commands/create_initial_data.py

@@ -11,6 +11,9 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
+import logging
+from urllib.parse import urlparse
+
 from ansible_base.rbac import permission_registry
 from ansible_base.rbac.models import DABPermission, RoleDefinition
 from django.contrib.contenttypes.models import ContentType
@@ -22,6 +25,7 @@
 from aap_eda.core.utils.credentials import inputs_to_store
 
 CRUD = ["add", "view", "change", "delete"]
+LOGGER = logging.getLogger(__name__)
 
 # FIXME(cutwater): Role descriptions were taken from the RBAC design document
 #  and must be updated.
@@ -408,21 +412,42 @@ def _copy_registry_credentials(self):
             name=enums.DefaultCredentialType.REGISTRY
         ).first()
         for cred in credentials:
+            de = models.DecisionEnvironment.objects.filter(
+                credential=cred
+            ).first()
+            host = "quay.io"
+            if de:
+                image_url = (
+                    de.image_url.replace("http://", "")
+                    .replace("https://", "")
+                    .replace("//", "")
+                )
+                host = urlparse(f"//{image_url}").hostname
             inputs = {
+                "host": host,
                 "username": cred.username,
                 "password": cred.secret.get_secret_value(),
             }
-            eda_cred = models.EdaCredential.objects.create(
+            eda_cred, created = models.EdaCredential.objects.get_or_create(
                 name=cred.name,
-                description=cred.description,
-                managed=False,
-                credential_type=cred_type,
-                inputs=inputs_to_store(inputs),
-            )
-            models.DecisionEnvironment.objects.filter(credential=cred).update(
-                eda_credential=eda_cred, credential=None
+                defaults={
+                    "description": cred.description,
+                    "managed": False,
+                    "credential_type": cred_type,
+                    "inputs": inputs_to_store(inputs),
+                },
             )
-            cred.delete()
+            if created:
+                models.DecisionEnvironment.objects.filter(
+                    credential=cred
+                ).update(eda_credential=eda_cred, credential=None)
+                cred.delete()
+            else:
+                info_msg = (
+                    f"Registry Credential {cred.name} already converted to "
+                    "EdaCredential. Skip the duplicated one."
+                )
+                LOGGER.info(info_msg)
 
         self.stdout.write(
             "All REGISTRY credentials are converted to Container Registry "
@@ -445,17 +470,26 @@ def _copy_scm_credentials(self):
                 "username": cred.username,
                 "password": cred.secret.get_secret_value(),
             }
-            eda_cred = models.EdaCredential.objects.create(
+            eda_cred, created = models.EdaCredential.objects.get_or_create(
                 name=cred.name,
-                description=cred.description,
-                managed=False,
-                credential_type=cred_type,
-                inputs=inputs_to_store(inputs),
-            )
-            models.Project.objects.filter(credential=cred).update(
-                eda_credential=eda_cred, credential=None
+                defaults={
+                    "description": cred.description,
+                    "managed": False,
+                    "credential_type": cred_type,
+                    "inputs": inputs_to_store(inputs),
+                },
             )
-            cred.delete()
+            if created:
+                models.Project.objects.filter(credential=cred).update(
+                    eda_credential=eda_cred, credential=None
+                )
+                cred.delete()
+            else:
+                info_msg = (
+                    f"Git Credential {cred.name} already converted to "
+                    "EdaCredential. Skip the duplicated one."
+                )
+                LOGGER.info(info_msg)
 
         self.stdout.write(
             "All GITHUB and GITLAB credentials are converted to Source "

tests/integration/core/test_create_initial_data.py

@@ -15,7 +15,9 @@
 import pytest
 from ansible_base.rbac.models import DABPermission, RoleDefinition
 
+from aap_eda.core import enums, models
 from aap_eda.core.management.commands.create_initial_data import Command
+from aap_eda.core.utils.credentials import inputs_from_store
 
 
 @pytest.mark.django_db
@@ -52,3 +54,73 @@ def test_remove_extra_permission():
     assert perm in auditor_role.permissions.all()
     Command().handle()
     assert perm not in auditor_role.permissions.all()
+
+
+def create_old_registry_credential():
+    credential = models.Credential.objects.create(
+        name="registry cred",
+        credential_type=enums.CredentialType.REGISTRY,
+        username="fred",
+        secret="mysec",
+    )
+    de = models.DecisionEnvironment.objects.create(
+        name="my DE",
+        image_url="private-reg.com/fred/de",
+        credential=credential,
+    )
+    return credential, de
+
+
+def create_old_git_credential():
+    credential = models.Credential.objects.create(
+        name="git cred",
+        credential_type=enums.CredentialType.GITHUB,
+        username="fred",
+        secret="mysec",
+    )
+    project = models.Project.objects.create(
+        name="my project",
+        url="github.com/fred/projects",
+        credential=credential,
+    )
+    return credential, project
+
+
+@pytest.mark.django_db
+def test_copy_registry_credentials(caplog):
+    credential, de = create_old_registry_credential()
+    Command().handle()
+
+    assert not models.Credential.objects.filter(id=credential.id).exists()
+    de.refresh_from_db()
+    assert de.eda_credential.name == credential.name
+    assert not de.eda_credential.managed
+    inputs = inputs_from_store(de.eda_credential.inputs.get_secret_value())
+    assert inputs["host"] == "private-reg.com"
+    assert inputs["username"] == "fred"
+    assert inputs["password"] == "mysec"
+
+    credential.id = None
+    credential.save()
+    Command().handle()
+    assert models.Credential.objects.filter(id=credential.id).exists()
+
+
+@pytest.mark.django_db
+def test_copy_project_credentials(caplog):
+    credential, project = create_old_git_credential()
+    Command().handle()
+
+    assert not models.Credential.objects.filter(id=credential.id).exists()
+    project.refresh_from_db()
+    assert project.eda_credential.name == credential.name
+    assert not project.eda_credential.managed
+    inputs = inputs_from_store(
+        project.eda_credential.inputs.get_secret_value()
+    )
+    assert inputs["username"] == "fred"
+    assert inputs["password"] == "mysec"
+    credential.id = None
+    credential.save()
+    Command().handle()
+    assert models.Credential.objects.filter(id=credential.id).exists()