From cdde9e98fa2921413a7faef9d4c476b2bc9dd0ad Mon Sep 17 00:00:00 2001 From: Martin Hauser Date: Thu, 4 Jun 2026 17:45:06 +0200 Subject: [PATCH] test(api): Add GraphQL nested filter and auto-filter tests Introduce declarative GraphQL filter test framework with `GraphQLFilterTest` and `GraphQLQueryTest` dataclasses. Implement auto-filter discovery from filter class annotations with per-field-kind test generators for string, numeric, date, range, and array lookups. Fixes #15569 --- netbox/dcim/tests/query_counts.json | 2 +- netbox/dcim/tests/test_api.py | 73 +- netbox/netbox/tests/test_graphql.py | 68 +- netbox/utilities/testing/api.py | 786 ++++++++++++++++++++- netbox/utilities/tests/test_api_graphql.py | 260 +++++++ 5 files changed, 1165 insertions(+), 24 deletions(-) create mode 100644 netbox/utilities/tests/test_api_graphql.py diff --git a/netbox/dcim/tests/query_counts.json b/netbox/dcim/tests/query_counts.json index 1d7cdda72..7861ee6b7 100644 --- a/netbox/dcim/tests/query_counts.json +++ b/netbox/dcim/tests/query_counts.json @@ -72,7 +72,7 @@ "rearporttemplate:api_list_objects": 12, "region:api_list_objects": 13, "region:list_objects_with_permission": 20, - "site:api_list_objects": 16, + "site:api_list_objects": 17, "site:list_objects_with_permission": 22, "sitegroup:api_list_objects": 13, "sitegroup:list_objects_with_permission": 20, diff --git a/netbox/dcim/tests/test_api.py b/netbox/dcim/tests/test_api.py index 63f968705..a9df054d8 100644 --- a/netbox/dcim/tests/test_api.py +++ b/netbox/dcim/tests/test_api.py @@ -20,6 +20,8 @@ from users.models import ObjectPermission, Token, User from utilities.testing import ( APITestCase, APIViewTestCases, + GraphQLFilterTest, + GraphQLQueryTest, create_test_device, create_test_nat_ip_pair, disable_logging, @@ -146,6 +148,19 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase): bulk_update_data = { 'status': 'planned', } + graphql_filter_tests = ( + GraphQLFilterTest( + name='tenant__name__exact', + filters='tenant: {name: {exact: "Tenant 1"}}', + expected=lambda qs: qs.filter(tenant__name='Tenant 1'), + permissions=('tenancy.view_tenant',), + ), + ) + + def assert_nested_locations_active(self, data): + site_data = data.get('site') or {} + location_names = sorted(location['name'] for location in site_data.get('locations', [])) + self.assertEqual(location_names, ['Site1 Active A', 'Site1 Active B']) @classmethod def setUpTestData(cls): @@ -160,15 +175,32 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase): SiteGroup.objects.create(name='Site Group 2', slug='site-group-2'), ) + tenant = Tenant.objects.create(name='Tenant 1', slug='tenant-1') + + # Site 1's tenant activates the dynamic tenant prefetch (+1 in api_list_objects baseline). sites = ( - Site(region=regions[0], group=groups[0], name='Site 1', slug='site-1'), + Site(region=regions[0], group=groups[0], tenant=tenant, name='Site 1', slug='site-1'), Site(region=regions[0], group=groups[0], name='Site 2', slug='site-2'), Site(region=regions[0], group=groups[0], name='Site 3', slug='site-3'), ) Site.objects.bulk_create(sites) + nested_site = Site.objects.get(slug='site-1') + cls.nested_site_pk = nested_site.pk + Location.objects.create( + site=nested_site, name='Site1 Active A', slug='site1-active-a', + status=LocationStatusChoices.STATUS_ACTIVE, + ) + Location.objects.create( + site=nested_site, name='Site1 Active B', slug='site1-active-b', + status=LocationStatusChoices.STATUS_ACTIVE, + ) + Location.objects.create( + site=nested_site, name='Site1 Planned', slug='site1-planned', + status=LocationStatusChoices.STATUS_PLANNED, + ) + rir = RIR.objects.create(name='RFC 6996', is_private=True) - tenant = Tenant.objects.create(name='Tenant 1', slug='tenant-1') asns = [ ASN(asn=65000 + i, rir=rir) for i in range(8) @@ -203,6 +235,19 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase): }, ] + cls.graphql_query_tests = ( + GraphQLQueryTest( + name='nested_locations_by_status', + query=( + '{ site(id: ' + str(cls.nested_site_pk) + ') { ' + 'locations(filters: {status: {exact: STATUS_ACTIVE}}) { name } ' + '} }' + ), + assert_result=cls.assert_nested_locations_active, + permissions=('dcim.view_location',), + ), + ) + def test_add_tags(self): """ Add tags to an existing object via the add_tags field. @@ -427,6 +472,16 @@ class LocationTestCase(APIViewTestCases.APIViewTestCase): 'description': 'New description', } user_permissions = ('dcim.view_site',) + graphql_filter_tests = ( + GraphQLFilterTest( + name='status__in_list', + filters='status: {in_list: [STATUS_PLANNED, STATUS_STAGING]}', + expected=lambda qs: qs.filter(status__in=[ + LocationStatusChoices.STATUS_PLANNED, + LocationStatusChoices.STATUS_STAGING, + ]), + ), + ) @classmethod def setUpTestData(cls): @@ -476,6 +531,20 @@ class LocationTestCase(APIViewTestCases.APIViewTestCase): parent=parent_locations[0], status=LocationStatusChoices.STATUS_ACTIVE, ) + Location.objects.create( + site=sites[0], + name='GraphQL Planned Location', + slug='graphql-planned-location', + parent=parent_locations[0], + status=LocationStatusChoices.STATUS_PLANNED, + ) + Location.objects.create( + site=sites[0], + name='GraphQL Staging Location', + slug='graphql-staging-location', + parent=parent_locations[0], + status=LocationStatusChoices.STATUS_STAGING, + ) cls.create_data = [ { diff --git a/netbox/netbox/tests/test_graphql.py b/netbox/netbox/tests/test_graphql.py index 2df3a7f98..f960890f7 100644 --- a/netbox/netbox/tests/test_graphql.py +++ b/netbox/netbox/tests/test_graphql.py @@ -10,7 +10,7 @@ from strawberry.schema.config import StrawberryConfig from dcim.choices import LocationStatusChoices from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Site, VirtualChassis -from extras.models import TableConfig +from extras.models import TableConfig, Tag from netbox.graphql.scalars import BigInt, BigIntScalar from netbox.graphql.schema import Query, get_schema_extensions from utilities.tables import get_table_for_model @@ -185,6 +185,72 @@ class GraphQLAPITestCase(APITestCase): self.assertNotIn('errors', data) self.assertEqual(len(data['data']['site']['locations']), 0) + @override_settings(LOGIN_REQUIRED=True) + def test_graphql_nested_filter_objects(self): + """ + Test filtering of nested GraphQL object lists. + """ + self.add_permissions('dcim.view_site', 'dcim.view_location', 'extras.view_tag') + + site = Site.objects.create( + name='Nested Filter Site', + slug='nested-filter-site' + ) + + # Location is MPTT-managed; bulk_create skips tree-init hooks. Use per-instance create. + Location.objects.create( + site=site, + name='Nested Active 1', + slug='nested-active-1', + status=LocationStatusChoices.STATUS_ACTIVE, + ) + Location.objects.create( + site=site, + name='Nested Active 2', + slug='nested-active-2', + status=LocationStatusChoices.STATUS_ACTIVE, + ) + Location.objects.create( + site=site, + name='Nested Planned', + slug='nested-planned', + status=LocationStatusChoices.STATUS_PLANNED, + ) + + planned = Tag.objects.create(name='Planned', slug='planned') + production = Tag.objects.create(name='Production', slug='production') + staging = Tag.objects.create(name='Staging', slug='staging') + site.tags.add(planned, production, staging) + + url = reverse('graphql') + query = f""" + {{ + site(id: {site.pk}) {{ + locations(filters: {{status: {{exact: STATUS_ACTIVE}}}}) {{ + name + }} + tags(filters: {{name: {{i_starts_with: "P"}}}}) {{ + name + }} + }} + }} + """ + + response = self.client.post(url, data={'query': query}, format="json", **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + data = json.loads(response.content) + self.assertNotIn('errors', data) + + self.assertEqual( + {location['name'] for location in data['data']['site']['locations']}, + {'Nested Active 1', 'Nested Active 2'} + ) + self.assertEqual( + {tag['name'] for tag in data['data']['site']['tags']}, + {'Planned', 'Production'} + ) + def test_graphql_integer_range_lookup(self): """ Test that range_lookup works for integer fields (e.g. vc_position). Regression test for #20468. diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py index 15383b415..938c6a3b7 100644 --- a/netbox/utilities/testing/api.py +++ b/netbox/utilities/testing/api.py @@ -1,21 +1,50 @@ import copy +import importlib import inspect import json +import types +import typing +from collections.abc import Callable +from dataclasses import dataclass +from decimal import Decimal +import strawberry import strawberry_django from django.conf import settings from django.contrib.contenttypes.models import ContentType +from django.contrib.postgres.fields import ArrayField +from django.db import models from django.test import override_settings from django.urls import reverse +from django.utils.module_loading import import_string from rest_framework import status from rest_framework.test import APIClient from strawberry.types.base import StrawberryList, StrawberryOptional from strawberry.types.lazy_type import LazyType from strawberry.types.union import StrawberryUnion +from strawberry_django import ( + BaseFilterLookup, + ComparisonFilterLookup, + DateFilterLookup, + DatetimeFilterLookup, + FilterLookup, + RangeLookup, + StrFilterLookup, + TimeFilterLookup, +) from core.choices import ObjectChangeActionChoices from core.models import ObjectChange, ObjectType from ipam.graphql.types import IPAddressFamilyType +from netbox.graphql.filter_lookups import ( + ArrayLookup, + BigIntegerLookup, + FloatLookup, + IntegerLookup, + IntegerRangeArrayLookup, + JSONFilter, + TreeNodeFilter, +) from netbox.models.features import ChangeLoggingMixin from users.constants import TOKEN_PREFIX from users.models import ObjectPermission, Token, User @@ -28,9 +57,48 @@ from .utils import disable_logging, disable_warnings, get_random_string __all__ = ( 'APITestCase', 'APIViewTestCases', + 'GraphQLFilterTest', + 'GraphQLQueryTest', ) +@dataclass(frozen=True) +class GraphQLFilterTest: + """ + Declarative GraphQL filter test case for APIViewTestCases.GraphQLTestCase. + + ``filters`` is the raw content to place inside the GraphQL ``filters`` input, + e.g. ``name: {i_contains: "site"}``. + + ``expected`` may be a callable accepting the model queryset, an ORM filter + dict, a queryset, an iterable of model instances, or an iterable of object + IDs. When omitted, the test only asserts that the filter returns at least one + result; this preserves compatibility with the legacy ``graphql_filter`` + attribute. + """ + name: str + filters: str + expected: object = None + permissions: tuple[str, ...] = () + + +@dataclass(frozen=True) +class GraphQLQueryTest: + """ + Declarative GraphQL query test case for model-specific complex queries. + + ``assert_result`` is called as ``assert_result(testcase, data)`` where + ``testcase`` is the running ``GraphQLTestCase`` instance (use it for + ``testcase.assertEqual`` etc.) and ``data`` is the decoded GraphQL + ``data`` object (the inner ``response.json()['data']``, not the full HTTP + response). + """ + name: str + query: str + assert_result: Callable + permissions: tuple[str, ...] = () + + # # REST/GraphQL API Tests # @@ -556,6 +624,21 @@ class APIViewTestCases: message=changelog_message) class GraphQLTestCase(APITestCase): + graphql_auto_filter_tests = True + graphql_auto_filter_exclude = () + + # Cap fields per lookup kind to keep test counts balanced across kinds + # (string fields shouldn't crowd out numeric/date/array fields). + graphql_auto_filter_fields_per_kind = 2 + + # Fail when auto mode is on and no tests were generated. + graphql_auto_filter_required = True + + # Additional explicit-list filter cases as GraphQLFilterTest instances. + graphql_filter_tests = () + + # Additional full-query cases (e.g. nested filters) as GraphQLQueryTest instances. + graphql_query_tests = () def _get_graphql_base_name(self): """ @@ -622,26 +705,627 @@ class APIViewTestCases: return query + @staticmethod + def _graphql_literal(value): + """ + Render a Python value as a GraphQL literal. + """ + if value is None: + return 'null' + if isinstance(value, bool): + return 'true' if value else 'false' + if isinstance(value, (int, float)): + return str(value) + if isinstance(value, Decimal): + return str(float(value)) + if isinstance(value, (list, tuple)): + items = ', '.join( + APIViewTestCases.GraphQLTestCase._graphql_literal(v) for v in value + ) + return f'[{items}]' + if isinstance(value, str): + return json.dumps(value) + + return json.dumps(str(value)) + + def _render_graphql_filter_value(self, params): + """ + Render the legacy graphql_filter dict value to a GraphQL filter value. + """ + if isinstance(params, str): + return params + + if not isinstance(params, dict): + return self._graphql_literal(params) + + lookup = params.get('lookup') + value = params['value'] + + if lookup: + return f'{{{lookup}: {self._graphql_literal(value)}}}' + + return self._graphql_literal(value) + + def _build_graphql_filter_string(self, **filters): + if not filters: + return '' + + filter_expressions = [ + f'{field_name}: {self._render_graphql_filter_value(params)}' + for field_name, params in filters.items() + ] + + return f'(filters: {{{", ".join(filter_expressions)}}})' + def _build_filtered_query(self, name, **filters): """ Create a filtered query: i.e. device_list(filters: {name: {i_contains: "akron"}}){. """ - # TODO: This should be extended to support AND, OR multi-lookups - if filters: - for field_name, params in filters.items(): - lookup = params['lookup'] - value = params['value'] - if lookup: - query = f'{{{lookup}: "{value}"}}' - filter_string = f'{field_name}: {query}' - else: - filter_string = f'{field_name}: "{value}"' - filter_string = f'(filters: {{{filter_string}}})' - else: - filter_string = '' + filter_string = self._build_graphql_filter_string(**filters) return self._build_query_with_filter(name, filter_string) + def _build_graphql_id_list_query(self, name, filters): + filter_string = f'(filters: {{{filters}}})' if filters else '' + selection = 'id' if self._graphql_type_exposes_id() else '__typename' + + return f""" + {{ + {name}{filter_string} {{ + {selection} + }} + }} + """ + + def _graphql_type_exposes_id(self): + """ + Return True when the model's GraphQL type exposes ``id`` as a + queryable selection. Some NetBox types (e.g. Notification, + Subscription) omit ``id`` from the output type; for those, the + assertion path falls back to length-only comparison. + """ + type_class = get_graphql_type_for_model(self.model) + strawberry_definition = getattr(type_class, '__strawberry_definition__', None) + if strawberry_definition is None: + return False + return any(field.name == 'id' for field in strawberry_definition.fields) + + def _get_model_graphql_filter_class(self, model=None): + """ + Return the model's GraphQL filter class, if one follows NetBox's + conventional .graphql.filters.Filter path. ``None`` if + the filter module (or any of its parent packages) is absent or the + class is not present in the module. Import errors originating + inside an existing filter module are re-raised. + """ + model = model or self.model + module_path = f'{model._meta.app_label}.graphql.filters' + class_name = f'{model.__name__}Filter' + + try: + module = importlib.import_module(module_path) + except ModuleNotFoundError as exc: + # Treat both ".graphql.filters" absent and any missing + # parent (e.g. ".graphql" or "") as "no conventional + # filter class". Real ImportErrors from inside an existing + # filter module still propagate. + if exc.name == module_path or module_path.startswith(f'{exc.name}.'): + return None + raise + + return getattr(module, class_name, None) + + def _get_graphql_filter_field_names(self): + """ + Return the names exposed by the model's GraphQL filter input, sourced + only from the conventional .graphql.filters.Filter path. + """ + filter_class = self._get_model_graphql_filter_class() + if filter_class is None: + return set() + + return self._collect_filter_class_annotation_names(filter_class) + + @staticmethod + def _collect_filter_class_annotation_names(filter_class): + field_names = set() + for cls in reversed(getattr(filter_class, '__mro__', ())): + field_names.update( + field_name for field_name in getattr(cls, '__annotations__', {}) + if not field_name.startswith('_') + ) + return field_names + + def _assert_graphql_filter_class_present(self, filter_fields, handwritten_tests=()): + """ + Raise when the model has no discoverable filter class or the class + declares no fields. Skipped when auto-filter generation is disabled, + the per-model opt-out attribute is set, or hand-written (legacy or + explicit) filter tests are declared for the model. + """ + if handwritten_tests: + return + if not getattr(self, 'graphql_auto_filter_required', True): + return + if not getattr(self, 'graphql_auto_filter_tests', True): + return + + label = self.model._meta.label + path = f'{self.model._meta.app_label}.graphql.filters.{self.model.__name__}Filter' + + filter_class = self._get_model_graphql_filter_class() + self.assertIsNotNone( + filter_class, + f'No GraphQL filter class found for {label} at {path}. ' + f'Set graphql_auto_filter_required = False on this test case if intentional.' + ) + self.assertTrue( + filter_fields, + f'GraphQL filter class for {label} declares no fields. ' + f'Set graphql_auto_filter_required = False on this test case if intentional.' + ) + + def _get_nonempty_field_value(self, field): + queryset = self._get_queryset() + + if getattr(field, 'null', False): + queryset = queryset.exclude(**{f'{field.name}__isnull': True}) + + if isinstance(field, (models.CharField, models.TextField)): + queryset = queryset.exclude(**{field.name: ''}) + + return queryset.values_list(field.name, flat=True).first() + + def _get_model_field_for_filter_field(self, field_name): + """ + Find the Django model field matching a filter field name. Filter + fields are declared with either the model field name (e.g. `name`) + or the FK attname (e.g. `tenant_id`). + """ + for field in self.model._meta.fields: + if field.name == field_name or getattr(field, 'attname', None) == field_name: + return field + return None + + def _iter_filter_class_annotations(self, filter_class): + """ + Yield (field_name, annotation) pairs for the filter class, walking + its MRO so inherited fields surface. Subclass annotations override + inherited ones (private `_`-prefixed names are skipped). + """ + annotations = {} + for cls in reversed(filter_class.__mro__): + annotations.update({ + name: ann for name, ann in getattr(cls, '__annotations__', {}).items() + if not name.startswith('_') + }) + yield from annotations.items() + + @staticmethod + def _unwrap_filter_annotation(annotation): + """ + Strip ``X | None`` / ``Optional[X]`` and ``Annotated[X, ...]`` + layers. Resolve `strawberry.lazy('...')` metadata so lazily-annotated + lookup types (e.g. ``Annotated['FloatLookup', strawberry.lazy('mod')] | None``) + are returned as the actual class. When an ``Annotated`` layer carries + multiple metadata entries, the first ``module``-bearing entry wins. + Returns None when the inner type cannot be resolved. + """ + if annotation is None: + return None + + lazy_module = None + # Cap iterations at 8: typical NetBox annotations nest at most 3 layers + # (Union > Annotated > ForwardRef). 8 is a generous safety net to + # prevent infinite loops on pathological / future annotation shapes. + for _ in range(8): + origin = typing.get_origin(annotation) + args = typing.get_args(annotation) + + if origin in (typing.Union, types.UnionType): + non_none = [a for a in args if a is not type(None)] + if len(non_none) != 1: + return None + annotation = non_none[0] + continue + + if hasattr(annotation, '__metadata__'): + for meta in annotation.__metadata__: + module_name = getattr(meta, 'module', None) + if module_name: + lazy_module = module_name + break + inner = args[0] if args else None + if inner is None: + return None + annotation = inner + continue + + break + + if isinstance(annotation, (str, typing.ForwardRef)): + if lazy_module is None: + return None + name = annotation.__forward_arg__ if isinstance(annotation, typing.ForwardRef) else annotation + try: + return import_string(f'{lazy_module}.{name}') + except ImportError: + return None + + return annotation + + @classmethod + def _classify_filter_annotation(cls, annotation): + """ + Resolve a filter field annotation to a (kind, kind_arg) tuple keyed + on the declared GraphQL lookup type. Returns (None, None) for + annotations the dispatcher does not handle (those fields are + silently skipped). + """ + annotation = cls._unwrap_filter_annotation(annotation) + if annotation is None or isinstance(annotation, str): + return None, None + + if annotation is strawberry.ID: + return 'id', None + + origin = typing.get_origin(annotation) + target = origin if isinstance(origin, type) else annotation + type_args = typing.get_args(annotation) + + if not isinstance(target, type): + return None, None + + if target in (IntegerLookup, BigIntegerLookup, FloatLookup): + return 'numeric', target + + # TreeNodeFilter schema requires {id, match_type}; skip auto-emit. + if target is TreeNodeFilter: + return None, None + + if issubclass(target, (DateFilterLookup, DatetimeFilterLookup, TimeFilterLookup)): + return 'date_lookup', None + + if target is RangeLookup or issubclass(target, RangeLookup): + return 'range_lookup', type_args[0] if type_args else None + + if issubclass(target, ArrayLookup): + return 'array_lookup', None + if target is IntegerRangeArrayLookup or issubclass(target, IntegerRangeArrayLookup): + return 'range_array_lookup', None + if target is JSONFilter: + # JSONFilter requires explicit (path, typed lookup); no general auto shape. + return None, None + + if issubclass(target, StrFilterLookup): + return 'str_lookup', None + if issubclass(target, ComparisonFilterLookup): + return 'comparison_lookup', type_args[0] if type_args else None + if issubclass(target, FilterLookup): + return 'filter_lookup', type_args[0] if type_args else None + # Enum-typed BaseFilterLookup needs an enum literal; skip auto-emit. + if issubclass(target, BaseFilterLookup): + return None, None + + return None, None + + def _emit_id_filter_tests(self, field_name, _kind_arg): + if field_name == 'id': + instance = self._get_queryset().first() + if instance is None: + return + yield GraphQLFilterTest( + name='id__exact', + filters=f'id: {self._graphql_literal(str(instance.pk))}', + expected=lambda qs, pk=instance.pk: qs.filter(pk=pk), + ) + return + + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None or not isinstance(model_field, models.ForeignKey): + return + queryset = self._get_queryset().exclude(**{f'{model_field.name}__isnull': True}) + value = queryset.values_list(model_field.attname, flat=True).first() + if value is None: + return + yield GraphQLFilterTest( + name=f'{field_name}__exact', + filters=f'{field_name}: {self._graphql_literal(str(value))}', + expected=lambda qs, attname=model_field.attname, v=value: qs.filter(**{attname: v}), + ) + + def _emit_str_lookup_filter_tests(self, field_name, _kind_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + value = self._get_nonempty_field_value(model_field) + if value in (None, ''): + return + value = str(value) + token = max(1, min(3, len(value))) + lookups = ( + ('exact', 'exact', value), + ('i_contains', 'icontains', value[:token]), + ('i_starts_with', 'istartswith', value[:token]), + ('i_ends_with', 'iendswith', value[-token:]), + ) + for lookup, orm_lookup, filter_value in lookups: + yield GraphQLFilterTest( + name=f'{field_name}__{lookup}', + filters=f'{field_name}: {{{lookup}: {self._graphql_literal(filter_value)}}}', + expected=( + lambda qs, fn=model_field.name, ol=orm_lookup, v=filter_value: + qs.filter(**{f'{fn}__{ol}': v}) + ), + ) + + def _emit_filter_lookup_filter_tests(self, field_name, type_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + value = self._get_nonempty_field_value(model_field) + if value is None: + return + if type_arg is bool or isinstance(value, bool): + yield GraphQLFilterTest( + name=f'{field_name}__exact', + filters=f'{field_name}: {{exact: {self._graphql_literal(value)}}}', + expected=lambda qs, fn=model_field.name, v=value: qs.filter(**{fn: v}), + ) + return + yield GraphQLFilterTest( + name=f'{field_name}__exact', + filters=f'{field_name}: {{exact: {self._graphql_literal(value)}}}', + expected=lambda qs, fn=model_field.name, v=value: qs.filter(**{f'{fn}__exact': v}), + ) + + def _emit_comparison_lookup_filter_tests(self, field_name, _type_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + value = self._get_nonempty_field_value(model_field) + if value is None: + return + yield GraphQLFilterTest( + name=f'{field_name}__exact', + filters=f'{field_name}: {{exact: {self._graphql_literal(value)}}}', + expected=lambda qs, fn=model_field.name, v=value: qs.filter(**{f'{fn}__exact': v}), + ) + + def _emit_numeric_filter_tests(self, field_name, _type_arg): + # NetBox numeric wrapper: {filter_lookup: {exact: N}}. + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + if isinstance(model_field, ArrayField): + return + value = self._get_nonempty_field_value(model_field) + if value is None: + return + if isinstance(value, Decimal): + value = float(value) + yield GraphQLFilterTest( + name=f'{field_name}__filter_lookup__exact', + filters=( + f'{field_name}: {{filter_lookup: ' + f'{{exact: {self._graphql_literal(value)}}}}}' + ), + expected=lambda qs, fn=model_field.name, v=value: qs.filter(**{f'{fn}__exact': v}), + ) + + def _emit_date_lookup_filter_tests(self, field_name, _kind_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + value = self._get_nonempty_field_value(model_field) + if value is None: + return + iso_value = value.isoformat() if hasattr(value, 'isoformat') else str(value) + yield GraphQLFilterTest( + name=f'{field_name}__exact', + filters=f'{field_name}: {{exact: "{iso_value}"}}', + expected=lambda qs, fn=model_field.name, v=value: qs.filter(**{fn: v}), + ) + + def _emit_range_lookup_filter_tests(self, field_name, _kind_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + aggregates = self._get_queryset().aggregate( + _min=models.Min(model_field.name), _max=models.Max(model_field.name), + ) + start, end = aggregates['_min'], aggregates['_max'] + if start is None or end is None or start == end: + return + yield GraphQLFilterTest( + name=f'{field_name}__range_lookup', + filters=( + f'{field_name}: {{range_lookup: ' + f'{{start: {self._graphql_literal(start)}, end: {self._graphql_literal(end)}}}}}' + ), + expected=( + lambda qs, fn=model_field.name, lo=start, hi=end: + qs.filter(**{f'{fn}__gte': lo, f'{fn}__lte': hi}) + ), + ) + + def _emit_array_lookup_filter_tests(self, field_name, _kind_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + if not isinstance(model_field, ArrayField): + return + queryset = self._get_queryset().exclude(**{field_name: []}) + sample = queryset.values_list(field_name, flat=True).first() + if not sample: + return + element = sample[0] + yield GraphQLFilterTest( + name=f'{field_name}__contains', + filters=( + f'{field_name}: {{contains: [{self._graphql_literal(element)}]}}' + ), + expected=( + lambda qs, fn=model_field.name, v=element: qs.filter(**{f'{fn}__contains': [v]}) + ), + ) + + def _emit_range_array_lookup_filter_tests(self, field_name, _kind_arg): + model_field = self._get_model_field_for_filter_field(field_name) + if model_field is None: + return + queryset = self._get_queryset().exclude(**{f'{field_name}__isnull': True}) + sample = queryset.values_list(field_name, flat=True).first() + if not sample: + return + first_range = sample[0] + lower = getattr(first_range, 'lower', None) + if lower is None: + return + yield GraphQLFilterTest( + name=f'{field_name}__contains', + filters=f'{field_name}: {{contains: {self._graphql_literal(lower)}}}', + expected=( + lambda qs, fn=model_field.name, v=lower: qs.filter(**{f'{fn}__range_contains': v}) + ), + ) + + def _iter_auto_graphql_filter_tests(self): + if not getattr(self, 'graphql_auto_filter_tests', True): + return + + filter_class = self._get_model_graphql_filter_class() + if filter_class is None: + return + + exclude = set(getattr(self, 'graphql_auto_filter_exclude', ())) + per_kind = self.graphql_auto_filter_fields_per_kind + + # Bucket eligible fields by lookup kind so per-kind budgeting balances coverage. + by_kind: dict[str, list[tuple[str, object]]] = {} + for field_name, annotation in self._iter_filter_class_annotations(filter_class): + if field_name in exclude: + continue + kind, kind_arg = self._classify_filter_annotation(annotation) + if kind is None: + continue + by_kind.setdefault(kind, []).append((field_name, kind_arg)) + + # Emit per-kind; the cap counts SUCCESSFUL emissions, not candidate fields, so + # early null/empty fields don't shadow later fields with usable fixture data. + for kind, fields in by_kind.items(): + emitter = getattr(self, f'_emit_{kind}_filter_tests', None) + if emitter is None: + continue + + emitted_fields = 0 + for field_name, kind_arg in fields: + tests = list(emitter(field_name, kind_arg)) + if not tests: + continue + yield from tests + emitted_fields += 1 + if emitted_fields >= per_kind: + break + + def _iter_legacy_graphql_filter_tests(self): + if not hasattr(self, 'graphql_filter'): + return + + filter_expressions = [ + f'{field_name}: {self._render_graphql_filter_value(params)}' + for field_name, params in self.graphql_filter.items() + ] + + yield GraphQLFilterTest( + name='graphql_filter', + filters=', '.join(filter_expressions), + ) + + def _coerce_graphql_filter_test(self, filter_test): + if isinstance(filter_test, GraphQLFilterTest): + return filter_test + + filter_test = dict(filter_test) + if 'filter' in filter_test and 'filters' not in filter_test: + filter_test['filters'] = filter_test.pop('filter') + + return GraphQLFilterTest(**filter_test) + + def _iter_explicit_graphql_filter_tests(self): + for filter_test in getattr(self, 'graphql_filter_tests', ()): + yield self._coerce_graphql_filter_test(filter_test) + + def _get_expected_id_set(self, filter_test): + expected = filter_test.expected + + if callable(expected): + expected = expected(self._get_queryset()) + + if isinstance(expected, dict): + expected = self._get_queryset().filter(**expected) + + if hasattr(expected, 'values_list'): + values = expected.distinct().values_list('pk', flat=True) + else: + values = [getattr(value, 'pk', value) for value in expected] + + return {str(value) for value in values} + + def _assert_graphql_filter_test(self, url, field_name, filter_test): + query = self._build_graphql_id_list_query(field_name, filter_test.filters) + + for permission in filter_test.permissions: + self.add_permissions(permission) + + response = self.client.post(url, data={'query': query}, format="json", **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + data = json.loads(response.content) + self.assertNotIn('errors', data) + + results = data['data'][field_name] + + if filter_test.expected is None: + self.assertGreater(len(results), 0) + return + + expected_ids = self._get_expected_id_set(filter_test) + + self.assertGreater( + len(expected_ids), 0, + msg=( + f'{self.model._meta.label}: filter "{filter_test.name}" produced an empty ' + f'expected set; the test would tautologically pass. Adjust fixtures or the ' + f'filter so the expected ORM queryset is non-empty.' + ), + ) + + if self._graphql_type_exposes_id(): + result_ids = [str(result['id']) for result in results] + self.assertEqual( + set(result_ids), expected_ids, + msg=f'{self.model._meta.label}: filter "{filter_test.name}" ID set mismatch', + ) + + self.assertEqual( + len(results), len(expected_ids), + msg=( + f'{self.model._meta.label}: filter "{filter_test.name}" result count mismatch ' + f'(GraphQL type does not expose id; comparing by length).' + ), + ) + + def _coerce_graphql_query_test(self, query_test): + if isinstance(query_test, GraphQLQueryTest): + return query_test + + query_test = dict(query_test) + if 'assertion' in query_test and 'assert_result' not in query_test: + query_test['assert_result'] = query_test.pop('assertion') + + return GraphQLQueryTest(**query_test) + def _build_query(self, name, **filters): """ Create a normal query - unfiltered or with a string query: i.e. site(name: "aaa"){. @@ -740,14 +1424,44 @@ class APIViewTestCases: self.assertNotIn('errors', data) self.assertEqual(len(data['data'][field_name]), self.model.objects.count()) + def _assert_graphql_filter_tests_exist(self, auto_tests, legacy_tests, explicit_tests): + """ + Fail loudly when auto mode is required and no GraphQL filter tests + (auto, legacy, or explicit) exist for the current model. + """ + if ( + getattr(self, 'graphql_auto_filter_tests', True) + and getattr(self, 'graphql_auto_filter_required', True) + and not auto_tests + and not legacy_tests + and not explicit_tests + ): + self.fail( + f'No GraphQL filter tests were generated for {self.model._meta.label}. ' + f'Set graphql_auto_filter_required = False or add explicit graphql_filter_tests ' + f'if intentional.' + ) + @override_settings(LOGIN_REQUIRED=True) def test_graphql_filter_objects(self): - if not hasattr(self, 'graphql_filter'): + legacy_tests = list(self._iter_legacy_graphql_filter_tests()) + explicit_tests = list(self._iter_explicit_graphql_filter_tests()) + + filter_fields = self._get_graphql_filter_field_names() + self._assert_graphql_filter_class_present( + filter_fields, handwritten_tests=[*legacy_tests, *explicit_tests] + ) + + auto_tests = list(self._iter_auto_graphql_filter_tests()) + + self._assert_graphql_filter_tests_exist(auto_tests, legacy_tests, explicit_tests) + + filter_tests = [*auto_tests, *legacy_tests, *explicit_tests] + if not filter_tests: return url = reverse('graphql') field_name = f'{self._get_graphql_base_name()}_list' - query = self._build_filtered_query(field_name, **self.graphql_filter) # Add object-level permission obj_perm = ObjectPermission( @@ -758,11 +1472,43 @@ class APIViewTestCases: obj_perm.users.add(self.user) obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) - response = self.client.post(url, data={'query': query}, format="json", **self.header) - self.assertHttpStatus(response, status.HTTP_200_OK) - data = json.loads(response.content) - self.assertNotIn('errors', data) - self.assertGreater(len(data['data'][field_name]), 0) + for filter_test in filter_tests: + with self.subTest(filter=filter_test.name): + self._assert_graphql_filter_test(url, field_name, filter_test) + + @override_settings(LOGIN_REQUIRED=True) + def test_graphql_extra_queries(self): + query_tests = [ + self._coerce_graphql_query_test(query_test) + for query_test in getattr(self, 'graphql_query_tests', ()) + ] + + if not query_tests: + return + + url = reverse('graphql') + + # Add object-level permission for this model. Additional permissions + # required by the query can be declared on the GraphQLQueryTest. + obj_perm = ObjectPermission( + name='Test permission', + actions=['view'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) + + for query_test in query_tests: + with self.subTest(query=query_test.name): + for permission in query_test.permissions: + self.add_permissions(permission) + + response = self.client.post(url, data={'query': query_test.query}, format="json", **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + data = json.loads(response.content) + self.assertNotIn('errors', data) + query_test.assert_result(self, data['data']) class APIViewTestCase( GetObjectViewTestCase, diff --git a/netbox/utilities/tests/test_api_graphql.py b/netbox/utilities/tests/test_api_graphql.py new file mode 100644 index 000000000..a9a368a28 --- /dev/null +++ b/netbox/utilities/tests/test_api_graphql.py @@ -0,0 +1,260 @@ +"""Tests for the GraphQL filter test framework in utilities/testing/api.py.""" + +import sys +import types +from decimal import Decimal +from typing import Annotated + +import strawberry +from django.test import TestCase + +from netbox.graphql.filter_lookups import FloatLookup +from utilities.testing.api import APIViewTestCases, GraphQLFilterTest + + +class GraphQLFilterAnnotationMROTestCase(TestCase): + """Cover MRO override, import error propagation, lazy annotation resolution, and the zero-auto-test gate.""" + + def test_subclass_annotation_overrides_base(self): + """Subclass annotations win over base in `_iter_filter_class_annotations`.""" + class Base: + __annotations__ = {'shared': int} + + class Sub(Base): + __annotations__ = {'shared': str} + + # Stand up a throwaway instance just to access the method as bound. + instance = APIViewTestCases.GraphQLTestCase() + pairs = dict(instance._iter_filter_class_annotations(Sub)) + self.assertEqual(pairs['shared'], str) + + def test_get_filter_class_propagates_real_import_errors(self): + """A broken import inside a model's filters module must surface, not silently return None.""" + broken = types.ModuleType('netbox_broken_filter_fixture.graphql.filters') + + def _raise(*args, **kwargs): + raise ImportError('simulated downstream breakage') + + broken.__getattr__ = _raise + # sys.modules mutation is safe under --parallel (separate processes), not threads. + sys.modules['netbox_broken_filter_fixture'] = types.ModuleType('netbox_broken_filter_fixture') + sys.modules['netbox_broken_filter_fixture.graphql'] = types.ModuleType( + 'netbox_broken_filter_fixture.graphql' + ) + sys.modules['netbox_broken_filter_fixture.graphql.filters'] = broken + + try: + class FakeMeta: + app_label = 'netbox_broken_filter_fixture' + + class FakeModel: + _meta = FakeMeta() + __name__ = 'BrokenModel' + + instance = APIViewTestCases.GraphQLTestCase() + with self.assertRaises(ImportError): + instance._get_model_graphql_filter_class(FakeModel) + finally: + for key in ( + 'netbox_broken_filter_fixture', + 'netbox_broken_filter_fixture.graphql', + 'netbox_broken_filter_fixture.graphql.filters', + ): + sys.modules.pop(key, None) + + def test_zero_auto_filter_tests_fails_loudly(self): + """Helper fails when auto mode is required and no tests of any kind exist.""" + + class FakeMeta: + label = 'fake.FakeModel' + + class FakeModel: + _meta = FakeMeta() + + class Case(APIViewTestCases.GraphQLTestCase): + model = FakeModel + graphql_auto_filter_tests = True + graphql_auto_filter_required = True + + case = Case() + with self.assertRaisesRegex(AssertionError, r'No GraphQL filter tests.*fake\.FakeModel'): + case._assert_graphql_filter_tests_exist([], [], []) + + def test_lazy_annotated_lookup_resolves(self): + """Annotated['FloatLookup', strawberry.lazy(...)] | None resolves to FloatLookup.""" + annotation = Annotated['FloatLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None + self.assertIs( + APIViewTestCases.GraphQLTestCase._unwrap_filter_annotation(annotation), + FloatLookup, + ) + + def test_str_lookup_emits_all_four_variants(self): + """`_emit_str_lookup_filter_tests` emits exact, i_contains, i_starts_with, i_ends_with.""" + captured_value = 'production' + + class FakeMeta: + label = 'fake.FakeModel' + app_label = 'fake' + + class Case(APIViewTestCases.GraphQLTestCase): + def _get_model_field_for_filter_field(self, field_name): + class FakeField: + name = field_name + return FakeField() + + def _get_nonempty_field_value(self, field): + return captured_value + + def _graphql_literal(self, value): + return f'"{value}"' + + case = Case() + tests = list(case._emit_str_lookup_filter_tests('name', None)) + self.assertEqual( + [t.name for t in tests], + ['name__exact', 'name__i_contains', 'name__i_starts_with', 'name__i_ends_with'], + ) + + def test_explicit_tests_satisfy_auto_required_gate(self): + """Helper does NOT fail when explicit tests exist, even if auto/legacy are empty.""" + + class FakeMeta: + label = 'fake.FakeModel' + + class FakeModel: + _meta = FakeMeta() + + class Case(APIViewTestCases.GraphQLTestCase): + model = FakeModel + graphql_auto_filter_tests = True + graphql_auto_filter_required = True + + case = Case() + # Should not raise. + case._assert_graphql_filter_tests_exist( + auto_tests=[], + legacy_tests=[], + explicit_tests=[GraphQLFilterTest(name='x', filters='x: 1')], + ) + + def test_graphql_literal_renders_lists(self): + """List and tuple values render as GraphQL list literals, not quoted strings.""" + literal = APIViewTestCases.GraphQLTestCase._graphql_literal + self.assertEqual(literal([1, 2, 3]), '[1, 2, 3]') + self.assertEqual(literal(('a', 'b')), '["a", "b"]') + self.assertEqual(literal([]), '[]') + + def test_graphql_literal_renders_decimal_as_number(self): + """Decimal values render as numeric literals, not quoted strings.""" + literal = APIViewTestCases.GraphQLTestCase._graphql_literal + self.assertEqual(literal(Decimal('1.23')), '1.23') + self.assertEqual(literal([Decimal('1.5'), Decimal('2.5')]), '[1.5, 2.5]') + + def test_per_kind_cap_counts_successful_emissions(self): + """Later candidate fields are tried until per-kind successful emissions reach the cap.""" + + class FakeMeta: + label = 'fake.FakeModel' + app_label = 'fake' + + class FakeModel: + _meta = FakeMeta() + __name__ = 'FakeModel' + + emit_calls = [] + + class Case(APIViewTestCases.GraphQLTestCase): + model = FakeModel + graphql_auto_filter_fields_per_kind = 2 + + def _get_model_graphql_filter_class(self, model=None): + class FilterClass: + __annotations__ = { + 'empty_field_1': str, + 'empty_field_2': str, + 'useful_field_1': str, + 'useful_field_2': str, + 'useful_field_3': str, + } + return FilterClass + + def _classify_filter_annotation(self, annotation): + return 'str_lookup', None + + def _emit_str_lookup_filter_tests(self, field_name, _kind_arg): + emit_calls.append(field_name) + if field_name.startswith('empty_'): + return iter(()) + return iter((GraphQLFilterTest(name=field_name, filters=f'{field_name}: "x"'),)) + + case = Case() + list(case._iter_auto_graphql_filter_tests()) + + # The candidate-counting bug stops at 'empty_field_1' and 'empty_field_2' (the slice + # captures the first 2). After the fix, the emitter is invoked on all 5 candidates + # in order until 2 SUCCESSFUL fields have emitted. + self.assertEqual( + emit_calls, + ['empty_field_1', 'empty_field_2', 'useful_field_1', 'useful_field_2'], + ) + + def test_get_filter_class_returns_none_when_parent_module_missing(self): + """When the parent `.graphql` package is absent, return None instead of re-raising.""" + + class FakeMeta: + app_label = 'netbox_missing_graphql_fixture' + + class FakeModel: + _meta = FakeMeta() + __name__ = 'BrokenModel' + + instance = APIViewTestCases.GraphQLTestCase() + # No `netbox_missing_graphql_fixture` package is registered in sys.modules, + # so import_module raises ModuleNotFoundError with exc.name == 'netbox_missing_graphql_fixture' + # (the parent), not the full path 'netbox_missing_graphql_fixture.graphql.filters'. + # The fix accepts both shapes. + self.assertIsNone(instance._get_model_graphql_filter_class(FakeModel)) + + def test_filter_class_assertion_skipped_with_handwritten_tests(self): + """Hand-written tests exempt a model from the conventional filter class requirement.""" + + class FakeMeta: + label = 'fake.FakeModel' + app_label = 'fake' + + class FakeModel: + _meta = FakeMeta() + __name__ = 'FakeModel' + + class Case(APIViewTestCases.GraphQLTestCase): + model = FakeModel + + def _get_model_graphql_filter_class(self, model=None): + return None + + case = Case() + # Should not raise despite the missing conventional filter class. + case._assert_graphql_filter_class_present( + set(), handwritten_tests=[GraphQLFilterTest(name='x', filters='x: 1')] + ) + + def test_filter_class_assertion_fails_without_filter_class(self): + """Missing conventional filter class raises when no hand-written tests exist.""" + + class FakeMeta: + label = 'fake.FakeModel' + app_label = 'fake' + + class FakeModel: + _meta = FakeMeta() + __name__ = 'FakeModel' + + class Case(APIViewTestCases.GraphQLTestCase): + model = FakeModel + + def _get_model_graphql_filter_class(self, model=None): + return None + + case = Case() + with self.assertRaisesRegex(AssertionError, r'No GraphQL filter class found for fake\.FakeModel'): + case._assert_graphql_filter_class_present(set())