Tempest plugin for the Neutron project.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1447 lines
56 KiB

# 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 time
import netaddr
from neutron_lib import constants as const
from oslo_log import log
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
LOG = log.getLogger(__name__)
class BaseNetworkTest(test.BaseTestCase):
"""Base class for Neutron tests that use the Tempest Neutron REST client
Per the Neutron API Guide, API v1.x was removed from the source code tree
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
external_network_id = CONF.network.public_network_id
def get_client_manager(cls, credential_type=None, roles=None,
manager = super(BaseNetworkTest, cls).get_client_manager(
# 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)
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)
def setup_credentials(cls):
# Create no network resources for these test.
super(BaseNetworkTest, cls).setup_credentials()
def setup_clients(cls):
super(BaseNetworkTest, cls).setup_clients()
cls.client = cls.os_primary.network_client
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.port_forwardings = []
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.admin_security_groups = []
cls.projects = []
cls.log_objects = []
cls.reserved_subnet_cidrs = set()
cls.keypairs = []
cls.trunks = []
cls.network_segment_ranges = []
cls.conntrack_helpers = []
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 port forwardings
for pf in cls.port_forwardings:
cls._try_delete_resource(cls.delete_port_forwarding, pf)
# Clean up floating IPs
for floating_ip in cls.floating_ips:
cls._try_delete_resource(cls.delete_floatingip, floating_ip)
# Clean up conntrack helpers
for cth in cls.conntrack_helpers:
cls._try_delete_resource(cls.delete_conntrack_helper, cth)
# Clean up routers
for router in cls.routers:
# Clean up metering label rules
for metering_label_rule in cls.metering_label_rules:
# Clean up metering labels
for metering_label in cls.metering_labels:
# Clean up flavors
for flavor in cls.flavors:
# Clean up service profiles
for service_profile in cls.service_profiles:
# Clean up ports
for port in cls.ports:
# Clean up subnets
for subnet in cls.subnets:
# Clean up admin subnets
for subnet in cls.admin_subnets:
# 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:
# Clean up security groups
for security_group in cls.security_groups:
# Clean up admin security groups
for security_group in cls.admin_security_groups:
for subnetpool in cls.subnetpools:
for subnetpool in cls.admin_subnetpools:
for address_scope in cls.address_scopes:
for address_scope in cls.admin_address_scopes:
for project in cls.projects:
# Clean up QoS rules
for qos_rule in cls.qos_rules:
# 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:
# Clean up log_objects
for log_object in cls.log_objects:
for keypair in cls.keypairs:
cls._try_delete_resource(cls.delete_keypair, keypair)
# Clean up network_segment_ranges
for network_segment_range in cls.network_segment_ranges:
super(BaseNetworkTest, cls).resource_cleanup()
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
:param delete_callable: delete method
:param args: arguments for delete method
:param kwargs: keyword arguments for delete method
delete_callable(*args, **kwargs)
# if resource is not found, this means it was deleted in the test
except lib_exc.NotFound:
def create_network(cls, network_name=None, client=None, external=None,
shared=None, provider_network_type=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
:param **kwargs: extra parameters to be forwarded to network service
name = (network_name or kwargs.pop('name', None) or
# 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
kwargs.update(project_id=project_id, tenant_id=project_id)
# Use default client
client = cls.client
network = client.create_network(name=name, **kwargs)['network']
network['client'] = client
return network
def delete_network(cls, network, client=None):
client = client or network.get('client') or cls.client
def create_shared_network(cls, network_name=None, **kwargs):
return cls.create_network(name=network_name, shared=True, **kwargs)
def create_subnet(cls, network, gateway='', cidr=None, mask_bits=None,
ip_version=None, client=None, reserve_cidr=True,
"""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 "" for "" 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
: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")
ip_version = gateway_ip.version
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))
kwargs['gateway_ip'] = None
body = client.create_subnet(
except lib_exc.BadRequest as e:
if 'overlaps with another subnet' not in str(e):
message = 'Available CIDR for subnet creation could not be found'
raise ValueError(message)
subnet = body['subnet']
if client is cls.client:
if reserve_cidr:
return subnet
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
if not cls.try_reserve_subnet_cidr(addr, **ipnetwork_kwargs):
raise ValueError('Subnet CIDR already reserved: {0!r}'.format(
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
: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
return True
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)
# 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'))
raise ValueError('Invalid IP version: {!r}'.format(ip_version))
if mask_bits:
subnet_cidrs = cidr.subnet(mask_bits)
subnet_cidrs = iter([cidr])
for subnet_cidr in subnet_cidrs:
if subnet_cidr not in cls.reserved_subnet_cidrs:
yield subnet_cidr
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
if CONF.network.port_profile and 'binding:profile' not in kwargs:
kwargs['binding:profile'] = CONF.network.port_profile
body = cls.client.create_port(network_id=network['id'],
port = body['port']
return port
def update_port(cls, port, **kwargs):
"""Wrapper utility that updates a test port."""
body = cls.client.update_port(port['id'],
return body['port']
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']
return router
def create_router(cls, *args, **kwargs):
return cls._create_router_with_client(cls.client, *args, **kwargs)
def create_admin_router(cls, *args, **kwargs):
return cls._create_router_with_client(cls.os_admin.network_client,
*args, **kwargs)
def create_floatingip(cls, external_network_id=None, port=None,
client=None, **kwargs):
"""Creates a floating IP.
Create a floating IP and schedule it for later deletion.
If a client is passed, then it is used for deleting the IP too.
:param external_network_id: network ID where to create
By default this is 'CONF.network.public_network_id'.
:param port: port to bind floating IP to
This is translated to 'port_id=port['id']'
By default it is None.
:param client: network client to be used for creating and cleaning up
the floating IP.
:param **kwargs: additional creation parameters to be forwarded to
networking server.
client = client or cls.client
external_network_id = (external_network_id or
if port:
port_id = kwargs.setdefault('port_id', port['id'])
if port_id != port['id']:
message = "Port ID specified twice: {!s} != {!s}".format(
port_id, port['id'])
raise ValueError(message)
fip = client.create_floatingip(external_network_id,
# save client to be used later in cls.delete_floatingip
# for final cleanup
fip['client'] = client
return fip
def delete_floatingip(cls, floating_ip, client=None):
"""Delete floating IP
:param client: Client to be used
If client is not given it will use the client used to create
the floating IP, or cls.client if unknown.
client = client or floating_ip.get('client') or cls.client
def create_port_forwarding(cls, fip_id, internal_port_id,
internal_port, external_port,
internal_ip_address=None, protocol="tcp",
"""Creates a port forwarding.
Create a port forwarding and schedule it for later deletion.
If a client is passed, then it is used for deleting the PF too.
:param fip_id: The ID of the floating IP address.
:param internal_port_id: The ID of the Neutron port associated to
the floating IP port forwarding.
:param internal_port: The TCP/UDP/other protocol port number of the
Neutron port fixed IP address associated to the floating ip
port forwarding.
:param external_port: The TCP/UDP/other protocol port number of
the port forwarding floating IP address.
:param internal_ip_address: The fixed IPv4 address of the Neutron
port associated to the floating IP port forwarding.
:param protocol: The IP protocol used in the floating IP port
:param client: network client to be used for creating and cleaning up
the floating IP port forwarding.
client = client or cls.client
pf = client.create_port_forwarding(
fip_id, internal_port_id, internal_port, external_port,
internal_ip_address, protocol)['port_forwarding']
# save ID of floating IP associated with port forwarding for final
# cleanup
pf['floatingip_id'] = fip_id
# save client to be used later in cls.delete_port_forwarding
# for final cleanup
pf['client'] = client
return pf
def delete_port_forwarding(cls, pf, client=None):
"""Delete port forwarding
:param client: Client to be used
If client is not given it will use the client used to create
the port forwarding, or cls.client if unknown.
client = client or pf.get('client') or cls.client
client.delete_port_forwarding(pf['floatingip_id'], pf['id'])
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
def add_extra_routes_atomic(cls, *args, **kwargs):
return cls.client.add_extra_routes_atomic(*args, **kwargs)
def remove_extra_routes_atomic(cls, *args, **kwargs):
return cls.client.remove_extra_routes_atomic(*args, **kwargs)
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']]
def create_qos_policy(cls, name, description=None, shared=False,
project_id=None, is_default=False):
"""Wrapper utility that returns a test QoS policy."""
body = cls.admin_client.create_qos_policy(
name, description, shared, project_id, is_default)
qos_policy = body['policy']
return qos_policy
def create_qos_bandwidth_limit_rule(cls, policy_id, max_kbps,
"""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']
return qos_rule
def create_qos_minimum_bandwidth_rule(cls, policy_id, min_kbps,
"""Wrapper utility that creates and returns a QoS min bw rule."""
body = cls.admin_client.create_minimum_bandwidth_rule(
policy_id, direction, min_kbps)
qos_rule = body['minimum_bandwidth_rule']
return qos_rule
def delete_router(cls, router, client=None):
client = client or cls.client
if 'routes' in router:
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:
router['id'], i['fixed_ips'][0]['subnet_id'])
except lib_exc.NotFound:
def create_address_scope(cls, name, is_admin=False, **kwargs):
if is_admin:
body = cls.admin_client.create_address_scope(name=name, **kwargs)
body = cls.client.create_address_scope(name=name, **kwargs)
return body['address_scope']
def create_subnetpool(cls, name, is_admin=False, client=None, **kwargs):
if client is None:
client = cls.admin_client if is_admin else cls.client
if is_admin:
body = client.create_subnetpool(name, **kwargs)
body = client.create_subnetpool(name, **kwargs)
return body['subnetpool']
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(
# Create a project will create a default security group.
sgs_list = cls.admin_client.list_security_groups(
for security_group in sgs_list:
# Make sure delete_security_group method will use
# the admin client for this group
security_group['client'] = cls.admin_client
return project
def create_security_group(cls, name=None, project=None, client=None,
if project:
client = client or cls.admin_client
project_id = kwargs.setdefault('project_id', project['id'])
tenant_id = kwargs.setdefault('tenant_id', project['id'])
if project_id != project['id'] or tenant_id != project['id']:
raise ValueError('Project ID specified multiple times')
client = client or cls.client
name = name or data_utils.rand_name(cls.__name__)
security_group = client.create_security_group(name=name, **kwargs)[
security_group['client'] = client
return security_group
def delete_security_group(cls, security_group, client=None):
client = client or security_group.get('client') or cls.client
def create_security_group_rule(cls, security_group=None, project=None,
client=None, ip_version=None, **kwargs):
if project:
client = client or cls.admin_client
project_id = kwargs.setdefault('project_id', project['id'])
tenant_id = kwargs.setdefault('tenant_id', project['id'])
if project_id != project['id'] or tenant_id != project['id']:
raise ValueError('Project ID specified multiple times')
if 'security_group_id' not in kwargs:
security_group = (security_group or
if security_group:
client = client or security_group.get('client')
security_group_id = kwargs.setdefault('security_group_id',
if security_group_id != security_group['id']:
raise ValueError('Security group ID specified multiple times.')
ip_version = ip_version or cls._ip_version
default_params = (
for key, value in default_params.items():
kwargs.setdefault(key, value)
client = client or cls.client
return client.create_security_group_rule(**kwargs)[
def get_security_group(cls, name='default', client=None):
client = client or cls.client
security_groups = client.list_security_groups()['security_groups']
for security_group in security_groups:
if security_group['name'] == name:
return security_group
raise ValueError("No such security group named {!r}".format(name))
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
return keypair
def delete_keypair(cls, keypair, client=None):
client = (client or keypair.get('client') or
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
return trunk
def delete_trunk(cls, trunk, client=None, detach_parent_port=True):
"""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
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'],
# 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():
return not parent_port['device_id']
if detach_parent_port and 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.
parent_port['device_id'], parent_port['id'])
def create_conntrack_helper(cls, router_id, helper, protocol, port,
"""Create a conntrack helper
Create a conntrack helper and schedule it for later deletion. If a
client is passed, then it is used for deleteing the CTH too.
:param router_id: The ID of the Neutron router associated to the
conntrack helper.
:param helper: The conntrack helper module alias
:param protocol: The conntrack helper IP protocol used in the conntrack
:param port: The conntrack helper IP protocol port number for the
conntrack helper.
:param client: network client to be used for creating and cleaning up
the conntrack helper.
client = client or cls.client
cth = client.create_conntrack_helper(router_id, helper, protocol,
# save ID of router associated with conntrack helper for final cleanup
cth['router_id'] = router_id
# save client to be used later in cls.delete_conntrack_helper for final
# cleanup
cth['client'] = client
return cth
def delete_conntrack_helper(cls, cth, client=None):
"""Delete conntrack helper
:param client: Client to be used
If client is not given it will use the client used to create the
conntrack helper, or cls.client if unknown.
client = client or cth.get('client') or cls.client
client.delete_conntrack_helper(cth['router_id'], cth['id'])
class BaseAdminNetworkTest(BaseNetworkTest):
credentials = ['primary', 'admin']
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
def create_metering_label(cls, name, description):
"""Wrapper utility that returns a test metering label."""
body = cls.admin_client.create_metering_label(
metering_label = body['metering_label']
return metering_label
def create_metering_label_rule(cls, remote_ip_prefix, direction,
"""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_rule = body['metering_label_rule']
return metering_label_rule
def create_network_segment_range(cls, name, shared,
project_id, network_type,
physical_network, minimum,
"""Wrapper utility that returns a test network segment range."""
network_segment_range_args = {'name': name,
'shared': shared,
'project_id': project_id,
'network_type': network_type,
'physical_network': physical_network,
'minimum': minimum,
'maximum': maximum}
body = cls.admin_client.create_network_segment_range(
network_segment_range = body['network_segment_range']
return network_segment_range
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,
flavor = body['flavor']
return flavor
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']
return service_profile
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']
return log_object
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:
[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:
cidr = subnet['cidr']
allocation_pools = subnet['allocation_pools']
iterators = []
if allocation_pools:
for allocation_pool in allocation_pools:
allocation_pool['start'], allocation_pool['end']))
net = netaddr.IPNetwork(cidr)
def _iterip():
for ip in net:
if ip not in (net.network, net.broadcast):
yield ip
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 create_provider_network(cls, physnet_name, start_segmentation_id,
segmentation_id = start_segmentation_id
for attempts in range(max_attempts):
return cls.create_network(
except lib_exc.Conflict:
segmentation_id += 1
if segmentation_id > 4095:
raise lib_exc.TempestException(
"No free segmentation id was found for provider "
"network creation!")
LOG.exception("Failed to create provider network after "
"%d attempts", max_attempts)
raise lib_exc.TimeoutException
def require_qos_rule_type(rule_type):
def decorator(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):
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):
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])
def plural_name(self):
return '%ss' % self.resource
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)
return method(*args, **kwargs)
def get_bare_url(self, url):
base_url = self.client.base_url
base_url_normalized = utils.normalize_url(base_url)
url_normalized = utils.normalize_url(url)
return url_normalized[len(base_url_normalized):]
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)
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)
def _test_list_sorts_asc(self):
def _test_list_sorts_desc(self):
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))
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)
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_))
return resources
def _test_list_pagination_with_marker(self):
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'])
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_))
# 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_))
self.assertSameOrder(resources, reversed(resources2))
return resources
def _test_list_pagination_with_href_links(self):
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'])
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_))
self.assertSameOrder(expected_resources, reversed(resources))
def _test_list_pagination_page_reverse_asc(self):
def _test_list_pagination_page_reverse_desc(self):
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)
# the last entry is not included in 2nd result when used as a
# marker
@tutils.requires_ext(extension="filter-validation", service="network")
def _test_list_validation_filters(
self, validation_args, filter_is_valid=True):
if not filter_is_valid:
self.assertRaises(lib_exc.BadRequest, self.list_method,
body = self.list_method(**validation_args)
resources = self._extract_resources(body)
for resource in resources:
self.assertIn(resource['name'], self.resource_names)