Follow-up to #804

This commit is contained in:
Jeremy Stretch 2017-01-17 12:32:54 -05:00
parent 398faf518c
commit ab706d2440
2 changed files with 53 additions and 69 deletions

View File

@ -298,13 +298,11 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel):
def get_absolute_url(self): def get_absolute_url(self):
return reverse('ipam:prefix', args=[self.pk]) return reverse('ipam:prefix', args=[self.pk])
def duplicates(self): def get_duplicates(self):
return Prefix.objects.filter( return Prefix.objects.filter(vrf=self.vrf, prefix=str(self.prefix)).exclude(pk=self.pk)
vrf=self.vrf,
prefix=str(self.prefix)
).exclude(pk=self.pk)
def clean(self): def clean(self):
# Disallow host masks # Disallow host masks
if self.prefix: if self.prefix:
if self.prefix.version == 4 and self.prefix.prefixlen == 32: if self.prefix.version == 4 and self.prefix.prefixlen == 32:
@ -316,13 +314,14 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel):
'prefix': "Cannot create host addresses (/128) as prefixes. Create an IPv6 address instead." 'prefix': "Cannot create host addresses (/128) as prefixes. Create an IPv6 address instead."
}) })
if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): # Enforce unique IP space if applicable
dupes = self.duplicates() if (self.vrf is None and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique):
if dupes: duplicate_prefixes = self.get_duplicates()
if duplicate_prefixes:
raise ValidationError({ raise ValidationError({
'prefix': "Duplicate prefix found in {}: {}".format( 'prefix': "Duplicate prefix found in {}: {}".format(
"VRF {}".format(self.vrf) if self.vrf else "global table", "VRF {}".format(self.vrf) if self.vrf else "global table",
dupes.first(), duplicate_prefixes.first(),
) )
}) })
@ -415,22 +414,19 @@ class IPAddress(CreatedUpdatedModel, CustomFieldModel):
def get_absolute_url(self): def get_absolute_url(self):
return reverse('ipam:ipaddress', args=[self.pk]) return reverse('ipam:ipaddress', args=[self.pk])
def duplicates(self): def get_duplicates(self):
return IPAddress.objects.filter( return IPAddress.objects.filter(vrf=self.vrf, address__net_host=str(self.address.ip)).exclude(pk=self.pk)
vrf=self.vrf,
address__net_host=str(self.address.ip)
).exclude(pk=self.pk)
def clean(self): def clean(self):
# Enforce unique IP space if applicable # Enforce unique IP space if applicable
if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): if (self.vrf is None and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique):
dupes = self.duplicates() duplicate_ips = self.get_duplicates()
if dupes: if duplicate_ips:
raise ValidationError({ raise ValidationError({
'address': "Duplicate IP Address found in {}: {}".format( 'address': "Duplicate IP address found in {}: {}".format(
"VRF {}".format(self.vrf) if self.vrf else "global table", "VRF {}".format(self.vrf) if self.vrf else "global table",
dupes.first(), duplicate_ips.first(),
) )
}) })

View File

@ -8,65 +8,53 @@ from django.core.exceptions import ValidationError
class TestPrefix(TestCase): class TestPrefix(TestCase):
fixtures = [ @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
'dcim',
'ipam'
]
def test_create(self):
prefix = Prefix.objects.create(
prefix=netaddr.IPNetwork('10.1.1.0/24'),
status=1
)
self.assertIsNone(prefix.clean())
@override_settings(ENFORCE_GLOBAL_UNIQUE=True)
def test_duplicate_global(self): def test_duplicate_global(self):
prefix = Prefix.objects.create( Prefix.objects.create(prefix=netaddr.IPNetwork('192.0.2.0/24'))
prefix=netaddr.IPNetwork('10.1.1.0/24'), duplicate_prefix = Prefix(prefix=netaddr.IPNetwork('192.0.2.0/24'))
status=1 self.assertIsNone(duplicate_prefix.clean())
)
self.assertRaises(ValidationError, prefix.clean)
@override_settings(ENFORCE_GLOBAL_UNIQUE=True) @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
def test_duplicate_global_unique(self):
Prefix.objects.create(prefix=netaddr.IPNetwork('192.0.2.0/24'))
duplicate_prefix = Prefix(prefix=netaddr.IPNetwork('192.0.2.0/24'))
self.assertRaises(ValidationError, duplicate_prefix.clean)
def test_duplicate_vrf(self): def test_duplicate_vrf(self):
pfx_kwargs = { vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
"prefix": netaddr.IPNetwork('10.1.1.0/24'), Prefix.objects.create(vrf=vrf, prefix=netaddr.IPNetwork('192.0.2.0/24'))
"status": 1, duplicate_prefix = Prefix(vrf=vrf, prefix=netaddr.IPNetwork('192.0.2.0/24'))
"vrf": VRF.objects.create(name='Test', rd='1:1'), self.assertIsNone(duplicate_prefix.clean())
}
Prefix.objects.create(**pfx_kwargs) def test_duplicate_vrf_unique(self):
dup_prefix = Prefix.objects.create(**pfx_kwargs) vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
self.assertRaises(ValidationError, dup_prefix.clean) Prefix.objects.create(vrf=vrf, prefix=netaddr.IPNetwork('192.0.2.0/24'))
duplicate_prefix = Prefix(vrf=vrf, prefix=netaddr.IPNetwork('192.0.2.0/24'))
self.assertRaises(ValidationError, duplicate_prefix.clean)
class TestIPAddress(TestCase): class TestIPAddress(TestCase):
fixtures = [ @override_settings(ENFORCE_GLOBAL_UNIQUE=False)
'dcim',
'ipam'
]
def test_create(self):
address = IPAddress.objects.create(
address=netaddr.IPNetwork('10.0.254.1/24'),
)
self.assertIsNone(address.clean())
@override_settings(ENFORCE_GLOBAL_UNIQUE=True)
def test_duplicate_global(self): def test_duplicate_global(self):
address = IPAddress.objects.create( IPAddress.objects.create(address=netaddr.IPNetwork('192.0.2.1/24'))
address=netaddr.IPNetwork('10.0.254.1/24'), duplicate_ip = IPAddress(address=netaddr.IPNetwork('192.0.2.1/24'))
) self.assertIsNone(duplicate_ip.clean())
self.assertRaises(ValidationError, address.clean)
@override_settings(ENFORCE_GLOBAL_UNIQUE=True) @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
def test_duplicate_global_unique(self):
IPAddress.objects.create(address=netaddr.IPNetwork('192.0.2.1/24'))
duplicate_ip = IPAddress(address=netaddr.IPNetwork('192.0.2.1/24'))
self.assertRaises(ValidationError, duplicate_ip.clean)
def test_duplicate_vrf(self): def test_duplicate_vrf(self):
pfx_kwargs = { vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=False)
"address": netaddr.IPNetwork('10.0.254.1/24'), IPAddress.objects.create(vrf=vrf, address=netaddr.IPNetwork('192.0.2.1/24'))
"status": 1, duplicate_ip = IPAddress(vrf=vrf, address=netaddr.IPNetwork('192.0.2.1/24'))
"vrf": VRF.objects.create(name='Test', rd='1:1'), self.assertIsNone(duplicate_ip.clean())
}
IPAddress.objects.create(**pfx_kwargs) def test_duplicate_vrf_unique(self):
dup_address = IPAddress.objects.create(**pfx_kwargs) vrf = VRF.objects.create(name='Test', rd='1:1', enforce_unique=True)
self.assertRaises(ValidationError, dup_address.clean) IPAddress.objects.create(vrf=vrf, address=netaddr.IPNetwork('192.0.2.1/24'))
duplicate_ip = IPAddress(vrf=vrf, address=netaddr.IPNetwork('192.0.2.1/24'))
self.assertRaises(ValidationError, duplicate_ip.clean)