diff --git a/netbox/ipam/tests/test_views.py b/netbox/ipam/tests/test_views.py index 10932ce72..5505d2a5c 100644 --- a/netbox/ipam/tests/test_views.py +++ b/netbox/ipam/tests/test_views.py @@ -1,6 +1,7 @@ import datetime from django.contrib.contenttypes.models import ContentType +from django.test import RequestFactory from django.urls import reverse from netaddr import IPNetwork @@ -8,8 +9,10 @@ from core.choices import ObjectChangeActionChoices from core.models import ObjectChange, ObjectType from dcim.constants import InterfaceTypeChoices from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site +from extras.models import SavedFilter from ipam.choices import * from ipam.models import * +from ipam.views import AggregatePrefixesView from netbox.choices import CSVDelimiterChoices, ImportFormatChoices from tenancy.models import Tenant from users.models import ObjectPermission @@ -353,6 +356,101 @@ class AggregateTestCase(ViewTestCases.PrimaryObjectViewTestCase): url = reverse('ipam:aggregate_prefixes', kwargs={'pk': aggregate.pk}) self.assertHttpStatus(self.client.get(url), 200) + def test_aggregate_prefixes_filter_suppresses_available_prefixes(self): + self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix') + + tenants = ( + Tenant(name='Aggregate Tenant 1', slug='aggregate-tenant-1'), + Tenant(name='Aggregate Tenant 2', slug='aggregate-tenant-2'), + ) + Tenant.objects.bulk_create(tenants) + + aggregate = Aggregate.objects.create( + prefix=IPNetwork('203.0.113.0/24'), + rir=RIR.objects.first() + ) + prefixes = ( + Prefix(prefix=IPNetwork('203.0.113.0/26'), tenant=tenants[0]), + Prefix(prefix=IPNetwork('203.0.113.64/26'), tenant=tenants[1]), + ) + Prefix.objects.bulk_create(prefixes) + + url = reverse('ipam:aggregate_prefixes', kwargs={'pk': aggregate.pk}) + response = self.client.get(url, {'tenant_id': tenants[0].pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, '203.0.113.0/26') + self.assertNotContains(response, '203.0.113.64/26') + + def test_aggregate_prefixes_saved_filter(self): + self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix') + + tenants = ( + Tenant(name='Aggregate Saved Tenant 1', slug='aggregate-saved-tenant-1'), + Tenant(name='Aggregate Saved Tenant 2', slug='aggregate-saved-tenant-2'), + ) + Tenant.objects.bulk_create(tenants) + + aggregate = Aggregate.objects.create( + prefix=IPNetwork('203.0.114.0/24'), + rir=RIR.objects.first() + ) + prefixes = ( + Prefix(prefix=IPNetwork('203.0.114.0/26'), tenant=tenants[0]), + Prefix(prefix=IPNetwork('203.0.114.64/26'), tenant=tenants[1]), + ) + Prefix.objects.bulk_create(prefixes) + + saved_filter = SavedFilter.objects.create( + name='Aggregate Tenant 1 prefixes', + slug='aggregate-tenant-1-prefixes', + parameters={ + 'tenant_id': [str(tenants[0].pk)], + }, + ) + saved_filter.object_types.add(ObjectType.objects.get_for_model(Prefix)) + + url = reverse('ipam:aggregate_prefixes', kwargs={'pk': aggregate.pk}) + response = self.client.get(url, {'filter_id': saved_filter.pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, '203.0.114.0/26') + self.assertNotContains(response, '203.0.114.64/26') + + def test_children_are_filtered_fallback(self): + """_children_are_filtered() rebuilds the queryset when prep_table_data() has not cached a result.""" + self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix') + + aggregate = Aggregate.objects.create( + prefix=IPNetwork('203.0.115.0/24'), + rir=RIR.objects.first() + ) + tenant = Tenant.objects.create(name='Aggregate Fallback Tenant', slug='aggregate-fallback-tenant') + Prefix.objects.create(prefix=IPNetwork('203.0.115.0/26'), tenant=tenant) + Prefix.objects.create(prefix=IPNetwork('203.0.115.64/26')) + + # No cached value: the fallback path rebuilds the filtered queryset and detects the filter. + view = AggregatePrefixesView() + request = RequestFactory().get('/', {'tenant_id': tenant.pk}) + request.user = self.user + self.assertFalse(hasattr(view, '_child_queryset_is_filtered')) + self.assertTrue(view._children_are_filtered(request, aggregate)) + + # No cached value and no filter: the fallback path reports no filtering. + view = AggregatePrefixesView() + request = RequestFactory().get('/') + request.user = self.user + self.assertFalse(view._children_are_filtered(request, aggregate)) + + # A cached value takes precedence over the actual request state. + view = AggregatePrefixesView() + view._set_children_filtered(False) + request = RequestFactory().get('/', {'tenant_id': tenant.pk}) + request.user = self.user + self.assertFalse(view._children_are_filtered(request, aggregate)) + class RoleTestCase(ViewTestCases.OrganizationalObjectViewTestCase): model = Role @@ -588,6 +686,63 @@ class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase): url = reverse('ipam:prefix_prefixes', kwargs={'pk': prefixes[0].pk}) self.assertHttpStatus(self.client.get(url), 200) + def test_prefix_prefixes_filter_suppresses_available_prefixes(self): + self.add_permissions('ipam.view_prefix') + + tenants = ( + Tenant(name='Prefix Tenant 1', slug='prefix-tenant-1'), + Tenant(name='Prefix Tenant 2', slug='prefix-tenant-2'), + ) + Tenant.objects.bulk_create(tenants) + + parent = Prefix.objects.create(prefix=IPNetwork('198.51.100.0/24')) + prefixes = ( + Prefix(prefix=IPNetwork('198.51.100.0/26'), tenant=tenants[0]), + Prefix(prefix=IPNetwork('198.51.100.64/26'), tenant=tenants[1]), + ) + Prefix.objects.bulk_create(prefixes) + + url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk}) + response = self.client.get(url, {'tenant_id': tenants[0].pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, '198.51.100.0/26') + self.assertNotContains(response, '198.51.100.64/26') + + def test_prefix_prefixes_saved_filter_suppresses_available_prefixes(self): + self.add_permissions('ipam.view_prefix') + + tenants = ( + Tenant(name='Prefix Saved Tenant 1', slug='prefix-saved-tenant-1'), + Tenant(name='Prefix Saved Tenant 2', slug='prefix-saved-tenant-2'), + ) + Tenant.objects.bulk_create(tenants) + + parent = Prefix.objects.create(prefix=IPNetwork('198.51.101.0/24')) + prefixes = ( + Prefix(prefix=IPNetwork('198.51.101.0/26'), tenant=tenants[0]), + Prefix(prefix=IPNetwork('198.51.101.64/26'), tenant=tenants[1]), + ) + Prefix.objects.bulk_create(prefixes) + + saved_filter = SavedFilter.objects.create( + name='Prefix Tenant 1 prefixes', + slug='prefix-tenant-1-prefixes', + parameters={ + 'tenant_id': [str(tenants[0].pk)], + }, + ) + saved_filter.object_types.add(ObjectType.objects.get_for_model(Prefix)) + + url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk}) + response = self.client.get(url, {'filter_id': saved_filter.pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, '198.51.101.0/26') + self.assertNotContains(response, '198.51.101.64/26') + def test_prefix_ipranges(self): self.add_permissions('ipam.view_prefix', 'ipam.view_iprange') prefix = Prefix.objects.create(prefix=IPNetwork('192.168.0.0/16')) @@ -616,6 +771,89 @@ class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase): url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk}) self.assertHttpStatus(self.client.get(url), 200) + def test_prefix_ipaddresses_filter(self): + self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange') + + tenants = ( + Tenant(name='IP Address Tenant 1', slug='ip-address-tenant-1'), + Tenant(name='IP Address Tenant 2', slug='ip-address-tenant-2'), + ) + Tenant.objects.bulk_create(tenants) + + prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24')) + ip_addresses = ( + IPAddress(address=IPNetwork('192.0.2.1/24'), tenant=tenants[0]), + IPAddress(address=IPNetwork('192.0.2.2/24'), tenant=tenants[1]), + ) + IPAddress.objects.bulk_create(ip_addresses) + + url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk}) + response = self.client.get(url, {'tenant_id': tenants[0].pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, '192.0.2.1/24') + self.assertNotContains(response, '192.0.2.2/24') + + def test_prefix_ipaddresses_saved_filter(self): + self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange') + + tenants = ( + Tenant(name='Saved Filter Tenant 1', slug='saved-filter-tenant-1'), + Tenant(name='Saved Filter Tenant 2', slug='saved-filter-tenant-2'), + ) + Tenant.objects.bulk_create(tenants) + + prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24')) + ip_addresses = ( + IPAddress(address=IPNetwork('192.0.2.1/24'), tenant=tenants[0]), + IPAddress(address=IPNetwork('192.0.2.2/24'), tenant=tenants[1]), + ) + IPAddress.objects.bulk_create(ip_addresses) + + saved_filter = SavedFilter.objects.create( + name='Tenant 1 IP addresses', + slug='tenant-1-ip-addresses', + parameters={ + 'tenant_id': [str(tenants[0].pk)], + }, + ) + saved_filter.object_types.add(ObjectType.objects.get_for_model(IPAddress)) + + url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk}) + response = self.client.get(url, {'filter_id': saved_filter.pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, '192.0.2.1/24') + self.assertNotContains(response, '192.0.2.2/24') + + def test_prefix_ipaddresses_unfiltered_shows_available_space(self): + """An unfiltered IP Addresses tab injects synthetic available-space rows.""" + self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange') + + prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29')) + IPAddress.objects.create(address=IPNetwork('192.0.2.1/29')) + + url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk}) + response = self.client.get(url) + + self.assertHttpStatus(response, 200) + self.assertGreater(len(response.context['table'].data), 1) + + def test_prefix_prefixes_unfiltered_shows_available_prefixes(self): + """An unfiltered Child Prefixes tab injects synthetic available-prefix rows.""" + self.add_permissions('ipam.view_prefix') + + parent = Prefix.objects.create(prefix=IPNetwork('198.51.102.0/24')) + Prefix.objects.create(prefix=IPNetwork('198.51.102.0/26')) + + url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk}) + response = self.client.get(url) + + self.assertHttpStatus(response, 200) + self.assertGreater(len(response.context['table'].data), 1) + def test_prefix_ipaddresses_with_single_address_range(self): self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange') # The IP Addresses tab annotates child IP addresses alongside any @@ -1130,6 +1368,70 @@ class VLANGroupTestCase(ViewTestCases.OrganizationalObjectViewTestCase): 'description': 'New description', } + def test_vlans_filter_suppresses_available_vlans(self): + self.add_permissions('ipam.view_vlangroup', 'ipam.view_vlan') + + group = VLANGroup.objects.create( + name='Filtered VLAN Group', + slug='filtered-vlan-group' + ) + vlans = ( + VLAN(group=group, vid=100, name='VLAN100'), + VLAN(group=group, vid=200, name='VLAN200'), + ) + VLAN.objects.bulk_create(vlans) + + url = reverse('ipam:vlangroup_vlans', kwargs={'pk': group.pk}) + response = self.client.get(url, {'vid': 100}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, 'VLAN100') + self.assertNotContains(response, 'VLAN200') + + def test_vlans_saved_filter_suppresses_available_vlans(self): + self.add_permissions('ipam.view_vlangroup', 'ipam.view_vlan') + + group = VLANGroup.objects.create( + name='Saved Filter VLAN Group', + slug='saved-filter-vlan-group' + ) + vlans = ( + VLAN(group=group, vid=100, name='VLAN100'), + VLAN(group=group, vid=200, name='VLAN200'), + ) + VLAN.objects.bulk_create(vlans) + + saved_filter = SavedFilter.objects.create( + name='VLAN 100', + slug='vlan-100', + parameters={ + 'vid': ['100'], + }, + ) + saved_filter.object_types.add(ObjectType.objects.get_for_model(VLAN)) + + url = reverse('ipam:vlangroup_vlans', kwargs={'pk': group.pk}) + response = self.client.get(url, {'filter_id': saved_filter.pk}) + + self.assertHttpStatus(response, 200) + self.assertEqual(len(response.context['table'].data), 1) + self.assertContains(response, 'VLAN100') + self.assertNotContains(response, 'VLAN200') + + def test_vlans_unfiltered_shows_available_vlans(self): + """An unfiltered VLANs tab injects synthetic available-VLAN rows.""" + self.add_permissions('ipam.view_vlangroup', 'ipam.view_vlan') + + group = VLANGroup.objects.create(name='Unfiltered VLAN Group', slug='unfiltered-vlan-group') + VLAN.objects.create(group=group, vid=1, name='VLAN0001') + + url = reverse('ipam:vlangroup_vlans', kwargs={'pk': group.pk}) + response = self.client.get(url) + + self.assertHttpStatus(response, 200) + self.assertGreater(len(response.context['table'].data), 1) + class VLANTestCase(ViewTestCases.PrimaryObjectViewTestCase): model = VLAN diff --git a/netbox/ipam/views.py b/netbox/ipam/views.py index 41f5a1832..98560cc8b 100644 --- a/netbox/ipam/views.py +++ b/netbox/ipam/views.py @@ -547,8 +547,64 @@ class AggregateView(generic.ObjectView): ) +class ChildAvailabilityMixin: + """ + Mixin for ObjectChildrenView subclasses that render synthetic "available" rows + (available IP space, prefixes, or VLANs) and must suppress them when the child + queryset has been narrowed by a direct or saved filter. + """ + + @staticmethod + def _where_signature(queryset): + # query.where is Django-internal, but it is the closest signal for "narrowed by a filter". + return str(queryset.query.where) + + def _set_children_filtered(self, is_filtered): + self._child_queryset_is_filtered = is_filtered + return is_filtered + + def _queryset_is_filtered(self, request, queryset, parent): + """ + Return True if the filtered child queryset differs from the unfiltered one. + + Compares WHERE clauses rather than testing queryset.query.where for truthiness, + because child querysets are already scoped to their parent object and carry WHERE + clauses before any user filter is applied. The result is cached on the view instance + so get_extra_context() can reuse it without rebuilding the queryset. + """ + if self.filterset is None: + return self._set_children_filtered(False) + + unfiltered = self.get_children(request, parent) + + return self._set_children_filtered( + self._where_signature(queryset) != self._where_signature(unfiltered) + ) + + def _children_are_filtered(self, request, parent): + """ + Return whether child objects are filtered. + + In the normal ObjectChildrenView flow prep_table_data() runs first and caches the + result, so this returns the cached value. Fall back to rebuilding the queryset for + direct calls where prep_table_data() has not run. + """ + if hasattr(self, '_child_queryset_is_filtered'): + return self._child_queryset_is_filtered + + if self.filterset is None: + return self._set_children_filtered(False) + + unfiltered = self.get_children(request, parent) + filtered = self.filterset(request.GET, unfiltered, request=request).qs + + return self._set_children_filtered( + self._where_signature(filtered) != self._where_signature(unfiltered) + ) + + @register_model_view(Aggregate, 'prefixes') -class AggregatePrefixesView(generic.ObjectChildrenView): +class AggregatePrefixesView(ChildAvailabilityMixin, generic.ObjectChildrenView): queryset = Aggregate.objects.all() child_model = Prefix table = tables.PrefixTable @@ -572,13 +628,21 @@ class AggregatePrefixesView(generic.ObjectChildrenView): show_available = bool(request.GET.get('show_available', 'true') == 'true') show_assigned = bool(request.GET.get('show_assigned', 'true') == 'true') + if self._queryset_is_filtered(request, queryset, parent): + show_available = False + return add_requested_prefixes(parent.prefix, queryset, show_available, show_assigned) def get_extra_context(self, request, instance): + show_available = ( + bool(request.GET.get('show_available', 'true') == 'true') and + not self._children_are_filtered(request, instance) + ) + return { 'bulk_querystring': f'within={instance.prefix}', 'first_available_prefix': instance.get_first_available_prefix(), - 'show_available': bool(request.GET.get('show_available', 'true') == 'true'), + 'show_available': show_available, 'show_assigned': bool(request.GET.get('show_assigned', 'true') == 'true'), } @@ -770,7 +834,7 @@ class PrefixView(generic.ObjectView): @register_model_view(Prefix, 'prefixes') -class PrefixPrefixesView(generic.ObjectChildrenView): +class PrefixPrefixesView(ChildAvailabilityMixin, generic.ObjectChildrenView): queryset = Prefix.objects.all() child_model = Prefix table = tables.PrefixTable @@ -794,13 +858,21 @@ class PrefixPrefixesView(generic.ObjectChildrenView): show_available = bool(request.GET.get('show_available', 'true') == 'true') show_assigned = bool(request.GET.get('show_assigned', 'true') == 'true') + if self._queryset_is_filtered(request, queryset, parent): + show_available = False + return add_requested_prefixes(parent.prefix, queryset, show_available, show_assigned) def get_extra_context(self, request, instance): + show_available = ( + bool(request.GET.get('show_available', 'true') == 'true') and + not self._children_are_filtered(request, instance) + ) + return { 'bulk_querystring': f"vrf_id={instance.vrf.pk if instance.vrf else '0'}&within={instance.prefix}", 'first_available_prefix': instance.get_first_available_prefix(), - 'show_available': bool(request.GET.get('show_available', 'true') == 'true'), + 'show_available': show_available, 'show_assigned': bool(request.GET.get('show_assigned', 'true') == 'true'), } @@ -833,7 +905,7 @@ class PrefixIPRangesView(generic.ObjectChildrenView): @register_model_view(Prefix, 'ipaddresses', path='ip-addresses') -class PrefixIPAddressesView(generic.ObjectChildrenView): +class PrefixIPAddressesView(ChildAvailabilityMixin, generic.ObjectChildrenView): queryset = Prefix.objects.all() child_model = IPAddress table = tables.AnnotatedIPAddressTable @@ -851,9 +923,10 @@ class PrefixIPAddressesView(generic.ObjectChildrenView): return parent.get_child_ips().restrict(request.user, 'view').prefetch_related('vrf', 'tenant', 'tenant__group') def prep_table_data(self, request, queryset, parent): - if not request.GET.get('q') and not get_table_ordering(request, self.table): + if not self._queryset_is_filtered(request, queryset, parent) and not get_table_ordering(request, self.table): return annotate_ip_space(parent) - return queryset + + return super().prep_table_data(request, queryset, parent) def get_extra_context(self, request, instance): return { @@ -1292,7 +1365,7 @@ class VLANGroupBulkDeleteView(generic.BulkDeleteView): @register_model_view(VLANGroup, 'vlans') -class VLANGroupVLANsView(generic.ObjectChildrenView): +class VLANGroupVLANsView(ChildAvailabilityMixin, generic.ObjectChildrenView): queryset = VLANGroup.objects.all() child_model = VLAN table = tables.VLANTable @@ -1312,9 +1385,11 @@ class VLANGroupVLANsView(generic.ObjectChildrenView): ) def prep_table_data(self, request, queryset, parent): - if not get_table_ordering(request, self.table): + # Skip synthetic available rows under active filters: filtered-out VLANs would otherwise look available. + if not self._queryset_is_filtered(request, queryset, parent) and not get_table_ordering(request, self.table): return add_available_vlans(queryset, parent) - return queryset + + return super().prep_table_data(request, queryset, parent) #