This commit is contained in:
parent
10ac9ffe8d
commit
052506fe9c
|
|
@ -1,4 +1,5 @@
|
|||
import datetime
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.db.backends.postgresql.psycopg_any import NumericRange
|
||||
|
|
@ -10,13 +11,16 @@ 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 extras.choices import CustomFieldTypeChoices
|
||||
from extras.models import CustomField, SavedFilter
|
||||
from ipam import filtersets
|
||||
from ipam.choices import *
|
||||
from ipam.models import *
|
||||
from ipam.views import AggregatePrefixesView
|
||||
from ipam.utils import AvailableIPSpace
|
||||
from ipam.views import AggregatePrefixesView, PrefixPrefixesView
|
||||
from netbox.choices import CSVDelimiterChoices, ImportFormatChoices
|
||||
from tenancy.models import Tenant
|
||||
from users.models import ObjectPermission
|
||||
from users.models import Group, ObjectPermission
|
||||
from utilities.testing import ViewTestCases, create_tags
|
||||
|
||||
|
||||
|
|
@ -420,37 +424,175 @@ class AggregateTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
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."""
|
||||
def test_aggregate_prefixes_custom_field_constraint_shows_available(self):
|
||||
"""A tenant custom-field permission constraint does not suppress available-prefix rows."""
|
||||
cf = CustomField.objects.create(name='integerCustomField', type=CustomFieldTypeChoices.TYPE_INTEGER)
|
||||
cf.object_types.set([ObjectType.objects.get_for_model(Tenant)])
|
||||
tenant = Tenant.objects.create(
|
||||
name='Agg CF Tenant', slug='agg-cf-tenant', custom_field_data={'integerCustomField': 1}
|
||||
)
|
||||
|
||||
aggregate = Aggregate.objects.create(prefix=IPNetwork('198.51.100.0/24'), rir=RIR.objects.first())
|
||||
child = Prefix.objects.create(prefix=IPNetwork('198.51.100.0/26'), tenant=tenant)
|
||||
|
||||
self.add_permissions('ipam.view_aggregate')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View prefixes', actions=['view'], constraints={'tenant__custom_field_data__integerCustomField': 1}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
||||
|
||||
self.assertIn(child, Prefix.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:aggregate_prefixes', kwargs={'pk': aggregate.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertTrue(response.context['show_available'])
|
||||
self.assertGreater(len(response.context['table'].data), 1)
|
||||
|
||||
def has_active_child_filters(self, **params):
|
||||
"""Run the child filter detector on a fresh view for the given query parameters."""
|
||||
view = AggregatePrefixesView()
|
||||
request = RequestFactory().get('/', params)
|
||||
request.user = self.user
|
||||
return view._has_active_child_filters(request)
|
||||
|
||||
def test_has_active_child_filters_declared_filters(self):
|
||||
"""A declared filter with a real value counts as active filtering."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
tenant = Tenant.objects.create(name='Declared Tenant', slug='declared-tenant')
|
||||
|
||||
self.assertTrue(self.has_active_child_filters(tenant_id=tenant.pk))
|
||||
self.assertTrue(self.has_active_child_filters(q='test'))
|
||||
# A boolean false is a value, not an absent filter.
|
||||
self.assertTrue(self.has_active_child_filters(is_pool='false'))
|
||||
|
||||
def test_has_active_child_filters_lookup_variants(self):
|
||||
"""Lookup variants generated by get_filters() count as active filtering."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
|
||||
aggregate = Aggregate.objects.create(
|
||||
prefix=IPNetwork('203.0.115.0/24'),
|
||||
rir=RIR.objects.first()
|
||||
self.assertTrue(self.has_active_child_filters(status__n=PrefixStatusChoices.STATUS_ACTIVE))
|
||||
self.assertTrue(self.has_active_child_filters(description__empty='true'))
|
||||
|
||||
def test_has_active_child_filters_custom_field_filters(self):
|
||||
"""Custom field filters registered on the filterset instance count as active filtering."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
cf = CustomField.objects.create(name='edge_cf', type=CustomFieldTypeChoices.TYPE_INTEGER)
|
||||
cf.object_types.set([ObjectType.objects.get_for_model(Prefix)])
|
||||
|
||||
self.assertTrue(self.has_active_child_filters(cf_edge_cf='1'))
|
||||
self.assertTrue(self.has_active_child_filters(cf_edge_cf__gte='1'))
|
||||
|
||||
def test_has_active_child_filters_saved_filter(self):
|
||||
"""A populated saved filter counts as active filtering by slug or by id."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
tenant = Tenant.objects.create(name='Saved Ref Tenant', slug='saved-ref-tenant')
|
||||
saved_filter = SavedFilter.objects.create(
|
||||
name='Saved ref', slug='saved-ref', parameters={'tenant_id': [str(tenant.pk)]}
|
||||
)
|
||||
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'))
|
||||
saved_filter.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
||||
|
||||
# No cached value: the fallback path rebuilds the filtered queryset and detects the filter.
|
||||
self.assertTrue(self.has_active_child_filters(filter=saved_filter.slug))
|
||||
self.assertTrue(self.has_active_child_filters(filter_id=saved_filter.pk))
|
||||
|
||||
def test_has_active_child_filters_unresolved_saved_filter(self):
|
||||
"""A saved filter reference that expands to nothing is not active filtering."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
saved_filter = SavedFilter.objects.create(name='Empty', slug='empty-saved-filter', parameters={})
|
||||
saved_filter.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
||||
|
||||
self.assertFalse(self.has_active_child_filters(filter_id=saved_filter.pk))
|
||||
self.assertFalse(self.has_active_child_filters(filter_id='99999999'))
|
||||
|
||||
def test_has_active_child_filters_without_values(self):
|
||||
"""A request with no parameters or only empty ones is not active filtering."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
|
||||
self.assertFalse(self.has_active_child_filters())
|
||||
self.assertFalse(self.has_active_child_filters(tenant_id=''))
|
||||
self.assertFalse(self.has_active_child_filters(q=''))
|
||||
|
||||
def test_has_active_child_filters_invalid_value(self):
|
||||
"""A filter value that fails validation is not applied, so it is not active filtering."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
|
||||
self.assertFalse(self.has_active_child_filters(tenant_id='99999999'))
|
||||
|
||||
def test_has_active_child_filters_valid_beside_invalid_value(self):
|
||||
"""A valid filter is still active when another submitted value is invalid."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
tenant = Tenant.objects.create(name='Mixed Tenant', slug='mixed-tenant')
|
||||
|
||||
self.assertTrue(self.has_active_child_filters(tenant_id=tenant.pk, status='not-a-valid-status'))
|
||||
|
||||
def test_has_active_child_filters_non_filter_params(self):
|
||||
"""Unknown parameters, display toggles, and table controls are not filters."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
|
||||
self.assertFalse(self.has_active_child_filters(not_a_filter='x'))
|
||||
self.assertFalse(self.has_active_child_filters(show_available='false'))
|
||||
self.assertFalse(self.has_active_child_filters(show_assigned='true'))
|
||||
self.assertFalse(self.has_active_child_filters(page='2', per_page='100', sort='prefix', tableconfig_id='1'))
|
||||
|
||||
def test_has_active_child_filters_control_beside_filter(self):
|
||||
"""A table control submitted alongside a filter does not mask the filter."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
tenant = Tenant.objects.create(name='Control Tenant', slug='control-tenant')
|
||||
|
||||
self.assertTrue(self.has_active_child_filters(page='2', tenant_id=tenant.pk))
|
||||
|
||||
def test_has_active_child_filters_without_filterset(self):
|
||||
"""A view without a filterset reports no active filters."""
|
||||
view = AggregatePrefixesView()
|
||||
request = RequestFactory().get('/', {'tenant_id': tenant.pk})
|
||||
view.filterset = None
|
||||
request = RequestFactory().get('/', {'tenant_id': '1'})
|
||||
request.user = self.user
|
||||
self.assertFalse(hasattr(view, '_child_queryset_is_filtered'))
|
||||
self.assertTrue(view._children_are_filtered(request, aggregate))
|
||||
self.assertFalse(view._has_active_child_filters(request))
|
||||
|
||||
def test_has_active_child_filters_reuses_view_filterset(self):
|
||||
"""The detector evaluates the FilterSet already bound by the view, not a fresh one."""
|
||||
aggregate = Aggregate.objects.create(prefix=IPNetwork('203.0.115.0/24'), rir=RIR.objects.first())
|
||||
tenant = Tenant.objects.create(name='Reuse Tenant', slug='reuse-tenant')
|
||||
|
||||
# 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})
|
||||
# The bound FilterSet is authoritative: it reports a filter the request itself does not carry.
|
||||
view.filterset_instance = AggregatePrefixesView.filterset(
|
||||
{'tenant_id': [str(tenant.pk)]}, view.get_children(request, aggregate), request=request
|
||||
)
|
||||
|
||||
request = RequestFactory().get('/', {'page': '2'})
|
||||
request.user = self.user
|
||||
self.assertFalse(view._children_are_filtered(request, aggregate))
|
||||
self.assertTrue(view._has_active_child_filters(request))
|
||||
|
||||
def test_child_tab_binds_filterset_once(self):
|
||||
"""A filtered child tab binds the FilterSet once; the detector reuses it instead of rebuilding."""
|
||||
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
||||
aggregate = Aggregate.objects.create(prefix=IPNetwork('203.0.116.0/24'), rir=RIR.objects.first())
|
||||
tenant = Tenant.objects.create(name='Bind Once Tenant', slug='bind-once-tenant')
|
||||
Prefix.objects.create(prefix=IPNetwork('203.0.116.0/26'), tenant=tenant)
|
||||
|
||||
# Count only data-bound instantiations: the filter form separately builds an unbound
|
||||
# FilterSet to resolve field modifiers, which is unrelated to filter detection.
|
||||
bound = []
|
||||
original_init = filtersets.PrefixFilterSet.__init__
|
||||
|
||||
def counting_init(fs, *args, **kwargs):
|
||||
if args or 'data' in kwargs:
|
||||
bound.append(fs)
|
||||
original_init(fs, *args, **kwargs)
|
||||
|
||||
url = reverse('ipam:aggregate_prefixes', kwargs={'pk': aggregate.pk})
|
||||
with patch.object(filtersets.PrefixFilterSet, '__init__', counting_init):
|
||||
response = self.client.get(url, {'tenant_id': tenant.pk})
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertEqual(len(bound), 1)
|
||||
self.assertFalse(response.context['show_available'])
|
||||
|
||||
|
||||
class RoleTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
|
||||
|
|
@ -829,18 +971,437 @@ class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
self.assertContains(response, '192.0.2.1/24')
|
||||
self.assertNotContains(response, '192.0.2.2/24')
|
||||
|
||||
def assertIPAvailabilityShown(self, response, visible_ip):
|
||||
"""The permitted IP renders as a real row and synthetic available-space rows are present."""
|
||||
records = list(response.context['table'].data)
|
||||
rendered_ip_pks = {r.pk for r in records if isinstance(r, IPAddress)}
|
||||
self.assertIn(visible_ip.pk, rendered_ip_pks)
|
||||
self.assertTrue(any(isinstance(r, AvailableIPSpace) for r in records))
|
||||
|
||||
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'))
|
||||
ip = 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)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_custom_field_constraint_shows_available(self):
|
||||
"""A permission constraint on a related object's custom field data does not suppress the available-IP rows."""
|
||||
cf = CustomField.objects.create(name='integerCustomField', type=CustomFieldTypeChoices.TYPE_INTEGER)
|
||||
cf.object_types.set([ObjectType.objects.get_for_model(Tenant)])
|
||||
|
||||
tenant = Tenant.objects.create(name='CF Tenant', slug='cf-tenant', custom_field_data={'integerCustomField': 1})
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), tenant=tenant)
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), tenant=tenant)
|
||||
|
||||
# The issue reports the JSON string "1", but an integer custom field stores an int, so a
|
||||
# string constraint would not match and would hide the parent. Use 1 and prove access below.
|
||||
constraint = {'tenant__custom_field_data__integerCustomField': 1}
|
||||
for model in (Prefix, IPAddress):
|
||||
obj_perm = ObjectPermission(
|
||||
name=f'View {model._meta.verbose_name}', actions=['view'], constraints=constraint
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(model))
|
||||
|
||||
# Self-verifying: the constraint must actually grant access to both objects, or the
|
||||
# availability assertion could pass through an unrestricted re-query instead of the fix.
|
||||
self.assertIn(prefix, Prefix.objects.restrict(self.user, 'view'))
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_related_field_constraint_shows_available(self):
|
||||
"""A permission constraint on a related object field does not suppress the available-IP rows."""
|
||||
tenant = Tenant.objects.create(name='Plain Constraint Tenant', slug='plain-constraint-tenant')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'), tenant=tenant)
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), tenant=tenant)
|
||||
|
||||
constraint = {'tenant__slug': 'plain-constraint-tenant'}
|
||||
for model in (Prefix, IPAddress):
|
||||
obj_perm = ObjectPermission(
|
||||
name=f'View {model._meta.verbose_name}', actions=['view'], constraints=constraint
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(model))
|
||||
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_constraint_with_direct_filter_suppresses(self):
|
||||
"""A direct filter still suppresses available-IP rows for a constrained user."""
|
||||
tenants = (
|
||||
Tenant(name='Constraint Direct 1', slug='constraint-direct-1'),
|
||||
Tenant(name='Constraint Direct 2', slug='constraint-direct-2'),
|
||||
)
|
||||
Tenant.objects.bulk_create(tenants)
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
|
||||
ip1 = IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), tenant=tenants[0])
|
||||
ip2 = IPAddress.objects.create(address=IPNetwork('192.0.2.2/24'), tenant=tenants[1])
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View IPs', actions=['view'], constraints={'tenant__slug__startswith': 'constraint-direct'}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
# Both IPs are visible under the constraint, so suppression below is due to the filter, not permissions.
|
||||
self.assertIn(ip1, IPAddress.objects.restrict(self.user, 'view'))
|
||||
self.assertIn(ip2, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
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_constraint_with_saved_filter_suppresses(self):
|
||||
"""A saved filter still suppresses available-IP rows for a constrained user."""
|
||||
tenants = (
|
||||
Tenant(name='Constraint Saved 1', slug='constraint-saved-1'),
|
||||
Tenant(name='Constraint Saved 2', slug='constraint-saved-2'),
|
||||
)
|
||||
Tenant.objects.bulk_create(tenants)
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
|
||||
ip1 = IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), tenant=tenants[0])
|
||||
ip2 = IPAddress.objects.create(address=IPNetwork('192.0.2.2/24'), tenant=tenants[1])
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View IPs', actions=['view'], constraints={'tenant__slug__startswith': 'constraint-saved'}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
# Both IPs are visible under the constraint, so suppression below is due to the filter, not permissions.
|
||||
self.assertIn(ip1, IPAddress.objects.restrict(self.user, 'view'))
|
||||
self.assertIn(ip2, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
saved_filter = SavedFilter.objects.create(
|
||||
name='Constraint saved tenant 1', slug='constraint-saved-tenant-1',
|
||||
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_direct_field_constraint_shows_available(self):
|
||||
"""A permission constraint on the IP Address status field does not suppress the available-IP rows."""
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), status=IPAddressStatusChoices.STATUS_ACTIVE)
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View IPs', actions=['view'], constraints={'status': IPAddressStatusChoices.STATUS_ACTIVE}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_own_custom_field_constraint_shows_available(self):
|
||||
"""A permission constraint on the IP Address custom field data does not suppress the available-IP rows."""
|
||||
cf = CustomField.objects.create(name='ip_cf', type=CustomFieldTypeChoices.TYPE_INTEGER)
|
||||
cf.object_types.set([ObjectType.objects.get_for_model(IPAddress)])
|
||||
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), custom_field_data={'ip_cf': 1})
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(name='View IPs', actions=['view'], constraints={'custom_field_data__ip_cf': 1})
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_multikey_constraint_shows_available(self):
|
||||
"""A permission constraint combining a direct and a related field does not suppress the available-IP rows."""
|
||||
tenant = Tenant.objects.create(name='Multi Key Tenant', slug='multi-key-tenant')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(
|
||||
address=IPNetwork('192.0.2.1/29'), status=IPAddressStatusChoices.STATUS_ACTIVE, tenant=tenant
|
||||
)
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View IPs',
|
||||
actions=['view'],
|
||||
constraints={'status': IPAddressStatusChoices.STATUS_ACTIVE, 'tenant__slug': 'multi-key-tenant'},
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_or_constraint_shows_available(self):
|
||||
"""A permission granting access through either of two constraints does not suppress the available-IP rows."""
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), status=IPAddressStatusChoices.STATUS_ACTIVE)
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View IPs',
|
||||
actions=['view'],
|
||||
constraints=[
|
||||
{'status': IPAddressStatusChoices.STATUS_ACTIVE},
|
||||
{'status': IPAddressStatusChoices.STATUS_RESERVED},
|
||||
],
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_address_startswith_constraint_shows_available(self):
|
||||
"""The #22539 address__startswith constraint does not suppress the available-IP rows."""
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.168.0.0/24'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.168.0.1/24'))
|
||||
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View IPs', actions=['view'], constraints={'address__startswith': '192.168.0.'}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_prefixes_show_available_false_skips_filter_detection(self):
|
||||
"""With availability already off, the filter detector never runs."""
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
parent = Prefix.objects.create(prefix=IPNetwork('198.51.104.0/24'))
|
||||
Prefix.objects.create(prefix=IPNetwork('198.51.104.0/26'))
|
||||
|
||||
view = PrefixPrefixesView()
|
||||
request = RequestFactory().get('/', {'show_available': 'false'})
|
||||
request.user = self.user
|
||||
view.prep_table_data(request, view.get_children(request, parent), parent)
|
||||
|
||||
self.assertFalse(hasattr(view, '_active_child_filters'))
|
||||
|
||||
def test_prefix_ipaddresses_valid_plus_invalid_filter_suppresses(self):
|
||||
"""A valid filter is applied and suppresses synthetic rows even when another filter value is invalid."""
|
||||
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress')
|
||||
tenant = Tenant.objects.create(name='VPI Tenant', slug='vpi-tenant')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
|
||||
matching = IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), tenant=tenant)
|
||||
IPAddress.objects.create(address=IPNetwork('192.0.2.2/24'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url, {'tenant_id': tenant.pk, 'status': 'not-a-valid-status'})
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
records = list(response.context['table'].data)
|
||||
self.assertFalse(any(isinstance(r, AvailableIPSpace) for r in records))
|
||||
self.assertEqual({r.pk for r in records if isinstance(r, IPAddress)}, {matching.pk})
|
||||
|
||||
def test_prefix_ipaddresses_htmx_direct_filter_suppresses(self):
|
||||
"""An HTMX table refresh with a direct filter suppresses available-IP rows."""
|
||||
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress')
|
||||
tenants = (
|
||||
Tenant(name='HTMX Tenant 1', slug='htmx-tenant-1'),
|
||||
Tenant(name='HTMX Tenant 2', slug='htmx-tenant-2'),
|
||||
)
|
||||
Tenant.objects.bulk_create(tenants)
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/24'))
|
||||
IPAddress.objects.create(address=IPNetwork('192.0.2.1/24'), tenant=tenants[0])
|
||||
IPAddress.objects.create(address=IPNetwork('192.0.2.2/24'), tenant=tenants[1])
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url, {'tenant_id': tenants[0].pk}, HTTP_HX_REQUEST='true')
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertEqual(len(response.context['table'].data), 1)
|
||||
|
||||
def test_prefix_ipaddresses_htmx_unfiltered_shows_available(self):
|
||||
"""An HTMX table refresh with no filter still injects available-IP rows."""
|
||||
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url, HTTP_HX_REQUEST='true')
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_group_constraint_shows_available(self):
|
||||
"""A constraint granted through a group does not suppress the available-IP rows."""
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
tenant = Tenant.objects.create(name='Group Grant', slug='group-grant')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), tenant=tenant)
|
||||
|
||||
group = Group.objects.create(name='IP Viewers')
|
||||
self.user.groups.add(group)
|
||||
obj_perm = ObjectPermission(name='View IPs', actions=['view'], constraints={'tenant__slug': 'group-grant'})
|
||||
obj_perm.save()
|
||||
obj_perm.groups.add(group)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
# The group grant must resolve, or the assertion below would be vacuous.
|
||||
self.assertIn(ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_ipaddresses_partial_visibility_shows_available(self):
|
||||
"""A constraint that hides one child IP keeps availability rows and omits the hidden IP."""
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
visible_tenant = Tenant.objects.create(name='Partial Vis', slug='partial-vis')
|
||||
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
visible_ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'), tenant=visible_tenant)
|
||||
hidden_ip = IPAddress.objects.create(address=IPNetwork('192.0.2.6/29'))
|
||||
|
||||
obj_perm = ObjectPermission(name='View IPs', actions=['view'], constraints={'tenant__slug': 'partial-vis'})
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
||||
|
||||
# Exactly one child IP is visible, so the constraint is doing real work.
|
||||
self.assertIn(visible_ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
self.assertNotIn(hidden_ip, IPAddress.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
# The hidden IP is omitted and its slot is counted as available space.
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, visible_ip)
|
||||
records = list(response.context['table'].data)
|
||||
self.assertNotIn(hidden_ip.pk, {r.pk for r in records if isinstance(r, IPAddress)})
|
||||
self.assertEqual(sum(r.size for r in records if isinstance(r, AvailableIPSpace)), 5)
|
||||
|
||||
def test_prefix_ipaddresses_partial_visibility_omits_hidden_range(self):
|
||||
"""A constraint on child ranges keeps the permitted range and omits the hidden one."""
|
||||
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress')
|
||||
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'))
|
||||
visible_range = IPRange.objects.create(
|
||||
start_address=IPNetwork('192.0.2.2/29'),
|
||||
end_address=IPNetwork('192.0.2.3/29'),
|
||||
size=2,
|
||||
mark_populated=True,
|
||||
description='visible',
|
||||
)
|
||||
hidden_range = IPRange.objects.create(
|
||||
start_address=IPNetwork('192.0.2.4/29'),
|
||||
end_address=IPNetwork('192.0.2.5/29'),
|
||||
size=2,
|
||||
mark_populated=True,
|
||||
)
|
||||
|
||||
obj_perm = ObjectPermission(name='View ranges', actions=['view'], constraints={'description': 'visible'})
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPRange))
|
||||
|
||||
# Exactly one child range is visible, so the constraint is doing real work.
|
||||
self.assertIn(visible_range, IPRange.objects.restrict(self.user, 'view'))
|
||||
self.assertNotIn(hidden_range, IPRange.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
records = list(response.context['table'].data)
|
||||
self.assertEqual({r.pk for r in records if isinstance(r, IPRange)}, {visible_range.pk})
|
||||
self.assertEqual(sum(r.size for r in records if isinstance(r, AvailableIPSpace)), 3)
|
||||
|
||||
def test_prefix_ipaddresses_sorted_suppresses_available(self):
|
||||
"""A sorted IP Addresses tab drops synthetic rows so ordering stays in SQL."""
|
||||
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url, {'sort': 'address'})
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
records = list(response.context['table'].data)
|
||||
self.assertEqual({r.pk for r in records if isinstance(r, IPAddress)}, {ip.pk})
|
||||
self.assertFalse(any(isinstance(r, AvailableIPSpace) for r in records))
|
||||
|
||||
def test_prefix_ipaddresses_empty_sort_shows_available(self):
|
||||
"""An empty sort value is not an ordering, so synthetic rows survive."""
|
||||
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange')
|
||||
prefix = Prefix.objects.create(prefix=IPNetwork('192.0.2.0/29'))
|
||||
ip = IPAddress.objects.create(address=IPNetwork('192.0.2.1/29'))
|
||||
|
||||
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
||||
response = self.client.get(url, {'sort': ''})
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertIPAvailabilityShown(response, ip)
|
||||
|
||||
def test_prefix_prefixes_unfiltered_shows_available_prefixes(self):
|
||||
"""An unfiltered Child Prefixes tab injects synthetic available-prefix rows."""
|
||||
|
|
@ -855,6 +1416,74 @@ class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
self.assertHttpStatus(response, 200)
|
||||
self.assertGreater(len(response.context['table'].data), 1)
|
||||
|
||||
def test_prefix_prefixes_custom_field_constraint_shows_available(self):
|
||||
"""A tenant custom-field permission constraint does not suppress available child-prefix rows."""
|
||||
cf = CustomField.objects.create(name='integerCustomField', type=CustomFieldTypeChoices.TYPE_INTEGER)
|
||||
cf.object_types.set([ObjectType.objects.get_for_model(Tenant)])
|
||||
tenant = Tenant.objects.create(
|
||||
name='Child CF Tenant', slug='child-cf-tenant', custom_field_data={'integerCustomField': 1}
|
||||
)
|
||||
|
||||
parent = Prefix.objects.create(prefix=IPNetwork('198.51.100.0/24'), tenant=tenant)
|
||||
child = Prefix.objects.create(prefix=IPNetwork('198.51.100.0/26'), tenant=tenant)
|
||||
|
||||
obj_perm = ObjectPermission(
|
||||
name='View prefixes', actions=['view'], constraints={'tenant__custom_field_data__integerCustomField': 1}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
||||
|
||||
self.assertIn(child, Prefix.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertTrue(response.context['show_available'])
|
||||
self.assertGreater(len(response.context['table'].data), 1)
|
||||
|
||||
def test_prefix_prefixes_available_only_shows_available(self):
|
||||
"""The Available button's parameters render synthetic rows and no assigned rows."""
|
||||
self.add_permissions('ipam.view_prefix')
|
||||
parent = Prefix.objects.create(prefix=IPNetwork('198.51.104.0/24'))
|
||||
Prefix.objects.create(prefix=IPNetwork('198.51.104.0/26'))
|
||||
|
||||
url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk})
|
||||
response = self.client.get(url, {'show_assigned': 'false', 'show_available': 'true'})
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
rendered = list(response.context['table'].data)
|
||||
self.assertTrue([p for p in rendered if p.pk is None])
|
||||
self.assertFalse([p for p in rendered if p.pk is not None])
|
||||
|
||||
def test_prefix_prefixes_partial_visibility_shows_available(self):
|
||||
"""A constraint that hides one child prefix does not suppress the available-prefix rows."""
|
||||
visible_tenant = Tenant.objects.create(name='PP Partial', slug='pp-partial')
|
||||
|
||||
# The parent carries the visible tenant, so the single constrained grant covers the tab itself.
|
||||
parent = Prefix.objects.create(prefix=IPNetwork('198.51.100.0/24'), tenant=visible_tenant)
|
||||
visible_child = Prefix.objects.create(prefix=IPNetwork('198.51.100.0/26'), tenant=visible_tenant)
|
||||
hidden_child = Prefix.objects.create(prefix=IPNetwork('198.51.100.64/26'))
|
||||
|
||||
obj_perm = ObjectPermission(
|
||||
name='View prefixes', actions=['view'], constraints={'tenant__slug': 'pp-partial'}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
||||
|
||||
self.assertIn(visible_child, Prefix.objects.restrict(self.user, 'view'))
|
||||
self.assertNotIn(hidden_child, Prefix.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
# Pins that a constraint does not suppress availability, not how the hidden child is handled.
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertTrue(response.context['show_available'])
|
||||
self.assertTrue([p for p in response.context['table'].data if p.pk is None])
|
||||
|
||||
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
|
||||
|
|
@ -1433,6 +2062,73 @@ class VLANGroupTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
|
|||
self.assertHttpStatus(response, 200)
|
||||
self.assertGreater(len(response.context['table'].data), 1)
|
||||
|
||||
def test_vlans_custom_field_constraint_shows_available(self):
|
||||
"""A tenant custom-field permission constraint does not suppress available-VLAN rows."""
|
||||
cf = CustomField.objects.create(name='integerCustomField', type=CustomFieldTypeChoices.TYPE_INTEGER)
|
||||
cf.object_types.set([ObjectType.objects.get_for_model(Tenant)])
|
||||
tenant = Tenant.objects.create(
|
||||
name='VLAN CF Tenant', slug='vlan-cf-tenant', custom_field_data={'integerCustomField': 1}
|
||||
)
|
||||
|
||||
group = VLANGroup.objects.create(name='CF VLAN Group', slug='cf-vlan-group')
|
||||
vlan = VLAN.objects.create(group=group, vid=10, name='VLAN0010', tenant=tenant)
|
||||
|
||||
self.add_permissions('ipam.view_vlangroup')
|
||||
obj_perm = ObjectPermission(
|
||||
name='View VLANs', actions=['view'], constraints={'tenant__custom_field_data__integerCustomField': 1}
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(VLAN))
|
||||
|
||||
self.assertIn(vlan, VLAN.objects.restrict(self.user, 'view'))
|
||||
|
||||
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)
|
||||
|
||||
def test_vlans_partial_visibility_shows_available(self):
|
||||
"""A constraint that hides one VLAN does not suppress the available-VLAN rows."""
|
||||
self.add_permissions('ipam.view_vlangroup')
|
||||
visible_tenant = Tenant.objects.create(name='VLAN Partial', slug='vlan-partial')
|
||||
|
||||
group = VLANGroup.objects.create(name='Partial VLAN Group', slug='partial-vlan-group')
|
||||
visible_vlan = VLAN.objects.create(group=group, vid=10, name='VLAN0010', tenant=visible_tenant)
|
||||
hidden_vlan = VLAN.objects.create(group=group, vid=20, name='VLAN0020')
|
||||
|
||||
obj_perm = ObjectPermission(name='View VLANs', actions=['view'], constraints={'tenant__slug': 'vlan-partial'})
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(VLAN))
|
||||
|
||||
self.assertIn(visible_vlan, VLAN.objects.restrict(self.user, 'view'))
|
||||
self.assertNotIn(hidden_vlan, VLAN.objects.restrict(self.user, 'view'))
|
||||
|
||||
url = reverse('ipam:vlangroup_vlans', kwargs={'pk': group.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
# Pins that a constraint does not suppress availability, not how the hidden child is handled.
|
||||
self.assertHttpStatus(response, 200)
|
||||
rendered = list(response.context['table'].data)
|
||||
self.assertIn(visible_vlan.vid, {r.vid for r in rendered if isinstance(r, VLAN)})
|
||||
self.assertTrue([r for r in rendered if isinstance(r, dict)]) # synthetic available VLANs present
|
||||
|
||||
def test_vlans_sorted_suppresses_available(self):
|
||||
"""A sorted VLANs tab drops synthetic available-VLAN rows."""
|
||||
self.add_permissions('ipam.view_vlangroup', 'ipam.view_vlan')
|
||||
group = VLANGroup.objects.create(name='Sorted VLAN Group', slug='sorted-vlan-group')
|
||||
vlan = VLAN.objects.create(group=group, vid=10, name='VLAN0010')
|
||||
|
||||
url = reverse('ipam:vlangroup_vlans', kwargs={'pk': group.pk})
|
||||
response = self.client.get(url, {'sort': 'vid'})
|
||||
|
||||
self.assertHttpStatus(response, 200)
|
||||
rendered = list(response.context['table'].data)
|
||||
self.assertEqual({r.vid for r in rendered if isinstance(r, VLAN)}, {vlan.vid})
|
||||
self.assertFalse([r for r in rendered if isinstance(r, dict)]) # no synthetic available VLANs
|
||||
|
||||
|
||||
class VLANTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
||||
model = VLAN
|
||||
|
|
|
|||
|
|
@ -67,14 +67,26 @@ def add_requested_prefixes(parent, prefix_list, show_available=True, show_assign
|
|||
return child_prefixes
|
||||
|
||||
|
||||
def annotate_ip_space(prefix):
|
||||
def annotate_ip_space(prefix, *, ip_addresses=None, ip_ranges=None):
|
||||
"""
|
||||
Return a prefix's child ranges and IPs interleaved with available space records.
|
||||
|
||||
:param prefix: Parent Prefix instance
|
||||
:param ip_addresses: Child IP addresses queryset (defaults to all child IPs)
|
||||
:param ip_ranges: Child IP ranges queryset (defaults to all populated child ranges)
|
||||
"""
|
||||
if ip_addresses is None:
|
||||
ip_addresses = prefix.get_child_ips()
|
||||
if ip_ranges is None:
|
||||
ip_ranges = prefix.get_child_ranges(mark_populated=True)
|
||||
|
||||
# Compile child objects
|
||||
records = []
|
||||
records.extend([
|
||||
(iprange.start_address.ip, iprange) for iprange in prefix.get_child_ranges(mark_populated=True)
|
||||
(iprange.start_address.ip, iprange) for iprange in ip_ranges
|
||||
])
|
||||
records.extend([
|
||||
(ip.address.ip, ip) for ip in prefix.get_child_ips()
|
||||
(ip.address.ip, ip) for ip in ip_addresses
|
||||
])
|
||||
records = sorted(records, key=lambda x: x[0])
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,11 @@
|
|||
import django_filters
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.core.exceptions import EmptyResultSet
|
||||
from django.db.models import Prefetch
|
||||
from django.db.models.expressions import RawSQL
|
||||
from django.shortcuts import get_object_or_404, redirect, render
|
||||
from django.urls import reverse
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from django_filters.constants import EMPTY_VALUES
|
||||
|
||||
from circuits.models import Provider
|
||||
from dcim.filtersets import InterfaceFilterSet
|
||||
|
|
@ -551,62 +552,63 @@ 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.
|
||||
(available IP space, prefixes, or VLANs) and must suppress them when the request
|
||||
activates a child object filter, so objects excluded by the filter are not
|
||||
misrepresented as available space.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _where_signature(queryset):
|
||||
# Compare compiled SQL rather than str(query.where): the WHERE tree embeds default
|
||||
# object reprs (memory addresses) for permission-constraint subqueries, so two
|
||||
# otherwise identical querysets built via restrict() never match (#22539).
|
||||
try:
|
||||
return queryset.query.get_compiler(using=queryset.db).as_sql()
|
||||
except EmptyResultSet:
|
||||
def _get_detection_filterset(self, request):
|
||||
"""
|
||||
Return the bound FilterSet used to evaluate the request, or None if the view declares no
|
||||
filterset. ObjectChildrenView.get() has already built one and validated its form while
|
||||
resolving the child queryset, so reuse it rather than paying for a second FilterSet:
|
||||
get_filters() regenerates every dynamic lookup variant and NetBoxModelFilterSet.__init__
|
||||
queries the custom fields for the model. Build one only for direct calls where get() has
|
||||
not run.
|
||||
"""
|
||||
if self.filterset_instance is not None:
|
||||
return self.filterset_instance
|
||||
if self.filterset is None:
|
||||
return None
|
||||
|
||||
def _set_children_filtered(self, is_filtered):
|
||||
self._child_queryset_is_filtered = is_filtered
|
||||
return is_filtered
|
||||
return self.filterset(request.GET, request=request)
|
||||
|
||||
def _queryset_is_filtered(self, request, queryset, parent):
|
||||
def _has_active_child_filters(self, request):
|
||||
"""
|
||||
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.
|
||||
Return True if the request supplies a valid, non-empty value for any filter declared by
|
||||
the view's filterset. Saved filters are expanded during filterset instantiation and
|
||||
dynamic custom field filters are registered on the instance, so both are detected.
|
||||
Permission constraints and parent scoping never appear in the request, so they cannot
|
||||
affect the result. The result is memoized because a single request evaluates it from
|
||||
both prep_table_data and get_extra_context.
|
||||
"""
|
||||
if self.filterset is None:
|
||||
return self._set_children_filtered(False)
|
||||
if hasattr(self, '_active_child_filters'):
|
||||
return self._active_child_filters
|
||||
|
||||
unfiltered = self.get_children(request, parent)
|
||||
self._active_child_filters = False
|
||||
|
||||
return self._set_children_filtered(
|
||||
self._where_signature(queryset) != self._where_signature(unfiltered)
|
||||
)
|
||||
# An empty request cannot activate a filter, so skip validating a form for nothing.
|
||||
if not request.GET:
|
||||
return False
|
||||
|
||||
def _children_are_filtered(self, request, parent):
|
||||
"""
|
||||
Return whether child objects are filtered.
|
||||
filterset = self._get_detection_filterset(request)
|
||||
if filterset is None:
|
||||
return False
|
||||
|
||||
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
|
||||
# A non-empty cleaned value means the request activated a declared filter. Emptiness
|
||||
# follows each filter's own semantics, so absent multi-value fields stay inactive.
|
||||
# This is a no-op for a reused FilterSet: .qs validated the form to build the queryset.
|
||||
filterset.form.is_valid()
|
||||
for name, value in filterset.form.cleaned_data.items():
|
||||
if isinstance(filterset.filters[name], django_filters.MultipleChoiceFilter):
|
||||
if value: # mirrors MultipleChoiceFilter.filter()
|
||||
self._active_child_filters = True
|
||||
break
|
||||
elif value not in EMPTY_VALUES: # mirrors Filter.filter()
|
||||
self._active_child_filters = True
|
||||
break
|
||||
|
||||
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)
|
||||
)
|
||||
return self._active_child_filters
|
||||
|
||||
|
||||
@register_model_view(Aggregate, 'prefixes')
|
||||
|
|
@ -634,7 +636,7 @@ class AggregatePrefixesView(ChildAvailabilityMixin, 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):
|
||||
if show_available and self._has_active_child_filters(request):
|
||||
show_available = False
|
||||
|
||||
return add_requested_prefixes(parent.prefix, queryset, show_available, show_assigned)
|
||||
|
|
@ -642,7 +644,7 @@ class AggregatePrefixesView(ChildAvailabilityMixin, generic.ObjectChildrenView):
|
|||
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)
|
||||
not self._has_active_child_filters(request)
|
||||
)
|
||||
|
||||
return {
|
||||
|
|
@ -864,7 +866,7 @@ class PrefixPrefixesView(ChildAvailabilityMixin, 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):
|
||||
if show_available and self._has_active_child_filters(request):
|
||||
show_available = False
|
||||
|
||||
return add_requested_prefixes(parent.prefix, queryset, show_available, show_assigned)
|
||||
|
|
@ -872,7 +874,7 @@ class PrefixPrefixesView(ChildAvailabilityMixin, generic.ObjectChildrenView):
|
|||
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)
|
||||
not self._has_active_child_filters(request)
|
||||
)
|
||||
|
||||
return {
|
||||
|
|
@ -929,8 +931,11 @@ class PrefixIPAddressesView(ChildAvailabilityMixin, 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 self._queryset_is_filtered(request, queryset, parent) and not get_table_ordering(request, self.table):
|
||||
return annotate_ip_space(parent)
|
||||
# Ordering is checked first: it reads request.GET directly, so a sorted request never
|
||||
# builds the detection filterset.
|
||||
if not get_table_ordering(request, self.table) and not self._has_active_child_filters(request):
|
||||
ip_ranges = parent.get_child_ranges(mark_populated=True).restrict(request.user, 'view')
|
||||
return annotate_ip_space(parent, ip_addresses=queryset, ip_ranges=ip_ranges)
|
||||
|
||||
return super().prep_table_data(request, queryset, parent)
|
||||
|
||||
|
|
@ -1392,7 +1397,7 @@ class VLANGroupVLANsView(ChildAvailabilityMixin, generic.ObjectChildrenView):
|
|||
|
||||
def prep_table_data(self, request, queryset, parent):
|
||||
# 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):
|
||||
if not get_table_ordering(request, self.table) and not self._has_active_child_filters(request):
|
||||
return add_available_vlans(queryset, parent)
|
||||
|
||||
return super().prep_table_data(request, queryset, parent)
|
||||
|
|
|
|||
|
|
@ -98,12 +98,14 @@ class ObjectChildrenView(ObjectView, ActionsMixin, TableMixin):
|
|||
table: The django-tables2 Table class used to render the child objects list
|
||||
filterset: A django-filter FilterSet that is applied to the queryset
|
||||
filterset_form: The form class used to render filter options
|
||||
filterset_instance: The bound FilterSet built for the current request (set during get())
|
||||
actions: An iterable of ObjectAction subclasses (see ActionsMixin)
|
||||
"""
|
||||
child_model = None
|
||||
table = None
|
||||
filterset = None
|
||||
filterset_form = None
|
||||
filterset_instance = None
|
||||
actions = (CloneObject, EditObject, DeleteObject, BulkEdit, BulkDelete)
|
||||
template_name = 'generic/object_children.html'
|
||||
|
||||
|
|
@ -142,7 +144,10 @@ class ObjectChildrenView(ObjectView, ActionsMixin, TableMixin):
|
|||
child_objects = self.get_children(request, instance)
|
||||
|
||||
if self.filterset:
|
||||
child_objects = self.filterset(request.GET, child_objects, request=request).qs
|
||||
# Retain the bound FilterSet so that prep_table_data() and get_extra_context() can
|
||||
# inspect the request's validated filter data without rebuilding it.
|
||||
self.filterset_instance = self.filterset(request.GET, child_objects, request=request)
|
||||
child_objects = self.filterset_instance.qs
|
||||
|
||||
# Determine the available actions
|
||||
actions = self.get_permitted_actions(request.user, model=self.child_model)
|
||||
|
|
|
|||
Loading…
Reference in New Issue