x/quark
Code Issues Proposed changes

Fixes bug in allocation pools

Browse Source
This commit is contained in:
Matt Dietz
2014-04-15 03:51:56 +00:00
parent 18e41596ad
commit 7cd38b67e9
2 changed files with 1 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
quark/db/models.py
View File

@@ -379,9 +379,7 @@ class IPPolicy(BASEV2, models.HasId, models.HasTenant):
@staticmethod @staticmethod
def get_ip_policy_cidrs(subnet): def get_ip_policy_cidrs(subnet):
ip_policy = subnet["ip_policy"] or \ ip_policy = subnet["ip_policy"] or {}
subnet["network"]["ip_policy"] or \
dict()
subnet_cidr = netaddr.IPNetwork(subnet["cidr"]) subnet_cidr = netaddr.IPNetwork(subnet["cidr"])
network_ip = subnet_cidr.network network_ip = subnet_cidr.network

27
quark/tests/test_ipam.py
View File

@@ -1141,33 +1141,6 @@ class TestQuarkIpPoliciesIpAllocation(QuarkIpamBaseTest):
version=4) version=4)
self.assertEqual(address[0]["address"], 281470681743362) self.assertEqual(address[0]["address"], 281470681743362)
def test_ip_policy_on_network(self):
net = dict(ip_policy=dict(exclude=[
models.IPPolicyCIDR(cidr="0.0.0.0/31")]))
subnet = dict(id=1, first_ip=0, last_ip=255,
cidr="0.0.0.0/24", ip_version=4,
next_auto_assign_ip=0, network=net,
ip_policy=None)
with self._stubs(subnets=[(subnet, 0)], addresses=[None, None]):
address = []
self.ipam.allocate_ip_address(self.context, address, 0, 0, 0,
version=4)
self.assertEqual(address[0]["address"], 281470681743362)
def test_ip_policy_on_network_exclusion_intersection(self):
net = dict(ip_policy=dict(exclude=[
models.IPPolicyCIDR(cidr="0.0.0.0/31"),
models.IPPolicyCIDR(cidr="0.0.0.254/32")]))
subnet = dict(id=1, first_ip=0, last_ip=255,
cidr="0.0.0.0/24", ip_version=4,
next_auto_assign_ip=0, network=net,
ip_policy=None)
with self._stubs(subnets=[(subnet, 0)], addresses=[None, None]):
address = []
self.ipam.allocate_ip_address(self.context, address, 0, 0, 0,
version=4)
self.assertEqual(address[0]["address"], 281470681743362)
def test_ip_policy_on_both_subnet_preferred(self): def test_ip_policy_on_both_subnet_preferred(self):
net = dict(ip_policy=dict(exclude=[ net = dict(ip_policy=dict(exclude=[
models.IPPolicyCIDR(cidr="0.0.0.0/31")])) models.IPPolicyCIDR(cidr="0.0.0.0/31")]))