x/quark
Code Issues Proposed changes

Merge pull request #466 from roaet/ip_addr_validation

Browse Source 
Added very basic validation to ip_addresses
This commit is contained in:
Amir Sadoughi
2015-10-13 10:29:51 -05:00
parent 10702f9bbe f21466b6ba
commit e15a093b97
2 changed files with 64 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
quark/plugin_modules/ip_addresses.py
View File

@@ -150,6 +150,9 @@ def create_ip_address(context, body):
LOG.info("create_ip_address for tenant %s" % context.tenant_id) LOG.info("create_ip_address for tenant %s" % context.tenant_id)
iptype = (ip_types.SHARED if _shared_ip_request(body) iptype = (ip_types.SHARED if _shared_ip_request(body)
else ip_types.FIXED) else ip_types.FIXED)
if 'ip_address' not in body:
raise exceptions.BadRequest(resource="ip_addresses",
msg="Invalid request body.")
if iptype == ip_types.FIXED and not CONF.QUARK.ipaddr_allow_fixed_ip: if iptype == ip_types.FIXED and not CONF.QUARK.ipaddr_allow_fixed_ip:
raise exceptions.BadRequest(resource="ip_addresses", raise exceptions.BadRequest(resource="ip_addresses",
msg="Only shared IPs may be made with " msg="Only shared IPs may be made with "
@@ -255,6 +258,9 @@ def update_ip_address(context, id, ip_address):
"""Due to NCP-1592 ensure that address_type cannot change after update.""" """Due to NCP-1592 ensure that address_type cannot change after update."""
LOG.info("update_ip_address %s for tenant %s" % (id, context.tenant_id)) LOG.info("update_ip_address %s for tenant %s" % (id, context.tenant_id))
ports = [] ports = []
if 'ip_address' not in ip_address:
raise exceptions.BadRequest(resource="ip_addresses",
msg="Invalid request body.")
with context.session.begin(): with context.session.begin():
address = db_api.ip_address_find(context, id=id, scope=db_api.ONE) address = db_api.ip_address_find(context, id=id, scope=db_api.ONE)
if not address: if not address:

58
quark/tests/functional/plugin_modules/test_shared_ips.py
View File

@@ -101,6 +101,40 @@ class QuarkSharedIPs(BaseFunctionalTest):
with self.assertRaises(self.disassociate_exception): with self.assertRaises(self.disassociate_exception):
ip_api.delete_ip_address(self.context, ip['id']) ip_api.delete_ip_address(self.context, ip['id'])
def test_update_shared_ip_with_plural_will_error(self):
with self._stubs(self.network, self.subnet, self.ports_info4) as (
net, sub, ports):
port_ids = [ports[0]['id'], ports[1]['id']]
shared_ip = {'ip_address': dict(port_ids=port_ids,
network_id=net['id'],
version=4)}
ip = ip_api.create_ip_address(self.context, shared_ip)
self.assertEqual(ip_types.SHARED, ip['type'])
port_ids = [ports[0]['id'], ports[3]['id']]
shared_ip = {'ip_addresses': dict(port_ids=port_ids)}
with self.assertRaises(exceptions.BadRequest):
ip_api.update_ip_address(self.context, ip['id'], shared_ip)
def test_update_shared_ip_with_garbage_will_error(self):
with self._stubs(self.network, self.subnet, self.ports_info4) as (
net, sub, ports):
port_ids = [ports[0]['id'], ports[1]['id']]
shared_ip = {'ip_address': dict(port_ids=port_ids,
network_id=net['id'],
version=4)}
ip = ip_api.create_ip_address(self.context, shared_ip)
self.assertEqual(ip_types.SHARED, ip['type'])
port_ids = [ports[0]['id'], ports[3]['id']]
shared_ip = {'delasdfkj': dict(port_ids=port_ids)}
with self.assertRaises(exceptions.BadRequest):
ip_api.update_ip_address(self.context, ip['id'], shared_ip)
def test_update_shared_ip_with_unowned_ports_is_okay(self): def test_update_shared_ip_with_unowned_ports_is_okay(self):
with self._stubs(self.network, self.subnet, self.ports_info4) as ( with self._stubs(self.network, self.subnet, self.ports_info4) as (
@@ -298,6 +332,30 @@ class QuarkSharedIPs(BaseFunctionalTest):
ports_ip = ip_api.get_ports_for_ip_address(self.context, ip['id']) ports_ip = ip_api.get_ports_for_ip_address(self.context, ip['id'])
self.assertEqual(2, len(ports_ip)) self.assertEqual(2, len(ports_ip))
def test_create_shared_ips_fails_with_plural_body(self):
with self._stubs(self.network, self.subnet, self.ports_info2) as (
net, sub, ports):
port_ids = [ports[0]['id'], ports[1]['id']]
shared_ip = {'ip_addresses': dict(port_ids=port_ids,
network_id=net['id'],
version=4)}
with self.assertRaises(exceptions.BadRequest):
ip_api.create_ip_address(self.context, shared_ip)
def test_create_shared_ips_fails_with_garbage_body(self):
with self._stubs(self.network, self.subnet, self.ports_info2) as (
net, sub, ports):
port_ids = [ports[0]['id'], ports[1]['id']]
shared_ip = {'derpie_derp': dict(port_ids=port_ids,
network_id=net['id'],
version=4)}
with self.assertRaises(exceptions.BadRequest):
ip_api.create_ip_address(self.context, shared_ip)
def test_shared_ip_in_fixed_ip_list(self): def test_shared_ip_in_fixed_ip_list(self):
with self._stubs(self.network, self.subnet, self.ports_info2) as ( with self._stubs(self.network, self.subnet, self.ports_info2) as (