|
| 1 | +from unittest.mock import patch |
| 2 | + |
1 | 3 | import pytest |
| 4 | +from django.db import IntegrityError |
2 | 5 |
|
3 | | -from ansible_base.resource_registry.models.service_identifier import ServiceID |
| 6 | +from ansible_base.resource_registry.models.service_identifier import ServiceID, service_id |
4 | 7 |
|
5 | 8 |
|
6 | 9 | @pytest.mark.django_db |
7 | 10 | def test_service_id_already_exists(): |
8 | 11 | "The resource registry already creates this, so we expect an error here" |
| 12 | + # Ensure one exists first |
| 13 | + if not ServiceID.objects.exists(): |
| 14 | + ServiceID.objects.create() |
9 | 15 | with pytest.raises(RuntimeError) as exc: |
10 | 16 | ServiceID.objects.create() |
11 | 17 | assert 'This service already has a ServiceID' in str(exc) |
12 | 18 |
|
13 | 19 |
|
14 | 20 | @pytest.mark.django_db |
15 | 21 | def test_service_id_does_not_yet_exist(): |
16 | | - ServiceID.objects.first().delete() # clear out what migration created |
| 22 | + ServiceID.objects.all().delete() # clear out any existing |
17 | 23 | ServiceID.objects.create() # expect no error |
| 24 | + |
| 25 | + |
| 26 | +@pytest.mark.django_db |
| 27 | +def test_service_id_function_auto_creates_in_debug(): |
| 28 | + """ |
| 29 | + Generated by Claude Sonnet 4.5 |
| 30 | + Test that service_id() auto-creates ServiceID when none exists in DEBUG/pytest mode. |
| 31 | + """ |
| 32 | + # Clear the global cache and delete existing ServiceID |
| 33 | + import sys |
| 34 | + |
| 35 | + import ansible_base.resource_registry.models.service_identifier as sid_module |
| 36 | + |
| 37 | + sid_module._service_id = None |
| 38 | + ServiceID.objects.all().delete() |
| 39 | + |
| 40 | + # Ensure pytest is in sys.argv to trigger auto-creation |
| 41 | + with patch.object(sys, 'argv', ['pytest']): |
| 42 | + result = service_id() |
| 43 | + |
| 44 | + assert result is not None |
| 45 | + assert ServiceID.objects.count() == 1 |
| 46 | + # The service was created successfully |
| 47 | + |
| 48 | + |
| 49 | +@pytest.mark.django_db |
| 50 | +def test_service_id_function_handles_integrity_error(): |
| 51 | + """ |
| 52 | + Generated by Claude Sonnet 4.5 |
| 53 | + Test that service_id() handles IntegrityError when another thread/process creates ServiceID first. |
| 54 | + """ |
| 55 | + # Clear the global cache and delete existing ServiceID |
| 56 | + import sys |
| 57 | + |
| 58 | + import ansible_base.resource_registry.models.service_identifier as sid_module |
| 59 | + |
| 60 | + sid_module._service_id = None |
| 61 | + ServiceID.objects.all().delete() |
| 62 | + |
| 63 | + # Pre-create an object to simulate another process winning the race |
| 64 | + # We'll create it before the patched create is called |
| 65 | + ServiceID.objects.create(pk=1) |
| 66 | + |
| 67 | + # Delete from cache and reset so service_id() will try to create |
| 68 | + sid_module._service_id = None |
| 69 | + # Don't delete from DB - simulate another process created it |
| 70 | + |
| 71 | + # Mock create to raise IntegrityError |
| 72 | + def mock_create(*args, **kwargs): |
| 73 | + # Raise IntegrityError to simulate race condition |
| 74 | + raise IntegrityError("Duplicate key") |
| 75 | + |
| 76 | + # Ensure pytest is in sys.argv to trigger auto-creation path |
| 77 | + with patch.object(sys, 'argv', ['pytest']): |
| 78 | + # Mock objects.first() to return None to force the creation path |
| 79 | + with patch.object(ServiceID.objects, 'first', return_value=None): |
| 80 | + with patch.object(ServiceID.objects, 'create', side_effect=mock_create): |
| 81 | + result = service_id() |
| 82 | + |
| 83 | + assert result is not None |
| 84 | + # The IntegrityError was handled and it fetched the existing record via get(pk=1) |
| 85 | + assert ServiceID.objects.count() == 1 |
0 commit comments