firewall: fix the Vlan's get_new_address() method, and add test

circle/firewall/models.py

@@ -36,7 +36,6 @@ from django.core.validators import MinValueValidator, MaxValueValidator
 from django.core.urlresolvers import reverse
 from django.core.urlresolvers import reverse
 import django.conf
 import django.conf
 from django.db.models.signals import post_save, post_delete
 from django.db.models.signals import post_save, post_delete
-from django.db.models import Q
 from celery.exceptions import TimeoutError
 from celery.exceptions import TimeoutError
 from netaddr import IPSet, EUI, IPNetwork, IPAddress, ipv6_full
 from netaddr import IPSet, EUI, IPNetwork, IPAddress, ipv6_full
 
 
@@ -500,12 +499,11 @@ class Vlan(AclBase, models.Model):
 
 
     def get_new_address(self):
     def get_new_address(self):
         hosts = self.host_set
         hosts = self.host_set
-        query = Q(shared_ip=False)\
-            | Q(external_ipv4__isnull=False)\
-            | Q(vlan=self)
-        ipv4_hosts = Host.objects.filter(query)
-
-        used_v4 = IPSet(ipv4_hosts.values_list('ipv4', flat=True))
+        used_ext_addrs = Host.objects.filter(
+            external_ipv4__isnull=False).values_list(
+            'external_ipv4', flat=True)
+        used_v4 = IPSet(hosts.values_list('ipv4', flat=True)).union(
+            used_ext_addrs).union([self.network4.ip])
         used_v6 = IPSet(hosts.exclude(ipv6__isnull=True)
         used_v6 = IPSet(hosts.exclude(ipv6__isnull=True)
                         .values_list('ipv6', flat=True))
                         .values_list('ipv6', flat=True))
 
 

circle/firewall/tests/test_firewall.py

@@ -102,6 +102,15 @@ class GetNewAddressTestCase(MockCeleryMixin, TestCase):
                  owner=self.u1).save()
                  owner=self.u1).save()
         self.assertRaises(ValidationError, self.vlan.get_new_address)
         self.assertRaises(ValidationError, self.vlan.get_new_address)
 
 
+    def test_all_addr_in_use2(self):
+        Host(hostname='h-xd', mac='01:02:03:04:05:06',
+             ipv4='10.0.0.6', vlan=self.vlan,
+             owner=self.u1).save()
+        Host(hostname='h-arni', mac='01:02:03:04:05:02',
+             ipv4='100.0.0.1', vlan=self.vlan, external_ipv4='10.0.0.2',
+             owner=self.u1).save()
+        self.assertRaises(ValidationError, self.vlan.get_new_address)
+
     def test_new_addr(self):
     def test_new_addr(self):
         used_v4 = IPSet(self.vlan.host_set.values_list('ipv4', flat=True))
         used_v4 = IPSet(self.vlan.host_set.values_list('ipv4', flat=True))
         assert self.vlan.get_new_address()['ipv4'] not in used_v4
         assert self.vlan.get_new_address()['ipv4'] not in used_v4