Get rid of pk=1 bad pattern and test it

Author: AlanCoding

ansible_base/resource_registry/models/service_identifier.py

@@ -30,10 +30,15 @@ def service_id():
             if settings.DEBUG or "pytest" in sys.argv:
                 try:
                     with transaction.atomic():
-                        obj = ServiceID.objects.create(pk=1)
+                        obj = ServiceID.objects.create()
+                        # Check if another process also created one during the race
+                        if ServiceID.objects.count() > 1:
+                            # We lost the race, delete ours and use the other
+                            obj.delete()
+                            obj = ServiceID.objects.first()
                 except IntegrityError:
                     # Another thread/process won the race—read it
-                    obj = ServiceID.objects.get(pk=1)
+                    obj = ServiceID.objects.first()
             else:
                 raise RuntimeError('Expected ServiceID to be created in data migrations but was not found')
         _service_id = str(obj.pk)

test_app/tests/resource_registry/models/test_service_id.py

@@ -62,7 +62,7 @@ def test_service_id_function_handles_integrity_error():
 
     # Pre-create an object to simulate another process winning the race
     # We'll create it before the patched create is called
-    ServiceID.objects.create(pk=1)
+    existing_obj = ServiceID.objects.create()
 
     # Delete from cache and reset so service_id() will try to create
     sid_module._service_id = None
@@ -75,11 +75,74 @@ def mock_create(*args, **kwargs):
 
     # Ensure pytest is in sys.argv to trigger auto-creation path
     with patch.object(sys, 'argv', ['pytest']):
-        # Mock objects.first() to return None to force the creation path
-        with patch.object(ServiceID.objects, 'first', return_value=None):
+        # Mock objects.first() to return None first time, then the actual object
+        with patch.object(ServiceID.objects, 'first', side_effect=[None, existing_obj]):
             with patch.object(ServiceID.objects, 'create', side_effect=mock_create):
                 result = service_id()
 
     assert result is not None
-    # The IntegrityError was handled and it fetched the existing record via get(pk=1)
+    # The IntegrityError was handled and it fetched the existing record via first()
     assert ServiceID.objects.count() == 1
+
+
+@pytest.mark.django_db
+def test_service_id_function_cleans_up_duplicate_on_race():
+    """
+    Generated by Claude Sonnet 4.5
+    Test that service_id() detects when two objects were created in a race and cleans up the duplicate.
+    """
+    import sys
+
+    import ansible_base.resource_registry.models.service_identifier as sid_module
+
+    sid_module._service_id = None
+    ServiceID.objects.all().delete()
+
+    # Create the "winner" object that another process created
+    winner_obj = ServiceID.objects.create()
+    winner_pk = winner_obj.pk
+
+    sid_module._service_id = None
+
+    # Track what we create and delete
+    created_obj = None
+    created_pk = None
+    deleted_pks = []
+
+    # Mock create to bypass save() exists check and track the created object
+    def mock_create(*args, **kwargs):
+        nonlocal created_obj, created_pk
+        import uuid
+
+        from django.db import models
+
+        created_obj = ServiceID(id=uuid.uuid4())
+        # Bypass the exists check in save()
+        models.Model.save(created_obj)
+        created_pk = created_obj.pk  # Capture pk before it gets deleted
+        return created_obj
+
+    # Track deletions
+    original_delete = ServiceID.delete
+
+    def track_delete(self, *args, **kwargs):
+        deleted_pks.append(self.pk)
+        return original_delete(self, *args, **kwargs)
+
+    with patch.object(sys, 'argv', ['pytest']):
+        # first() returns None to trigger creation path, then winner_obj for cleanup
+        with patch.object(ServiceID.objects, 'first', side_effect=[None, winner_obj]):
+            with patch.object(ServiceID.objects, 'create', side_effect=mock_create):
+                # count() returns 2 to trigger cleanup, then actual count
+                with patch.object(ServiceID.objects, 'count', side_effect=[2, 1]):
+                    with patch.object(ServiceID, 'delete', track_delete):
+                        result = service_id()
+
+    # Verify our created object was deleted
+    assert created_obj is not None
+    assert created_pk in deleted_pks
+    # Verify we returned the winner's ID
+    assert result == str(winner_pk)
+    # Verify only one object remains in the actual database
+    assert ServiceID.objects.count() == 1
+    assert ServiceID.objects.filter(pk=winner_pk).exists()