def _allocate_ips_for_port(self, context, port):
        """Allocate IP addresses for the port.

        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:
            configured_ips = self._test_fixed_ips_for_port(context,
                                                           p["network_id"],
                                                           p['fixed_ips'],
                                                           p['device_owner'])
            ips = self._allocate_fixed_ips(context,
                                           configured_ips,
                                           p['mac_address'])

            # 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)
                elif 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:
                    result = IpamNonPluggableBackend._generate_ip(context,
                                                                  subnets)
                    ips.append({'ip_address': result['ip_address'],
                                'subnet_id': result['subnet_id']})

        for subnet in v6_stateless:
            # IP addresses for IPv6 SLAAC and DHCPv6-stateless subnets
            # are implicitly included.
            ip_address = self._calculate_ipv6_eui64_addr(context, subnet,
                                                         p['mac_address'])
            ips.append({'ip_address': ip_address.format(),
                        'subnet_id': subnet['id']})

        return ips

    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, subnets):
        """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, subnets)

            is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
            if ('ip_address' in fixed and
                    subnet['cidr'] != n_const.PROVISIONAL_IPV6_PD_PREFIX):
                if (is_auto_addr_subnet and device_owner not in
                        constants.ROUTER_INTERFACE_OWNERS):
                    raise ipam_exc.AllocationOnAutoAddressSubnet(
                        ip=fixed['ip_address'], subnet_id=subnet['id'])
                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, device_owner)
        return fixed_ip_list

    def _allocate_fixed_ips(self, context, fixed_ips, mac_address):
        """Allocate IP addresses according to the configured fixed_ips."""
        ips = []

        # we need to start with entries that asked for a specific IP in case
        # those IPs happen to be next in the line for allocation for ones that
        # didn't ask for a specific IP
        fixed_ips.sort(key=lambda x: 'ip_address' not in x)
        for fixed in fixed_ips:
            subnet = self._get_subnet(context, fixed['subnet_id'])
            is_auto_addr = ipv6_utils.is_auto_address_subnet(subnet)
            if 'ip_address' in fixed:
                if not is_auto_addr:
                    # Remove the IP address from the allocation pool
                    IpamNonPluggableBackend._allocate_specific_ip(
                        context, fixed['subnet_id'], fixed['ip_address'])
                ips.append({'ip_address': fixed['ip_address'],
                            'subnet_id': fixed['subnet_id']})
            # Only subnet ID is specified => need to generate IP
            # from subnet
            else:
                if is_auto_addr:
                    ip_address = self._calculate_ipv6_eui64_addr(context,
                                                                 subnet,
                                                                 mac_address)
                    ips.append({'ip_address': ip_address.format(),
                                'subnet_id': subnet['id']})
                else:
                    subnets = [subnet]
                    # IP address allocation
                    result = self._generate_ip(context, subnets)
                    ips.append({'ip_address': result['ip_address'],
                                'subnet_id': result['subnet_id']})
        return ips

    def _get_changed_ips_for_port(self, context, original_ips,
                                  new_ips, device_owner):
        """Calculate changes in IPs for the port."""
        # the new_ips contain all of the fixed_ips that are to be updated
        if len(new_ips) > cfg.CONF.max_fixed_ips_per_port:
            msg = _('Exceeded maximum amount of fixed ips per port')
            raise n_exc.InvalidInput(error_message=msg)

        # These ips are still on the port and haven't been removed
        prev_ips = []

        # Remove all of the intersecting elements
        for original_ip in original_ips[:]:
            for new_ip in new_ips[:]:
                if ('ip_address' in new_ip and
                    original_ip['ip_address'] == new_ip['ip_address']):
                    original_ips.remove(original_ip)
                    new_ips.remove(new_ip)
                    prev_ips.append(original_ip)
                    break
            else:
                # For ports that are not router ports, retain any automatic
                # (non-optional, e.g. IPv6 SLAAC) addresses.
                if device_owner not in constants.ROUTER_INTERFACE_OWNERS:
                    subnet = self._get_subnet(context,
                                              original_ip['subnet_id'])
                    if (ipv6_utils.is_auto_address_subnet(subnet)):
                        original_ips.remove(original_ip)
                        prev_ips.append(original_ip)
        return self.Changes(add=new_ips,
                            original=prev_ips,
                            remove=original_ips)

    def setup(self, network):
        """Create and initialize a device for network's DHCP on this host."""
        port = self.setup_dhcp_port(network)
        self._update_dhcp_port(network, port)
        interface_name = self.get_interface_name(network, port)

        if ip_lib.ensure_device_is_ready(interface_name, namespace=network.namespace):
            LOG.debug("Reusing existing device: %s.", interface_name)
        else:
            self.driver.plug(network.id, port.id, interface_name, port.mac_address, namespace=network.namespace)
            self.fill_dhcp_udp_checksums(namespace=network.namespace)
        ip_cidrs = []
        for fixed_ip in port.fixed_ips:
            subnet = fixed_ip.subnet
            if not ipv6_utils.is_auto_address_subnet(subnet):
                net = netaddr.IPNetwork(subnet.cidr)
                ip_cidr = "%s/%s" % (fixed_ip.ip_address, net.prefixlen)
                ip_cidrs.append(ip_cidr)

        if self.conf.enable_isolated_metadata and self.conf.use_namespaces:
            ip_cidrs.append(METADATA_DEFAULT_CIDR)

        self.driver.init_l3(interface_name, ip_cidrs, namespace=network.namespace)

        # ensure that the dhcp interface is first in the list
        if network.namespace is None:
            device = ip_lib.IPDevice(interface_name)
            device.route.pullup_route(interface_name, ip_version=constants.IP_VERSION_4)

        if self.conf.use_namespaces:
            self._set_default_route(network, interface_name)

        return interface_name

    def create_subnet(self, context, subnet):

        s = subnet["subnet"]
        cidr = s.get("cidr", attributes.ATTR_NOT_SPECIFIED)
        prefixlen = s.get("prefixlen", attributes.ATTR_NOT_SPECIFIED)
        has_cidr = attributes.is_attr_set(cidr)
        has_prefixlen = attributes.is_attr_set(prefixlen)

        if has_cidr and has_prefixlen:
            msg = _("cidr and prefixlen must not be supplied together")
            raise n_exc.BadRequest(resource="subnets", msg=msg)

        if has_cidr:
            # turn the CIDR into a proper subnet
            net = netaddr.IPNetwork(s["cidr"])
            subnet["subnet"]["cidr"] = "%s/%s" % (net.network, net.prefixlen)

        s["tenant_id"] = self._get_tenant_id_for_create(context, s)
        subnetpool_id = self._get_subnetpool_id(s)
        if not subnetpool_id:
            if not has_cidr:
                msg = _("A cidr must be specified in the absence of a " "subnet pool")
                raise n_exc.BadRequest(resource="subnets", msg=msg)
            # Create subnet from the implicit(AKA null) pool
            created_subnet = self._create_subnet_from_implicit_pool(context, subnet)
        else:
            created_subnet = self._create_subnet_from_pool(context, subnet, subnetpool_id)

        # 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(created_subnet):
            self._add_auto_addrs_on_network_ports(context, created_subnet)

        return created_subnet

 def _update_router_gw_ports(self, context, network, subnet):
     l3plugin = manager.NeutronManager.get_service_plugins().get(
             service_constants.L3_ROUTER_NAT)
     if l3plugin:
         gw_ports = self._get_router_gw_ports_by_network(context,
                 network['id'])
         router_ids = [p['device_id'] for p in gw_ports]
         ctx_admin = ctx.get_admin_context()
         ext_subnets_dict = {s['id']: s for s in network['subnets']}
         for id in router_ids:
             router = l3plugin.get_router(ctx_admin, id)
             external_gateway_info = router['external_gateway_info']
             # Get all stateful (i.e. non-SLAAC/DHCPv6-stateless) fixed ips
             fips = [f for f in external_gateway_info['external_fixed_ips']
                     if not ipv6_utils.is_auto_address_subnet(
                         ext_subnets_dict[f['subnet_id']])]
             num_fips = len(fips)
             # Don't add the fixed IP to the port if it already
             # has a stateful fixed IP of the same IP version
             if num_fips > 1:
                 continue
             if num_fips == 1 and netaddr.IPAddress(
                     fips[0]['ip_address']).version == subnet['ip_version']:
                 continue
             external_gateway_info['external_fixed_ips'].append(
                                          {'subnet_id': subnet['id']})
             info = {'router': {'external_gateway_info':
                 external_gateway_info}}
             l3plugin.update_router(context, id, info)

    def _is_ip_required_by_subnet(self, context, subnet_id, device_owner):
        # For ports that are not router ports, retain any automatic
        # (non-optional, e.g. IPv6 SLAAC) addresses.
        if device_owner in constants.ROUTER_INTERFACE_OWNERS:
            return True

        subnet = self._get_subnet(context, subnet_id)
        return not ipv6_utils.is_auto_address_subnet(subnet)

    def setup(self, network):
        """Create and initialize a device for network's DHCP on this host."""
        port = self.setup_dhcp_port(network)
        self._update_dhcp_port(network, port)
        interface_name = self.get_interface_name(network, port)

        if ip_lib.ensure_device_is_ready(interface_name, namespace=network.namespace):
            LOG.debug("Reusing existing device: %s.", interface_name)
        else:
            try:
                self.driver.plug(
                    network.id,
                    port.id,
                    interface_name,
                    port.mac_address,
                    namespace=network.namespace,
                    mtu=network.get("mtu"),
                )
            except Exception:
                with excutils.save_and_reraise_exception():
                    LOG.exception(_LE("Unable to plug DHCP port for " "network %s. Releasing port."), network.id)
                    self.plugin.release_dhcp_port(network.id, port.device_id)

            self.fill_dhcp_udp_checksums(namespace=network.namespace)
        ip_cidrs = []
        for fixed_ip in port.fixed_ips:
            subnet = fixed_ip.subnet
            if not ipv6_utils.is_auto_address_subnet(subnet):
                net = netaddr.IPNetwork(subnet.cidr)
                ip_cidr = "%s/%s" % (fixed_ip.ip_address, net.prefixlen)
                ip_cidrs.append(ip_cidr)

        if self.driver.use_gateway_ips:
            # For each DHCP-enabled subnet, add that subnet's gateway
            # IP address to the Linux device for the DHCP port.
            for subnet in network.subnets:
                if not subnet.enable_dhcp:
                    continue
                gateway = subnet.gateway_ip
                if gateway:
                    net = netaddr.IPNetwork(subnet.cidr)
                    ip_cidrs.append("%s/%s" % (gateway, net.prefixlen))

        if self.conf.enable_isolated_metadata:
            ip_cidrs.append(METADATA_DEFAULT_CIDR)

        self.driver.init_l3(interface_name, ip_cidrs, namespace=network.namespace)

        self._set_default_route(network, interface_name)
        try:
            self._cleanup_stale_devices(network, port)
        except Exception:
            # catch everything as we don't want to fail because of
            # cleanup step
            LOG.error(_LE("Exception during stale dhcp device cleanup"))

        return interface_name

    def _is_ip_required_by_subnet(self, context, subnet_id, device_owner):
        # For ports that are not router ports, retain any automatic
        # (non-optional, e.g. IPv6 SLAAC) addresses.
        # NOTE: Need to check the SNAT ports for DVR routers here since
        # they consume an IP.
        if device_owner in const.ROUTER_INTERFACE_OWNERS_SNAT:
            return True

        subnet = self._get_subnet(context, subnet_id)
        return not (ipv6_utils.is_auto_address_subnet(subnet) and
                    not ipv6_utils.is_ipv6_pd_enabled(subnet))

 def _validate_eui64_applicable(self, subnet):
     # Per RFC 4862, section 5.5.3, prefix length and interface
     # id together should be equal to 128. Currently neutron supports
     # EUI64 interface id only, thus limiting the prefix
     # length to be 64 only.
     if ipv6_utils.is_auto_address_subnet(subnet):
         if netaddr.IPNetwork(subnet['cidr']).prefixlen != 64:
             msg = _('Invalid CIDR %s for IPv6 address mode. '
                     'OpenStack uses the EUI-64 address format, '
                     'which requires the prefix to be /64.')
             raise n_exc.InvalidInput(
                 error_message=(msg % subnet['cidr']))

    def setup(self, network):
        """Create and initialize a device for network's DHCP on this host."""
        port = self.setup_dhcp_port(network)
        interface_name = self.get_interface_name(network, port)

        if ip_lib.ensure_device_is_ready(interface_name,
                                         namespace=network.namespace):
            LOG.debug('Reusing existing device: %s.', interface_name)
        else:
            self.driver.plug(network.id,
                             port.id,
                             interface_name,
                             port.mac_address,
                             namespace=network.namespace)
            self.fill_dhcp_udp_checksums(namespace=network.namespace)
        ip_cidrs = []
        for fixed_ip in port.fixed_ips:
            subnet = fixed_ip.subnet
            if not ipv6_utils.is_auto_address_subnet(subnet):
                net = netaddr.IPNetwork(subnet.cidr)
                ip_cidr = '%s/%s' % (fixed_ip.ip_address, net.prefixlen)
                ip_cidrs.append(ip_cidr)
        LOG.debug("ip_cidrs = %s" % ip_cidrs)

        if self.driver.subnet_ip_usage is constants.USE_GATEWAY_IPS:
            # For each DHCP-enabled subnet, add that subnet's gateway
            # IP address to the Linux device for the DHCP port..
            for subnet in network.subnets:
                if not subnet.enable_dhcp:
                    continue
                gateway = subnet.gateway_ip
                if gateway:
                    net = netaddr.IPNetwork(subnet.cidr)
                    ip_cidrs.append('%s/%s' % (gateway, net.prefixlen))
        LOG.debug("ip_cidrs = %s" % ip_cidrs)

        if (self.conf.enable_isolated_metadata and
            self.conf.use_namespaces):
            ip_cidrs.append(METADATA_DEFAULT_CIDR)

        self.driver.init_l3(interface_name, ip_cidrs,
                            namespace=network.namespace)

        # ensure that the dhcp interface is first in the list
        if network.namespace is None:
            device = ip_lib.IPDevice(interface_name)
            device.route.pullup_route(interface_name,
                                      ip_version=constants.IP_VERSION_4)

        if self.conf.use_namespaces:
            self._set_default_route(network, interface_name)

        return interface_name

    def _classify_subnets(self, context, subnets):
        """Split into v4, v6 stateless and v6 stateful subnets"""

        v4, v6_stateful, v6_stateless = [], [], []
        for subnet in subnets:
            if subnet['ip_version'] == 4:
                v4.append(subnet)
            elif not ipv6_utils.is_auto_address_subnet(subnet):
                v6_stateful.append(subnet)
            else:
                v6_stateless.append(subnet)
        return v4, v6_stateful, v6_stateless

    def _create_subnet(self, context, subnet, subnetpool_id):
        s = subnet["subnet"]

        with context.session.begin(subtransactions=True):
            network = self._get_network(context, s["network_id"])
            subnet = self._allocate_subnet(context, network, s, subnetpool_id)
        if hasattr(network, "external") and network.external:
            self._update_router_gw_ports(context, network, subnet)
        # 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._add_auto_addrs_on_network_ports(context, subnet)
        return self._make_subnet_dict(subnet)

    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_set = []
        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 and
                subnet['cidr'] != constants.PROVISIONAL_IPV6_PD_PREFIX):
                # Ensure that the IP's are unique
                if not IpamNonPluggableBackend._check_unique_ip(
                        context, network_id,
                        subnet['id'], fixed['ip_address']):
                    raise n_exc.IpAddressInUse(net_id=network_id,
                                               ip_address=fixed['ip_address'])

                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_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_SNAT or
                    ipv6_utils.is_ipv6_pd_enabled(subnet) or
                    not is_auto_addr_subnet):
                    fixed_ip_set.append({'subnet_id': subnet['id']})

        self._validate_max_ips_per_port(fixed_ip_set)
        return fixed_ip_set

    def delete_subnet(self, context, id):
        with context.session.begin(subtransactions=True):
            subnet = self._get_subnet(context, id)

            # Make sure the subnet isn't used by other resources
            _check_subnet_not_used(context, id)

            # Delete all network owned ports
            qry_network_ports = (
                context.session.query(models_v2.IPAllocation).
                filter_by(subnet_id=subnet['id']).
                join(models_v2.Port))
            # Remove network owned ports, and delete IP allocations
            # for IPv6 addresses which were automatically generated
            # via SLAAC
            is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
            if is_auto_addr_subnet:
                self._subnet_check_ip_allocations_internal_router_ports(
                        context, id)
            else:
                qry_network_ports = (
                    qry_network_ports.filter(models_v2.Port.device_owner.
                    in_(AUTO_DELETE_PORT_OWNERS)))
            network_ports = qry_network_ports.all()
            if network_ports:
                for port in network_ports:
                    context.session.delete(port)
            # Check if there are more IP allocations, unless
            # is_auto_address_subnet is True. In that case the check is
            # unnecessary. This additional check not only would be wasteful
            # for this class of subnet, but is also error-prone since when
            # the isolation level is set to READ COMMITTED allocations made
            # concurrently will be returned by this query
            if not is_auto_addr_subnet:
                alloc = self._subnet_check_ip_allocations(context, id)
                if alloc:
                    LOG.info(_LI("Found port (%(port_id)s, %(ip)s) having IP "
                                 "allocation on subnet "
                                 "%(subnet)s, cannot delete"),
                             {'ip': alloc.ip_address,
                              'port_id': alloc.port_id,
                              'subnet': id})
                    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 test_combinations(self):
     Mode = collections.namedtuple('Mode', "addr_mode ra_mode "
                                           "is_auto_address")
     subnets = [
         Mode(None, None, False),
         Mode(constants.DHCPV6_STATEFUL, None, False),
         Mode(constants.DHCPV6_STATELESS, None, True),
         Mode(constants.IPV6_SLAAC, None, True),
         Mode(None, constants.DHCPV6_STATEFUL, False),
         Mode(None, constants.DHCPV6_STATELESS, True),
         Mode(None, constants.IPV6_SLAAC, True),
         Mode(constants.DHCPV6_STATEFUL, constants.DHCPV6_STATEFUL, False),
         Mode(constants.DHCPV6_STATELESS, constants.DHCPV6_STATELESS, True),
         Mode(constants.IPV6_SLAAC, constants.IPV6_SLAAC, True),
     ]
     for subnet in subnets:
         self.subnet['ipv6_address_mode'] = subnet.addr_mode
         self.subnet['ipv6_ra_mode'] = subnet.ra_mode
         self.assertEqual(subnet.is_auto_address,
                          ipv6_utils.is_auto_address_subnet(self.subnet))

def delete_subnet(self, context, id):
    # REVISIT(rkukura) The super(Ml2Plugin, self).delete_subnet()
    # function is not used because it deallocates the subnet's addresses
    # from ports in the DB without invoking the derived class's
    # update_port(), preventing mechanism drivers from being called.
    # This approach should be revisited when the API layer is reworked
    # during icehouse.

    LOG.debug("Deleting subnet %s", id)
    session = context.session
    attempt = 0
    while True:
        attempt += 1
        LOG.info(i18n._LI("Attempt %(attempt)s to delete subnet %(subnet)s"),
                 {'attempt': attempt, 'subnet': id})
        if attempt > 100:
            raise InfiniteLoopError()
        with session.begin(subtransactions=True):
            record = self._get_subnet(context, id)
            subnet = self._make_subnet_dict(record, None, context=context)
            qry_allocated = (session.query(models_v2.IPAllocation).
                             filter_by(subnet_id=id).
                             join(models_v2.Port))
            is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
            # Remove network owned ports, and delete IP allocations
            # for IPv6 addresses which were automatically generated
            # via SLAAC
            if is_auto_addr_subnet:
                self._subnet_check_ip_allocations_internal_router_ports(
                        context, id)
            else:
                qry_allocated = (
                    qry_allocated.filter(models_v2.Port.device_owner.
                    in_(db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS)))
            allocated = qry_allocated.all()
            # Delete all the IPAllocation that can be auto-deleted
            if allocated:
                for x in allocated:
                    session.delete(x)
            LOG.debug("Ports to auto-deallocate: %s", allocated)
            # Check if there are more IP allocations, unless
            # is_auto_address_subnet is True. In that case the check is
            # unnecessary. This additional check not only would be wasteful
            # for this class of subnet, but is also error-prone since when
            # the isolation level is set to READ COMMITTED allocations made
            # concurrently will be returned by this query
            if not is_auto_addr_subnet:
                alloc = self._subnet_check_ip_allocations(context, id)
                if alloc:
                    user_alloc = self._subnet_get_user_allocation(
                        context, id)
                    if user_alloc:
                        LOG.info(i18n._LI("Found port (%(port_id)s, %(ip)s) "
                            "having IP allocation on subnet "
                            "%(subnet)s, cannot delete"),
                            {'ip': user_alloc.ip_address,
                             'port_id': user_alloc.port_id,
                             'subnet': id})
                        raise exc.SubnetInUse(subnet_id=id)
                    else:
                        # allocation found and it was DHCP port
                        # that appeared after autodelete ports were
                        # removed - need to restart whole operation
                        raise os_db_exception.RetryRequest(
                            exc.SubnetInUse(subnet_id=id))

            db_base_plugin_v2._check_subnet_not_used(context, id)

            # If allocated is None, then all the IPAllocation were
            # correctly deleted during the previous pass.
            if not allocated:
                network = self.get_network(context, subnet['network_id'])
                mech_context = driver_context.SubnetContext(self, context,
                                                            subnet,
                                                            network)
                self.mechanism_manager.delete_subnet_precommit(
                    mech_context)

                LOG.debug("Deleting subnet record")
                session.delete(record)

                # The super(Ml2Plugin, self).delete_subnet() is not called,
                # so need to manually call delete_subnet for pluggable ipam
                self.ipam.delete_subnet(context, id)

                LOG.debug("Committing transaction")
                break

        for a in allocated:
            if a.port:
                # calling update_port() for each allocation to remove the
                # IP from the port and call the MechanismDrivers
                data = {attributes.PORT:
                        {'fixed_ips': [{'subnet_id': ip.subnet_id,
                                        'ip_address': ip.ip_address}
                                       for ip in a.port.fixed_ips
                                       if ip.subnet_id != id]}}
                try:
                    self.update_port(context, a.port_id, data)
                except exc.PortNotFound:
#.........这里部分代码省略.........

    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_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']

            is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
            if 'ip_address' in fixed:
                # Ensure that the IP's are unique