x/quark
Code Issues Proposed changes

Implemented the floating ip update API method

Browse Source
This commit is contained in:
Alan Quillin
2015-06-05 13:58:01 -05:00
parent ed60148090
commit 55d0f0685d
7 changed files with 387 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
quark/db/api.py
View File

@@ -299,6 +299,12 @@ def ip_address_delete(context, addr):
context.session.delete(addr)
def ip_address_deallocate(context, address, **kwargs):
kwargs["_deallocated"] = 1
kwargs["deallocated_at"] = timeutils.utcnow()
return ip_address_update(context, address, **kwargs)
@scoped
def ip_address_find(context, lock_mode=False, **filters):
query = context.session.query(models.IPAddress)
@@ -898,12 +904,11 @@ def floating_ip_find(context, lock_mode=False, limit=None, sorts=None,
limit, sorts, marker)
def floating_ip_associate_fixed_ip(context, floating_ip, fixed_ip,
enable=True):
assoc = models.FloatingToFixedIPAssociation()
assoc.floating_ip_address_id = floating_ip.id
assoc.fixed_ip_address_id = fixed_ip.id
assoc.enabled = enable
context.session.add(assoc)
def floating_ip_associate_fixed_ip(context, floating_ip, fixed_ip):
floating_ip.fixed_ip = fixed_ip
return floating_ip
def floating_ip_disassociate_fixed_ip(context, floating_ip):
floating_ip.fixed_ip = None
return floating_ip

20
quark/drivers/unicorn_driver.py
View File

@@ -59,8 +59,19 @@ class UnicornDriver(object):
LOG.error("register_floating_ip: %s" % msg)
raise ex.RegisterFloatingIpFailure(id=floating_ip.id)
def update_floating_ip(self, floating_ip):
pass
def update_floating_ip(self, floating_ip, port, fixed_ip):
url = "%s/%s" % (CONF.QUARK.floating_ip_base_url,
floating_ip["address_readable"])
req = self._build_request_body(floating_ip, port, fixed_ip)
LOG.info("Calling unicorn to register floating ip: %s %s" % (url, req))
r = requests.put(url, data=json.dumps(req))
if r.status_code != 200:
msg = "Unexpected status from unicorn API: Status Code %s, " \
"Message: %s" % (r.status_code, r.json())
LOG.error("register_floating_ip: %s" % msg)
raise ex.RegisterFloatingIpFailure(id=floating_ip.id)
def remove_floating_ip(self, floating_ip):
url = "%s/%s" % (CONF.QUARK.floating_ip_base_url,
@@ -69,7 +80,10 @@ class UnicornDriver(object):
LOG.info("Calling unicorn to remove floating ip: %s" % url)
r = requests.delete(url)
if r.status_code != 204:
if r.status_code == 404:
LOG.warn("The floating IP %s does not exist in the unicorn system."
% floating_ip.address_readable)
elif r.status_code != 204:
msg = "Unexpected status from unicorn API: Status Code %s, " \
"Message: %s" % (r.status_code, r.json())
LOG.error("remove_floating_ip: %s" % msg)

10
quark/exceptions.py
View File

@@ -167,3 +167,13 @@ class NoAvailableFixedIPsForPort(exceptions.Conflict):
class PortDoesNotHaveAGateway(exceptions.Conflict):
message = _("Port %(port_id) does not have a gateway")
class PortAlreadyAssociatedToFloatingIP(exceptions.BadRequest):
message = _("Port %(port_id) is already associated with "
"floating IP %(flip_id)")
class FloatingIPUpdateNoPortIdSupplied(exceptions.BadRequest):
message = _("When no port is currently associated to the floating if, "
"port_id is required but was not supplied")

3
quark/plugin.py
View File

@@ -413,7 +413,8 @@ class Plugin(neutron_plugin_base_v2.NeutronPluginBaseV2,
@sessioned
def update_floatingip(self, context, id, floatingip):
return floating_ips.update_floatingip(context, id, floatingip)
return floating_ips.update_floatingip(context, id,
floatingip["floatingip"])
@sessioned
def get_floatingip(self, context, id, fields=None):

161
quark/plugin_modules/floating_ips.py
View File

@@ -20,7 +20,7 @@ from oslo_log import log as logging
from quark.db import api as db_api
from quark.db import ip_types
from quark.drivers import floating_ip_registry as registry
from quark import exceptions as quark_exceptions
from quark import exceptions as qex
from quark import ipam
from quark import plugin_views as v
@@ -40,6 +40,17 @@ CONF.register_opts(quark_router_opts, "QUARK")
def create_floatingip(context, content):
"""Allocate or reallocate a floating IP.
:param context: neutron api request context.
:param content: dictionary describing the floating ip, with keys
as listed in the RESOURCE_ATTRIBUTE_MAP object in
neutron/api/v2/attributes.py. All keys will be populated.
:returns: Dictionary containing details for the new floating IP. If values
are declared in the fields parameter, then only those keys will be
present.
"""
LOG.info("create_floatingip %s for tenant %s and body %s" %
(id, context.tenant_id, content))
tenant_id = content.get("tenant_id")
@@ -69,12 +80,12 @@ def create_floatingip(context, content):
raise exceptions.PortNotFound(port_id=port_id)
if not port.ip_addresses or len(port.ip_addresses) == 0:
raise quark_exceptions.NoAvailableFixedIPsForPort(port_id=port_id)
raise qex.NoAvailableFixedIPsForPort(port_id=port_id)
if not fixed_ip_address:
fixed_ip = _get_next_available_fixed_ip(port)
if not fixed_ip:
raise quark_exceptions.NoAvailableFixedIPsForPort(
raise qex.NoAvailableFixedIPsForPort(
port_id=port_id)
else:
fixed_ip = next((ip for ip in port.ip_addresses
@@ -83,13 +94,13 @@ def create_floatingip(context, content):
None)
if not fixed_ip:
raise quark_exceptions.FixedIpDoesNotExistsForPort(
raise qex.FixedIpDoesNotExistsForPort(
fixed_ip=fixed_ip_address, port_id=port_id)
if any(ip for ip in port.ip_addresses
if (ip.get("address_type") == ip_types.FLOATING and
ip.fixed_ip["address_readable"] == fixed_ip_address)):
raise quark_exceptions.PortAlreadyContainsFloatingIp(
raise qex.PortAlreadyContainsFloatingIp(
port_id=port_id)
new_addresses = []
@@ -106,55 +117,145 @@ def create_floatingip(context, content):
ip_addresses=ip_addresses,
address_type=ip_types.FLOATING)
floating_ip = new_addresses[0]
flip = new_addresses[0]
if fixed_ip and port:
with context.session.begin():
floating_ip = db_api.floating_ip_associate_fixed_ip(context,
floating_ip,
flip = db_api.port_associate_ip(context, [port], flip, [port_id])
flip = db_api.floating_ip_associate_fixed_ip(context, flip,
fixed_ip)
flip_driver_type = CONF.QUARK.default_floating_ip_driver
flip_driver = registry.DRIVER_REGISTRY.get_driver(flip_driver_type)
flip_driver.register_floating_ip(floating_ip, port, fixed_ip)
flip_driver.register_floating_ip(flip, port, fixed_ip)
return v._make_floating_ip_dict(floating_ip)
return v._make_floating_ip_dict(flip, port_id)
def update_floatingip(context, id, body):
def update_floatingip(context, id, content):
"""Update an existing floating IP.
:param context: neutron api request context.
:param id: id of the floating ip
:param content: dictionary with keys indicating fields to update.
valid keys are those that have a value of True for 'allow_put'
as listed in the RESOURCE_ATTRIBUTE_MAP object in
neutron/api/v2/attributes.py.
:returns: Dictionary containing details for the new floating IP. If values
are declared in the fields parameter, then only those keys will be
present.
"""
LOG.info("update_floatingip %s for tenant %s and body %s" %
(id, context.tenant_id, body))
(id, context.tenant_id, content))
# floating_ip_dict = body.get("ip_address")
#
# if "port_id" not in floating_ip_dict:
# raise exceptions.BadRequest(resource="floating_ip",
# msg="port_id is required.")
if "port_id" not in content:
raise exceptions.BadRequest(resource="floating_ip",
msg="port_id is required.")
# port_id = floating_ip_dict.get("port_id")
port_id = content.get("port_id")
port = None
fixed_ip = None
current_port = None
flip = None
raise NotImplementedError()
with context.session.begin():
flip = db_api.floating_ip_find(context, id=id, scope=db_api.ONE)
current_ports = flip.ports
if current_ports and len(current_ports) > 0:
current_port = current_ports[0]
if not port_id and not current_port:
raise qex.FloatingIPUpdateNoPortIdSupplied()
if port_id:
port = db_api.port_find(context, id=port_id, scope=db_api.ONE)
if not port:
raise exceptions.PortNotFound(port_id=port_id)
if current_port and current_port.id == port_id:
d = dict(flip_id=id, port_id=port_id)
raise qex.PortAlreadyAssociatedToFloatingIP(**d)
fixed_ip = _get_next_available_fixed_ip(port)
LOG.info("new fixed ip: %s" % fixed_ip)
if not fixed_ip:
raise qex.NoAvailableFixedIPsForPort(port_id=port_id)
LOG.info("current ports: %s" % current_ports)
if current_port:
flip = db_api.port_disassociate_ip(context, [current_port], flip)
if flip.fixed_ip:
flip = db_api.floating_ip_disassociate_fixed_ip(context, flip)
if port:
flip = db_api.port_associate_ip(context, [port], flip, [port_id])
flip = db_api.floating_ip_associate_fixed_ip(context, flip,
fixed_ip)
flip_driver_type = CONF.QUARK.default_floating_ip_driver
flip_driver = registry.DRIVER_REGISTRY.get_driver(flip_driver_type)
if port:
if current_port:
flip_driver.update_floating_ip(flip, port, fixed_ip)
else:
flip_driver.register_floating_ip(flip, port, fixed_ip)
else:
flip_driver.remove_floating_ip(flip)
# Note(alanquillin) The ports parameters on the model is not
# properly getting cleaned up when removed. Manually cleaning them up.
# Need to fix the db api to correctly update the model.
if not port:
flip.ports = []
return v._make_floating_ip_dict(flip)
def delete_floatingip(context, id):
"""deallocate a floating IP.
:param context: neutron api request context.
:param id: id of the floating ip
"""
LOG.info("delete_floatingip %s for tenant %s" % (id, context.tenant_id))
filters = {"address_type": ip_types.FLOATING, "_deallocated": False}
floating_ip = db_api.floating_ip_find(context, id=id, scope=db_api.ONE,
**filters)
if not floating_ip:
raise quark_exceptions.FloatingIpNotFound(id=id)
flip = db_api.floating_ip_find(context, id=id, scope=db_api.ONE, **filters)
if not flip:
raise qex.FloatingIpNotFound(id=id)
current_ports = flip.ports
current_port = None
if current_ports and len(current_ports) > 0:
current_port = current_ports[0]
with context.session.begin():
strategy_name = flip.network.get("ipam_strategy")
ipam_driver = ipam.IPAM_REGISTRY.get_strategy(strategy_name)
ipam_driver.deallocate_ip_address(context, flip)
if current_port:
flip = db_api.port_disassociate_ip(context, [current_port],
flip)
if flip.fixed_ip:
flip = db_api.floating_ip_disassociate_fixed_ip(context, flip)
db_api.ip_address_deallocate(context, flip)
if flip.fixed_ip:
driver_type = CONF.QUARK.default_floating_ip_driver
driver = registry.DRIVER_REGISTRY.get_driver(driver_type)
driver.remove_floating_ip(floating_ip)
strategy_name = floating_ip.network.get("ipam_strategy")
ipam_driver = ipam.IPAM_REGISTRY.get_strategy(strategy_name)
ipam_driver.deallocate_ip_address(context, floating_ip)
driver.remove_floating_ip(flip)
def get_floatingip(context, id, fields=None):
@@ -179,7 +280,7 @@ def get_floatingip(context, id, fields=None):
**filters)
if not floating_ip:
raise quark_exceptions.FloatingIpNotFound(id=id)
raise qex.FloatingIpNotFound(id=id)
return v._make_floating_ip_dict(floating_ip)

4
quark/plugin_views.py
View File

@@ -276,13 +276,15 @@ def _make_ip_policy_dict(ipp):
"exclude": [ippc["cidr"] for ippc in ipp["exclude"]]}
def _make_floating_ip_dict(flip):
def _make_floating_ip_dict(flip, port_id=None):
if not port_id:
ports = flip.ports
port_id = None
if ports and len(ports) > 0:
port_id = None if not ports[0] else ports[0].id
fixed_ip = flip.fixed_ip
return {"id": flip.get("id"),
"floating_network_id": flip.get("network_id"),
"router_id": CONF.QUARK.floating_ip_router_id,

215
quark/tests/plugin_modules/test_floating_ips.py
View File

@@ -18,10 +18,10 @@ import contextlib
import datetime
import mock
import netaddr
from neutron.common import exceptions
from neutron.common import exceptions as ex
from quark.db import models
from quark import exceptions as quark_exceptions
from quark import exceptions as q_ex
from quark.plugin_modules import floating_ips
from quark.tests import test_quark_plugin
@@ -36,10 +36,14 @@ class TestRemoveFloatingIPs(test_quark_plugin.TestQuarkPlugin):
with contextlib.nested(
mock.patch("quark.db.api.floating_ip_find"),
mock.patch("quark.db.api.floating_ip_disassociate_fixed_ip"),
mock.patch("quark.db.api.port_disassociate_ip"),
mock.patch("quark.db.api.ip_address_deallocate"),
mock.patch("quark.ipam.QuarkIpam.deallocate_ip_address"),
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".remove_floating_ip")
) as (flip_find, mock_dealloc, mock_remove_flip):
) as (flip_find, db_fixed_ip_disassoc, db_port_disassoc, db_dealloc,
mock_dealloc, mock_remove_flip):
flip_find.return_value = flip_model
yield
@@ -52,7 +56,7 @@ class TestRemoveFloatingIPs(test_quark_plugin.TestQuarkPlugin):
def test_delete_floating_by_when_ip_address_does_not_exists_fails(self):
with self._stubs():
with self.assertRaises(quark_exceptions.FloatingIpNotFound):
with self.assertRaises(q_ex.FloatingIpNotFound):
self.plugin.delete_floatingip(self.context, 1)
@@ -213,6 +217,10 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
def _alloc_ip(context, new_addr, net_id, port_m, *args, **kwargs):
new_addr.append(flip_model)
def _port_assoc(context, ports, addr, enable_port=None):
addr.ports = ports
return addr
def _flip_fixed_ip_assoc(context, addr, fixed_ip):
addr.fixed_ip = fixed_ip
return addr
@@ -224,13 +232,16 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".register_floating_ip"),
mock.patch("quark.db.api.port_associate_ip"),
mock.patch("quark.db.api.floating_ip_associate_fixed_ip")
) as (flip_find, net_find, port_find, alloc_ip, mock_reg_flip, assoc):
) as (flip_find, net_find, port_find, alloc_ip, mock_reg_flip,
port_assoc, fixed_ip_assoc):
flip_find.return_value = flip_model
net_find.return_value = net_model
port_find.return_value = port_model
alloc_ip.side_effect = _alloc_ip
assoc.side_effect = _flip_fixed_ip_assoc
port_assoc.side_effect = _port_assoc
fixed_ip_assoc.side_effect = _flip_fixed_ip_assoc
yield
def test_create_with_a_port(self):
@@ -329,14 +340,14 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
def test_create_without_network_id_fails(self):
with self._stubs():
with self.assertRaises(exceptions.BadRequest):
with self.assertRaises(ex.BadRequest):
request = dict(port_id=2, floating_ip_address="10.0.0.1")
self.plugin.create_floatingip(self.context,
dict(floatingip=request))
def test_create_with_invalid_network_fails(self):
with self._stubs():
with self.assertRaises(exceptions.NetworkNotFound):
with self.assertRaises(ex.NetworkNotFound):
request = dict(floating_network_id=123,
port_id=2, floating_ip_address="10.0.0.1")
self.plugin.create_floatingip(self.context,
@@ -347,7 +358,7 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
ipam_strategy="ANY")
with self._stubs(network=network):
with self.assertRaises(exceptions.PortNotFound):
with self.assertRaises(ex.PortNotFound):
request = dict(floating_network_id=network["id"],
port_id=2, floating_ip_address="10.0.0.1")
self.plugin.create_floatingip(self.context,
@@ -367,7 +378,7 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
with self._stubs(port=port, ips=fixed_ips, network=network):
with self.assertRaises(
quark_exceptions.FixedIpDoesNotExistsForPort):
q_ex.FixedIpDoesNotExistsForPort):
request = dict(floating_network_id=network["id"],
port_id=port["id"],
fixed_ip_address="192.168.0.2")
@@ -400,7 +411,7 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
with self._stubs(port=port, ips=ips, network=network):
with self.assertRaises(
quark_exceptions.PortAlreadyContainsFloatingIp):
q_ex.PortAlreadyContainsFloatingIp):
request = dict(floating_network_id=network["id"],
port_id=port["id"],
fixed_ip_address="192.168.0.1")
@@ -415,7 +426,7 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
with self._stubs(port=port, network=network):
with self.assertRaises(
quark_exceptions.NoAvailableFixedIPsForPort):
q_ex.NoAvailableFixedIPsForPort):
request = dict(floating_network_id=network["id"],
port_id=port["id"])
self.plugin.create_floatingip(self.context,
@@ -444,8 +455,186 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
with self._stubs(port=port, ips=ips, network=network):
with self.assertRaises(
quark_exceptions.NoAvailableFixedIPsForPort):
q_ex.NoAvailableFixedIPsForPort):
request = dict(floating_network_id=network["id"],
port_id=port["id"])
self.plugin.create_floatingip(self.context,
dict(floatingip=request))
class TestUpdateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, flip=None, curr_port=None, new_port=None):
curr_port_model = None
if curr_port:
curr_port_model = models.Port()
curr_port_model.update(curr_port)
new_port_model = None
if new_port:
new_port_model = models.Port()
new_port_model.update(new_port)
fixed_ips = new_port.get("fixed_ips")
if fixed_ips:
new_port_model.ip_addresses = []
for ip in fixed_ips:
addr = netaddr.IPAddress(ip)
fixed_ip_model = models.IPAddress()
fixed_ip_model.update(dict(address_readable=ip,
address=int(addr), version=4,
address_type="fixed"))
new_port_model.ip_addresses.append(fixed_ip_model)
flip_model = None
if flip:
flip_model = models.IPAddress()
flip_model.update(flip)
if curr_port_model:
flip_model.ports = [curr_port_model]
fixed_ip = flip.get("fixed_ip_address")
if fixed_ip:
addr = netaddr.IPAddress(fixed_ip)
fixed_ip_model = models.IPAddress()
fixed_ip_model.update(dict(address_readable=fixed_ip,
address=int(addr), version=4,
address_type="fixed"))
flip_model.fixed_ip = fixed_ip_model
def _find_port(context, id, **kwargs):
return (curr_port_model if (curr_port_model and
id == curr_port_model.id)
else new_port_model)
def _flip_assoc(context, addr, fixed_ip):
addr.fixed_ip = fixed_ip
return addr
def _flip_disassoc(context, addr):
addr.fixed_ip = None
return addr
def _port_assoc(context, ports, addr, enable_ports=None):
addr.ports = ports
return addr
def _port_dessoc(context, ports, addr):
addr.associations = []
addr.ports = []
return addr
with contextlib.nested(
mock.patch("quark.db.api.floating_ip_find"),
mock.patch("quark.db.api.port_find"),
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".register_floating_ip"),
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".update_floating_ip"),
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".remove_floating_ip"),
mock.patch("quark.db.api.port_associate_ip"),
mock.patch("quark.db.api.port_disassociate_ip"),
mock.patch("quark.db.api.floating_ip_associate_fixed_ip"),
mock.patch("quark.db.api.floating_ip_disassociate_fixed_ip")
) as (flip_find, port_find, reg_flip, update_flip, rem_flip,
port_assoc, port_dessoc, flip_assoc, flip_dessoc):
flip_find.return_value = flip_model
port_find.side_effect = _find_port
port_assoc.side_effect = _port_assoc
port_dessoc.side_effect = _port_dessoc
flip_assoc.side_effect = _flip_assoc
flip_dessoc.side_effect = _flip_disassoc
yield
def test_update_with_new_port_and_no_previous_port(self):
new_port = dict(id="2", fixed_ips=["192.168.0.1"])
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip, new_port=new_port):
content = dict(port_id=new_port["id"])
ret = self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
self.assertEqual(ret["fixed_ip_address"], "192.168.0.1")
self.assertEqual(ret["port_id"], new_port["id"])
def test_update_with_new_port(self):
curr_port = dict(id="1")
new_port = dict(id="2", fixed_ips=["192.168.0.1"])
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip, curr_port=curr_port, new_port=new_port):
content = dict(port_id=new_port["id"])
ret = self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
self.assertEqual(ret["fixed_ip_address"], "192.168.0.1")
self.assertEqual(ret["port_id"], new_port["id"])
def test_update_with_no_port(self):
curr_port = dict(id="1")
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip, curr_port=curr_port):
content = dict(port_id=None)
ret = self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
self.assertEqual(ret.get("fixed_ip_address"), None)
self.assertEqual(ret.get("port_id"), None)
def test_update_with_non_existent_port_should_fail(self):
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip):
with self.assertRaises(ex.PortNotFound):
content = dict(port_id="123")
self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
def test_update_with_port_with_no_fixed_ip_avail_should_fail(self):
new_port = dict(id="123")
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip, new_port=new_port):
with self.assertRaises(q_ex.NoAvailableFixedIPsForPort):
content = dict(port_id="123")
self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
def test_update_with_same_port_should_fail(self):
new_port = dict(id="123")
curr_port = dict(id="123")
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip, new_port=new_port, curr_port=curr_port):
with self.assertRaises(q_ex.PortAlreadyAssociatedToFloatingIP):
content = dict(port_id="123")
self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
def test_update_with_no_port_and_no_previous_port_should_fail(self):
addr = netaddr.IPAddress("10.0.0.1")
flip = dict(id="3", fixed_ip_address="172.16.1.1", address=int(addr),
address_readable=str(addr))
with self._stubs(flip=flip):
with self.assertRaises(q_ex.FloatingIPUpdateNoPortIdSupplied):
content = dict(port_id=None)
self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
def test_update_with_missing_port_id_param_should_fail(self):
with self._stubs():
with self.assertRaises(ex.BadRequest):
content = {}
self.plugin.update_floatingip(self.context, "123",
dict(floatingip=content))