|
1 | 1 | import pytest |
| 2 | +from django import VERSION |
2 | 3 | from django.contrib.contenttypes.models import ContentType |
3 | 4 |
|
4 | | -from ansible_base.resource_registry.models import Resource |
5 | | -from test_app.models import Organization |
| 5 | +from ansible_base.resource_registry.models import Resource, ResourceType |
| 6 | +from test_app.models import Inventory, Organization |
6 | 7 |
|
7 | 8 |
|
8 | 9 | @pytest.mark.django_db |
@@ -100,3 +101,114 @@ def test_resource_field_filtering(organization): |
100 | 101 |
|
101 | 102 | org = Organization.objects.get(resource__name=organization.name) |
102 | 103 | assert org.resource.pk == resource.pk |
| 104 | + |
| 105 | + |
| 106 | +@pytest.mark.django_db |
| 107 | +def test_resource_field_prefetch_related_across_foreign_key(organization, organization_1, organization_2, django_assert_num_queries): |
| 108 | + """ |
| 109 | + Generated by Claude Code (claude-sonnet-4-5@20250929) |
| 110 | + Test that prefetch_related works across a ForeignKey to a model with a resource field. |
| 111 | + This tests prefetching organization__resource on Inventory objects. |
| 112 | + """ |
| 113 | + # Create inventory objects linked to organizations |
| 114 | + Inventory.objects.create(name="Inventory 1", organization=organization) |
| 115 | + Inventory.objects.create(name="Inventory 2", organization=organization_1) |
| 116 | + Inventory.objects.create(name="Inventory 3", organization=organization_2) |
| 117 | + |
| 118 | + org_ctype = ContentType.objects.get_for_model(Organization) |
| 119 | + |
| 120 | + # Prefetch organization__resource should result in 3 queries: |
| 121 | + # 1. Fetch all Inventory objects |
| 122 | + # 2. Fetch all related Organization objects |
| 123 | + # 3. Fetch all related Resource objects |
| 124 | + with django_assert_num_queries(3) as captured: |
| 125 | + inventory_qs = list(Inventory.objects.prefetch_related("organization__resource").all()) |
| 126 | + |
| 127 | + # Verify the queries were as expected |
| 128 | + assert "test_app_inventory" in captured[0]["sql"] |
| 129 | + assert "test_app_organization" in captured[1]["sql"] |
| 130 | + assert "dab_resource_registry_resource" in captured[2]["sql"] |
| 131 | + |
| 132 | + assert len(inventory_qs) == 3 |
| 133 | + |
| 134 | + # Collect resource pks for later verification |
| 135 | + resource_pks = {} |
| 136 | + with django_assert_num_queries(0): |
| 137 | + for inv in inventory_qs: |
| 138 | + assert inv.organization is not None |
| 139 | + assert inv.organization.resource is not None |
| 140 | + # Verify the resource data is correct |
| 141 | + assert inv.organization.name == inv.organization.resource.name |
| 142 | + assert str(inv.organization.pk) == inv.organization.resource.object_id |
| 143 | + resource_pks[inv.organization.pk] = inv.organization.resource.pk |
| 144 | + |
| 145 | + # Verify the resources match what's in the database |
| 146 | + for org_pk, resource_pk in resource_pks.items(): |
| 147 | + expected_resource = Resource.objects.get(object_id=org_pk, content_type=org_ctype) |
| 148 | + assert resource_pk == expected_resource.pk |
| 149 | + |
| 150 | + |
| 151 | +@pytest.mark.django_db |
| 152 | +def test_resource_field_prefetch_resource_type_from_organization(organization, organization_1, organization_2, django_assert_num_queries): |
| 153 | + """ |
| 154 | + Generated by Claude Code (claude-sonnet-4-5@20250929) |
| 155 | + Test that prefetch_related works from Organization through resource to resource_type. |
| 156 | + This tests prefetching resource__content_type__resource_type on Organization objects. |
| 157 | + """ |
| 158 | + org_ctype = ContentType.objects.get_for_model(Organization) |
| 159 | + |
| 160 | + # Prefetch resource__content_type__resource_type should result in 4 queries: |
| 161 | + # 1. Fetch all Organization objects |
| 162 | + # 2. Fetch all related Resource objects |
| 163 | + # 3. Fetch all related ContentType objects |
| 164 | + # 4. Fetch all related ResourceType objects |
| 165 | + with django_assert_num_queries(4) as captured: |
| 166 | + org_qs = list(Organization.objects.prefetch_related("resource__content_type__resource_type").all()) |
| 167 | + |
| 168 | + # Verify the queries were as expected |
| 169 | + assert "test_app_organization" in captured[0]["sql"] |
| 170 | + assert "dab_resource_registry_resource" in captured[1]["sql"] |
| 171 | + assert "django_content_type" in captured[2]["sql"] |
| 172 | + assert "dab_resource_registry_resourcetype" in captured[3]["sql"] |
| 173 | + |
| 174 | + assert len(org_qs) > 2 |
| 175 | + |
| 176 | + # Collect resource type data for later verification |
| 177 | + resource_type_data = {} |
| 178 | + with django_assert_num_queries(0): |
| 179 | + for org in org_qs: |
| 180 | + assert org.resource is not None |
| 181 | + assert org.resource.content_type is not None |
| 182 | + assert org.resource.content_type.resource_type is not None |
| 183 | + # Verify the resource type is correct |
| 184 | + resource_type = org.resource.content_type.resource_type |
| 185 | + assert resource_type.content_type == org_ctype |
| 186 | + resource_type_data[org.pk] = resource_type.pk |
| 187 | + |
| 188 | + # Verify the resource types match what's in the database |
| 189 | + expected_resource_type = ResourceType.objects.get(content_type=org_ctype) |
| 190 | + for org_pk, resource_type_pk in resource_type_data.items(): |
| 191 | + assert resource_type_pk == expected_resource_type.pk |
| 192 | + |
| 193 | + |
| 194 | +@pytest.mark.skipif(VERSION[0] < 5, reason='get_prefetch_querysets() is only valid for Django 5+') |
| 195 | +@pytest.mark.django_db |
| 196 | +@pytest.mark.parametrize( |
| 197 | + "querysets_arg", |
| 198 | + [ |
| 199 | + pytest.param([Resource.objects.none(), Resource.objects.none()], id="two_querysets"), |
| 200 | + pytest.param([Resource.objects.none(), Resource.objects.none(), Resource.objects.none()], id="three_querysets"), |
| 201 | + ], |
| 202 | +) |
| 203 | +def test_resource_field_get_prefetch_querysets_invalid_length(organization, querysets_arg): |
| 204 | + """ |
| 205 | + Generated by Claude Code (claude-sonnet-4-5@20250929) |
| 206 | + Test that get_prefetch_querysets raises ValueError when querysets argument has invalid length. |
| 207 | + The method expects exactly 1 queryset if querysets is provided, so multiple querysets should raise. |
| 208 | + """ |
| 209 | + # Get the forward descriptor for the resource field |
| 210 | + descriptor = Organization.resource.field.forward_related_accessor_class(Organization.resource.field) |
| 211 | + |
| 212 | + # Call get_prefetch_querysets with invalid querysets argument |
| 213 | + with pytest.raises(ValueError, match='querysets argument of get_prefetch_querysets\\(\\) should have a length of 1\\.'): |
| 214 | + descriptor.get_prefetch_querysets([organization], querysets=querysets_arg) |
0 commit comments