AAP-51352 Allow running tests with --nomigrations (ansible#853)

I was able to test locally with

```
pytest test_app/tests/ --nomigrations
```

I got 5 failures, 2 were due to local dependency config issues, and 3
were some specifically from the resource_registry app related to this,
but were directly touching `ServiceID` in some way so those just aren't
going to work with this and that's okay.

Author: AlanCoding

ansible_base/resource_registry/models/service_identifier.py

@@ -1,6 +1,8 @@
+import sys
 import uuid
 
-from django.db import models
+from django.conf import settings
+from django.db import IntegrityError, models, transaction
 
 
 class ServiceID(models.Model):
@@ -23,5 +25,21 @@ def save(self, *args, **kwargs):
 def service_id():
     global _service_id
     if not _service_id:
-        _service_id = str(ServiceID.objects.first().pk)
+        obj = ServiceID.objects.first()
+        if obj is None:
+            if settings.DEBUG or "pytest" in sys.argv:
+                try:
+                    with transaction.atomic():
+                        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.first()
+            else:
+                raise RuntimeError('Expected ServiceID to be created in data migrations but was not found')
+        _service_id = str(obj.pk)
     return _service_id

test_app/tests/conftest.py

@@ -13,6 +13,7 @@
 from django.contrib.contenttypes.models import ContentType
 from django.db.migrations.recorder import MigrationRecorder
 from django.db.models.signals import post_migrate
+from django.db.utils import ProgrammingError
 from django.test.client import RequestFactory
 from django.test.utils import override_settings
 from drf_spectacular.generators import SchemaGenerator
@@ -47,7 +48,13 @@ def test_migrations_okay(apps=None, app_config=None, **kwargs):
         return  # so that it is only ran once
     disk_steps = defaultdict(set)
     app_exceptions = {'default': 'auth', 'social_auth': 'social_django'}
-    for app in MigrationRecorder.Migration.objects.values_list('app', flat=True).distinct():
+
+    try:
+        app_list = list(MigrationRecorder.Migration.objects.values_list('app', flat=True).distinct())
+    except ProgrammingError:
+        return
+
+    for app in app_list:
         if app in app_exceptions:
             continue
         try:

test_app/tests/resource_registry/models/test_service_id.py

@@ -1,17 +1,148 @@
+from unittest.mock import patch
+
 import pytest
+from django.db import IntegrityError
 
-from ansible_base.resource_registry.models.service_identifier import ServiceID
+from ansible_base.resource_registry.models.service_identifier import ServiceID, service_id
 
 
 @pytest.mark.django_db
 def test_service_id_already_exists():
     "The resource registry already creates this, so we expect an error here"
+    # Ensure one exists first
+    if not ServiceID.objects.exists():
+        ServiceID.objects.create()
     with pytest.raises(RuntimeError) as exc:
         ServiceID.objects.create()
     assert 'This service already has a ServiceID' in str(exc)
 
 
 @pytest.mark.django_db
 def test_service_id_does_not_yet_exist():
-    ServiceID.objects.first().delete()  # clear out what migration created
+    ServiceID.objects.all().delete()  # clear out any existing
     ServiceID.objects.create()  # expect no error
+
+
+@pytest.mark.django_db
+def test_service_id_function_auto_creates_in_debug():
+    """
+    Generated by Claude Sonnet 4.5
+    Test that service_id() auto-creates ServiceID when none exists in DEBUG/pytest mode.
+    """
+    # Clear the global cache and delete existing ServiceID
+    import sys
+
+    import ansible_base.resource_registry.models.service_identifier as sid_module
+
+    sid_module._service_id = None
+    ServiceID.objects.all().delete()
+
+    # Ensure pytest is in sys.argv to trigger auto-creation
+    with patch.object(sys, 'argv', ['pytest']):
+        result = service_id()
+
+    assert result is not None
+    assert ServiceID.objects.count() == 1
+    # The service was created successfully
+
+
+@pytest.mark.django_db
+def test_service_id_function_handles_integrity_error():
+    """
+    Generated by Claude Sonnet 4.5
+    Test that service_id() handles IntegrityError when another thread/process creates ServiceID first.
+    """
+    # Clear the global cache and delete existing ServiceID
+    import sys
+
+    import ansible_base.resource_registry.models.service_identifier as sid_module
+
+    sid_module._service_id = None
+    ServiceID.objects.all().delete()
+
+    # Pre-create an object to simulate another process winning the race
+    # We'll create it before the patched create is called
+    existing_obj = ServiceID.objects.create()
+
+    # Delete from cache and reset so service_id() will try to create
+    sid_module._service_id = None
+    # Don't delete from DB - simulate another process created it
+
+    # Mock create to raise IntegrityError
+    def mock_create(*args, **kwargs):
+        # Raise IntegrityError to simulate race condition
+        raise IntegrityError("Duplicate key")
+
+    # Ensure pytest is in sys.argv to trigger auto-creation path
+    with patch.object(sys, 'argv', ['pytest']):
+        # 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 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()