neutron_tempest_plugin/api/base.py

# Copyright 2012 OpenStack Foundation
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import functools
import math

import netaddr
from neutron_lib import constants as const
from tempest.common import utils as tutils
from tempest.lib.common.utils import data_utils
from tempest.lib import exceptions as lib_exc
from tempest import test

from neutron_tempest_plugin.api import clients
from neutron_tempest_plugin.common import constants
from neutron_tempest_plugin.common import utils
from neutron_tempest_plugin import config
from neutron_tempest_plugin import exceptions

CONF = config.CONF


class BaseNetworkTest(test.BaseTestCase):

    """
    Base class for the Neutron tests that use the Tempest Neutron REST client

    Per the Neutron API Guide, API v1.x was removed from the source code tree
    (docs.openstack.org/api/openstack-network/2.0/content/Overview-d1e71.html)
    Therefore, v2.x of the Neutron API is assumed. It is also assumed that the
    following options are defined in the [network] section of etc/tempest.conf:

        project_network_cidr with a block of cidr's from which smaller blocks
        can be allocated for tenant networks

        project_network_mask_bits with the mask bits to be used to partition
        the block defined by tenant-network_cidr

    Finally, it is assumed that the following option is defined in the
    [service_available] section of etc/tempest.conf

        neutron as True
    """

    force_tenant_isolation = False
    credentials = ['primary']

    # Default to ipv4.
    _ip_version = const.IP_VERSION_4

    # Derive from BaseAdminNetworkTest class to have this initialized
    admin_client = None

    @classmethod
    def get_client_manager(cls, credential_type=None, roles=None,
                           force_new=None):
        manager = super(BaseNetworkTest, cls).get_client_manager(
            credential_type=credential_type,
            roles=roles,
            force_new=force_new
        )
        # Neutron uses a different clients manager than the one in the Tempest
        # save the original in case mixed tests need it
        if credential_type == 'primary':
            cls.os_tempest = manager
        return clients.Manager(manager.credentials)

    @classmethod
    def skip_checks(cls):
        super(BaseNetworkTest, cls).skip_checks()
        if not CONF.service_available.neutron:
            raise cls.skipException("Neutron support is required")
        if (cls._ip_version == const.IP_VERSION_6 and
                not CONF.network_feature_enabled.ipv6):
            raise cls.skipException("IPv6 Tests are disabled.")
        for req_ext in getattr(cls, 'required_extensions', []):
            if not tutils.is_extension_enabled(req_ext, 'network'):
                msg = "%s extension not enabled." % req_ext
                raise cls.skipException(msg)

    @classmethod
    def setup_credentials(cls):
        # Create no network resources for these test.
        cls.set_network_resources()
        super(BaseNetworkTest, cls).setup_credentials()

    @classmethod
    def setup_clients(cls):
        super(BaseNetworkTest, cls).setup_clients()
        cls.client = cls.os_primary.network_client

    @classmethod
    def resource_setup(cls):
        super(BaseNetworkTest, cls).resource_setup()

        cls.networks = []
        cls.admin_networks = []
        cls.subnets = []
        cls.admin_subnets = []
        cls.ports = []
        cls.routers = []
        cls.floating_ips = []
        cls.metering_labels = []
        cls.service_profiles = []
        cls.flavors = []
        cls.metering_label_rules = []
        cls.qos_rules = []
        cls.qos_policies = []
        cls.ethertype = "IPv" + str(cls._ip_version)
        cls.address_scopes = []
        cls.admin_address_scopes = []
        cls.subnetpools = []
        cls.admin_subnetpools = []
        cls.security_groups = []
        cls.projects = []
        cls.log_objects = []
        cls.reserved_subnet_cidrs = set()
        cls.keypairs = []
        cls.trunks = []

    @classmethod
    def resource_cleanup(cls):
        if CONF.service_available.neutron:
            # Clean up trunks
            for trunk in cls.trunks:
                cls._try_delete_resource(cls.delete_trunk, trunk)

            # Clean up floating IPs
            for floating_ip in cls.floating_ips:
                cls._try_delete_resource(cls.client.delete_floatingip,
                                         floating_ip['id'])
            # Clean up routers
            for router in cls.routers:
                cls._try_delete_resource(cls.delete_router,
                                         router)
            # Clean up metering label rules
            for metering_label_rule in cls.metering_label_rules:
                cls._try_delete_resource(
                    cls.admin_client.delete_metering_label_rule,
                    metering_label_rule['id'])
            # Clean up metering labels
            for metering_label in cls.metering_labels:
                cls._try_delete_resource(
                    cls.admin_client.delete_metering_label,
                    metering_label['id'])
            # Clean up flavors
            for flavor in cls.flavors:
                cls._try_delete_resource(
                    cls.admin_client.delete_flavor,
                    flavor['id'])
            # Clean up service profiles
            for service_profile in cls.service_profiles:
                cls._try_delete_resource(
                    cls.admin_client.delete_service_profile,
                    service_profile['id'])
            # Clean up ports
            for port in cls.ports:
                cls._try_delete_resource(cls.client.delete_port,
                                         port['id'])
            # Clean up subnets
            for subnet in cls.subnets:
                cls._try_delete_resource(cls.client.delete_subnet,
                                         subnet['id'])
            # Clean up admin subnets
            for subnet in cls.admin_subnets:
                cls._try_delete_resource(cls.admin_client.delete_subnet,
                                         subnet['id'])
            # Clean up networks
            for network in cls.networks:
                cls._try_delete_resource(cls.delete_network, network)

            # Clean up admin networks
            for network in cls.admin_networks:
                cls._try_delete_resource(cls.admin_client.delete_network,
                                         network['id'])

            # Clean up security groups
            for secgroup in cls.security_groups:
                cls._try_delete_resource(cls.client.delete_security_group,
                                         secgroup['id'])

            for subnetpool in cls.subnetpools:
                cls._try_delete_resource(cls.client.delete_subnetpool,
                                         subnetpool['id'])

            for subnetpool in cls.admin_subnetpools:
                cls._try_delete_resource(cls.admin_client.delete_subnetpool,
                                         subnetpool['id'])

            for address_scope in cls.address_scopes:
                cls._try_delete_resource(cls.client.delete_address_scope,
                                         address_scope['id'])

            for address_scope in cls.admin_address_scopes:
                cls._try_delete_resource(
                    cls.admin_client.delete_address_scope,
                    address_scope['id'])

            for project in cls.projects:
                cls._try_delete_resource(
                    cls.identity_admin_client.delete_project,
                    project['id'])

            # Clean up QoS rules
            for qos_rule in cls.qos_rules:
                cls._try_delete_resource(cls.admin_client.delete_qos_rule,
                                         qos_rule['id'])
            # Clean up QoS policies
            # as all networks and ports are already removed, QoS policies
            # shouldn't be "in use"
            for qos_policy in cls.qos_policies:
                cls._try_delete_resource(cls.admin_client.delete_qos_policy,
                                         qos_policy['id'])

            # Clean up log_objects
            for log_object in cls.log_objects:
                cls._try_delete_resource(cls.admin_client.delete_log,
                                         log_object['id'])

            for keypair in cls.keypairs:
                cls._try_delete_resource(cls.delete_keypair, keypair)

        super(BaseNetworkTest, cls).resource_cleanup()

    @classmethod
    def _try_delete_resource(cls, delete_callable, *args, **kwargs):
        """Cleanup resources in case of test-failure

        Some resources are explicitly deleted by the test.
        If the test failed to delete a resource, this method will execute
        the appropriate delete methods. Otherwise, the method ignores NotFound
        exceptions thrown for resources that were correctly deleted by the
        test.

        :param delete_callable: delete method
        :param args: arguments for delete method
        :param kwargs: keyword arguments for delete method
        """
        try:
            delete_callable(*args, **kwargs)
        # if resource is not found, this means it was deleted in the test
        except lib_exc.NotFound:
            pass

    @classmethod
    def create_network(cls, network_name=None, client=None, external=None,
                       shared=None, provider_network_type=None,
                       provider_physical_network=None,
                       provider_segmentation_id=None, **kwargs):
        """Create a network.

        When client is not provider and admin_client is attribute is not None
        (for example when using BaseAdminNetworkTest base class) and using any
        of the convenience parameters (external, shared, provider_network_type,
        provider_physical_network and provider_segmentation_id) it silently
        uses admin_client. If the network is not shared then it uses the same
        project_id as regular client.

        :param network_name: Human-readable name of the network

        :param client: client to be used for connecting to network service

        :param external: indicates whether the network has an external routing
        facility that's not managed by the networking service.

        :param shared: indicates whether this resource is shared across all
        projects. By default, only administrative users can change this value.
        If True and admin_client attribute is not None, then the network is
        created under administrative project.

        :param provider_network_type: the type of physical network that this
        network should be mapped to. For example, 'flat', 'vlan', 'vxlan', or
        'gre'. Valid values depend on a networking back-end.

        :param provider_physical_network: the physical network where this
        network should be implemented. The Networking API v2.0 does not provide
        a way to list available physical networks. For example, the Open
        vSwitch plug-in configuration file defines a symbolic name that maps to
        specific bridges on each compute host.

        :param provider_segmentation_id: The ID of the isolated segment on the
        physical network. The network_type attribute defines the segmentation
        model. For example, if the network_type value is 'vlan', this ID is a
        vlan identifier. If the network_type value is 'gre', this ID is a gre
        key.

        :param **kwargs: extra parameters to be forwarded to network service
        """

        name = (network_name or kwargs.pop('name', None) or
                data_utils.rand_name('test-network-'))

        # translate convenience parameters
        admin_client_required = False
        if provider_network_type:
            admin_client_required = True
            kwargs['provider:network_type'] = provider_network_type
        if provider_physical_network:
            admin_client_required = True
            kwargs['provider:physical_network'] = provider_physical_network
        if provider_segmentation_id:
            admin_client_required = True
            kwargs['provider:segmentation_id'] = provider_segmentation_id
        if external is not None:
            admin_client_required = True
            kwargs['router:external'] = bool(external)
        if shared is not None:
            admin_client_required = True
            kwargs['shared'] = bool(shared)

        if not client:
            if admin_client_required and cls.admin_client:
                # For convenience silently switch to admin client
                client = cls.admin_client
                if not shared:
                    # Keep this network visible from current project
                    project_id = (kwargs.get('project_id') or
                                  kwargs.get('tenant_id') or
                                  cls.client.tenant_id)
                    kwargs.update(project_id=project_id, tenant_id=project_id)
            else:
                # Use default client
                client = cls.client

        network = client.create_network(name=name, **kwargs)['network']
        network['client'] = client
        cls.networks.append(network)
        return network

    @classmethod
    def delete_network(cls, network, client=None):
        client = client or network.get('client') or cls.client
        client.delete_network(network['id'])

    @classmethod
    def create_shared_network(cls, network_name=None, **kwargs):
        return cls.create_network(name=network_name, shared=True, **kwargs)

    @classmethod
    def create_network_keystone_v3(cls, network_name=None, project_id=None,
                                   tenant_id=None, client=None):
        params = {}
        if project_id:
            params['project_id'] = project_id
        if tenant_id:
            params['tenant_id'] = tenant_id
        return cls.create_network(name=network_name, client=client, **params)

    @classmethod
    def create_subnet(cls, network, gateway='', cidr=None, mask_bits=None,
                      ip_version=None, client=None, reserve_cidr=True,
                      **kwargs):
        """Wrapper utility that returns a test subnet.

        Convenient wrapper for client.create_subnet method. It reserves and
        allocates CIDRs to avoid creating overlapping subnets.

        :param network: network where to create the subnet
        network['id'] must contain the ID of the network

        :param gateway: gateway IP address
        It can be a str or a netaddr.IPAddress
        If gateway is not given, then it will use default address for
        given subnet CIDR, like "192.168.0.1" for "192.168.0.0/24" CIDR
        if gateway is given as None then no gateway will be assigned

        :param cidr: CIDR of the subnet to create
        It can be either None, a str or a netaddr.IPNetwork instance

        :param mask_bits: CIDR prefix length
        It can be either None or a numeric value.
        If cidr parameter is given then mask_bits is used to determinate a
        sequence of valid CIDR to use as generated.
        Please see netaddr.IPNetwork.subnet method documentation[1]

        :param ip_version: ip version of generated subnet CIDRs
        It can be None, IP_VERSION_4 or IP_VERSION_6
        It has to match given either given CIDR and gateway

        :param ip_version: numeric value (either IP_VERSION_4 or IP_VERSION_6)
        this value must match CIDR and gateway IP versions if any of them is
        given

        :param client: client to be used to connect to network service

        :param reserve_cidr: if True then it reserves assigned CIDR to avoid
        using the same CIDR for further subnets in the scope of the same
        test case class

        :param **kwargs: optional parameters to be forwarded to wrapped method

        [1] http://netaddr.readthedocs.io/en/latest/tutorial_01.html#supernets-and-subnets  # noqa
        """

        # allow tests to use admin client
        if not client:
            client = cls.client

        if gateway:
            gateway_ip = netaddr.IPAddress(gateway)
            if ip_version:
                if ip_version != gateway_ip.version:
                    raise ValueError(
                        "Gateway IP version doesn't match IP version")
            else:
                ip_version = gateway_ip.version
        else:
            ip_version = ip_version or cls._ip_version

        for subnet_cidr in cls.get_subnet_cidrs(
                ip_version=ip_version, cidr=cidr, mask_bits=mask_bits):
            if gateway is not None:
                kwargs['gateway_ip'] = str(gateway or (subnet_cidr.ip + 1))
            try:
                body = client.create_subnet(
                    network_id=network['id'],
                    cidr=str(subnet_cidr),
                    ip_version=subnet_cidr.version,
                    **kwargs)
                break
            except lib_exc.BadRequest as e:
                if 'overlaps with another subnet' not in str(e):
                    raise
        else:
            message = 'Available CIDR for subnet creation could not be found'
            raise ValueError(message)
        subnet = body['subnet']
        if client is cls.client:
            cls.subnets.append(subnet)
        else:
            cls.admin_subnets.append(subnet)
        if reserve_cidr:
            cls.reserve_subnet_cidr(subnet_cidr)
        return subnet

    @classmethod
    def reserve_subnet_cidr(cls, addr, **ipnetwork_kwargs):
        """Reserve given subnet CIDR making sure it is not used by create_subnet

        :param addr: the CIDR address to be reserved
        It can be a str or netaddr.IPNetwork instance

        :param **ipnetwork_kwargs: optional netaddr.IPNetwork constructor
        parameters
        """

        if not cls.try_reserve_subnet_cidr(addr, **ipnetwork_kwargs):
            raise ValueError('Subnet CIDR already reserved: %r'.format(
                addr))

    @classmethod
    def try_reserve_subnet_cidr(cls, addr, **ipnetwork_kwargs):
        """Reserve given subnet CIDR if it hasn't been reserved before

        :param addr: the CIDR address to be reserved
        It can be a str or netaddr.IPNetwork instance

        :param **ipnetwork_kwargs: optional netaddr.IPNetwork constructor
        parameters

        :return: True if it wasn't reserved before, False elsewhere.
        """

        subnet_cidr = netaddr.IPNetwork(addr, **ipnetwork_kwargs)
        if subnet_cidr in cls.reserved_subnet_cidrs:
            return False
        else:
            cls.reserved_subnet_cidrs.add(subnet_cidr)
            return True

    @classmethod
    def get_subnet_cidrs(
            cls, cidr=None, mask_bits=None, ip_version=None):
        """Iterate over a sequence of unused subnet CIDR for IP version

        :param cidr: CIDR of the subnet to create
        It can be either None, a str or a netaddr.IPNetwork instance

        :param mask_bits: CIDR prefix length
        It can be either None or a numeric value.
        If cidr parameter is given then mask_bits is used to determinate a
        sequence of valid CIDR to use as generated.
        Please see netaddr.IPNetwork.subnet method documentation[1]

        :param ip_version: ip version of generated subnet CIDRs
        It can be None, IP_VERSION_4 or IP_VERSION_6
        It has to match given CIDR if given

        :return: iterator over reserved CIDRs of type netaddr.IPNetwork

        [1] http://netaddr.readthedocs.io/en/latest/tutorial_01.html#supernets-and-subnets  # noqa
        """

        if cidr:
            # Generate subnet CIDRs starting from given CIDR
            # checking it is of requested IP version
            cidr = netaddr.IPNetwork(cidr, version=ip_version)
        else:
            # Generate subnet CIDRs starting from configured values
            ip_version = ip_version or cls._ip_version
            if ip_version == const.IP_VERSION_4:
                mask_bits = mask_bits or config.safe_get_config_value(
                    'network', 'project_network_mask_bits')
                cidr = netaddr.IPNetwork(config.safe_get_config_value(
                    'network', 'project_network_cidr'))
            elif ip_version == const.IP_VERSION_6:
                mask_bits = config.safe_get_config_value(
                    'network', 'project_network_v6_mask_bits')
                cidr = netaddr.IPNetwork(config.safe_get_config_value(
                    'network', 'project_network_v6_cidr'))
            else:
                raise ValueError('Invalid IP version: {!r}'.format(ip_version))

        if mask_bits:
            subnet_cidrs = cidr.subnet(mask_bits)
        else:
            subnet_cidrs = iter([cidr])

        for subnet_cidr in subnet_cidrs:
            if subnet_cidr not in cls.reserved_subnet_cidrs:
                yield subnet_cidr

    @classmethod
    def create_port(cls, network, **kwargs):
        """Wrapper utility that returns a test port."""
        if CONF.network.port_vnic_type and 'binding:vnic_type' not in kwargs:
            kwargs['binding:vnic_type'] = CONF.network.port_vnic_type
        body = cls.client.create_port(network_id=network['id'],
                                      **kwargs)
        port = body['port']
        cls.ports.append(port)
        return port

    @classmethod
    def update_port(cls, port, **kwargs):
        """Wrapper utility that updates a test port."""
        body = cls.client.update_port(port['id'],
                                      **kwargs)
        return body['port']

    @classmethod
    def _create_router_with_client(
        cls, client, router_name=None, admin_state_up=False,
        external_network_id=None, enable_snat=None, **kwargs
    ):
        ext_gw_info = {}
        if external_network_id:
            ext_gw_info['network_id'] = external_network_id
        if enable_snat is not None:
            ext_gw_info['enable_snat'] = enable_snat
        body = client.create_router(
            router_name, external_gateway_info=ext_gw_info,
            admin_state_up=admin_state_up, **kwargs)
        router = body['router']
        cls.routers.append(router)
        return router

    @classmethod
    def create_router(cls, *args, **kwargs):
        return cls._create_router_with_client(cls.client, *args, **kwargs)

    @classmethod
    def create_admin_router(cls, *args, **kwargs):
        return cls._create_router_with_client(cls.os_admin.network_client,
                                              *args, **kwargs)

    @classmethod
    def create_floatingip(cls, external_network_id):
        """Wrapper utility that returns a test floating IP."""
        body = cls.client.create_floatingip(
            floating_network_id=external_network_id)
        fip = body['floatingip']
        cls.floating_ips.append(fip)
        return fip

    @classmethod
    def create_router_interface(cls, router_id, subnet_id):
        """Wrapper utility that returns a router interface."""
        interface = cls.client.add_router_interface_with_subnet_id(
            router_id, subnet_id)
        return interface

    @classmethod
    def get_supported_qos_rule_types(cls):
        body = cls.client.list_qos_rule_types()
        return [rule_type['type'] for rule_type in body['rule_types']]

    @classmethod
    def create_qos_policy(cls, name, description=None, shared=False,
                          tenant_id=None, is_default=False):
        """Wrapper utility that returns a test QoS policy."""
        body = cls.admin_client.create_qos_policy(
            name, description, shared, tenant_id, is_default)
        qos_policy = body['policy']
        cls.qos_policies.append(qos_policy)
        return qos_policy

    @classmethod
    def create_qos_bandwidth_limit_rule(cls, policy_id, max_kbps,
                                        max_burst_kbps,
                                        direction=const.EGRESS_DIRECTION):
        """Wrapper utility that returns a test QoS bandwidth limit rule."""
        body = cls.admin_client.create_bandwidth_limit_rule(
            policy_id, max_kbps, max_burst_kbps, direction)
        qos_rule = body['bandwidth_limit_rule']
        cls.qos_rules.append(qos_rule)
        return qos_rule

    @classmethod
    def delete_router(cls, router, client=None):
        client = client or cls.client
        body = client.list_router_interfaces(router['id'])
        interfaces = [port for port in body['ports']
                      if port['device_owner'] in const.ROUTER_INTERFACE_OWNERS]
        for i in interfaces:
            try:
                client.remove_router_interface_with_subnet_id(
                    router['id'], i['fixed_ips'][0]['subnet_id'])
            except lib_exc.NotFound:
                pass
        client.delete_router(router['id'])

    @classmethod
    def create_address_scope(cls, name, is_admin=False, **kwargs):
        if is_admin:
            body = cls.admin_client.create_address_scope(name=name, **kwargs)
            cls.admin_address_scopes.append(body['address_scope'])
        else:
            body = cls.client.create_address_scope(name=name, **kwargs)
            cls.address_scopes.append(body['address_scope'])
        return body['address_scope']

    @classmethod
    def create_subnetpool(cls, name, is_admin=False, **kwargs):
        if is_admin:
            body = cls.admin_client.create_subnetpool(name, **kwargs)
            cls.admin_subnetpools.append(body['subnetpool'])
        else:
            body = cls.client.create_subnetpool(name, **kwargs)
            cls.subnetpools.append(body['subnetpool'])
        return body['subnetpool']

    @classmethod
    def create_project(cls, name=None, description=None):
        test_project = name or data_utils.rand_name('test_project_')
        test_description = description or data_utils.rand_name('desc_')
        project = cls.identity_admin_client.create_project(
            name=test_project,
            description=test_description)['project']
        cls.projects.append(project)
        return project

    @classmethod
    def create_security_group(cls, name, **kwargs):
        body = cls.client.create_security_group(name=name, **kwargs)
        cls.security_groups.append(body['security_group'])
        return body['security_group']

    @classmethod
    def create_keypair(cls, client=None, name=None, **kwargs):
        client = client or cls.os_primary.keypairs_client
        name = name or data_utils.rand_name('keypair-test')
        keypair = client.create_keypair(name=name, **kwargs)['keypair']

        # save client for later cleanup
        keypair['client'] = client
        cls.keypairs.append(keypair)
        return keypair

    @classmethod
    def delete_keypair(cls, keypair, client=None):
        client = (client or keypair.get('client') or
                  cls.os_primary.keypairs_client)
        client.delete_keypair(keypair_name=keypair['name'])

    @classmethod
    def create_trunk(cls, port=None, subports=None, client=None, **kwargs):
        """Create network trunk

        :param port: dictionary containing parent port ID (port['id'])
        :param client: client to be used for connecting to networking service
        :param **kwargs: extra parameters to be forwarded to network service

        :returns: dictionary containing created trunk details
        """
        client = client or cls.client

        if port:
            kwargs['port_id'] = port['id']

        trunk = client.create_trunk(subports=subports, **kwargs)['trunk']
        # Save client reference for later deletion
        trunk['client'] = client
        cls.trunks.append(trunk)
        return trunk

    @classmethod
    def delete_trunk(cls, trunk, client=None):
        """Delete network trunk

        :param trunk: dictionary containing trunk ID (trunk['id'])

        :param client: client to be used for connecting to networking service
        """
        client = client or trunk.get('client') or cls.client
        trunk.update(client.show_trunk(trunk['id'])['trunk'])

        if not trunk['admin_state_up']:
            # Cannot touch trunk before admin_state_up is True
            client.update_trunk(trunk['id'], admin_state_up=True)
        if trunk['sub_ports']:
            # Removes trunk ports before deleting it
            cls._try_delete_resource(client.remove_subports, trunk['id'],
                                     trunk['sub_ports'])

        # we have to detach the interface from the server before
        # the trunk can be deleted.
        parent_port = {'id': trunk['port_id']}

        def is_parent_port_detached():
            parent_port.update(client.show_port(parent_port['id'])['port'])
            return not parent_port['device_id']

        if not is_parent_port_detached():
            # this could probably happen when trunk is deleted and parent port
            # has been assigned to a VM that is still running. Here we are
            # assuming that device_id points to such VM.
            cls.os_primary.compute.InterfacesClient().delete_interface(
                parent_port['device_id'], parent_port['id'])
            utils.wait_until_true(is_parent_port_detached)

        client.delete_trunk(trunk['id'])


class BaseAdminNetworkTest(BaseNetworkTest):

    credentials = ['primary', 'admin']

    @classmethod
    def setup_clients(cls):
        super(BaseAdminNetworkTest, cls).setup_clients()
        cls.admin_client = cls.os_admin.network_client
        cls.identity_admin_client = cls.os_admin.projects_client

    @classmethod
    def create_metering_label(cls, name, description):
        """Wrapper utility that returns a test metering label."""
        body = cls.admin_client.create_metering_label(
            description=description,
            name=data_utils.rand_name("metering-label"))
        metering_label = body['metering_label']
        cls.metering_labels.append(metering_label)
        return metering_label

    @classmethod
    def create_metering_label_rule(cls, remote_ip_prefix, direction,
                                   metering_label_id):
        """Wrapper utility that returns a test metering label rule."""
        body = cls.admin_client.create_metering_label_rule(
            remote_ip_prefix=remote_ip_prefix, direction=direction,
            metering_label_id=metering_label_id)
        metering_label_rule = body['metering_label_rule']
        cls.metering_label_rules.append(metering_label_rule)
        return metering_label_rule

    @classmethod
    def create_flavor(cls, name, description, service_type):
        """Wrapper utility that returns a test flavor."""
        body = cls.admin_client.create_flavor(
            description=description, service_type=service_type,
            name=name)
        flavor = body['flavor']
        cls.flavors.append(flavor)
        return flavor

    @classmethod
    def create_service_profile(cls, description, metainfo, driver):
        """Wrapper utility that returns a test service profile."""
        body = cls.admin_client.create_service_profile(
            driver=driver, metainfo=metainfo, description=description)
        service_profile = body['service_profile']
        cls.service_profiles.append(service_profile)
        return service_profile

    @classmethod
    def create_log(cls, name, description=None,
                   resource_type='security_group', resource_id=None,
                   target_id=None, event='ALL', enabled=True):
        """Wrapper utility that returns a test log object."""
        log_args = {'name': name,
                    'description': description,
                    'resource_type': resource_type,
                    'resource_id': resource_id,
                    'target_id': target_id,
                    'event': event,
                    'enabled': enabled}
        body = cls.admin_client.create_log(**log_args)
        log_object = body['log']
        cls.log_objects.append(log_object)
        return log_object

    @classmethod
    def get_unused_ip(cls, net_id, ip_version=None):
        """Get an unused ip address in a allocation pool of net"""
        body = cls.admin_client.list_ports(network_id=net_id)
        ports = body['ports']
        used_ips = []
        for port in ports:
            used_ips.extend(
                [fixed_ip['ip_address'] for fixed_ip in port['fixed_ips']])
        body = cls.admin_client.list_subnets(network_id=net_id)
        subnets = body['subnets']

        for subnet in subnets:
            if ip_version and subnet['ip_version'] != ip_version:
                continue
            cidr = subnet['cidr']
            allocation_pools = subnet['allocation_pools']
            iterators = []
            if allocation_pools:
                for allocation_pool in allocation_pools:
                    iterators.append(netaddr.iter_iprange(
                        allocation_pool['start'], allocation_pool['end']))
            else:
                net = netaddr.IPNetwork(cidr)

                def _iterip():
                    for ip in net:
                        if ip not in (net.network, net.broadcast):
                            yield ip
                iterators.append(iter(_iterip()))

            for iterator in iterators:
                for ip in iterator:
                    if str(ip) not in used_ips:
                        return str(ip)

        message = (
            "net(%s) has no usable IP address in allocation pools" % net_id)
        raise exceptions.InvalidConfiguration(message)


def require_qos_rule_type(rule_type):
    def decorator(f):
        @functools.wraps(f)
        def wrapper(self, *func_args, **func_kwargs):
            if rule_type not in self.get_supported_qos_rule_types():
                raise self.skipException(
                    "%s rule type is required." % rule_type)
            return f(self, *func_args, **func_kwargs)
        return wrapper
    return decorator


def _require_sorting(f):
    @functools.wraps(f)
    def inner(self, *args, **kwargs):
        if not tutils.is_extension_enabled("sorting", "network"):
            self.skipTest('Sorting feature is required')
        return f(self, *args, **kwargs)
    return inner


def _require_pagination(f):
    @functools.wraps(f)
    def inner(self, *args, **kwargs):
        if not tutils.is_extension_enabled("pagination", "network"):
            self.skipTest('Pagination feature is required')
        return f(self, *args, **kwargs)
    return inner


class BaseSearchCriteriaTest(BaseNetworkTest):

    # This should be defined by subclasses to reflect resource name to test
    resource = None

    field = 'name'

    # NOTE(ihrachys): some names, like those starting with an underscore (_)
    # are sorted differently depending on whether the plugin implements native
    # sorting support, or not. So we avoid any such cases here, sticking to
    # alphanumeric. Also test a case when there are multiple resources with the
    # same name
    resource_names = ('test1', 'abc1', 'test10', '123test') + ('test1',)

    force_tenant_isolation = True

    list_kwargs = {}

    list_as_admin = False

    def assertSameOrder(self, original, actual):
        # gracefully handle iterators passed
        original = list(original)
        actual = list(actual)
        self.assertEqual(len(original), len(actual))
        for expected, res in zip(original, actual):
            self.assertEqual(expected[self.field], res[self.field])

    @utils.classproperty
    def plural_name(self):
        return '%ss' % self.resource

    @property
    def list_client(self):
        return self.admin_client if self.list_as_admin else self.client

    def list_method(self, *args, **kwargs):
        method = getattr(self.list_client, 'list_%s' % self.plural_name)
        kwargs.update(self.list_kwargs)
        return method(*args, **kwargs)

    def get_bare_url(self, url):
        base_url = self.client.base_url
        self.assertTrue(url.startswith(base_url))
        return url[len(base_url):]

    @classmethod
    def _extract_resources(cls, body):
        return body[cls.plural_name]

    def _test_list_sorts(self, direction):
        sort_args = {
            'sort_dir': direction,
            'sort_key': self.field
        }
        body = self.list_method(**sort_args)
        resources = self._extract_resources(body)
        self.assertNotEmpty(
            resources, "%s list returned is empty" % self.resource)
        retrieved_names = [res[self.field] for res in resources]
        expected = sorted(retrieved_names)
        if direction == constants.SORT_DIRECTION_DESC:
            expected = list(reversed(expected))
        self.assertEqual(expected, retrieved_names)

    @_require_sorting
    def _test_list_sorts_asc(self):
        self._test_list_sorts(constants.SORT_DIRECTION_ASC)

    @_require_sorting
    def _test_list_sorts_desc(self):
        self._test_list_sorts(constants.SORT_DIRECTION_DESC)

    @_require_pagination
    def _test_list_pagination(self):
        for limit in range(1, len(self.resource_names) + 1):
            pagination_args = {
                'limit': limit,
            }
            body = self.list_method(**pagination_args)
            resources = self._extract_resources(body)
            self.assertEqual(limit, len(resources))

    @_require_pagination
    def _test_list_no_pagination_limit_0(self):
        pagination_args = {
            'limit': 0,
        }
        body = self.list_method(**pagination_args)
        resources = self._extract_resources(body)
        self.assertGreaterEqual(len(resources), len(self.resource_names))

    def _test_list_pagination_iteratively(self, lister):
        # first, collect all resources for later comparison
        sort_args = {
            'sort_dir': constants.SORT_DIRECTION_ASC,
            'sort_key': self.field
        }
        body = self.list_method(**sort_args)
        expected_resources = self._extract_resources(body)
        self.assertNotEmpty(expected_resources)

        resources = lister(
            len(expected_resources), sort_args
        )

        # finally, compare that the list retrieved in one go is identical to
        # the one containing pagination results
        self.assertSameOrder(expected_resources, resources)

    def _list_all_with_marker(self, niterations, sort_args):
        # paginate resources one by one, using last fetched resource as a
        # marker
        resources = []
        for i in range(niterations):
            pagination_args = sort_args.copy()
            pagination_args['limit'] = 1
            if resources:
                pagination_args['marker'] = resources[-1]['id']
            body = self.list_method(**pagination_args)
            resources_ = self._extract_resources(body)
            self.assertEqual(1, len(resources_))
            resources.extend(resources_)
        return resources

    @_require_pagination
    @_require_sorting
    def _test_list_pagination_with_marker(self):
        self._test_list_pagination_iteratively(self._list_all_with_marker)

    def _list_all_with_hrefs(self, niterations, sort_args):
        # paginate resources one by one, using next href links
        resources = []
        prev_links = {}

        for i in range(niterations):
            if prev_links:
                uri = self.get_bare_url(prev_links['next'])
            else:
                sort_args.update(self.list_kwargs)
                uri = self.list_client.build_uri(
                    self.plural_name, limit=1, **sort_args)
            prev_links, body = self.list_client.get_uri_with_links(
                self.plural_name, uri
            )
            resources_ = self._extract_resources(body)
            self.assertEqual(1, len(resources_))
            resources.extend(resources_)

        # The last element is empty and does not contain 'next' link
        uri = self.get_bare_url(prev_links['next'])
        prev_links, body = self.client.get_uri_with_links(
            self.plural_name, uri
        )
        self.assertNotIn('next', prev_links)

        # Now walk backwards and compare results
        resources2 = []
        for i in range(niterations):
            uri = self.get_bare_url(prev_links['previous'])
            prev_links, body = self.list_client.get_uri_with_links(
                self.plural_name, uri
            )
            resources_ = self._extract_resources(body)
            self.assertEqual(1, len(resources_))
            resources2.extend(resources_)

        self.assertSameOrder(resources, reversed(resources2))

        return resources

    @_require_pagination
    @_require_sorting
    def _test_list_pagination_with_href_links(self):
        self._test_list_pagination_iteratively(self._list_all_with_hrefs)

    @_require_pagination
    @_require_sorting
    def _test_list_pagination_page_reverse_with_href_links(
            self, direction=constants.SORT_DIRECTION_ASC):
        pagination_args = {
            'sort_dir': direction,
            'sort_key': self.field,
        }
        body = self.list_method(**pagination_args)
        expected_resources = self._extract_resources(body)

        page_size = 2
        pagination_args['limit'] = page_size

        prev_links = {}
        resources = []
        num_resources = len(expected_resources)
        niterations = int(math.ceil(float(num_resources) / page_size))
        for i in range(niterations):
            if prev_links:
                uri = self.get_bare_url(prev_links['previous'])
            else:
                pagination_args.update(self.list_kwargs)
                uri = self.list_client.build_uri(
                    self.plural_name, page_reverse=True, **pagination_args)
            prev_links, body = self.list_client.get_uri_with_links(
                self.plural_name, uri
            )
            resources_ = self._extract_resources(body)
            self.assertGreaterEqual(page_size, len(resources_))
            resources.extend(reversed(resources_))

        self.assertSameOrder(expected_resources, reversed(resources))

    @_require_pagination
    @_require_sorting
    def _test_list_pagination_page_reverse_asc(self):
        self._test_list_pagination_page_reverse(
            direction=constants.SORT_DIRECTION_ASC)

    @_require_pagination
    @_require_sorting
    def _test_list_pagination_page_reverse_desc(self):
        self._test_list_pagination_page_reverse(
            direction=constants.SORT_DIRECTION_DESC)

    def _test_list_pagination_page_reverse(self, direction):
        pagination_args = {
            'sort_dir': direction,
            'sort_key': self.field,
            'limit': 3,
        }
        body = self.list_method(**pagination_args)
        expected_resources = self._extract_resources(body)

        pagination_args['limit'] -= 1
        pagination_args['marker'] = expected_resources[-1]['id']
        pagination_args['page_reverse'] = True
        body = self.list_method(**pagination_args)

        self.assertSameOrder(
            # the last entry is not included in 2nd result when used as a
            # marker
            expected_resources[:-1],
            self._extract_resources(body))

    def _test_list_validation_filters(self):
        validation_args = {
            'unknown_filter': 'value',
        }
        body = self.list_method(**validation_args)
        resources = self._extract_resources(body)
        for resource in resources:
            self.assertIn(resource['name'], self.resource_names)