x/quark
Code Issues Proposed changes

Added subnet cidr validation from quantum

Browse Source 
* Copied subnet cidr validation from quantum
* Added test failure for no subnet found
* Added unit tests for overlapping subnets validation
* Fixed cover venv for quark - quantum dep
* Fixed test failures: incorrect use of list as defaullt paramete
This commit is contained in:
Amir Sadoughi
2013-05-07 12:37:24 -05:00
parent 85821b89d3
commit 921df1ad71
3 changed files with 102 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
quark/plugin.py
View File

@@ -215,6 +215,35 @@ class Plugin(quantum_plugin_base_v2.QuantumPluginBaseV2):
"port_ids": [port["id"] for port in address["ports"]],
"subnet_id": address["subnet_id"]}
def _validate_subnet_cidr(self, context, network, new_subnet_cidr):
"""Validate the CIDR for a subnet.
Verifies the specified CIDR does not overlap with the ones defined
for the other subnets specified for this network, or with any other
CIDR if overlapping IPs are disabled.
"""
new_subnet_ipset = netaddr.IPSet([new_subnet_cidr])
if CONF.allow_overlapping_ips:
subnet_list = network.subnets
else:
subnet_list = db_api.subnet_find(context.elevated())
for subnet in subnet_list:
if (netaddr.IPSet([subnet.cidr]) & new_subnet_ipset):
# don't give out details of the overlapping subnet
err_msg = (_("Requested subnet with cidr: %(cidr)s for "
"network: %(network_id)s overlaps with another "
"subnet") %
{'cidr': new_subnet_cidr,
'network_id': network.id})
LOG.error(_("Validation for CIDR: %(new_cidr)s failed - "
"overlaps with subnet %(subnet_id)s "
"(CIDR: %(cidr)s)"),
{'new_cidr': new_subnet_cidr,
'subnet_id': subnet.id,
'cidr': subnet.cidr})
raise exceptions.InvalidInput(error_message=err_msg)
def create_subnet(self, context, subnet):
"""Create a subnet.
@@ -234,7 +263,7 @@ class Plugin(quantum_plugin_base_v2.QuantumPluginBaseV2):
raise exceptions.NetworkNotFound(net_id=net_id)
s = subnet["subnet"]
self._validate_subnet_cidr(context, net, s["cidr"])
cidr = netaddr.IPNetwork(s["cidr"])
gateway_ip = s.pop("gateway_ip")
if gateway_ip is attributes.ATTR_NOT_SPECIFIED:

75
quark/tests/test_quark_plugin.py
View File

@@ -74,6 +74,8 @@ class TestQuarkAPIExtensions(TestQuarkPlugin):
class TestQuarkGetSubnets(TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, subnets=None, routes=None):
if routes is None:
routes = []
route_models = []
for route in routes:
r = models.Route()
@@ -88,11 +90,13 @@ class TestQuarkGetSubnets(TestQuarkPlugin):
s = models.Subnet()
s.update(s_dict)
subnet_models.append(s)
else:
elif subnets:
mod = models.Subnet()
mod.update(subnets)
mod["routes"] = route_models
subnet_models = mod
else:
subnet_models = None
with mock.patch("quark.db.api.subnet_find") as subnet_find:
subnet_find.return_value = subnet_models
@@ -119,6 +123,11 @@ class TestQuarkGetSubnets(TestQuarkPlugin):
for key in route.keys():
self.assertEqual(routes[0][key], route[key])
def test_subnet_show_fail(self):
with self._stubs():
with self.assertRaises(exceptions.SubnetNotFound):
self.plugin.get_subnet(self.context, 1)
def test_subnet_show(self):
subnet_id = str(uuid.uuid4())
route = dict(id=1, cidr="0.0.0.0/0", gateway="192.168.0.1",
@@ -141,6 +150,58 @@ class TestQuarkGetSubnets(TestQuarkPlugin):
self.assertEqual(routes[0][key], route[key])
class TestQuarkCreateSubnetOverlapping(TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, subnets=None):
if subnets is None:
subnets = []
subnet_models = []
for subnet in subnets:
s = models.Subnet()
s.update(subnet)
subnet_models.append(s)
network = models.Network()
network.update(dict(id=1, subnets=subnet_models))
with contextlib.nested(
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_create")
) as (net_find, subnet_find, subnet_create):
net_find.return_value = network
subnet_find.return_value = subnet_models
yield subnet_create
def test_create_subnet_overlapping_true(self):
cfg.CONF.set_override('allow_overlapping_ips', True)
with self._stubs() as subnet_create:
s = dict(subnet=dict(
gateway_ip=quantum_attrs.ATTR_NOT_SPECIFIED,
dns_nameservers=quantum_attrs.ATTR_NOT_SPECIFIED,
cidr="192.168.1.1/8",
network_id=1))
self.plugin.create_subnet(self.context, s)
self.assertEqual(subnet_create.call_count, 1)
def test_create_subnet_overlapping_false(self):
cfg.CONF.set_override('allow_overlapping_ips', False)
with self._stubs() as subnet_create:
s = dict(subnet=dict(
gateway_ip=quantum_attrs.ATTR_NOT_SPECIFIED,
dns_nameservers=quantum_attrs.ATTR_NOT_SPECIFIED,
cidr="192.168.1.1/8",
network_id=1))
self.plugin.create_subnet(self.context, s)
self.assertEqual(subnet_create.call_count, 1)
def test_create_subnet_overlapping_conflict(self):
cfg.CONF.set_override('allow_overlapping_ips', False)
with self._stubs(subnets=[dict(cidr="192.168.10.1/24")]):
with self.assertRaises(exceptions.InvalidInput):
s = dict(subnet=dict(cidr="192.168.1.1/8",
network_id=1))
self.plugin.create_subnet(self.context, s)
# TODO(amir): Refactor the tests to test individual subnet attributes.
# * copy.deepcopy was necessary to maintain tests on keys, which is a bit ugly.
# * workaround is also in place for lame ATTR_NOT_SPECIFIED object()
@@ -362,13 +423,13 @@ class TestQuarkCreateSubnet(TestQuarkPlugin):
class TestQuarkUpdateSubnet(TestQuarkPlugin):
DEFAULT_ROUTE = [dict(destination="0.0.0.0/0",
nexthop="172.16.0.1")]
@contextlib.contextmanager
def _stubs(self, host_routes=[], new_routes=None, find_routes=True):
def _stubs(self, host_routes=None, new_routes=None, find_routes=True):
if host_routes is None:
host_routes = []
elif not host_routes:
host_routes = [dict(destination="0.0.0.0/0",
nexthop="172.16.0.1")]
if new_routes:
new_routes = [models.Route(cidr=r["destination"],
gateway=r["nexthop"],
@@ -425,6 +486,7 @@ class TestQuarkUpdateSubnet(TestQuarkPlugin):
def test_update_subnet_dns_nameservers(self):
with self._stubs(
host_routes=self.DEFAULT_ROUTE
) as (dns_delete, dns_create,
route_update, route_delete, route_create):
req = dict(subnet=dict(dns_nameservers=["1.1.1.2"]))
@@ -439,6 +501,7 @@ class TestQuarkUpdateSubnet(TestQuarkPlugin):
def test_update_subnet_routes(self):
with self._stubs(
host_routes=self.DEFAULT_ROUTE
) as (dns_delete, dns_create,
route_update, route_delete, route_create):
req = dict(subnet=dict(
@@ -458,6 +521,7 @@ class TestQuarkUpdateSubnet(TestQuarkPlugin):
def test_update_subnet_gateway_ip_with_default_route_in_db(self):
with self._stubs(
host_routes=self.DEFAULT_ROUTE,
new_routes=[dict(destination="0.0.0.0/0", nexthop="1.2.3.4")]
) as (dns_delete, dns_create,
route_update, route_delete, route_create):
@@ -519,6 +583,7 @@ class TestQuarkUpdateSubnet(TestQuarkPlugin):
def test_update_subnet_gateway_ip_with_default_route_in_args(self):
with self._stubs(
host_routes=self.DEFAULT_ROUTE
) as (dns_delete, dns_create,
route_update, route_delete, route_create):
req = dict(subnet=dict(

2
tox.ini
View File

@@ -30,6 +30,8 @@ setenv = VIRTUAL_ENV={envdir}
NOSE_COVER_HTML=1
NOSE_COVER_HTML_DIR=.cover-report
NOSE_COVER_MIN_PERCENTAGE=70
deps = -r{toxinidir}/requirements.txt
-r{toxinidir}/test-requirements.txt
commands = nosetests --cover-package=quark --cover-erase {posargs}
[testenv:venv]