openstack/neutron
Code Issues Proposed changes

Merge "Add Pluggable IPAM Backend Part 2"

Browse Source
This commit is contained in:
Jenkins 2015-07-24 17:18:14 +00:00 committed by Gerrit Code Review
commit 20f099150f
16 changed files with 806 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/common/constants.py
View File

@ -45,6 +45,9 @@ DEVICE_OWNER_LOADBALANCERV2 = "neutron:LOADBALANCERV2"
# DEVICE_OWNER_ROUTER_HA_INTF is a special case and so is not included.
ROUTER_INTERFACE_OWNERS = (DEVICE_OWNER_ROUTER_INTF,
DEVICE_OWNER_DVR_INTERFACE)
ROUTER_INTERFACE_OWNERS_SNAT = (DEVICE_OWNER_ROUTER_INTF,
DEVICE_OWNER_DVR_INTERFACE,
DEVICE_OWNER_ROUTER_SNAT)
L3_AGENT_MODE_DVR = 'dvr'
L3_AGENT_MODE_DVR_SNAT = 'dvr_snat'
L3_AGENT_MODE_LEGACY = 'legacy'

43
neutron/db/db_base_plugin_v2.py
View File

@ -36,6 +36,7 @@ from neutron import context as ctx
from neutron.db import api as db_api
from neutron.db import db_base_plugin_common
from neutron.db import ipam_non_pluggable_backend
from neutron.db import ipam_pluggable_backend
from neutron.db import models_v2
from neutron.db import rbac_db_models as rbac_db
from neutron.db import sqlalchemyutils
@ -101,7 +102,10 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
self.nova_notifier.record_port_status_changed)
def set_ipam_backend(self):
self.ipam = ipam_non_pluggable_backend.IpamNonPluggableBackend()
if cfg.CONF.ipam_driver:
self.ipam = ipam_pluggable_backend.IpamPluggableBackend()
else:
self.ipam = ipam_non_pluggable_backend.IpamNonPluggableBackend()
def _validate_host_route(self, route, ip_version):
try:
@ -470,10 +474,10 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
with context.session.begin(subtransactions=True):
network = self._get_network(context, s["network_id"])
subnet = self.ipam.allocate_subnet(context,
network,
s,
subnetpool_id)
subnet, ipam_subnet = self.ipam.allocate_subnet(context,
network,
s,
subnetpool_id)
if hasattr(network, 'external') and network.external:
self._update_router_gw_ports(context,
network,
@ -481,7 +485,8 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
# If this subnet supports auto-addressing, then update any
# internal ports on the network with addresses for this subnet.
if ipv6_utils.is_auto_address_subnet(subnet):
self.ipam.add_auto_addrs_on_network_ports(context, subnet)
self.ipam.add_auto_addrs_on_network_ports(context, subnet,
ipam_subnet)
return self._make_subnet_dict(subnet, context=context)
def _get_subnetpool_id(self, subnet):
@ -561,21 +566,24 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
s['ip_version'] = db_subnet.ip_version
s['cidr'] = db_subnet.cidr
s['id'] = db_subnet.id
s['tenant_id'] = db_subnet.tenant_id
self._validate_subnet(context, s, cur_subnet=db_subnet)
db_pools = [netaddr.IPRange(p['first_ip'], p['last_ip'])
for p in db_subnet.allocation_pools]
range_pools = None
if s.get('allocation_pools') is not None:
# Convert allocation pools to IPRange to simplify future checks
range_pools = self.ipam.pools_to_ip_range(s['allocation_pools'])
s['allocation_pools'] = range_pools
if s.get('gateway_ip') is not None:
if s.get('allocation_pools') is not None:
allocation_pools = [{'start': p['start'], 'end': p['end']}
for p in s['allocation_pools']]
else:
allocation_pools = [{'start': p['first_ip'],
'end': p['last_ip']}
for p in db_subnet.allocation_pools]
self.ipam.validate_gw_out_of_pools(s["gateway_ip"],
allocation_pools)
pools = range_pools if range_pools is not None else db_pools
self.ipam.validate_gw_out_of_pools(s["gateway_ip"], pools)
with context.session.begin(subtransactions=True):
subnet, changes = self.ipam.update_db_subnet(context, id, s)
subnet, changes = self.ipam.update_db_subnet(context, id, s,
db_pools)
result = self._make_subnet_dict(subnet, context=context)
# Keep up with fields that changed
result.update(changes)
@ -654,6 +662,9 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
raise n_exc.SubnetInUse(subnet_id=id)
context.session.delete(subnet)
# Delete related ipam subnet manually,
# since there is no FK relationship
self.ipam.delete_subnet(context, id)
def get_subnet(self, context, id, fields=None):
subnet = self._get_subnet(context, id)

106
neutron/db/ipam_backend_mixin.py
View File

@ -52,6 +52,24 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
return str(netaddr.IPNetwork(cidr_net).network + 1)
return subnet.get('gateway_ip')
@staticmethod
def pools_to_ip_range(ip_pools):
ip_range_pools = []
for ip_pool in ip_pools:
try:
ip_range_pools.append(netaddr.IPRange(ip_pool['start'],
ip_pool['end']))
except netaddr.AddrFormatError:
LOG.info(_LI("Found invalid IP address in pool: "
"%(start)s - %(end)s:"),
{'start': ip_pool['start'],
'end': ip_pool['end']})
raise n_exc.InvalidAllocationPool(pool=ip_pool)
return ip_range_pools
def delete_subnet(self, context, subnet_id):
pass
def validate_pools_with_subnetpool(self, subnet):
"""Verifies that allocation pools are set correctly
@ -140,22 +158,23 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
def _update_subnet_allocation_pools(self, context, subnet_id, s):
context.session.query(models_v2.IPAllocationPool).filter_by(
subnet_id=subnet_id).delete()
new_pools = [models_v2.IPAllocationPool(first_ip=p['start'],
last_ip=p['end'],
pools = ((netaddr.IPAddress(p.first, p.version).format(),
netaddr.IPAddress(p.last, p.version).format())
for p in s['allocation_pools'])
new_pools = [models_v2.IPAllocationPool(first_ip=p[0],
last_ip=p[1],
subnet_id=subnet_id)
for p in s['allocation_pools']]
for p in pools]
context.session.add_all(new_pools)
# Call static method with self to redefine in child
# (non-pluggable backend)
self._rebuild_availability_ranges(context, [s])
# Gather new pools for result:
result_pools = [{'start': pool['start'],
'end': pool['end']}
for pool in s['allocation_pools']]
# Gather new pools for result
result_pools = [{'start': p[0], 'end': p[1]} for p in pools]
del s['allocation_pools']
return result_pools
def update_db_subnet(self, context, subnet_id, s):
def update_db_subnet(self, context, subnet_id, s, oldpools):
changes = {}
if "dns_nameservers" in s:
changes['dns_nameservers'] = (
@ -239,38 +258,23 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
LOG.debug("Performing IP validity checks on allocation pools")
ip_sets = []
for ip_pool in ip_pools:
try:
start_ip = netaddr.IPAddress(ip_pool['start'])
end_ip = netaddr.IPAddress(ip_pool['end'])
except netaddr.AddrFormatError:
LOG.info(_LI("Found invalid IP address in pool: "
"%(start)s - %(end)s:"),
{'start': ip_pool['start'],
'end': ip_pool['end']})
raise n_exc.InvalidAllocationPool(pool=ip_pool)
start_ip = netaddr.IPAddress(ip_pool.first, ip_pool.version)
end_ip = netaddr.IPAddress(ip_pool.last, ip_pool.version)
if (start_ip.version != subnet.version or
end_ip.version != subnet.version):
LOG.info(_LI("Specified IP addresses do not match "
"the subnet IP version"))
raise n_exc.InvalidAllocationPool(pool=ip_pool)
if end_ip < start_ip:
LOG.info(_LI("Start IP (%(start)s) is greater than end IP "
"(%(end)s)"),
{'start': ip_pool['start'], 'end': ip_pool['end']})
raise n_exc.InvalidAllocationPool(pool=ip_pool)
if start_ip < subnet_first_ip or end_ip > subnet_last_ip:
LOG.info(_LI("Found pool larger than subnet "
"CIDR:%(start)s - %(end)s"),
{'start': ip_pool['start'],
'end': ip_pool['end']})
{'start': start_ip, 'end': end_ip})
raise n_exc.OutOfBoundsAllocationPool(
pool=ip_pool,
subnet_cidr=subnet_cidr)
# Valid allocation pool
# Create an IPSet for it for easily verifying overlaps
ip_sets.append(netaddr.IPSet(netaddr.IPRange(
ip_pool['start'],
ip_pool['end']).cidrs()))
ip_sets.append(netaddr.IPSet(ip_pool.cidrs()))
LOG.debug("Checking for overlaps among allocation pools "
"and gateway ip")
@ -291,22 +295,54 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
pool_2=r_range,
subnet_cidr=subnet_cidr)
def _validate_max_ips_per_port(self, fixed_ip_list):
if len(fixed_ip_list) > cfg.CONF.max_fixed_ips_per_port:
msg = _('Exceeded maximim amount of fixed ips per port')
raise n_exc.InvalidInput(error_message=msg)
def _get_subnet_for_fixed_ip(self, context, fixed, network_id):
if 'subnet_id' in fixed:
subnet = self._get_subnet(context, fixed['subnet_id'])
if subnet['network_id'] != network_id:
msg = (_("Failed to create port on network %(network_id)s"
", because fixed_ips included invalid subnet "
"%(subnet_id)s") %
{'network_id': network_id,
'subnet_id': fixed['subnet_id']})
raise n_exc.InvalidInput(error_message=msg)
# Ensure that the IP is valid on the subnet
if ('ip_address' in fixed and
not ipam_utils.check_subnet_ip(subnet['cidr'],
fixed['ip_address'])):
raise n_exc.InvalidIpForSubnet(ip_address=fixed['ip_address'])
return subnet
if 'ip_address' not in fixed:
msg = _('IP allocation requires subnet_id or ip_address')
raise n_exc.InvalidInput(error_message=msg)
filter = {'network_id': [network_id]}
subnets = self._get_subnets(context, filters=filter)
for subnet in subnets:
if ipam_utils.check_subnet_ip(subnet['cidr'],
fixed['ip_address']):
return subnet
raise n_exc.InvalidIpForNetwork(ip_address=fixed['ip_address'])
def _prepare_allocation_pools(self, allocation_pools, cidr, gateway_ip):
"""Returns allocation pools represented as list of IPRanges"""
if not attributes.is_attr_set(allocation_pools):
return ipam_utils.generate_pools(cidr, gateway_ip)
self._validate_allocation_pools(allocation_pools, cidr)
ip_range_pools = self.pools_to_ip_range(allocation_pools)
self._validate_allocation_pools(ip_range_pools, cidr)
if gateway_ip:
self.validate_gw_out_of_pools(gateway_ip, allocation_pools)
return [netaddr.IPRange(p['start'], p['end'])
for p in allocation_pools]
self.validate_gw_out_of_pools(gateway_ip, ip_range_pools)
return ip_range_pools
def validate_gw_out_of_pools(self, gateway_ip, pools):
for allocation_pool in pools:
pool_range = netaddr.IPRange(
allocation_pool['start'],
allocation_pool['end'])
for pool_range in pools:
if netaddr.IPAddress(gateway_ip) in pool_range:
raise n_exc.GatewayConflictWithAllocationPools(
pool=pool_range,

67
neutron/db/ipam_non_pluggable_backend.py
View File

@ -14,7 +14,6 @@
# under the License.
import netaddr
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log as logging
from sqlalchemy import and_
@ -29,7 +28,6 @@ from neutron.db import ipam_backend_mixin
from neutron.db import models_v2
from neutron.ipam import requests as ipam_req
from neutron.ipam import subnet_alloc
from neutron.ipam import utils as ipam_utils
LOG = logging.getLogger(__name__)
@ -242,49 +240,17 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
"""
fixed_ip_set = []
for fixed in fixed_ips:
found = False
if 'subnet_id' not in fixed:
if 'ip_address' not in fixed:
msg = _('IP allocation requires subnet_id or ip_address')
raise n_exc.InvalidInput(error_message=msg)
filter = {'network_id': [network_id]}
subnets = self._get_subnets(context, filters=filter)
for subnet in subnets:
if ipam_utils.check_subnet_ip(subnet['cidr'],
fixed['ip_address']):
found = True
subnet_id = subnet['id']
break
if not found:
raise n_exc.InvalidIpForNetwork(
ip_address=fixed['ip_address'])
else:
subnet = self._get_subnet(context, fixed['subnet_id'])
if subnet['network_id'] != network_id:
msg = (_("Failed to create port on network %(network_id)s"
", because fixed_ips included invalid subnet "
"%(subnet_id)s") %
{'network_id': network_id,
'subnet_id': fixed['subnet_id']})
raise n_exc.InvalidInput(error_message=msg)
subnet_id = subnet['id']
subnet = self._get_subnet_for_fixed_ip(context, fixed, network_id)
is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
if 'ip_address' in fixed:
# Ensure that the IP's are unique
if not IpamNonPluggableBackend._check_unique_ip(
context, network_id,
subnet_id, fixed['ip_address']):
subnet['id'], fixed['ip_address']):
raise n_exc.IpAddressInUse(net_id=network_id,
ip_address=fixed['ip_address'])
# Ensure that the IP is valid on the subnet
if (not found and
not ipam_utils.check_subnet_ip(subnet['cidr'],
fixed['ip_address'])):
raise n_exc.InvalidIpForSubnet(
ip_address=fixed['ip_address'])
if (is_auto_addr_subnet and
device_owner not in
constants.ROUTER_INTERFACE_OWNERS):
@ -292,23 +258,20 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
"assigned to a port on subnet %(id)s since the "
"subnet is configured for automatic addresses") %
{'address': fixed['ip_address'],
'id': subnet_id})
'id': subnet['id']})
raise n_exc.InvalidInput(error_message=msg)
fixed_ip_set.append({'subnet_id': subnet_id,
fixed_ip_set.append({'subnet_id': subnet['id'],
'ip_address': fixed['ip_address']})
else:
# A scan for auto-address subnets on the network is done
# separately so that all such subnets (not just those
# listed explicitly here by subnet ID) are associated
# with the port.
if (device_owner in constants.ROUTER_INTERFACE_OWNERS or
device_owner == constants.DEVICE_OWNER_ROUTER_SNAT or
if (device_owner in constants.ROUTER_INTERFACE_OWNERS_SNAT or
not is_auto_addr_subnet):
fixed_ip_set.append({'subnet_id': subnet_id})
fixed_ip_set.append({'subnet_id': subnet['id']})
if len(fixed_ip_set) > cfg.CONF.max_fixed_ips_per_port:
msg = _('Exceeded maximim amount of fixed ips per port')
raise n_exc.InvalidInput(error_message=msg)
self._validate_max_ips_per_port(fixed_ip_set)
return fixed_ip_set
def _allocate_fixed_ips(self, context, fixed_ips, mac_address):
@ -382,8 +345,7 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
net_id_filter = {'network_id': [p['network_id']]}
subnets = self._get_subnets(context, filters=net_id_filter)
is_router_port = (
p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS or
p['device_owner'] == constants.DEVICE_OWNER_ROUTER_SNAT)
p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS_SNAT)
fixed_configured = p['fixed_ips'] is not attributes.ATTR_NOT_SPECIFIED
if fixed_configured:
@ -431,17 +393,16 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
return ips
def add_auto_addrs_on_network_ports(self, context, subnet):
def add_auto_addrs_on_network_ports(self, context, subnet, ipam_subnet):
"""For an auto-address subnet, add addrs for ports on the net."""
with context.session.begin(subtransactions=True):
network_id = subnet['network_id']
port_qry = context.session.query(models_v2.Port)
for port in port_qry.filter(
ports = port_qry.filter(
and_(models_v2.Port.network_id == network_id,
models_v2.Port.device_owner !=
constants.DEVICE_OWNER_ROUTER_SNAT,
~models_v2.Port.device_owner.in_(
constants.ROUTER_INTERFACE_OWNERS))):
constants.ROUTER_INTERFACE_OWNERS_SNAT)))
for port in ports:
ip_address = self._calculate_ipv6_eui64_addr(
context, subnet, port['mac_address'])
allocated = models_v2.IPAllocation(network_id=network_id,
@ -505,4 +466,6 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
subnet['dns_nameservers'],
subnet['host_routes'],
subnet_request)
return subnet
# ipam_subnet is not expected to be allocated for non pluggable ipam,
# so just return None for it (second element in returned tuple)
return subnet, None

324
neutron/db/ipam_pluggable_backend.py
View File

@ -13,13 +13,22 @@
# License for the specific language governing permissions and limitations
# under the License.
import netaddr
from oslo_db import exception as db_exc
from oslo_log import log as logging
from oslo_utils import excutils
from sqlalchemy import and_
from neutron.api.v2 import attributes
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.db import ipam_backend_mixin
from neutron.db import models_v2
from neutron.i18n import _LE
from neutron.ipam import driver
from neutron.ipam import exceptions as ipam_exc
from neutron.ipam import requests as ipam_req
LOG = logging.getLogger(__name__)
@ -110,7 +119,6 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
ip_address, ip_subnet = self._ipam_allocate_single_ip(
context, ipam_driver, port, ip_list)
allocated.append({'ip_address': ip_address,
'subnet_cidr': ip_subnet['subnet_cidr'],
'subnet_id': ip_subnet['subnet_id']})
except Exception:
with excutils.save_and_reraise_exception():
@ -127,3 +135,317 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
"external system for %s"), addresses)
return allocated
def _ipam_update_allocation_pools(self, context, ipam_driver, subnet):
self._validate_allocation_pools(subnet['allocation_pools'],
subnet['cidr'])
factory = ipam_driver.get_subnet_request_factory()
subnet_request = factory.get_request(context, subnet, None)
ipam_driver.update_subnet(subnet_request)
def delete_subnet(self, context, subnet_id):
ipam_driver = driver.Pool.get_instance(None, context)
ipam_driver.remove_subnet(subnet_id)
def allocate_ips_for_port_and_store(self, context, port, port_id):
network_id = port['port']['network_id']
ips = []
try:
ips = self._allocate_ips_for_port(context, port)
for ip in ips:
ip_address = ip['ip_address']
subnet_id = ip['subnet_id']
IpamPluggableBackend._store_ip_allocation(
context, ip_address, network_id,
subnet_id, port_id)
except Exception:
with excutils.save_and_reraise_exception():
if ips:
LOG.debug("An exception occurred during port creation."
"Reverting IP allocation")
ipam_driver = driver.Pool.get_instance(None, context)
self._ipam_deallocate_ips(context, ipam_driver,
port['port'], ips,
revert_on_fail=False)
def _allocate_ips_for_port(self, context, port):
"""Allocate IP addresses for the port. IPAM version.
If port['fixed_ips'] is set to 'ATTR_NOT_SPECIFIED', allocate IP
addresses for the port. If port['fixed_ips'] contains an IP address or
a subnet_id then allocate an IP address accordingly.
"""
p = port['port']
ips = []
v6_stateless = []
net_id_filter = {'network_id': [p['network_id']]}
subnets = self._get_subnets(context, filters=net_id_filter)
is_router_port = (
p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS_SNAT)
fixed_configured = p['fixed_ips'] is not attributes.ATTR_NOT_SPECIFIED
if fixed_configured:
ips = self._test_fixed_ips_for_port(context,
p["network_id"],
p['fixed_ips'],
p['device_owner'])
# For ports that are not router ports, implicitly include all
# auto-address subnets for address association.
if not is_router_port:
v6_stateless += [subnet for subnet in subnets
if ipv6_utils.is_auto_address_subnet(subnet)]
else:
# Split into v4, v6 stateless and v6 stateful subnets
v4 = []
v6_stateful = []
for subnet in subnets:
if subnet['ip_version'] == 4:
v4.append(subnet)
else:
if ipv6_utils.is_auto_address_subnet(subnet):
if not is_router_port:
v6_stateless.append(subnet)
else:
v6_stateful.append(subnet)
version_subnets = [v4, v6_stateful]
for subnets in version_subnets:
if subnets:
ips.append([{'subnet_id': s['id']}
for s in subnets])
for subnet in v6_stateless:
# IP addresses for IPv6 SLAAC and DHCPv6-stateless subnets
# are implicitly included.
ips.append({'subnet_id': subnet['id'],
'subnet_cidr': subnet['cidr'],
'eui64_address': True,
'mac': p['mac_address']})
ipam_driver = driver.Pool.get_instance(None, context)
return self._ipam_allocate_ips(context, ipam_driver, p, ips)
def _test_fixed_ips_for_port(self, context, network_id, fixed_ips,
device_owner):
"""Test fixed IPs for port.
Check that configured subnets are valid prior to allocating any
IPs. Include the subnet_id in the result if only an IP address is
configured.
:raises: InvalidInput, IpAddressInUse, InvalidIpForNetwork,
InvalidIpForSubnet
"""
fixed_ip_list = []
for fixed in fixed_ips:
subnet = self._get_subnet_for_fixed_ip(context, fixed, network_id)
is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
if 'ip_address' in fixed:
if (is_auto_addr_subnet and device_owner not in
constants.ROUTER_INTERFACE_OWNERS):
msg = (_("IPv6 address %(address)s can not be directly "
"assigned to a port on subnet %(id)s since the "
"subnet is configured for automatic addresses") %
{'address': fixed['ip_address'],
'id': subnet['id']})
raise n_exc.InvalidInput(error_message=msg)
fixed_ip_list.append({'subnet_id': subnet['id'],
'ip_address': fixed['ip_address']})
else:
# A scan for auto-address subnets on the network is done
# separately so that all such subnets (not just those
# listed explicitly here by subnet ID) are associated
# with the port.
if (device_owner in constants.ROUTER_INTERFACE_OWNERS_SNAT or
not is_auto_addr_subnet):
fixed_ip_list.append({'subnet_id': subnet['id']})
self._validate_max_ips_per_port(fixed_ip_list)
return fixed_ip_list
def _update_ips_for_port(self, context, port,
original_ips, new_ips, mac):
"""Add or remove IPs from the port. IPAM version"""
added = []
removed = []
changes = self._get_changed_ips_for_port(
context, original_ips, new_ips, port['device_owner'])
# Check if the IP's to add are OK
to_add = self._test_fixed_ips_for_port(
context, port['network_id'], changes.add,
port['device_owner'])
ipam_driver = driver.Pool.get_instance(None, context)
if changes.remove:
removed = self._ipam_deallocate_ips(context, ipam_driver, port,
changes.remove)
if to_add:
added = self._ipam_allocate_ips(context, ipam_driver,
changes, to_add)
return self.Changes(add=added,
original=changes.original,
remove=removed)
def save_allocation_pools(self, context, subnet, allocation_pools):
for pool in allocation_pools:
first_ip = str(netaddr.IPAddress(pool.first, pool.version))
last_ip = str(netaddr.IPAddress(pool.last, pool.version))
ip_pool = models_v2.IPAllocationPool(subnet=subnet,
first_ip=first_ip,
last_ip=last_ip)
context.session.add(ip_pool)
def update_port_with_ips(self, context, db_port, new_port, new_mac):
changes = self.Changes(add=[], original=[], remove=[])
if 'fixed_ips' in new_port:
original = self._make_port_dict(db_port,
process_extensions=False)
changes = self._update_ips_for_port(context,
db_port,
original["fixed_ips"],
new_port['fixed_ips'],
new_mac)
try:
# Check if the IPs need to be updated
network_id = db_port['network_id']
for ip in changes.add:
self._store_ip_allocation(
context, ip['ip_address'], network_id,
ip['subnet_id'], db_port.id)
for ip in changes.remove:
self._delete_ip_allocation(context, network_id,
ip['subnet_id'], ip['ip_address'])
self._update_db_port(context, db_port, new_port, network_id,
new_mac)
except Exception:
with excutils.save_and_reraise_exception():
if 'fixed_ips' in new_port:
LOG.debug("An exception occurred during port update.")
ipam_driver = driver.Pool.get_instance(None, context)
if changes.add:
LOG.debug("Reverting IP allocation.")
self._ipam_deallocate_ips(context, ipam_driver,
db_port, changes.add,
revert_on_fail=False)
if changes.remove:
LOG.debug("Reverting IP deallocation.")
self._ipam_allocate_ips(context, ipam_driver,
db_port, changes.remove,
revert_on_fail=False)
return changes
def delete_port(self, context, id):
# Get fixed_ips list before port deletion
port = self._get_port(context, id)
ipam_driver = driver.Pool.get_instance(None, context)
super(IpamPluggableBackend, self).delete_port(context, id)
# Deallocating ips via IPAM after port is deleted locally.
# So no need to do rollback actions on remote server
# in case of fail to delete port locally
self._ipam_deallocate_ips(context, ipam_driver, port,
port['fixed_ips'])
def update_db_subnet(self, context, id, s, old_pools):
ipam_driver = driver.Pool.get_instance(None, context)
if "allocation_pools" in s:
self._ipam_update_allocation_pools(context, ipam_driver, s)
try:
subnet, changes = super(IpamPluggableBackend,
self).update_db_subnet(context, id,
s, old_pools)
except Exception:
with excutils.save_and_reraise_exception():
if "allocation_pools" in s and old_pools:
LOG.error(
_LE("An exception occurred during subnet update."
"Reverting allocation pool changes"))
s['allocation_pools'] = old_pools
self._ipam_update_allocation_pools(context, ipam_driver, s)
return subnet, changes
def add_auto_addrs_on_network_ports(self, context, subnet, ipam_subnet):
"""For an auto-address subnet, add addrs for ports on the net."""
with context.session.begin(subtransactions=True):
network_id = subnet['network_id']
port_qry = context.session.query(models_v2.Port)
ports = port_qry.filter(
and_(models_v2.Port.network_id == network_id,
~models_v2.Port.device_owner.in_(
constants.ROUTER_INTERFACE_OWNERS_SNAT)))
for port in ports:
ip_request = ipam_req.AutomaticAddressRequest(
prefix=subnet['cidr'],
mac=port['mac_address'])
ip_address = ipam_subnet.allocate(ip_request)
allocated = models_v2.IPAllocation(network_id=network_id,
port_id=port['id'],
ip_address=ip_address,
subnet_id=subnet['id'])
try:
# Do the insertion of each IP allocation entry within
# the context of a nested transaction, so that the entry
# is rolled back independently of other entries whenever
# the corresponding port has been deleted.
with context.session.begin_nested():
context.session.add(allocated)
except db_exc.DBReferenceError:
LOG.debug("Port %s was deleted while updating it with an "
"IPv6 auto-address. Ignoring.", port['id'])
LOG.debug("Reverting IP allocation for %s", ip_address)
# Do not fail if reverting allocation was unsuccessful
try:
ipam_subnet.deallocate(ip_address)
except Exception:
LOG.debug("Reverting IP allocation failed for %s",
ip_address)
def allocate_subnet(self, context, network, subnet, subnetpool_id):
subnetpool = None
if subnetpool_id:
subnetpool = self._get_subnetpool(context, subnetpool_id)
self._validate_ip_version_with_subnetpool(subnet, subnetpool)
# gateway_ip and allocation pools should be validated or generated
# only for specific request
if subnet['cidr'] is not attributes.ATTR_NOT_SPECIFIED:
subnet['gateway_ip'] = self._gateway_ip_str(subnet,
subnet['cidr'])
subnet['allocation_pools'] = self._prepare_allocation_pools(
subnet['allocation_pools'],
subnet['cidr'],
subnet['gateway_ip'])
ipam_driver = driver.Pool.get_instance(subnetpool, context)
subnet_factory = ipam_driver.get_subnet_request_factory()
subnet_request = subnet_factory.get_request(context, subnet,
subnetpool)
ipam_subnet = ipam_driver.allocate_subnet(subnet_request)
# get updated details with actually allocated subnet
subnet_request = ipam_subnet.get_details()
try:
subnet = self._save_subnet(context,
network,
self._make_subnet_args(
subnet_request,
subnet,
subnetpool_id),
subnet['dns_nameservers'],
subnet['host_routes'],
subnet_request)
except Exception:
# Note(pbondar): Third-party ipam servers can't rely
# on transaction rollback, so explicit rollback call needed.
# IPAM part rolled back in exception handling
# and subnet part is rolled back by transaction rollback.
with excutils.save_and_reraise_exception():
LOG.debug("An exception occurred during subnet creation."
"Reverting subnet allocation.")
self.delete_subnet(context, subnet_request.subnet_id)
return subnet, ipam_subnet

11
neutron/ipam/driver.py
View File

@ -148,14 +148,3 @@ class Subnet(object):
:returns: An instance of SpecificSubnetRequest with the subnet detail.
"""
@abc.abstractmethod
def associate_neutron_subnet(self, subnet_id):
"""Associate the IPAM subnet with a neutron subnet.
This operation should be performed to attach a neutron subnet to the
current subnet instance. In some cases IPAM subnets may be created
independently of neutron subnets and associated at a later stage.
:param subnet_id: neutron subnet identifier.
"""

17
neutron/ipam/drivers/neutrondb_ipam/db_api.py
View File

@ -54,11 +54,18 @@ class IpamSubnetManager(object):
session.add(ipam_subnet)
return self._ipam_subnet_id
def associate_neutron_id(self, session, neutron_subnet_id):
session.query(db_models.IpamSubnet).filter_by(
id=self._ipam_subnet_id).update(
{'neutron_subnet_id': neutron_subnet_id})
self._neutron_subnet_id = neutron_subnet_id
@classmethod
def delete(cls, session, neutron_subnet_id):
"""Delete IPAM subnet.
IPAM subnet no longer has foreign key to neutron subnet,
so need to perform delete manually
:param session: database sesssion
:param neutron_subnet_id: neutron subnet id associated with ipam subnet
"""
return session.query(db_models.IpamSubnet).filter_by(
neutron_subnet_id=neutron_subnet_id).delete()
def create_pool(self, session, pool_start, pool_end):
"""Create an allocation pool and availability ranges for the subnet.

36
neutron/ipam/drivers/neutrondb_ipam/driver.py
View File

@ -56,7 +56,7 @@ class NeutronDbSubnet(ipam_base.Subnet):
ipam_subnet_id = uuidutils.generate_uuid()
subnet_manager = ipam_db_api.IpamSubnetManager(
ipam_subnet_id,
None)
subnet_request.subnet_id)
# Create subnet resource
session = ctx.session
subnet_manager.create(session)
@ -76,8 +76,7 @@ class NeutronDbSubnet(ipam_base.Subnet):
allocation_pools=pools,
gateway_ip=subnet_request.gateway_ip,
tenant_id=subnet_request.tenant_id,
subnet_id=subnet_request.subnet_id,
subnet_id_not_set=True)
subnet_id=subnet_request.subnet_id)
@classmethod
def load(cls, neutron_subnet_id, ctx):
@ -88,7 +87,7 @@ class NeutronDbSubnet(ipam_base.Subnet):
ipam_subnet = ipam_db_api.IpamSubnetManager.load_by_neutron_subnet_id(
ctx.session, neutron_subnet_id)
if not ipam_subnet:
LOG.error(_LE("Unable to retrieve IPAM subnet as the referenced "
LOG.error(_LE("IPAM subnet referenced to "
"Neutron subnet %s does not exist"),
neutron_subnet_id)
raise n_exc.SubnetNotFound(subnet_id=neutron_subnet_id)
@ -113,7 +112,7 @@ class NeutronDbSubnet(ipam_base.Subnet):
def __init__(self, internal_id, ctx, cidr=None,
allocation_pools=None, gateway_ip=None, tenant_id=None,
subnet_id=None, subnet_id_not_set=False):
subnet_id=None):
# NOTE: In theory it could have been possible to grant the IPAM
# driver direct access to the database. While this is possible,
# it would have led to duplicate code and/or non-trivial
@ -124,7 +123,7 @@ class NeutronDbSubnet(ipam_base.Subnet):
self._pools = allocation_pools
self._gateway_ip = gateway_ip
self._tenant_id = tenant_id
self._subnet_id = None if subnet_id_not_set else subnet_id
self._subnet_id = subnet_id
self.subnet_manager = ipam_db_api.IpamSubnetManager(internal_id,
self._subnet_id)
self._context = ctx
@ -363,17 +362,6 @@ class NeutronDbSubnet(ipam_base.Subnet):
self._tenant_id, self.subnet_manager.neutron_id,
self._cidr, self._gateway_ip, self._pools)
def associate_neutron_subnet(self, subnet_id):
"""Set neutron identifier for this subnet"""
session = self._context.session
if self._subnet_id:
raise
# IPAMSubnet does not have foreign key to Subnet,
# so need verify subnet existence.
NeutronDbSubnet._fetch_subnet(self._context, subnet_id)
self.subnet_manager.associate_neutron_id(session, subnet_id)
self._subnet_id = subnet_id
class NeutronDbPool(subnet_alloc.SubnetAllocator):
"""Subnet pools backed by Neutron Database.
@ -429,10 +417,16 @@ class NeutronDbPool(subnet_alloc.SubnetAllocator):
subnet.update_allocation_pools(subnet_request.allocation_pools)
return subnet
def remove_subnet(self, subnet):
def remove_subnet(self, subnet_id):
"""Remove data structures for a given subnet.
All the IPAM-related data are cleared when a subnet is deleted thanks
to cascaded foreign key relationships.
IPAM-related data has no foreign key relationships to neutron subnet,
so removing ipam subnet manually
"""
pass
count = ipam_db_api.IpamSubnetManager.delete(self._context.session,
subnet_id)
if count < 1:
LOG.error(_LE("IPAM subnet referenced to "
"Neutron subnet %s does not exist"),
subnet_id)
raise n_exc.SubnetNotFound(subnet_id=subnet_id)

3
neutron/ipam/subnet_alloc.py
View File

@ -193,9 +193,6 @@ class IpamSubnet(driver.Subnet):
def get_details(self):
return self._req
def associate_neutron_subnet(self, subnet_id):
pass
class SubnetPoolReader(object):
'''Class to assist with reading a subnetpool, loading defaults, and

35
neutron/tests/functional/db/test_ipam.py
View File

@ -24,6 +24,7 @@ from neutron import context
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db import model_base
from neutron.db import models_v2
from neutron.ipam.drivers.neutrondb_ipam import db_models as ipam_models
from neutron.tests import base
from neutron.tests.common import base as common_base
@ -47,9 +48,13 @@ class IpamTestCase(object):
Base class for tests that aim to test ip allocation.
"""
def configure_test(self):
def configure_test(self, use_pluggable_ipam=False):
model_base.BASEV2.metadata.create_all(self.engine)
cfg.CONF.set_override('notify_nova_on_port_status_changes', False)
if use_pluggable_ipam:
self._turn_on_pluggable_ipam()
else:
self._turn_off_pluggable_ipam()
self.plugin = base_plugin.NeutronDbPluginV2()
self.cxt = get_admin_test_context(self.engine.url)
self.addCleanup(self.cxt._session.close)
@ -60,6 +65,16 @@ class IpamTestCase(object):
self._create_network()
self._create_subnet()
def _turn_off_pluggable_ipam(self):
cfg.CONF.set_override('ipam_driver', None)
self.ip_availability_range = models_v2.IPAvailabilityRange
def _turn_on_pluggable_ipam(self):
cfg.CONF.set_override('ipam_driver', 'internal')
DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
self.setup_coreplugin(DB_PLUGIN_KLASS)
self.ip_availability_range = ipam_models.IpamAvailabilityRange
def result_set_to_dicts(self, resultset, keys):
dicts = []
for item in resultset:
@ -75,7 +90,7 @@ class IpamTestCase(object):
def assert_ip_avail_range_matches(self, expected):
result_set = self.cxt.session.query(
models_v2.IPAvailabilityRange).all()
self.ip_availability_range).all()
keys = ['first_ip', 'last_ip']
actual = self.result_set_to_dicts(result_set, keys)
self.assertEqual(expected, actual)
@ -218,3 +233,19 @@ class TestIpamPsql(common_base.PostgreSQLTestCase,
def setUp(self):
super(TestIpamPsql, self).setUp()
self.configure_test()
class TestPluggableIpamMySql(common_base.MySQLTestCase,
base.BaseTestCase, IpamTestCase):
def setUp(self):
super(TestPluggableIpamMySql, self).setUp()
self.configure_test(use_pluggable_ipam=True)
class TestPluggableIpamPsql(common_base.PostgreSQLTestCase,
base.BaseTestCase, IpamTestCase):
def setUp(self):
super(TestPluggableIpamPsql, self).setUp()
self.configure_test(use_pluggable_ipam=True)

5
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -3278,6 +3278,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
[{'start': '10.0.0.2', 'end': '10.0.0.254'},
{'end': '10.0.0.254'}],
None,
[{'start': '10.0.0.200', 'end': '10.0.3.20'}],
[{'start': '10.0.2.250', 'end': '10.0.3.5'}],
[{'start': '10.0.2.10', 'end': '10.0.2.5'}],
[{'start': '10.0.0.2', 'end': '10.0.0.3'},
{'start': '10.0.0.2', 'end': '10.0.0.3'}]]
tenant_id = network['network']['tenant_id']
@ -3816,7 +3819,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
return orig(s, instance)
mock.patch.object(orm.Session, 'add',
new=db_ref_err_for_ipalloc).start()
mock.patch.object(non_ipam.IpamNonPluggableBackend,
mock.patch.object(db_base_plugin_common.DbBasePluginCommon,
'_get_subnet',
return_value=mock.Mock()).start()
# Add an IPv6 auto-address subnet to the network

258
neutron/tests/unit/db/test_ipam_pluggable_backend.py
View File

@ -15,18 +15,49 @@
import mock
import netaddr
import webob.exc
from oslo_config import cfg
from oslo_utils import uuidutils
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.db import ipam_backend_mixin
from neutron.db import ipam_pluggable_backend
from neutron.ipam import requests as ipam_req
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_base
class UseIpamMixin(object):
def setUp(self):
cfg.CONF.set_override("ipam_driver", 'internal')
super(UseIpamMixin, self).setUp()
class TestIpamHTTPResponse(UseIpamMixin, test_db_base.TestV2HTTPResponse):
pass
class TestIpamPorts(UseIpamMixin, test_db_base.TestPortsV2):
pass
class TestIpamNetworks(UseIpamMixin, test_db_base.TestNetworksV2):
pass
class TestIpamSubnets(UseIpamMixin, test_db_base.TestSubnetsV2):
pass
class TestIpamSubnetPool(UseIpamMixin, test_db_base.TestSubnetPoolsV2):
pass
class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
def setUp(self):
cfg.CONF.set_override("ipam_driver", 'internal')
super(TestDbBasePluginIpam, self).setUp()
self.tenant_id = uuidutils.generate_uuid()
self.subnet_id = uuidutils.generate_uuid()
@ -56,6 +87,11 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend()
return mocks
def _prepare_mocks_with_pool_mock(self, pool_mock):
mocks = self._prepare_mocks()
pool_mock.get_instance.return_value = mocks['driver']
return mocks
def _get_allocate_mock(self, auto_ip='10.0.0.2',
fail_ip='127.0.0.1',
error_message='SomeError'):
@ -233,3 +269,225 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
self._validate_allocate_calls(ips[:-1], mocks)
# Deallocate should be called for the first ip only
mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
@mock.patch('neutron.ipam.driver.Pool')
def test_create_subnet_over_ipam(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
cidr = '192.168.0.0/24'
allocation_pools = [{'start': '192.168.0.2', 'end': '192.168.0.254'}]
with self.subnet(allocation_pools=allocation_pools,
cidr=cidr):
pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
assert mocks['driver'].allocate_subnet.called
request = mocks['driver'].allocate_subnet.call_args[0][0]
self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
@mock.patch('neutron.ipam.driver.Pool')
def test_create_subnet_over_ipam_with_rollback(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
mocks['driver'].allocate_subnet.side_effect = ValueError
cidr = '10.0.2.0/24'
with self.network() as network:
self._create_subnet(self.fmt, network['network']['id'],
cidr, expected_res_status=500)
pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
assert mocks['driver'].allocate_subnet.called
request = mocks['driver'].allocate_subnet.call_args[0][0]
self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
# Verify no subnet was created for network
req = self.new_show_request('networks', network['network']['id'])
res = req.get_response(self.api)
net = self.deserialize(self.fmt, res)
self.assertEqual(0, len(net['network']['subnets']))
@mock.patch('neutron.ipam.driver.Pool')
def test_ipam_subnet_deallocated_if_create_fails(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
cidr = '10.0.2.0/24'
with mock.patch.object(
ipam_backend_mixin.IpamBackendMixin, '_save_subnet',
side_effect=ValueError), self.network() as network:
self._create_subnet(self.fmt, network['network']['id'],
cidr, expected_res_status=500)
pool_mock.get_instance.assert_any_call(None, mock.ANY)
self.assertEqual(2, pool_mock.get_instance.call_count)
assert mocks['driver'].allocate_subnet.called
request = mocks['driver'].allocate_subnet.call_args[0][0]
self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
# Verify remove ipam subnet was called
mocks['driver'].remove_subnet.assert_called_once_with(
self.subnet_id)
@mock.patch('neutron.ipam.driver.Pool')
def test_update_subnet_over_ipam(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
cidr = '10.0.0.0/24'
allocation_pools = [{'start': '10.0.0.2', 'end': '10.0.0.254'}]
with self.subnet(allocation_pools=allocation_pools,
cidr=cidr) as subnet:
data = {'subnet': {'allocation_pools': [
{'start': '10.0.0.10', 'end': '10.0.0.20'},
{'start': '10.0.0.30', 'end': '10.0.0.40'}]}}
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(res.status_code, 200)
pool_mock.get_instance.assert_any_call(None, mock.ANY)
self.assertEqual(2, pool_mock.get_instance.call_count)
assert mocks['driver'].update_subnet.called
request = mocks['driver'].update_subnet.call_args[0][0]
self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
ip_ranges = [netaddr.IPRange(p['start'],
p['end']) for p in data['subnet']['allocation_pools']]
self.assertEqual(ip_ranges, request.allocation_pools)
@mock.patch('neutron.ipam.driver.Pool')
def test_delete_subnet_over_ipam(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
gateway_ip = '10.0.0.1'
cidr = '10.0.0.0/24'
res = self._create_network(fmt=self.fmt, name='net',
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
pool_mock.get_instance.assert_any_call(None, mock.ANY)
self.assertEqual(2, pool_mock.get_instance.call_count)
mocks['driver'].remove_subnet.assert_called_once_with(
subnet['subnet']['id'])
@mock.patch('neutron.ipam.driver.Pool')
def test_delete_subnet_over_ipam_with_rollback(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
mocks['driver'].remove_subnet.side_effect = ValueError
gateway_ip = '10.0.0.1'
cidr = '10.0.0.0/24'
res = self._create_network(fmt=self.fmt, name='net',
admin_state_up=True)
network = self.deserialize(self.fmt, res)
subnet = self._make_subnet(self.fmt, network, gateway_ip,
cidr, ip_version=4)
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(res.status_int, webob.exc.HTTPServerError.code)
pool_mock.get_instance.assert_any_call(None, mock.ANY)
self.assertEqual(2, pool_mock.get_instance.call_count)
mocks['driver'].remove_subnet.assert_called_once_with(
subnet['subnet']['id'])
# Verify subnet was recreated after failed ipam call
subnet_req = self.new_show_request('subnets',
subnet['subnet']['id'])
raw_res = subnet_req.get_response(self.api)
sub_res = self.deserialize(self.fmt, raw_res)
self.assertIn(sub_res['subnet']['cidr'], cidr)
self.assertIn(sub_res['subnet']['gateway_ip'],
gateway_ip)
@mock.patch('neutron.ipam.driver.Pool')
def test_create_port_ipam(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
auto_ip = '10.0.0.2'
expected_calls = [{'ip_address': ''}]
mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
auto_ip=auto_ip)
with self.subnet() as subnet:
with self.port(subnet=subnet) as port:
ips = port['port']['fixed_ips']
self.assertEqual(1, len(ips))
self.assertEqual(ips[0]['ip_address'], auto_ip)
self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
self._validate_allocate_calls(expected_calls, mocks)
@mock.patch('neutron.ipam.driver.Pool')
def test_create_port_ipam_with_rollback(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
mocks['subnet'].allocate.side_effect = ValueError
with self.network() as network:
with self.subnet(network=network):
net_id = network['network']['id']
data = {
'port': {'network_id': net_id,
'tenant_id': network['network']['tenant_id']}}
port_req = self.new_create_request('ports', data)
res = port_req.get_response(self.api)
self.assertEqual(res.status_int,
webob.exc.HTTPServerError.code)
# verify no port left after failure
req = self.new_list_request('ports', self.fmt,
"network_id=%s" % net_id)
res = self.deserialize(self.fmt, req.get_response(self.api))
self.assertEqual(0, len(res['ports']))
@mock.patch('neutron.ipam.driver.Pool')
def test_update_port_ipam(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
auto_ip = '10.0.0.2'
new_ip = '10.0.0.15'
expected_calls = [{'ip_address': ip} for ip in ['', new_ip]]
mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
auto_ip=auto_ip)
with self.subnet() as subnet:
with self.port(subnet=subnet) as port:
ips = port['port']['fixed_ips']
self.assertEqual(1, len(ips))
self.assertEqual(ips[0]['ip_address'], auto_ip)
# Update port with another new ip
data = {"port": {"fixed_ips": [{
'subnet_id': subnet['subnet']['id'],
'ip_address': new_ip}]}}
req = self.new_update_request('ports', data,
port['port']['id'])
res = self.deserialize(self.fmt, req.get_response(self.api))
ips = res['port']['fixed_ips']
self.assertEqual(1, len(ips))
self.assertEqual(new_ip, ips[0]['ip_address'])
# Allocate should be called for the first two networks
self._validate_allocate_calls(expected_calls, mocks)
# Deallocate should be called for the first ip only
mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
@mock.patch('neutron.ipam.driver.Pool')
def test_delete_port_ipam(self, pool_mock):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
auto_ip = '10.0.0.2'
mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
auto_ip=auto_ip)
with self.subnet() as subnet:
with self.port(subnet=subnet) as port:
ips = port['port']['fixed_ips']
self.assertEqual(1, len(ips))
self.assertEqual(ips[0]['ip_address'], auto_ip)
req = self.new_delete_request('ports', port['port']['id'])
res = req.get_response(self.api)
self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
mocks['subnet'].deallocate.assert_called_once_with(auto_ip)