Handle duplicate service slugs

setup.cfg

@@ -45,7 +45,6 @@ install_requires =
     isodate
     notifications-api-common>=0.3.0
     zgw-consumers>=0.35.1
-    zgw-consumers-oas
     oyaml
     PyJWT>=2.0.0
     requests
@@ -61,6 +60,7 @@ tests_require =
     black
     requests-mock
     freezegun
+    zgw-consumers-oas
 
 [options.packages.find]
 include =
@@ -80,6 +80,7 @@ tests =
     black
     requests-mock
     freezegun
+    zgw-consumers-oas
 pep8 = flake8
 coverage = pytest-cov
 docs =

tests/test_migrations.py

@@ -1,4 +1,7 @@
+from typing import Callable, Generator
+
 from django.apps import apps as global_apps
+from django.apps.registry import Apps
 from django.db import connection
 from django.db.migrations.executor import MigrationExecutor
 from django.test import override_settings
@@ -13,93 +16,88 @@ class BaseMigrationTest:
     migrate_to = None  # The migration we want to test
     setting_overrides = {}
 
-    @pytest.fixture(autouse=True)
-    def setup_migration(self, db):
+    @pytest.fixture
+    def setup_migration(self, db) -> Generator:
         """
         Setup the migration test by reversing to `migrate_from` state,
         then applying the `migrate_to` state.
         """
-        assert self.app is not None, "You must define the `app` attribute"
-        assert self.migrate_from is not None, "You must define `migrate_from`"
-        assert self.migrate_to is not None, "You must define `migrate_to`"
 
-        # Step 1: Set up the MigrationExecutor
-        executor = MigrationExecutor(connection)
+        def _migrate(callback: Callable) -> None:
+            assert self.app is not None, "You must define the `app` attribute"
+            assert self.migrate_from is not None, "You must define `migrate_from`"
+            assert self.migrate_to is not None, "You must define `migrate_to`"
+
+            # Step 1: Set up the MigrationExecutor
+            executor = MigrationExecutor(connection)
 
-        # Step 2: Reverse to the starting migration state
-        migrate_from = [(self.app, self.migrate_from)]
-        migrate_to = [(self.app, self.migrate_to)]
-        old_migrate_state = executor.migrate(migrate_from)
-        old_apps = old_migrate_state.apps
+            # Step 2: Reverse to the starting migration state
+            migrate_from = [(self.app, self.migrate_from)]
+            migrate_to = [(self.app, self.migrate_to)]
+            old_migrate_state = executor.migrate(migrate_from)
+            old_apps = old_migrate_state.apps
 
-        # Step 3: Run the `setup_before_migration` hook with old state models
-        self.setup_before_migration(old_apps)
+            # Step 3: Call `callback` old state models
+            callback(old_apps)
 
-        # Step 4: Apply the migration we want to test with any settings overrides
-        with override_settings(**self.setting_overrides):
+            # Step 4: Apply the migration we want to test with any settings overrides
             executor = MigrationExecutor(connection)
             executor.loader.build_graph()  # reload the graph in case of dependency changes
             executor.migrate(migrate_to)
 
-        # Set `self.apps` to the post-migration state
-        self.apps = executor.loader.project_state(migrate_to).apps
-
-    def setup_before_migration(self, apps):
-        """
-        Hook to set up the state before the migration. Subclasses can
-        override this to create data or perform other setup actions.
-        """
-        pass
+        yield _migrate
 
 
 class TestMigrateAPICredentialToService(BaseMigrationTest):
     app = "vng_api_common"
     migrate_from = "0005_auto_20190614_1346"
     migrate_to = "0006_delete_apicredential"
 
-    def setup_before_migration(self, apps):
-        Service = apps.get_model("zgw_consumers", "Service")
-        APICredential = apps.get_model("vng_api_common", "APICredential")
-
-        APICredential.objects.create(
-            api_root="https://example.com/zaken/api/v1/",
-            label="Zaken API",
-            client_id="user",
-            secret="secret",
-            user_id="user",
-            user_representation="User",
-        )
+    def test_migrate_api_credentials_to_services(self, setup_migration: Callable):
+        def migration_callback(apps: Apps) -> None:
+            Service = apps.get_model("zgw_consumers", "Service")
+            APICredential = apps.get_model("vng_api_common", "APICredential")
+
+            APICredential.objects.create(
+                api_root="https://example.com/zaken/api/v1/",
+                label="Zaken API",
+                client_id="user",
+                secret="secret",
+                user_id="user",
+                user_representation="User",
+            )
+
+            APICredential.objects.create(
+                api_root="https://example.com/api/v1/",
+                label="Selectielijst API",
+                client_id="user2",
+                secret="secret2",
+                user_id="user2",
+                user_representation="User2",
+            )
+
+            APICredential.objects.create(
+                api_root="https://example.com/documenten/api/v1/",
+                label="Documenten API",
+                client_id="user3",
+                secret="secret3",
+                user_id="user3",
+                user_representation="User3",
+            )
+            Service.objects.create(
+                api_root="https://example.com/documenten/api/v1/",
+                label="Documenten",
+                slug="already-exists",
+                api_type=APITypes.drc,
+                auth_type=AuthTypes.zgw,
+                client_id="user4",
+                secret="secret4",
+                user_id="user4",
+                user_representation="User4",
+            )
+
+        setup_migration(migration_callback)
 
-        APICredential.objects.create(
-            api_root="https://example.com/api/v1/",
-            label="Selectielijst API",
-            client_id="user2",
-            secret="secret2",
-            user_id="user2",
-            user_representation="User2",
-        )
-
-        APICredential.objects.create(
-            api_root="https://example.com/documenten/api/v1/",
-            label="Documenten API",
-            client_id="user3",
-            secret="secret3",
-            user_id="user3",
-            user_representation="User3",
-        )
-        Service.objects.create(
-            api_root="https://example.com/documenten/api/v1/",
-            label="Documenten",
-            slug="already-exists",
-            api_type=APITypes.drc,
-            auth_type=AuthTypes.zgw,
-            client_id="user4",
-            secret="secret4",
-            user_id="user4",
-            user_representation="User4",
-        )
-
-    def test_migrate_api_credentials_to_services(self):
         Service = global_apps.get_model("zgw_consumers", "Service")
 
         zaken_service = Service.objects.get(slug="zaken-api")
@@ -136,33 +134,93 @@ def test_migrate_api_credentials_to_services(self):
         assert documenten_service.user_id == "user4"
         assert documenten_service.user_representation == "User4"
 
+    def test_duplicate_service_slug(self, setup_migration):
+        """
+        Test that the migration handles duplicate slugs correctly
+        """
+
+        def migration_callback(apps):
+            Service = apps.get_model("zgw_consumers", "Service")
+            APICredential = apps.get_model("vng_api_common", "APICredential")
+
+            APICredential.objects.create(
+                api_root="https://example.com/documenten/api/v1/",
+                label="Documenten API",
+                client_id="user3",
+                secret="secret3",
+                user_id="user3",
+                user_representation="User3",
+            )
+
+            Service.objects.create(
+                api_root="https://example.com/documenten/api/v2/",
+                label="Documenten",
+                slug="documenten-api",
+                api_type=APITypes.drc,
+                auth_type=AuthTypes.zgw,
+                client_id="user4",
+                secret="secret4",
+                user_id="user4",
+                user_representation="User4",
+            )
+
+        setup_migration(migration_callback)
+
+        Service = global_apps.get_model("zgw_consumers", "Service")
+
+        service_v1 = Service.objects.get(slug="documenten-api-2")
+
+        assert service_v1.api_root == "https://example.com/documenten/api/v1/"
+        assert service_v1.label == "Documenten API"
+        assert service_v1.api_type == APITypes.drc
+        assert service_v1.auth_type == AuthTypes.zgw
+        assert service_v1.client_id == "user3"
+        assert service_v1.secret == "secret3"
+        assert service_v1.user_id == "user3"
+        assert service_v1.user_representation == "User3"
+
+        service_v2 = Service.objects.get(slug="documenten-api")
+
+        assert service_v2.api_root == "https://example.com/documenten/api/v2/"
+        assert service_v2.label == "Documenten"
+        assert service_v2.api_type == APITypes.drc
+        assert service_v2.auth_type == AuthTypes.zgw
+        assert service_v2.client_id == "user4"
+        assert service_v2.secret == "secret4"
+        assert service_v2.user_id == "user4"
+        assert service_v2.user_representation == "User4"
+
 
 class TestMigrateAuthorizationsConfigAPIRootToService(BaseMigrationTest):
     app = "authorizations"
     migrate_from = "0015_auto_20220318_1608"
     migrate_to = "0016_remove_authorizationsconfig_api_root_and_more"
 
-    def setup_before_migration(self, apps):
-        Service = apps.get_model("zgw_consumers", "Service")
-        AuthorizationsConfig = apps.get_model("authorizations", "AuthorizationsConfig")
-
-        config, _ = AuthorizationsConfig.objects.get_or_create()
-        config.api_root = "https://example.com/autorisaties/api/v1/"
-        config.save()
-
-        self.service = Service.objects.create(
-            api_root="https://example.com/autorisaties/api/v1/",
-            label="Autorisaties",
-            slug="autorisaties",
-            api_type=APITypes.ac,
-            auth_type=AuthTypes.zgw,
-            client_id="user",
-            secret="secret",
-            user_id="user",
-            user_representation="User",
-        )
+    def test_migrate_api_root_to_service(self, setup_migration: Callable):
+        def migration_callback(apps: Apps) -> None:
+            Service = apps.get_model("zgw_consumers", "Service")
+            AuthorizationsConfig = apps.get_model(
+                "authorizations", "AuthorizationsConfig"
+            )
+
+            config, _ = AuthorizationsConfig.objects.get_or_create()
+            config.api_root = "https://example.com/autorisaties/api/v1/"
+            config.save()
+
+            self.service = Service.objects.create(
+                api_root="https://example.com/autorisaties/api/v1/",
+                label="Autorisaties",
+                slug="autorisaties",
+                api_type=APITypes.ac,
+                auth_type=AuthTypes.zgw,
+                client_id="user",
+                secret="secret",
+                user_id="user",
+                user_representation="User",
+            )
+
+        setup_migration(migration_callback)
 
-    def test_migrate_api_root_to_service(self):
         AuthorizationsConfig = global_apps.get_model(
             "authorizations", "AuthorizationsConfig"
         )

vng_api_common/migrations/0006_delete_apicredential.py

@@ -1,16 +1,17 @@
 # Generated by Django 5.1.2 on 2024-10-24 13:51
 
 import logging
+from typing import Set
 
-from django.db import IntegrityError, migrations
+from django.db import migrations, models
 from django.utils.text import slugify
 
 from zgw_consumers.constants import APITypes, AuthTypes
 
 logger = logging.getLogger(__name__)
 
 
-def get_api_type(api_root: str) -> APITypes:
+def _get_api_type(api_root: str) -> APITypes:
     mapping = {
         "/autorisaties/api/": APITypes.ac,
         "/zaken/api/": APITypes.zrc,
@@ -26,38 +27,50 @@ def get_api_type(api_root: str) -> APITypes:
     return APITypes.orc
 
 
+def _get_service_slug(credential: models.Model, existing_slugs: Set[str]) -> str:
+    default_slug: str = slugify(credential.label)
+
+    if default_slug not in existing_slugs or not existing_slugs:
+        return default_slug
+
+    count = 2
+    slug = f"{default_slug}-{count}"
+
+    while slug in existing_slugs:
+        count += 1
+        slug = f"{default_slug}-{count}"
+
+    return slug
+
+
 def migrate_credentials_to_service(apps, _) -> None:
     APICredential = apps.get_model("vng_api_common", "APICredential")
     Service = apps.get_model("zgw_consumers", "Service")
 
     credentials = APICredential.objects.all()
 
+    existings_service_slugs = set(Service.objects.values_list("slug", flat=True))
+
     for credential in credentials:
         logger.info(f"Creating Service for {credential.client_id}")
 
-        service_slug = slugify(credential.label)
-
-        try:
-            _, created = Service.objects.get_or_create(
-                api_root=credential.api_root,
-                defaults=dict(
-                    label=credential.label,
-                    slug=service_slug,
-                    api_type=get_api_type(credential.api_root),
-                    auth_type=AuthTypes.zgw,
-                    client_id=credential.client_id,
-                    secret=credential.secret,
-                    user_id=credential.user_id,
-                    user_representation=credential.user_representation,
-                ),
-            )
-        except IntegrityError:
-            logger.warning(
-                f"Unable to create Service for {credential.api_root}. Check the"
-                "`service slug` field on the existing Service's to verify an existing"
-                " Service already exists."
-            )
-            continue
+        service_slug = _get_service_slug(credential, existings_service_slugs)
+
+        _, created = Service.objects.get_or_create(
+            api_root=credential.api_root,
+            defaults=dict(
+                label=credential.label,
+                slug=service_slug,
+                api_type=_get_api_type(credential.api_root),
+                auth_type=AuthTypes.zgw,
+                client_id=credential.client_id,
+                secret=credential.secret,
+                user_id=credential.user_id,
+                user_representation=credential.user_representation,
+            ),
+        )
+
+        existings_service_slugs.add(service_slug)
 
         if created:
             logger.info(f"Created new Service for {credential.api_root}")