2987 lines
125 KiB
Python
2987 lines
125 KiB
Python
import datetime
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.db.backends.postgresql.psycopg_any import NumericRange
|
|
from django.test import RequestFactory, override_settings
|
|
from django.urls import reverse
|
|
from netaddr import IPNetwork
|
|
|
|
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, Region, Site
|
|
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.utils import AvailableIPSpace
|
|
from ipam.views import AggregatePrefixesView, PrefixPrefixesView
|
|
from netbox.choices import CSVDelimiterChoices, ImportFormatChoices
|
|
from tenancy.models import Tenant
|
|
from users.models import Group, ObjectPermission, Owner
|
|
from utilities.testing import ViewTestCases, create_tags, post_data
|
|
|
|
|
|
class ASNRangeTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = ASNRange
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
rirs = [
|
|
RIR(name='RIR 1', slug='rir-1', is_private=True),
|
|
RIR(name='RIR 2', slug='rir-2', is_private=True),
|
|
]
|
|
RIR.objects.bulk_create(rirs)
|
|
|
|
tenants = [
|
|
Tenant(name='Tenant 1', slug='tenant-1'),
|
|
Tenant(name='Tenant 2', slug='tenant-2'),
|
|
]
|
|
Tenant.objects.bulk_create(tenants)
|
|
|
|
asn_ranges = (
|
|
ASNRange(name='ASN Range 1', slug='asn-range-1', rir=rirs[0], tenant=tenants[0], start=100, end=199),
|
|
ASNRange(name='ASN Range 2', slug='asn-range-2', rir=rirs[0], tenant=tenants[0], start=200, end=299),
|
|
ASNRange(name='ASN Range 3', slug='asn-range-3', rir=rirs[0], tenant=tenants[0], start=300, end=399),
|
|
)
|
|
ASNRange.objects.bulk_create(asn_ranges)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'ASN Range X',
|
|
'slug': 'asn-range-x',
|
|
'rir': rirs[1].pk,
|
|
'tenant': tenants[1].pk,
|
|
'start': 1000,
|
|
'end': 1099,
|
|
'description': 'A new ASN range',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name,slug,rir,tenant,start,end,description",
|
|
f"ASN Range 4,asn-range-4,{rirs[1].name},{tenants[1].name},400,499,Fourth range",
|
|
f"ASN Range 5,asn-range-5,{rirs[1].name},{tenants[1].name},500,599,Fifth range",
|
|
f"ASN Range 6,asn-range-6,{rirs[1].name},{tenants[1].name},600,699,Sixth range",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,description",
|
|
f"{asn_ranges[0].pk},New description 1",
|
|
f"{asn_ranges[1].pk},New description 2",
|
|
f"{asn_ranges[2].pk},New description 3",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'rir': rirs[1].pk,
|
|
'description': 'Next description',
|
|
}
|
|
|
|
|
|
class ASNTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = ASN
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
rirs = [
|
|
RIR(name='RIR 1', slug='rir-1', is_private=True),
|
|
RIR(name='RIR 2', slug='rir-2', is_private=True),
|
|
]
|
|
RIR.objects.bulk_create(rirs)
|
|
|
|
roles = (
|
|
Role(name='Role 1', slug='role-1'),
|
|
Role(name='Role 2', slug='role-2'),
|
|
)
|
|
Role.objects.bulk_create(roles)
|
|
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2')
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
tenants = (
|
|
Tenant(name='Tenant 1', slug='tenant-1'),
|
|
Tenant(name='Tenant 2', slug='tenant-2'),
|
|
)
|
|
Tenant.objects.bulk_create(tenants)
|
|
|
|
asns = (
|
|
ASN(asn=65001, rir=rirs[0], role=roles[0], tenant=tenants[0]),
|
|
ASN(asn=65002, rir=rirs[1], role=roles[1], tenant=tenants[1]),
|
|
ASN(asn=4200000001, rir=rirs[0], role=roles[0], tenant=tenants[0]),
|
|
ASN(asn=4200000002, rir=rirs[1], role=roles[1], tenant=tenants[1]),
|
|
)
|
|
ASN.objects.bulk_create(asns)
|
|
|
|
asns[0].sites.set([sites[0]])
|
|
asns[1].sites.set([sites[1]])
|
|
asns[2].sites.set([sites[0]])
|
|
asns[3].sites.set([sites[1]])
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'asn': 65000,
|
|
'rir': rirs[0].pk,
|
|
'role': roles[0].pk,
|
|
'tenant': tenants[0].pk,
|
|
'site': sites[0].pk,
|
|
'description': 'A new ASN',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"asn,rir,role",
|
|
f"65003,RIR 1,{roles[0].name}",
|
|
f"65004,RIR 2,{roles[1].name}",
|
|
f"4200000003,RIR 1,{roles[0].name}",
|
|
f"4200000004,RIR 2,{roles[1].name}",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,description",
|
|
f"{asns[0].pk},New description1",
|
|
f"{asns[1].pk},New description2",
|
|
f"{asns[2].pk},New description3",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'rir': rirs[1].pk,
|
|
'role': roles[1].pk,
|
|
'description': 'Next description',
|
|
}
|
|
|
|
|
|
class VRFTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = VRF
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
tenants = (
|
|
Tenant(name='Tenant A', slug='tenant-a'),
|
|
Tenant(name='Tenant B', slug='tenant-b'),
|
|
)
|
|
Tenant.objects.bulk_create(tenants)
|
|
|
|
vrfs = (
|
|
VRF(name='VRF 1', rd='65000:1'),
|
|
VRF(name='VRF 2', rd='65000:2'),
|
|
VRF(name='VRF 3', rd='65000:3'),
|
|
)
|
|
VRF.objects.bulk_create(vrfs)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'VRF X',
|
|
'rd': '65000:999',
|
|
'tenant': tenants[0].pk,
|
|
'enforce_unique': True,
|
|
'description': 'A new VRF',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name",
|
|
"VRF 4",
|
|
"VRF 5",
|
|
"VRF 6",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name",
|
|
f"{vrfs[0].pk},VRF 7",
|
|
f"{vrfs[1].pk},VRF 8",
|
|
f"{vrfs[2].pk},VRF 9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'tenant': tenants[1].pk,
|
|
'enforce_unique': False,
|
|
'description': 'New description',
|
|
}
|
|
|
|
|
|
class RouteTargetTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = RouteTarget
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
tenants = (
|
|
Tenant(name='Tenant A', slug='tenant-a'),
|
|
Tenant(name='Tenant B', slug='tenant-b'),
|
|
)
|
|
Tenant.objects.bulk_create(tenants)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
route_targets = (
|
|
RouteTarget(name='65000:1001', tenant=tenants[0]),
|
|
RouteTarget(name='65000:1002', tenant=tenants[1]),
|
|
RouteTarget(name='65000:1003'),
|
|
)
|
|
RouteTarget.objects.bulk_create(route_targets)
|
|
|
|
cls.form_data = {
|
|
'name': '65000:100',
|
|
'description': 'A new route target',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name,tenant,description",
|
|
"65000:1004,Tenant A,Foo",
|
|
"65000:1005,Tenant B,Bar",
|
|
"65000:1006,,No tenant",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{route_targets[0].pk},65000:1007,New description1",
|
|
f"{route_targets[1].pk},65000:1008,New description2",
|
|
f"{route_targets[2].pk},65000:1009,New description3",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'tenant': tenants[1].pk,
|
|
'description': 'New description',
|
|
}
|
|
|
|
|
|
class RIRTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
|
|
model = RIR
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
rirs = (
|
|
RIR(name='RIR 1', slug='rir-1'),
|
|
RIR(name='RIR 2', slug='rir-2'),
|
|
RIR(name='RIR 3', slug='rir-3'),
|
|
)
|
|
RIR.objects.bulk_create(rirs)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'RIR X',
|
|
'slug': 'rir-x',
|
|
'is_private': True,
|
|
'description': 'A new RIR',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name,slug,description",
|
|
"RIR 4,rir-4,Fourth RIR",
|
|
"RIR 5,rir-5,Fifth RIR",
|
|
"RIR 6,rir-6,Sixth RIR",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{rirs[0].pk},RIR 7,Fourth RIR7",
|
|
f"{rirs[1].pk},RIR 8,Fifth RIR8",
|
|
f"{rirs[2].pk},RIR 9,Sixth RIR9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'description': 'New description',
|
|
}
|
|
|
|
|
|
class AggregateTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = Aggregate
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
rirs = (
|
|
RIR(name='RIR 1', slug='rir-1'),
|
|
RIR(name='RIR 2', slug='rir-2'),
|
|
)
|
|
RIR.objects.bulk_create(rirs)
|
|
|
|
aggregates = (
|
|
Aggregate(prefix=IPNetwork('10.1.0.0/16'), rir=rirs[0]),
|
|
Aggregate(prefix=IPNetwork('10.2.0.0/16'), rir=rirs[0]),
|
|
Aggregate(prefix=IPNetwork('10.3.0.0/16'), rir=rirs[0]),
|
|
)
|
|
Aggregate.objects.bulk_create(aggregates)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'prefix': IPNetwork('10.99.0.0/16'),
|
|
'rir': rirs[1].pk,
|
|
'date_added': datetime.date(2020, 1, 1),
|
|
'description': 'A new aggregate',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"prefix,rir",
|
|
"10.4.0.0/16,RIR 1",
|
|
"10.5.0.0/16,RIR 1",
|
|
"10.6.0.0/16,RIR 1",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,description",
|
|
f"{aggregates[0].pk},New description1",
|
|
f"{aggregates[1].pk},New description2",
|
|
f"{aggregates[2].pk},New description3",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'rir': rirs[1].pk,
|
|
'date_added': datetime.date(2020, 1, 1),
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_aggregate_prefixes(self):
|
|
self.add_permissions('ipam.view_aggregate', 'ipam.view_prefix')
|
|
rir = RIR.objects.first()
|
|
aggregate = Aggregate.objects.create(prefix=IPNetwork('192.168.0.0/16'), rir=rir)
|
|
prefixes = (
|
|
Prefix(prefix=IPNetwork('192.168.1.0/24')),
|
|
Prefix(prefix=IPNetwork('192.168.2.0/24')),
|
|
Prefix(prefix=IPNetwork('192.168.3.0/24')),
|
|
)
|
|
Prefix.objects.bulk_create(prefixes)
|
|
self.assertEqual(aggregate.get_child_prefixes().count(), 3)
|
|
|
|
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_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')
|
|
|
|
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)]}
|
|
)
|
|
saved_filter.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
|
|
|
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()
|
|
view.filterset = None
|
|
request = RequestFactory().get('/', {'tenant_id': '1'})
|
|
request.user = self.user
|
|
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')
|
|
|
|
view = AggregatePrefixesView()
|
|
request = RequestFactory().get('/')
|
|
request.user = self.user
|
|
|
|
# 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.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):
|
|
model = Role
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
roles = (
|
|
Role(name='Role 1', slug='role-1'),
|
|
Role(name='Role 2', slug='role-2'),
|
|
Role(name='Role 3', slug='role-3'),
|
|
)
|
|
Role.objects.bulk_create(roles)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'Role X',
|
|
'slug': 'role-x',
|
|
'weight': 200,
|
|
'description': 'A new role',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name,slug,weight",
|
|
"Role 4,role-4,1000",
|
|
"Role 5,role-5,1000",
|
|
"Role 6,role-6,1000",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{roles[0].pk},Role 7,New description7",
|
|
f"{roles[1].pk},Role 8,New description8",
|
|
f"{roles[2].pk},Role 9,New description9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'description': 'New description',
|
|
}
|
|
|
|
|
|
class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = Prefix
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
vrfs = (
|
|
VRF(name='VRF 1', rd='65000:1'),
|
|
VRF(name='VRF 2', rd='65000:2'),
|
|
)
|
|
VRF.objects.bulk_create(vrfs)
|
|
|
|
roles = (
|
|
Role(name='Role 1', slug='role-1'),
|
|
Role(name='Role 2', slug='role-2'),
|
|
)
|
|
Role.objects.bulk_create(roles)
|
|
|
|
prefixes = (
|
|
Prefix(prefix=IPNetwork('10.1.0.0/16'), vrf=vrfs[0], scope=sites[0], role=roles[0]),
|
|
Prefix(prefix=IPNetwork('10.2.0.0/16'), vrf=vrfs[0], scope=sites[0], role=roles[0]),
|
|
Prefix(prefix=IPNetwork('10.3.0.0/16'), vrf=vrfs[0], scope=sites[0], role=roles[0]),
|
|
)
|
|
Prefix.objects.bulk_create(prefixes)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'prefix': IPNetwork('192.0.2.0/24'),
|
|
'scope_content_type': ContentType.objects.get_for_model(Site).pk,
|
|
'scope_object_id': sites[1].pk,
|
|
'vrf': vrfs[1].pk,
|
|
'tenant': None,
|
|
'vlan': None,
|
|
'status': PrefixStatusChoices.STATUS_RESERVED,
|
|
'role': roles[1].pk,
|
|
'is_pool': True,
|
|
'description': 'A new prefix',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
site = sites[0]
|
|
cls.csv_data = {
|
|
'default': (
|
|
"vrf,prefix,status,scope_type,scope_id",
|
|
f"VRF 1,10.4.0.0/16,active,dcim.site,{site.pk}",
|
|
f"VRF 1,10.5.0.0/16,active,dcim.site,{site.pk}",
|
|
f"VRF 1,10.6.0.0/16,active,dcim.site,{site.pk}",
|
|
),
|
|
'scope_name': (
|
|
"vrf,prefix,status,scope_type,scope_name",
|
|
f"VRF 1,10.4.0.0/16,active,dcim.site,{site.name}",
|
|
f"VRF 1,10.5.0.0/16,active,dcim.site,{site.name}",
|
|
f"VRF 1,10.6.0.0/16,active,dcim.site,{site.name}",
|
|
),
|
|
}
|
|
|
|
cls.csv_update_data = (
|
|
"id,description,status",
|
|
f"{prefixes[0].pk},New description 7,{PrefixStatusChoices.STATUS_RESERVED}",
|
|
f"{prefixes[1].pk},New description 8,{PrefixStatusChoices.STATUS_RESERVED}",
|
|
f"{prefixes[2].pk},New description 9,{PrefixStatusChoices.STATUS_RESERVED}",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'vrf': vrfs[1].pk,
|
|
'tenant': None,
|
|
'status': PrefixStatusChoices.STATUS_RESERVED,
|
|
'role': roles[1].pk,
|
|
'is_pool': False,
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_bulk_edit_htmx_dependent_field_refresh_skips_validation(self):
|
|
"""An HTMX content-type change (no _apply) re-renders the bulk-edit form without validation errors."""
|
|
prefix = Prefix.objects.create(prefix=IPNetwork('10.99.0.0/24'))
|
|
self.add_permissions('ipam.view_prefix', 'ipam.change_prefix')
|
|
|
|
data = {
|
|
'pk': [prefix.pk],
|
|
'scope_content_type': ContentType.objects.get_for_model(Site).pk,
|
|
# The client-side hx-on::config-request clears the paired object id on a type change.
|
|
'scope_object_id': '',
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data, headers={'HX-Request': 'true'})
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertNotContains(response, 'Please select a site')
|
|
# The object selector is rebuilt for the new type rather than erroring out.
|
|
self.assertContains(response, 'name="scope_object_id"')
|
|
self.assertContains(response, 'data-url="/api/dcim/sites/"')
|
|
|
|
def test_bulk_edit_apply_still_validates_incomplete_scope(self):
|
|
"""A real apply with a content type but no object still surfaces the validation error."""
|
|
prefix = Prefix.objects.create(prefix=IPNetwork('10.99.1.0/24'))
|
|
self.add_permissions('ipam.view_prefix', 'ipam.change_prefix')
|
|
|
|
data = {
|
|
'pk': [prefix.pk],
|
|
'_apply': '1',
|
|
'scope_content_type': ContentType.objects.get_for_model(Site).pk,
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data)
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertContains(response, 'Please select a site')
|
|
|
|
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
|
def test_bulk_edit_applies_scope(self):
|
|
"""A bulk edit which sets a scope persists the generic foreign key to every selected object."""
|
|
site = Site.objects.first()
|
|
prefixes = (
|
|
Prefix.objects.create(prefix=IPNetwork('10.98.0.0/24')),
|
|
Prefix.objects.create(prefix=IPNetwork('10.98.1.0/24')),
|
|
)
|
|
self.add_permissions('ipam.view_prefix', 'ipam.change_prefix')
|
|
|
|
data = {
|
|
'pk': [p.pk for p in prefixes],
|
|
'_apply': '1',
|
|
'scope_content_type': ContentType.objects.get_for_model(Site).pk,
|
|
'scope_object_id': site.pk,
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
for prefix in prefixes:
|
|
prefix.refresh_from_db()
|
|
self.assertEqual(prefix.scope_type, ContentType.objects.get_for_model(Site))
|
|
self.assertEqual(prefix.scope_id, site.pk)
|
|
self.assertEqual(prefix.scope, site)
|
|
|
|
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
|
def test_bulk_edit_nullifies_scope(self):
|
|
"""Nullifying the scope in a bulk edit clears both concrete GFK columns on every object."""
|
|
# The prefixes created in setUpTestData are already scoped to a Site.
|
|
prefixes = Prefix.objects.filter(scope_id__isnull=False)
|
|
self.assertTrue(prefixes.exists())
|
|
self.add_permissions('ipam.view_prefix', 'ipam.change_prefix')
|
|
|
|
data = {
|
|
'pk': [p.pk for p in prefixes],
|
|
'_apply': '1',
|
|
'_nullify': ['scope'],
|
|
# An empty scope alongside _nullify must not trigger the incomplete-scope validation error.
|
|
'scope_content_type': '',
|
|
'scope_object_id': '',
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
for prefix in prefixes:
|
|
prefix.refresh_from_db()
|
|
self.assertIsNone(prefix.scope_type_id)
|
|
self.assertIsNone(prefix.scope_id)
|
|
|
|
def test_scope_object_selector_restricted_by_permissions(self):
|
|
"""A constrained user cannot assign a scope object outside their permitted object set."""
|
|
sites = Site.objects.all()[:2]
|
|
permitted_site, forbidden_site = sites[0], sites[1]
|
|
|
|
# Grant add/view but constrain viewable Sites to a single object, mirroring how the object
|
|
# selector's queryset is narrowed by restrict_form_fields() in a real request.
|
|
self.add_permissions('ipam.add_prefix')
|
|
site_perm = ObjectPermission(
|
|
name='Restricted sites', actions=['view'], constraints={'pk': permitted_site.pk}
|
|
)
|
|
site_perm.save()
|
|
site_perm.users.add(self.user)
|
|
site_perm.object_types.add(ObjectType.objects.get_for_model(Site))
|
|
|
|
site_ct = ContentType.objects.get_for_model(Site)
|
|
|
|
# A minimal payload isolates the scope restriction from unrelated FK field permissions.
|
|
base = {
|
|
'prefix': '10.97.0.0/24',
|
|
'status': PrefixStatusChoices.STATUS_ACTIVE,
|
|
'scope_content_type': site_ct.pk,
|
|
}
|
|
|
|
# Assigning the forbidden site is rejected: it is not among the user's permitted choices.
|
|
data = post_data({**base, 'scope_object_id': forbidden_site.pk})
|
|
response = self.client.post(self._get_url('add'), data)
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertFalse(Prefix.objects.filter(prefix='10.97.0.0/24').exists())
|
|
|
|
# Assigning the permitted site succeeds.
|
|
data = post_data({**base, 'prefix': '10.97.1.0/24', 'scope_object_id': permitted_site.pk})
|
|
response = self.client.post(self._get_url('add'), data)
|
|
self.assertHttpStatus(response, 302)
|
|
prefix = Prefix.objects.get(prefix='10.97.1.0/24')
|
|
self.assertEqual(prefix.scope, permitted_site)
|
|
|
|
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
|
def test_scope_rejects_object_id_from_other_content_type(self):
|
|
"""A submitted object ID belonging to a different content type is rejected, not silently accepted."""
|
|
region = Region.objects.create(name='Region 1', slug='region-1')
|
|
self.add_permissions('ipam.add_prefix')
|
|
|
|
# Choose a Site pk which is not also a valid Region pk. The two tables' sequences are
|
|
# independent and are not rolled back between test classes, so a collision would otherwise
|
|
# make the "forbidden" pk a legitimate Region and depend on test ordering.
|
|
site = Site.objects.exclude(pk__in=Region.objects.values_list('pk', flat=True)).first()
|
|
self.assertIsNotNone(site)
|
|
|
|
# Region content type paired with a Site's pk: the object must be validated against the
|
|
# selected type, so a Site pk cannot resurface as a Region.
|
|
data = post_data({
|
|
**self.form_data,
|
|
'prefix': IPNetwork('10.96.0.0/24'),
|
|
'scope_content_type': ContentType.objects.get_for_model(Region).pk,
|
|
'scope_object_id': site.pk,
|
|
})
|
|
response = self.client.post(self._get_url('add'), data)
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertFalse(Prefix.objects.filter(prefix='10.96.0.0/24').exists())
|
|
|
|
# A matching Region pk under the Region content type validates.
|
|
data['prefix'] = '10.96.1.0/24'
|
|
data['scope_object_id'] = region.pk
|
|
response = self.client.post(self._get_url('add'), data)
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(Prefix.objects.get(prefix='10.96.1.0/24').scope, region)
|
|
|
|
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
|
def test_scope_rejects_malformed_input(self):
|
|
"""Malformed scope input is rejected as invalid rather than raising a server error."""
|
|
site = Site.objects.first()
|
|
site_ct = ContentType.objects.get_for_model(Site).pk
|
|
self.add_permissions('ipam.add_prefix')
|
|
|
|
malformed = (
|
|
{'scope_content_type': 'not-a-number', 'scope_object_id': site.pk},
|
|
{'scope_content_type': '-1', 'scope_object_id': site.pk},
|
|
{'scope_content_type': site_ct, 'scope_object_id': 'not-a-number'},
|
|
# An object ID which overflows PositiveBigIntegerField: it reaches the DB as a filter
|
|
# argument, so a missing guard would surface a driver error (500) rather than a clean rejection.
|
|
{'scope_content_type': site_ct, 'scope_object_id': '9' * 30},
|
|
)
|
|
for i, scope in enumerate(malformed):
|
|
with self.subTest(scope=scope):
|
|
data = post_data({**self.form_data, 'prefix': IPNetwork(f'10.95.{i}.0/24'), **scope})
|
|
response = self.client.post(self._get_url('add'), data)
|
|
# Rejected with a re-rendered form (200), never a 500, and no object created.
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertFalse(Prefix.objects.filter(prefix=f'10.95.{i}.0/24').exists())
|
|
# The rejection lands on the scope field, not on some unrelated field.
|
|
self.assertIn('scope', response.context['form'].errors)
|
|
|
|
def test_bulk_add_ipv4_prefixes(self):
|
|
"""Test bulk creating IPv4 prefixes using a pattern."""
|
|
self.add_permissions('ipam.view_prefix')
|
|
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
|
|
obj_perm.save()
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
|
|
|
initial_count = Prefix.objects.count()
|
|
url = reverse('ipam:prefix_bulk_add')
|
|
data = {
|
|
'pattern': '10.0.[0-2].0/24',
|
|
'status': PrefixStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(url, data)
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(Prefix.objects.count(), initial_count + 3)
|
|
|
|
for i in range(3):
|
|
self.assertTrue(Prefix.objects.filter(prefix=IPNetwork(f'10.0.{i}.0/24')).exists())
|
|
|
|
def test_bulk_add_ipv6_prefixes(self):
|
|
"""Test bulk creating IPv6 prefixes using a pattern."""
|
|
self.add_permissions('ipam.view_prefix')
|
|
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
|
|
obj_perm.save()
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
|
|
|
initial_count = Prefix.objects.count()
|
|
url = reverse('ipam:prefix_bulk_add')
|
|
data = {
|
|
'pattern': 'fd00:db8:[0-3]::/48',
|
|
'status': PrefixStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(url, data)
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(Prefix.objects.count(), initial_count + 4)
|
|
|
|
for i in range(4):
|
|
self.assertTrue(Prefix.objects.filter(prefix=IPNetwork(f'fd00:db8:{i}::/48')).exists())
|
|
|
|
def test_bulk_add_ipv6_prefixes_uppercase_hex(self):
|
|
"""Test bulk creating IPv6 prefixes using uppercase hex in the pattern."""
|
|
self.add_permissions('ipam.view_prefix')
|
|
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
|
|
obj_perm.save()
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
|
|
|
initial_count = Prefix.objects.count()
|
|
url = reverse('ipam:prefix_bulk_add')
|
|
data = {
|
|
'pattern': 'fd00:0:0:[48-4F]00::/56',
|
|
'status': PrefixStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(url, data)
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(Prefix.objects.count(), initial_count + 8)
|
|
|
|
expected_hex = ['48', '49', '4a', '4b', '4c', '4d', '4e', '4f']
|
|
for h in expected_hex:
|
|
prefix_str = f'fd00:0:0:{h}00::/56'
|
|
self.assertTrue(
|
|
Prefix.objects.filter(prefix=IPNetwork(prefix_str)).exists(),
|
|
f'Expected prefix {prefix_str} was not created'
|
|
)
|
|
|
|
def test_bulk_add_prefixes_with_changelog_message(self):
|
|
self.add_permissions('ipam.view_prefix')
|
|
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
|
|
obj_perm.save()
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix))
|
|
|
|
changelog_message = 'Bulk-created prefixes'
|
|
prefixes = [IPNetwork(f'198.18.{i}.0/24') for i in range(3)]
|
|
url = reverse('ipam:prefix_bulk_add')
|
|
data = {
|
|
'pattern': '198.18.[0-2].0/24',
|
|
'status': PrefixStatusChoices.STATUS_ACTIVE,
|
|
'changelog_message': changelog_message,
|
|
}
|
|
|
|
response = self.client.post(url, data)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
created_prefixes = list(Prefix.objects.filter(prefix__in=prefixes))
|
|
self.assertEqual(len(created_prefixes), len(prefixes))
|
|
|
|
objectchanges = ObjectChange.objects.filter(
|
|
action=ObjectChangeActionChoices.ACTION_CREATE,
|
|
changed_object_type=ContentType.objects.get_for_model(Prefix),
|
|
changed_object_id__in=[obj.pk for obj in created_prefixes],
|
|
)
|
|
self.assertEqual(objectchanges.count(), len(prefixes))
|
|
for objectchange in objectchanges:
|
|
self.assertEqual(objectchange.message, changelog_message)
|
|
|
|
def test_prefix_prefixes(self):
|
|
self.add_permissions('ipam.view_prefix')
|
|
prefixes = (
|
|
Prefix(prefix=IPNetwork('192.168.0.0/16')),
|
|
Prefix(prefix=IPNetwork('192.168.1.0/24')),
|
|
Prefix(prefix=IPNetwork('192.168.2.0/24')),
|
|
Prefix(prefix=IPNetwork('192.168.3.0/24')),
|
|
)
|
|
Prefix.objects.bulk_create(prefixes)
|
|
self.assertEqual(prefixes[0].get_child_prefixes().count(), 3)
|
|
|
|
url = reverse('ipam:prefix_prefixes', kwargs={'pk': prefixes[0].pk})
|
|
self.assertHttpStatus(self.client.get(url), 200)
|
|
|
|
def test_prefix_prefixes_add_links_include_scope_params(self):
|
|
"""Child-prefix Add links pre-populate scope via the GenericObjectChoiceField subwidget params."""
|
|
self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
|
|
|
|
site = Site.objects.create(name='Scope Site', slug='scope-site')
|
|
parent = Prefix.objects.create(prefix=IPNetwork('203.0.113.0/24'), scope=site)
|
|
Prefix.objects.create(prefix=IPNetwork('203.0.113.0/26'), scope=site)
|
|
|
|
url = reverse('ipam:prefix_prefixes', kwargs={'pk': parent.pk})
|
|
response = self.client.get(url)
|
|
|
|
self.assertHttpStatus(response, 200)
|
|
scope_ct = ContentType.objects.get_for_model(Site)
|
|
# The new GenericObjectChoiceField reads these subwidget-named query params.
|
|
self.assertContains(response, f'scope_content_type={scope_ct.pk}')
|
|
self.assertContains(response, f'scope_object_id={site.pk}')
|
|
|
|
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'))
|
|
ip_ranges = (
|
|
IPRange(start_address='192.168.0.1/24', end_address='192.168.0.100/24', size=99),
|
|
IPRange(start_address='192.168.1.1/24', end_address='192.168.1.100/24', size=99),
|
|
IPRange(start_address='192.168.2.1/24', end_address='192.168.2.100/24', size=99),
|
|
)
|
|
IPRange.objects.bulk_create(ip_ranges)
|
|
self.assertEqual(prefix.get_child_ranges().count(), 3)
|
|
|
|
url = reverse('ipam:prefix_ipranges', kwargs={'pk': prefix.pk})
|
|
self.assertHttpStatus(self.client.get(url), 200)
|
|
|
|
def test_prefix_ipaddresses(self):
|
|
self.add_permissions('ipam.view_prefix', 'ipam.view_ipaddress', 'ipam.view_iprange')
|
|
prefix = Prefix.objects.create(prefix=IPNetwork('192.168.0.0/16'))
|
|
ip_addresses = (
|
|
IPAddress(address=IPNetwork('192.168.0.1/16')),
|
|
IPAddress(address=IPNetwork('192.168.0.2/16')),
|
|
IPAddress(address=IPNetwork('192.168.0.3/16')),
|
|
)
|
|
IPAddress.objects.bulk_create(ip_addresses)
|
|
self.assertEqual(prefix.get_child_ips().count(), 3)
|
|
|
|
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 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'))
|
|
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.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."""
|
|
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_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
|
|
# mark-populated child IP ranges. Make sure a single-address range
|
|
# (start_address == end_address) renders without errors and is shown
|
|
# in its range-like display form rather than as a plain IP address.
|
|
prefix = Prefix.objects.create(prefix=IPNetwork('192.168.0.0/16'))
|
|
IPAddress.objects.create(address=IPNetwork('192.168.0.1/16'))
|
|
IPRange.objects.create(
|
|
start_address=IPNetwork('192.168.0.50/16'),
|
|
end_address=IPNetwork('192.168.0.50/16'),
|
|
mark_populated=True,
|
|
)
|
|
|
|
url = reverse('ipam:prefix_ipaddresses', kwargs={'pk': prefix.pk})
|
|
response = self.client.get(url)
|
|
self.assertHttpStatus(response, 200)
|
|
# The single-address range is rendered with both endpoints, not as
|
|
# 192.168.0.50/16 (which would make it indistinguishable from an
|
|
# IPAddress row in this mixed-record view).
|
|
self.assertContains(response, '192.168.0.50-192.168.0.50/16')
|
|
|
|
def test_prefix_import(self):
|
|
"""
|
|
Custom import test for YAML-based imports (versus CSV)
|
|
"""
|
|
self.add_permissions('dcim.view_site', 'ipam.view_vlan')
|
|
site = Site.objects.get(name='Site 1')
|
|
IMPORT_DATA = f"""
|
|
prefix: 10.1.1.0/24
|
|
status: active
|
|
vlan: 101
|
|
scope_type: dcim.site
|
|
scope_id: {site.pk}
|
|
"""
|
|
# Note, a site is not tied to the VLAN to verify the fix for #12622
|
|
VLAN.objects.create(vid=101, name='VLAN101')
|
|
|
|
# Add all required permissions to the test user
|
|
self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
|
|
|
|
form_data = {
|
|
'data': IMPORT_DATA,
|
|
'format': 'yaml'
|
|
}
|
|
response = self.client.post(reverse('ipam:prefix_bulk_import'), data=form_data, follow=True)
|
|
self.assertHttpStatus(response, 200)
|
|
|
|
prefix = Prefix.objects.get(prefix='10.1.1.0/24')
|
|
self.assertEqual(prefix.status, PrefixStatusChoices.STATUS_ACTIVE)
|
|
self.assertEqual(prefix.vlan.vid, 101)
|
|
self.assertEqual(prefix.scope, site)
|
|
|
|
def test_prefix_import_with_scope_name(self):
|
|
"""
|
|
Test YAML-based import using scope_name instead of scope_id.
|
|
"""
|
|
self.add_permissions('dcim.view_site')
|
|
site = Site.objects.get(name='Site 1')
|
|
IMPORT_DATA = """
|
|
prefix: 10.1.3.0/24
|
|
status: active
|
|
scope_type: dcim.site
|
|
scope_name: Site 1
|
|
"""
|
|
# Add all required permissions to the test user
|
|
self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
|
|
|
|
form_data = {
|
|
'data': IMPORT_DATA,
|
|
'format': 'yaml'
|
|
}
|
|
response = self.client.post(reverse('ipam:prefix_bulk_import'), data=form_data, follow=True)
|
|
self.assertHttpStatus(response, 200)
|
|
|
|
prefix = Prefix.objects.get(prefix='10.1.3.0/24')
|
|
self.assertEqual(prefix.status, PrefixStatusChoices.STATUS_ACTIVE)
|
|
self.assertEqual(prefix.scope, site)
|
|
|
|
def test_prefix_import_with_vlan_group(self):
|
|
"""
|
|
This test covers a unique import edge case where VLAN group is specified during the import.
|
|
"""
|
|
self.add_permissions('dcim.view_site', 'ipam.view_vlan', 'ipam.view_vlangroup')
|
|
site = Site.objects.get(name='Site 1')
|
|
IMPORT_DATA = f"""
|
|
prefix: 10.1.2.0/24
|
|
status: active
|
|
scope_type: dcim.site
|
|
scope_id: {site.pk}
|
|
vlan_group: Group 1
|
|
vlan: 102
|
|
"""
|
|
vlan_group = VLANGroup.objects.create(name='Group 1', slug='group-1', scope=Site.objects.get(name="Site 1"))
|
|
VLAN.objects.create(vid=102, name='VLAN102', group=vlan_group)
|
|
|
|
# Add all required permissions to the test user
|
|
self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
|
|
|
|
form_data = {
|
|
'data': IMPORT_DATA,
|
|
'format': 'yaml'
|
|
}
|
|
response = self.client.post(reverse('ipam:prefix_bulk_import'), data=form_data, follow=True)
|
|
self.assertHttpStatus(response, 200)
|
|
|
|
prefix = Prefix.objects.get(prefix='10.1.2.0/24')
|
|
self.assertEqual(prefix.status, PrefixStatusChoices.STATUS_ACTIVE)
|
|
self.assertEqual(prefix.vlan.vid, 102)
|
|
self.assertEqual(prefix.scope, site)
|
|
|
|
def test_prefix_import_with_vlan_site_multiple_vlans_same_vid(self):
|
|
"""
|
|
Test import when multiple VLANs exist with the same vid but different sites.
|
|
Ref: #20560
|
|
"""
|
|
self.add_permissions('dcim.view_site', 'ipam.view_vlan')
|
|
site1 = Site.objects.get(name='Site 1')
|
|
site2 = Site.objects.get(name='Site 2')
|
|
|
|
# Create VLANs with the same vid but different sites
|
|
vlan1 = VLAN.objects.create(vid=1, name='VLAN1-Site1', site=site1)
|
|
VLAN.objects.create(vid=1, name='VLAN1-Site2', site=site2) # Create ambiguity
|
|
|
|
# Import prefix with vlan_site specified
|
|
IMPORT_DATA = f"""
|
|
prefix: 10.11.0.0/22
|
|
status: active
|
|
scope_type: dcim.site
|
|
scope_id: {site1.pk}
|
|
vlan_site: {site1.name}
|
|
vlan: 1
|
|
description: LOC02-MGMT
|
|
"""
|
|
|
|
# Add all required permissions to the test user
|
|
self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
|
|
|
|
form_data = {
|
|
'data': IMPORT_DATA,
|
|
'format': 'yaml'
|
|
}
|
|
response = self.client.post(reverse('ipam:prefix_bulk_import'), data=form_data, follow=True)
|
|
self.assertHttpStatus(response, 200)
|
|
|
|
# Verify the prefix was created with the correct VLAN
|
|
prefix = Prefix.objects.get(prefix='10.11.0.0/22')
|
|
self.assertEqual(prefix.vlan, vlan1)
|
|
|
|
def test_prefix_import_with_vlan_site_and_global_vlan(self):
|
|
"""
|
|
Test import when a global VLAN (no site) and site-specific VLAN exist with same vid.
|
|
When vlan_site is specified, should prefer the site-specific VLAN.
|
|
Ref: #20560
|
|
"""
|
|
self.add_permissions('dcim.view_site', 'ipam.view_vlan')
|
|
site1 = Site.objects.get(name='Site 1')
|
|
|
|
# Create a global VLAN (no site) and a site-specific VLAN with the same vid
|
|
VLAN.objects.create(vid=10, name='VLAN10-Global', site=None) # Create ambiguity
|
|
vlan_site = VLAN.objects.create(vid=10, name='VLAN10-Site1', site=site1)
|
|
|
|
# Import prefix with vlan_site specified
|
|
IMPORT_DATA = f"""
|
|
prefix: 10.12.0.0/22
|
|
status: active
|
|
scope_type: dcim.site
|
|
scope_id: {site1.pk}
|
|
vlan_site: {site1.name}
|
|
vlan: 10
|
|
description: Test Site-Specific VLAN
|
|
"""
|
|
|
|
# Add all required permissions to the test user
|
|
self.add_permissions('ipam.view_prefix', 'ipam.add_prefix')
|
|
|
|
form_data = {
|
|
'data': IMPORT_DATA,
|
|
'format': 'yaml'
|
|
}
|
|
response = self.client.post(reverse('ipam:prefix_bulk_import'), data=form_data, follow=True)
|
|
self.assertHttpStatus(response, 200)
|
|
|
|
# Verify the prefix was created with the site-specific VLAN (not the global one)
|
|
prefix = Prefix.objects.get(prefix='10.12.0.0/22')
|
|
self.assertEqual(prefix.vlan, vlan_site)
|
|
|
|
|
|
class IPRangeTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = IPRange
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
vrfs = (
|
|
VRF(name='VRF 1', rd='65000:1'),
|
|
VRF(name='VRF 2', rd='65000:2'),
|
|
)
|
|
VRF.objects.bulk_create(vrfs)
|
|
|
|
roles = (
|
|
Role(name='Role 1', slug='role-1'),
|
|
Role(name='Role 2', slug='role-2'),
|
|
)
|
|
Role.objects.bulk_create(roles)
|
|
|
|
ip_ranges = (
|
|
IPRange(start_address='192.168.0.10/24', end_address='192.168.0.100/24', size=91),
|
|
IPRange(start_address='192.168.1.10/24', end_address='192.168.1.100/24', size=91),
|
|
IPRange(start_address='192.168.2.10/24', end_address='192.168.2.100/24', size=91),
|
|
IPRange(start_address='192.168.3.10/24', end_address='192.168.3.100/24', size=91),
|
|
IPRange(start_address='192.168.4.10/24', end_address='192.168.4.100/24', size=91),
|
|
)
|
|
IPRange.objects.bulk_create(ip_ranges)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'start_address': IPNetwork('192.0.5.10/24'),
|
|
'end_address': IPNetwork('192.0.5.100/24'),
|
|
'vrf': vrfs[1].pk,
|
|
'tenant': None,
|
|
'vlan': None,
|
|
'status': IPRangeStatusChoices.STATUS_RESERVED,
|
|
'role': roles[1].pk,
|
|
'is_pool': True,
|
|
'description': 'A new IP range',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"vrf,start_address,end_address,status",
|
|
"VRF 1,10.1.0.1/16,10.1.9.254/16,active",
|
|
"VRF 1,10.2.0.1/16,10.2.9.254/16,active",
|
|
"VRF 1,10.3.0.1/16,10.3.9.254/16,active",
|
|
# Single-address range (start == end)
|
|
"VRF 1,10.4.0.1/16,10.4.0.1/16,active",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,description,status",
|
|
f"{ip_ranges[0].pk},New description 7,{IPRangeStatusChoices.STATUS_RESERVED}",
|
|
f"{ip_ranges[1].pk},New description 8,{IPRangeStatusChoices.STATUS_RESERVED}",
|
|
f"{ip_ranges[2].pk},New description 9,{IPRangeStatusChoices.STATUS_RESERVED}",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'vrf': vrfs[1].pk,
|
|
'tenant': None,
|
|
'status': IPRangeStatusChoices.STATUS_RESERVED,
|
|
'role': roles[1].pk,
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_iprange_ipaddresses(self):
|
|
self.add_permissions('ipam.view_iprange', 'ipam.view_ipaddress')
|
|
iprange = IPRange.objects.create(
|
|
start_address=IPNetwork('192.168.0.1/24'),
|
|
end_address=IPNetwork('192.168.0.100/24'),
|
|
size=99
|
|
)
|
|
ip_addresses = (
|
|
IPAddress(address=IPNetwork('192.168.0.1/24')),
|
|
IPAddress(address=IPNetwork('192.168.0.2/24')),
|
|
IPAddress(address=IPNetwork('192.168.0.3/24')),
|
|
)
|
|
IPAddress.objects.bulk_create(ip_addresses)
|
|
self.assertEqual(iprange.get_child_ips().count(), 3)
|
|
|
|
url = reverse('ipam:iprange_ipaddresses', kwargs={'pk': iprange.pk})
|
|
self.assertHttpStatus(self.client.get(url), 200)
|
|
|
|
def test_create_single_address_range(self):
|
|
# Exercise the UI form path with start_address == end_address. The
|
|
# generic test_create_object_with_permission covers the multi-address
|
|
# case via cls.form_data; this test mirrors that flow for the single-
|
|
# address case so both paths stay covered.
|
|
self.add_permissions('ipam.add_iprange')
|
|
form_data = {
|
|
'start_address': '192.0.6.10/24',
|
|
'end_address': '192.0.6.10/24',
|
|
'status': IPRangeStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
initial_count = IPRange.objects.count()
|
|
|
|
response = self.client.post(reverse('ipam:iprange_add'), data=form_data)
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(IPRange.objects.count(), initial_count + 1)
|
|
|
|
iprange = IPRange.objects.order_by('pk').last()
|
|
self.assertEqual(str(iprange.start_address), '192.0.6.10/24')
|
|
self.assertEqual(str(iprange.end_address), '192.0.6.10/24')
|
|
self.assertEqual(iprange.size, 1)
|
|
|
|
|
|
class IPAddressTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = IPAddress
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
vrfs = (
|
|
VRF(name='VRF 1', rd='65000:1'),
|
|
VRF(name='VRF 2', rd='65000:2'),
|
|
)
|
|
VRF.objects.bulk_create(vrfs)
|
|
|
|
ipaddresses = (
|
|
IPAddress(address=IPNetwork('192.0.2.1/24'), vrf=vrfs[0]),
|
|
IPAddress(address=IPNetwork('192.0.2.2/24'), vrf=vrfs[0]),
|
|
IPAddress(address=IPNetwork('192.0.2.3/24'), vrf=vrfs[0]),
|
|
)
|
|
IPAddress.objects.bulk_create(ipaddresses)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
fhrp_groups = (
|
|
FHRPGroup(
|
|
name='FHRP Group 1',
|
|
protocol=FHRPGroupProtocolChoices.PROTOCOL_HSRP,
|
|
group_id=10
|
|
),
|
|
FHRPGroup(
|
|
name='FHRP Group 2',
|
|
protocol=FHRPGroupProtocolChoices.PROTOCOL_HSRP,
|
|
group_id=20
|
|
),
|
|
FHRPGroup(
|
|
name='FHRP Group 3',
|
|
protocol=FHRPGroupProtocolChoices.PROTOCOL_HSRP,
|
|
group_id=30
|
|
),
|
|
)
|
|
FHRPGroup.objects.bulk_create(fhrp_groups)
|
|
cls.form_data = {
|
|
'vrf': vrfs[1].pk,
|
|
'address': IPNetwork('192.0.2.99/24'),
|
|
'tenant': None,
|
|
'status': IPAddressStatusChoices.STATUS_RESERVED,
|
|
'role': IPAddressRoleChoices.ROLE_ANYCAST,
|
|
'nat_inside': None,
|
|
'dns_name': 'example',
|
|
'description': 'A new IP address',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"vrf,address,status,fhrp_group",
|
|
"VRF 1,192.0.2.4/24,active,FHRP Group 1",
|
|
"VRF 1,192.0.2.5/24,active,FHRP Group 2",
|
|
"VRF 1,192.0.2.6/24,active,FHRP Group 3",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,description,status",
|
|
f"{ipaddresses[0].pk},New description 7,{IPAddressStatusChoices.STATUS_RESERVED}",
|
|
f"{ipaddresses[1].pk},New description 8,{IPAddressStatusChoices.STATUS_RESERVED}",
|
|
f"{ipaddresses[2].pk},New description 9,{IPAddressStatusChoices.STATUS_RESERVED}",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'vrf': vrfs[1].pk,
|
|
'tenant': None,
|
|
'status': IPAddressStatusChoices.STATUS_RESERVED,
|
|
'role': IPAddressRoleChoices.ROLE_ANYCAST,
|
|
'dns_name': 'example',
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_bulk_add_ipaddresses_with_changelog_message(self):
|
|
self.add_permissions('ipam.view_ipaddress', 'ipam.view_vrf')
|
|
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
|
|
obj_perm.save()
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
|
|
|
|
vrf = VRF.objects.get(name='VRF 1')
|
|
changelog_message = 'Bulk-created IP addresses'
|
|
addresses = [IPNetwork(f'198.51.100.{i}/24') for i in range(10, 13)]
|
|
url = reverse('ipam:ipaddress_bulk_add')
|
|
data = {
|
|
'pattern': '198.51.100.[10-12]/24',
|
|
'vrf': vrf.pk,
|
|
'status': IPAddressStatusChoices.STATUS_ACTIVE,
|
|
'changelog_message': changelog_message,
|
|
}
|
|
|
|
response = self.client.post(url, data)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
created_addresses = list(IPAddress.objects.filter(address__in=addresses, vrf=vrf))
|
|
self.assertEqual(len(created_addresses), len(addresses))
|
|
|
|
objectchanges = ObjectChange.objects.filter(
|
|
action=ObjectChangeActionChoices.ACTION_CREATE,
|
|
changed_object_type=ContentType.objects.get_for_model(IPAddress),
|
|
changed_object_id__in=[obj.pk for obj in created_addresses],
|
|
)
|
|
self.assertEqual(objectchanges.count(), len(addresses))
|
|
for objectchange in objectchanges:
|
|
self.assertEqual(objectchange.message, changelog_message)
|
|
|
|
|
|
class FHRPGroupTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = FHRPGroup
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
fhrp_groups = (
|
|
FHRPGroup(
|
|
protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2,
|
|
group_id=10,
|
|
auth_type=FHRPGroupAuthTypeChoices.AUTHENTICATION_PLAINTEXT,
|
|
auth_key='foobar123',
|
|
),
|
|
FHRPGroup(
|
|
protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP3,
|
|
group_id=20,
|
|
auth_type=FHRPGroupAuthTypeChoices.AUTHENTICATION_MD5,
|
|
auth_key='foobar123',
|
|
),
|
|
FHRPGroup(
|
|
protocol=FHRPGroupProtocolChoices.PROTOCOL_HSRP,
|
|
group_id=30
|
|
),
|
|
)
|
|
FHRPGroup.objects.bulk_create(fhrp_groups)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'protocol': FHRPGroupProtocolChoices.PROTOCOL_VRRP2,
|
|
'group_id': 99,
|
|
'auth_type': FHRPGroupAuthTypeChoices.AUTHENTICATION_MD5,
|
|
'auth_key': 'abc123def456',
|
|
'description': 'Blah blah blah',
|
|
'name': 'test123 name',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"protocol,group_id,auth_type,auth_key,description",
|
|
"vrrp2,40,plaintext,foobar123,Foo",
|
|
"vrrp3,50,md5,foobar123,Bar",
|
|
"hsrp,60,,,",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{fhrp_groups[0].pk},FHRP Group 1,New description 1",
|
|
f"{fhrp_groups[1].pk},FHRP Group 2,New description 2",
|
|
f"{fhrp_groups[2].pk},FHRP Group 3,New description 3",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'protocol': FHRPGroupProtocolChoices.PROTOCOL_CARP,
|
|
}
|
|
|
|
|
|
class VLANGroupTestCase(ViewTestCases.OrganizationalObjectViewTestCase):
|
|
model = VLANGroup
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
vlan_groups = (
|
|
VLANGroup(name='VLAN Group 1', slug='vlan-group-1', scope=sites[0]),
|
|
VLANGroup(name='VLAN Group 2', slug='vlan-group-2', scope=sites[0]),
|
|
VLANGroup(name='VLAN Group 3', slug='vlan-group-3', scope=sites[0]),
|
|
)
|
|
VLANGroup.objects.bulk_create(vlan_groups)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'VLAN Group X',
|
|
'slug': 'vlan-group-x',
|
|
'description': 'A new VLAN group',
|
|
'vid_ranges': '100-199,300-399',
|
|
'scope_content_type': ContentType.objects.get_for_model(Site).pk,
|
|
'scope_object_id': sites[1].pk,
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = {
|
|
'default': (
|
|
"name,slug,scope_type,scope_id,description",
|
|
"VLAN Group 4,vlan-group-4,,,Fourth VLAN group",
|
|
f"VLAN Group 5,vlan-group-5,dcim.site,{sites[0].pk},Fifth VLAN group",
|
|
f"VLAN Group 6,vlan-group-6,dcim.site,{sites[1].pk},Sixth VLAN group",
|
|
),
|
|
'scope_name': (
|
|
"name,slug,scope_type,scope_name,description",
|
|
"VLAN Group 4,vlan-group-4,,,Fourth VLAN group",
|
|
f"VLAN Group 5,vlan-group-5,dcim.site,{sites[0].name},Fifth VLAN group",
|
|
f"VLAN Group 6,vlan-group-6,dcim.site,{sites[1].name},Sixth VLAN group",
|
|
),
|
|
}
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{vlan_groups[0].pk},VLAN Group 7,Fourth VLAN group7",
|
|
f"{vlan_groups[1].pk},VLAN Group 8,Fifth VLAN group8",
|
|
f"{vlan_groups[2].pk},VLAN Group 9,Sixth VLAN group9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'description': 'New description',
|
|
'scope_content_type': ContentType.objects.get_for_model(Site).pk,
|
|
'scope_object_id': sites[1].pk,
|
|
}
|
|
|
|
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)
|
|
|
|
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
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
sites = (
|
|
Site(name='Site 1', slug='site-1'),
|
|
Site(name='Site 2', slug='site-2'),
|
|
)
|
|
Site.objects.bulk_create(sites)
|
|
|
|
vlangroups = (
|
|
VLANGroup(name='VLAN Group 1', slug='vlan-group-1', scope=sites[0]),
|
|
VLANGroup(name='VLAN Group 2', slug='vlan-group-2', scope=sites[1]),
|
|
)
|
|
VLANGroup.objects.bulk_create(vlangroups)
|
|
|
|
roles = (
|
|
Role(name='Role 1', slug='role-1'),
|
|
Role(name='Role 2', slug='role-2'),
|
|
)
|
|
Role.objects.bulk_create(roles)
|
|
|
|
vlans = (
|
|
VLAN(group=vlangroups[0], vid=101, name='VLAN101', site=sites[0], role=roles[0]),
|
|
VLAN(group=vlangroups[0], vid=102, name='VLAN102', site=sites[0], role=roles[0]),
|
|
VLAN(group=vlangroups[0], vid=103, name='VLAN103', site=sites[0], role=roles[0]),
|
|
)
|
|
VLAN.objects.bulk_create(vlans)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'site': sites[1].pk,
|
|
'group': vlangroups[1].pk,
|
|
'vid': 999,
|
|
'name': 'VLAN999',
|
|
'tenant': None,
|
|
'status': VLANStatusChoices.STATUS_RESERVED,
|
|
'role': roles[1].pk,
|
|
'description': 'A new VLAN',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"vid,name,status",
|
|
"104,VLAN104,active",
|
|
"105,VLAN105,active",
|
|
"106,VLAN106,active",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{vlans[0].pk},VLAN107,New description 7",
|
|
f"{vlans[1].pk},VLAN108,New description 8",
|
|
f"{vlans[2].pk},VLAN109,New description 9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'site': sites[1].pk,
|
|
'group': vlangroups[1].pk,
|
|
'tenant': None,
|
|
'status': VLANStatusChoices.STATUS_RESERVED,
|
|
'role': roles[1].pk,
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_bulk_add_vlans(self):
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
group = VLANGroup.objects.get(name='VLAN Group 1')
|
|
initial_count = VLAN.objects.count()
|
|
expected_vids = (110, 120, 121, 122)
|
|
|
|
form_data = {
|
|
'pattern': '110,120-122',
|
|
'group': group.pk,
|
|
'name': 'Pool-{vid}',
|
|
'status': VLANStatusChoices.STATUS_RESERVED,
|
|
}
|
|
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(VLAN.objects.count(), initial_count + len(expected_vids))
|
|
|
|
for vid in expected_vids:
|
|
self.assertTrue(
|
|
VLAN.objects.filter(
|
|
group=group,
|
|
vid=vid,
|
|
name=f'Pool-{vid}'
|
|
).exists()
|
|
)
|
|
|
|
def test_bulk_add_vlans_rolls_back_on_duplicate_name(self):
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
group = VLANGroup.objects.get(name='VLAN Group 1')
|
|
initial_count = VLAN.objects.count()
|
|
|
|
form_data = {
|
|
'pattern': '110-112',
|
|
'group': group.pk,
|
|
'name': 'Duplicate name',
|
|
'status': VLANStatusChoices.STATUS_RESERVED,
|
|
}
|
|
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertEqual(VLAN.objects.count(), initial_count)
|
|
self.assertFalse(VLAN.objects.filter(group=group, vid=110).exists())
|
|
|
|
def test_bulk_add_vlans_rolls_back_when_any_id_outside_group_range(self):
|
|
self.add_permissions('ipam.add_vlan')
|
|
|
|
group = VLANGroup.objects.create(
|
|
name='Restricted VLAN Group',
|
|
slug='restricted-vlan-group',
|
|
vid_ranges=[NumericRange(200, 204)] # Valid VIDs: 200-203
|
|
)
|
|
initial_count = VLAN.objects.count()
|
|
|
|
form_data = {
|
|
'pattern': '200-203,500',
|
|
'group': group.pk,
|
|
'name': 'Restricted-{vid}',
|
|
'status': VLANStatusChoices.STATUS_RESERVED,
|
|
}
|
|
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertEqual(VLAN.objects.count(), initial_count)
|
|
self.assertFalse(VLAN.objects.filter(group=group, vid=200).exists())
|
|
self.assertFalse(VLAN.objects.filter(group=group, vid=203).exists())
|
|
self.assertFalse(VLAN.objects.filter(group=group, vid=500).exists())
|
|
|
|
def test_bulk_add_vlans_pattern_shapes(self):
|
|
"""Single values, multiple values, ranges, and combinations create the expected VLANs."""
|
|
self.add_permissions('ipam.add_vlan')
|
|
# The combination runs against a second group: subTests share one transaction, and VIDs
|
|
# 10 & 20 would otherwise collide with the multiple-values case via the (group, vid) constraint.
|
|
cases = (
|
|
('500', (500,), 'VLAN Group 1'),
|
|
('5,10,20', (5, 10, 20), 'VLAN Group 1'),
|
|
('600-605', tuple(range(600, 606)), 'VLAN Group 1'),
|
|
('1,10-20,300-305', (1, *range(10, 21), *range(300, 306)), 'VLAN Group 2'),
|
|
)
|
|
for pattern, expected_vids, group_name in cases:
|
|
with self.subTest(pattern=pattern):
|
|
group = VLANGroup.objects.get(name=group_name)
|
|
initial_count = VLAN.objects.count()
|
|
form_data = {
|
|
'pattern': pattern,
|
|
'group': group.pk,
|
|
'name': 'Pool-{vid}',
|
|
'status': VLANStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(VLAN.objects.count(), initial_count + len(expected_vids))
|
|
for vid in expected_vids:
|
|
self.assertTrue(VLAN.objects.filter(group=group, vid=vid, name=f'Pool-{vid}').exists())
|
|
|
|
def test_bulk_add_vlans_invalid_pattern(self):
|
|
"""An invalid pattern re-renders the form with a pattern error and creates nothing."""
|
|
self.add_permissions('ipam.add_vlan')
|
|
initial_count = VLAN.objects.count()
|
|
|
|
for pattern in ('abc', '20-10', '0', '4095', '10-'):
|
|
with self.subTest(pattern=pattern):
|
|
form_data = {
|
|
'pattern': pattern,
|
|
'name': 'Pool-{vid}',
|
|
'status': VLANStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertIn('pattern', response.context['form'].errors)
|
|
self.assertEqual(VLAN.objects.count(), initial_count)
|
|
|
|
def test_bulk_add_vlans_static_name_without_group(self):
|
|
"""A static name (no {vid} placeholder) is permitted across VLANs not assigned to a group."""
|
|
self.add_permissions('ipam.add_vlan')
|
|
initial_count = VLAN.objects.count()
|
|
|
|
form_data = {
|
|
'pattern': '710-712',
|
|
'name': 'Same name',
|
|
'status': VLANStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(VLAN.objects.count(), initial_count + 3)
|
|
self.assertEqual(VLAN.objects.filter(name='Same name').count(), 3)
|
|
|
|
def test_bulk_add_vlans_rolls_back_on_constrained_permission(self):
|
|
"""Bulk creation rolls back when a generated VLAN falls outside the user's add constraints."""
|
|
obj_perm = ObjectPermission(
|
|
name='Test permission',
|
|
actions=['add'],
|
|
constraints={'vid__lt': 120}
|
|
)
|
|
obj_perm.save()
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(VLAN))
|
|
|
|
initial_count = VLAN.objects.count()
|
|
form_data = {
|
|
'pattern': '110,120-122',
|
|
'name': 'Pool-{vid}',
|
|
'status': VLANStatusChoices.STATUS_ACTIVE,
|
|
}
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertEqual(VLAN.objects.count(), initial_count)
|
|
self.assertTrue(response.context['form'].non_field_errors())
|
|
|
|
def test_bulk_add_vlans_propagates_field_errors(self):
|
|
"""A per-object validation error on a non-pattern field is reported on the bulk-create form."""
|
|
self.add_permissions('ipam.add_vlan')
|
|
initial_count = VLAN.objects.count()
|
|
|
|
form_data = {
|
|
'pattern': '800',
|
|
'name': 'Pool-{vid}',
|
|
'status': VLANStatusChoices.STATUS_ACTIVE,
|
|
'qinq_role': VLANQinQRoleChoices.ROLE_CUSTOMER, # Requires an SVLAN
|
|
}
|
|
response = self.client.post(reverse('ipam:vlan_bulk_add'), form_data)
|
|
|
|
self.assertHttpStatus(response, 200)
|
|
self.assertEqual(VLAN.objects.count(), initial_count)
|
|
self.assertTrue(response.context['form'].non_field_errors())
|
|
|
|
|
|
class VLANTranslationPolicyTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = VLANTranslationPolicy
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
vlan_translation_policies = (
|
|
VLANTranslationPolicy(
|
|
name='Policy 1',
|
|
description='foobar1',
|
|
),
|
|
VLANTranslationPolicy(
|
|
name='Policy 2',
|
|
description='foobar2',
|
|
),
|
|
VLANTranslationPolicy(
|
|
name='Policy 3',
|
|
description='foobar3',
|
|
),
|
|
)
|
|
VLANTranslationPolicy.objects.bulk_create(vlan_translation_policies)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'Policy999',
|
|
'description': 'A new VLAN Translation Policy',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name,description",
|
|
"Policy101,foobar1",
|
|
"Policy102,foobar2",
|
|
"Policy103,foobar3",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{vlan_translation_policies[0].pk},Policy101,New description 1",
|
|
f"{vlan_translation_policies[1].pk},Policy102,New description 2",
|
|
f"{vlan_translation_policies[2].pk},Policy103,New description 3",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'description': 'New description',
|
|
}
|
|
|
|
|
|
class VLANTranslationRuleTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = VLANTranslationRule
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
vlan_translation_policies = (
|
|
VLANTranslationPolicy(
|
|
name='Policy 1',
|
|
description='foobar1',
|
|
),
|
|
VLANTranslationPolicy(
|
|
name='Policy 2',
|
|
description='foobar2',
|
|
),
|
|
VLANTranslationPolicy(
|
|
name='Policy 3',
|
|
description='foobar3',
|
|
),
|
|
)
|
|
VLANTranslationPolicy.objects.bulk_create(vlan_translation_policies)
|
|
|
|
vlan_translation_rules = (
|
|
VLANTranslationRule(
|
|
policy=vlan_translation_policies[0],
|
|
local_vid=100,
|
|
remote_vid=200,
|
|
),
|
|
VLANTranslationRule(
|
|
policy=vlan_translation_policies[0],
|
|
local_vid=101,
|
|
remote_vid=201,
|
|
),
|
|
VLANTranslationRule(
|
|
policy=vlan_translation_policies[1],
|
|
local_vid=102,
|
|
remote_vid=202,
|
|
),
|
|
)
|
|
VLANTranslationRule.objects.bulk_create(vlan_translation_rules)
|
|
|
|
cls.form_data = {
|
|
'policy': vlan_translation_policies[0].pk,
|
|
'local_vid': 300,
|
|
'remote_vid': 400,
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"policy,local_vid,remote_vid",
|
|
f"{vlan_translation_policies[0].name},103,203",
|
|
f"{vlan_translation_policies[0].name},104,204",
|
|
f"{vlan_translation_policies[1].name},105,205",
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,local_vid,remote_vid",
|
|
f"{vlan_translation_rules[0].pk},105,205",
|
|
f"{vlan_translation_rules[1].pk},106,206",
|
|
f"{vlan_translation_rules[2].pk},107,207",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'policy': vlan_translation_policies[2].pk,
|
|
}
|
|
|
|
|
|
class ServiceTemplateTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = ServiceTemplate
|
|
# port_mappings is an ArrayField, but the form submits it as a JSON string rather than a list, so
|
|
# the value isn't directly comparable to the stored field during the view test's edit assertions
|
|
validation_excluded_fields = ('port_mappings',)
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
service_templates = (
|
|
ServiceTemplate(name='Service Template 1', port_mappings=['tcp/101']),
|
|
ServiceTemplate(name='Service Template 2', port_mappings=['tcp/102']),
|
|
ServiceTemplate(name='Service Template 3', port_mappings=['tcp/103']),
|
|
)
|
|
ServiceTemplate.objects.bulk_create(service_templates)
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'name': 'Service Template X',
|
|
'port_mappings': '[{"protocol": "udp", "ports": "104,105"}]',
|
|
'description': 'A new service template',
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"name,port_mappings,description",
|
|
"Service Template 4,tcp/1,First service template",
|
|
"Service Template 5,tcp/2,Second service template",
|
|
'Service Template 6,"udp/3,tcp/4",Third service template',
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{service_templates[0].pk},Service Template 7,First service template7",
|
|
f"{service_templates[1].pk},Service Template 8,Second service template8",
|
|
f"{service_templates[2].pk},Service Template 9,Third service template9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_port_mappings_stored_from_form(self):
|
|
# port_mappings is in validation_excluded_fields (the form submits JSON rows, not a list), so the
|
|
# standard create/edit view tests can't compare it. Assert the form round-trip explicitly.
|
|
self.add_permissions('ipam.add_servicetemplate', 'ipam.change_servicetemplate')
|
|
|
|
# Create: a row's comma-separated ports expand into individual mappings
|
|
data = {
|
|
'name': 'Mappings Via Form',
|
|
'port_mappings': '[{"protocol": "udp", "ports": "104,105"}]',
|
|
}
|
|
self.assertHttpStatus(self.client.post(self._get_url('add'), data=post_data(data)), 302)
|
|
template = ServiceTemplate.objects.get(name='Mappings Via Form')
|
|
self.assertEqual(template.port_mappings, ['udp/104', 'udp/105'])
|
|
|
|
# Edit: multiple rows, a range, and a non-canonical port are expanded and normalized
|
|
data['port_mappings'] = '[{"protocol": "tcp", "ports": "080,8000-8002"}, {"protocol": "udp", "ports": "53"}]'
|
|
self.assertHttpStatus(
|
|
self.client.post(self._get_url('edit', template), data=post_data(data)), 302
|
|
)
|
|
template.refresh_from_db()
|
|
self.assertEqual(template.port_mappings, ['tcp/80', 'tcp/8000', 'tcp/8001', 'tcp/8002', 'udp/53'])
|
|
|
|
def test_bulk_edit_port_mappings(self):
|
|
# Bulk add/remove port mappings across selected templates (tags-style). ServiceTemplate is where
|
|
# the add/remove fields are declared (ServiceBulkEditForm inherits them), so cover it directly.
|
|
self.add_permissions('ipam.view_servicetemplate', 'ipam.change_servicetemplate')
|
|
templates = list(
|
|
ServiceTemplate.objects.filter(name__in=['Service Template 1', 'Service Template 2']).order_by('name')
|
|
)
|
|
data = {
|
|
'pk': [t.pk for t in templates],
|
|
'add_port_mappings': '[{"protocol": "udp", "ports": "53"}]',
|
|
'remove_port_mappings': '[{"protocol": "tcp", "ports": "101"}]',
|
|
'_apply': '',
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Service Template 1 (was tcp/101): tcp/101 removed, udp/53 added
|
|
self.assertEqual(ServiceTemplate.objects.get(pk=templates[0].pk).port_mappings, ['udp/53'])
|
|
# Service Template 2 (was tcp/102): remove is a no-op, udp/53 added
|
|
self.assertEqual(
|
|
ServiceTemplate.objects.get(pk=templates[1].pk).port_mappings, ['tcp/102', 'udp/53']
|
|
)
|
|
|
|
def test_bulk_edit_removing_all_mappings_is_rejected(self):
|
|
# Emptying an object's mappings must fail validation rather than persist an invalid object.
|
|
self.add_permissions('ipam.view_servicetemplate', 'ipam.change_servicetemplate')
|
|
template = ServiceTemplate.objects.get(name='Service Template 1')
|
|
data = {
|
|
'pk': [template.pk],
|
|
'remove_port_mappings': '[{"protocol": "tcp", "ports": "101"}]',
|
|
'_apply': '',
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data)
|
|
self.assertHttpStatus(response, 200) # Re-rendered with the error, not a redirect
|
|
template.refresh_from_db()
|
|
self.assertEqual(template.port_mappings, ['tcp/101'])
|
|
|
|
|
|
class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|
model = Service
|
|
# TODO, related to #9816, cannot validate GFK
|
|
# port_mappings is an ArrayField, but the form submits it as a JSON string rather than a list, so
|
|
# the value isn't directly comparable to the stored field during the view test's edit assertions
|
|
validation_excluded_fields = ('device', 'port_mappings')
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
|
|
site = Site.objects.create(name='Site 1', slug='site-1')
|
|
manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
|
|
devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1')
|
|
role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
|
|
device = Device.objects.create(name='Device 1', site=site, device_type=devicetype, role=role)
|
|
interface = Interface.objects.create(device=device, name='Interface 1', type=InterfaceTypeChoices.TYPE_VIRTUAL)
|
|
fhrp_group = FHRPGroup.objects.create(
|
|
name='Group 1', group_id=1234, protocol=FHRPGroupProtocolChoices.PROTOCOL_CARP
|
|
)
|
|
|
|
services = (
|
|
Service(parent=device, name='Service 1', port_mappings=['tcp/101']),
|
|
Service(parent=device, name='Service 2', port_mappings=['tcp/102']),
|
|
Service(parent=device, name='Service 3', port_mappings=['tcp/103']),
|
|
)
|
|
Service.objects.bulk_create(services)
|
|
|
|
ip_addresses = (
|
|
IPAddress(assigned_object=interface, address='192.0.2.1/24'),
|
|
IPAddress(assigned_object=interface, address='192.0.2.2/24'),
|
|
IPAddress(assigned_object=fhrp_group, address='192.0.2.3/24'),
|
|
)
|
|
IPAddress.objects.bulk_create(ip_addresses)
|
|
|
|
owner = Owner.objects.create(name='Owner 1')
|
|
|
|
tags = create_tags('Alpha', 'Bravo', 'Charlie')
|
|
|
|
cls.form_data = {
|
|
'parent_content_type': ContentType.objects.get_for_model(Device).pk,
|
|
'parent_object_id': device.pk,
|
|
'name': 'Service X',
|
|
'port_mappings': '[{"protocol": "tcp", "ports": "104,105"}, {"protocol": "udp", "ports": "104"}]',
|
|
'ipaddresses': [],
|
|
'description': 'A new service',
|
|
'owner': owner.pk,
|
|
'tags': [t.pk for t in tags],
|
|
}
|
|
|
|
cls.csv_data = (
|
|
"parent_object_type,parent,name,port_mappings,ipaddresses,description",
|
|
"dcim.device,Device 1,Service 1,tcp/1,192.0.2.1/24,First service",
|
|
"dcim.device,Device 1,Service 2,tcp/2,192.0.2.2/24,Second service",
|
|
"dcim.device,Device 1,Service 3,udp/3,,Third service",
|
|
'ipam.fhrpgroup,Group 1,Service 4,"tcp/4,udp/4",192.0.2.3/24,Fourth service',
|
|
)
|
|
|
|
cls.csv_update_data = (
|
|
"id,name,description",
|
|
f"{services[0].pk},Service 7,First service7",
|
|
f"{services[1].pk},Service 8,Second service8",
|
|
f"{services[2].pk},Service 9,Third service9",
|
|
)
|
|
|
|
cls.bulk_edit_data = {
|
|
'description': 'New description',
|
|
}
|
|
|
|
def test_unassigned_ip_addresses(self):
|
|
self.add_permissions('ipam.view_service', 'dcim.view_device', 'ipam.view_ipaddress')
|
|
device = Device.objects.first()
|
|
addr = IPAddress.objects.create(address='192.0.2.4/24')
|
|
csv_data = (
|
|
"parent_object_type,parent_object_id,name,port_mappings,ipaddresses,description",
|
|
f"dcim.device,{device.pk},Service 11,tcp/10,{addr.address},Eleventh service",
|
|
)
|
|
|
|
initial_count = self._get_queryset().count()
|
|
data = {
|
|
'data': '\n'.join(csv_data),
|
|
'format': ImportFormatChoices.CSV,
|
|
'csv_delimiter': CSVDelimiterChoices.AUTO,
|
|
}
|
|
|
|
# Assign model-level permission
|
|
obj_perm = ObjectPermission.objects.create(name='Test permission', actions=['add'])
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
|
|
|
|
# Test POST with permission
|
|
response = self.client.post(self._get_url('bulk_import'), data)
|
|
|
|
self.assertHttpStatus(response, 200)
|
|
form_errors = response.context['form'].errors
|
|
self.assertEqual(len(form_errors), 1)
|
|
self.assertIn(addr.address, form_errors['__all__'][0])
|
|
self.assertEqual(self._get_queryset().count(), initial_count)
|
|
|
|
def test_alternate_csv_import(self):
|
|
self.add_permissions('ipam.view_service', 'dcim.view_device', 'ipam.view_ipaddress')
|
|
device = Device.objects.first()
|
|
interface = device.interfaces.first()
|
|
addr = IPAddress.objects.create(assigned_object=interface, address='192.0.2.3/24')
|
|
csv_data = (
|
|
"parent_object_type,parent_object_id,name,port_mappings,ipaddresses,description",
|
|
f"dcim.device,{device.pk},Service 11,tcp/10,{addr.address},Eleventh service",
|
|
)
|
|
|
|
initial_count = self._get_queryset().count()
|
|
data = {
|
|
'data': '\n'.join(csv_data),
|
|
'format': ImportFormatChoices.CSV,
|
|
'csv_delimiter': CSVDelimiterChoices.AUTO,
|
|
}
|
|
|
|
# Assign model-level permission
|
|
obj_perm = ObjectPermission.objects.create(name='Test permission', actions=['add'])
|
|
obj_perm.users.add(self.user)
|
|
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
|
|
|
|
# Test POST with permission
|
|
response = self.client.post(self._get_url('bulk_import'), data)
|
|
|
|
if response.status_code != 302:
|
|
self.assertEqual(response.context['form'].errors, {}) # debugging aid
|
|
self.assertHttpStatus(response, 302)
|
|
self.assertEqual(self._get_queryset().count(), initial_count + len(csv_data) - 1)
|
|
|
|
def test_create_from_template(self):
|
|
self.add_permissions(
|
|
'ipam.view_service',
|
|
'ipam.add_service',
|
|
'dcim.view_device',
|
|
'ipam.view_servicetemplate',
|
|
)
|
|
|
|
device = Device.objects.first()
|
|
service_template = ServiceTemplate.objects.create(
|
|
name='HTTP',
|
|
port_mappings=['tcp/80'],
|
|
description='Hypertext transfer protocol'
|
|
)
|
|
|
|
request = {
|
|
'path': self._get_url('add'),
|
|
'data': {
|
|
'parent_content_type': ContentType.objects.get_for_model(Device).pk,
|
|
'parent_object_id': device.pk,
|
|
'service_template': service_template.pk,
|
|
},
|
|
}
|
|
|
|
self.assertHttpStatus(self.client.post(**request), 302)
|
|
instance = self._get_queryset().order_by('pk').last()
|
|
self.assertEqual(instance.parent, device)
|
|
self.assertEqual(instance.name, service_template.name)
|
|
self.assertEqual(instance.description, service_template.description)
|
|
# Port mappings should be copied from the template
|
|
self.assertEqual(instance.port_mappings, ['tcp/80'])
|
|
|
|
# EXEMPT_VIEW_PERMISSIONS matches the standard create/edit view tests: without it the form's related
|
|
# object fields (parent, tags) reject choices the test user has no view permission for.
|
|
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
|
def test_port_mappings_stored_from_form(self):
|
|
# port_mappings is in validation_excluded_fields (the form submits JSON rows, not a list), so the
|
|
# standard create/edit view tests can't compare it. Assert the form round-trip explicitly, using a
|
|
# multi-protocol payload — the whole point of the field — including the same port on two protocols.
|
|
self.add_permissions('ipam.add_service', 'ipam.change_service')
|
|
|
|
data = {**self.form_data, 'name': 'Mappings Via Form'}
|
|
self.assertHttpStatus(self.client.post(self._get_url('add'), data=post_data(data)), 302)
|
|
service = Service.objects.get(name='Mappings Via Form')
|
|
self.assertEqual(service.port_mappings, ['tcp/104', 'tcp/105', 'udp/104'])
|
|
|
|
# Edit: a range and a non-canonical port are expanded and normalized
|
|
data['port_mappings'] = '[{"protocol": "tcp", "ports": "080,8000-8002"}]'
|
|
self.assertHttpStatus(
|
|
self.client.post(self._get_url('edit', service), data=post_data(data)), 302
|
|
)
|
|
service.refresh_from_db()
|
|
self.assertEqual(service.port_mappings, ['tcp/80', 'tcp/8000', 'tcp/8001', 'tcp/8002'])
|
|
|
|
def test_bulk_edit_port_mappings(self):
|
|
# Bulk add/remove port mappings across selected services (tags-style)
|
|
self.add_permissions('ipam.view_service', 'ipam.change_service')
|
|
services = list(Service.objects.filter(name__in=['Service 1', 'Service 2']).order_by('name'))
|
|
data = {
|
|
'pk': [s.pk for s in services],
|
|
'add_port_mappings': '[{"protocol": "udp", "ports": "53"}]',
|
|
'remove_port_mappings': '[{"protocol": "tcp", "ports": "101"}]',
|
|
'_apply': '',
|
|
}
|
|
response = self.client.post(self._get_url('bulk_edit'), data)
|
|
self.assertHttpStatus(response, 302)
|
|
|
|
# Service 1 (was tcp/101): tcp/101 removed, udp/53 added
|
|
service1 = Service.objects.get(pk=services[0].pk)
|
|
self.assertEqual(service1.port_mappings, ['udp/53'])
|
|
# Service 2 (was tcp/102): remove is a no-op, udp/53 added
|
|
service2 = Service.objects.get(pk=services[1].pk)
|
|
self.assertEqual(service2.port_mappings, ['tcp/102', 'udp/53'])
|