Fixes #22210: Respect filters when rendering IPAM child availability views (#22327)

* fix(ipam): Honor filters for child availability views

Retain the instantiated child FilterSet on ObjectChildrenView and expose
whether child object filters are active. Use this in IPAM child views to
avoid rendering synthetic availability rows when the child queryset has
been filtered.

This ensures Saved Filters and direct filters are respected on
Prefix IP Address, Child Prefix, Aggregate Prefix,
and VLAN Group VLAN tabs.

Fixes #22210

* refactor(ipam): Replace has_active_filters with ChildAvailabilityMixin

Extracts filter detection logic from ObjectChildrenView into a dedicated
ChildAvailabilityMixin. Compares WHERE clause signatures between
filtered and unfiltered querysets instead of inspecting filter
parameters, improving reliability when child querysets are pre-scoped
to parent objects.
This commit is contained in:
Martin Hauser 2026-06-02 22:30:17 +02:00 committed by GitHub
parent b1ebd93349
commit 3172e47904
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 387 additions and 10 deletions

View File

@ -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

View File

@ -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)
#