openstack/neutron-tempest-plugin
Code Issues Proposed changes
5e33282fc2
neutron-tempest-plugin/neutron_tempest_plugin/api/base.py
Takashi Kajinami 938f2c7fc8 Remove transition logic for old tempest option renaming 
This logic was required to adopt to the renaming in tempest[1] which
was done in 12.0.0 release. Now the minimum tempest version required
is much higher so we no longer need this compatibility consideration.

[1] ed6e586b9f8f0ada10af7711f297afa01b2b7754

Change-Id: I5d13f9146b00b2fcd5c2e96071ab914b8a56a089
2024-11-23 02:34:24 +09:00

1682 lines
65 KiB
Python
Raw Blame History

# 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
(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
external_network_id = CONF.network.public_network_id
__is_driver_ovn = None
@classmethod
def _is_driver_ovn(cls):
ovn_agents = cls.os_admin.network_client.list_agents(
binary='ovn-controller')['agents']
return len(ovn_agents) > 0
@property
def is_driver_ovn(self):
if self.__is_driver_ovn is None:
if hasattr(self, 'os_admin'):
self.__is_driver_ovn = self._is_driver_ovn()
return self.__is_driver_ovn
@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.port_forwardings = []
cls.local_ips = []
cls.local_ip_associations = []
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_groups = []
cls.admin_address_groups = []
cls.address_scopes = []
cls.admin_address_scopes = []
cls.subnetpools = []
cls.admin_subnetpools = []
cls.security_groups = []
cls.admin_security_groups = []
cls.sg_rule_templates = []
cls.projects = []
cls.log_objects = []
cls.reserved_subnet_cidrs = set()
cls.keypairs = []
cls.trunks = []
cls.network_segment_ranges = []
cls.conntrack_helpers = []
cls.ndp_proxies = []
@classmethod
def reserve_external_subnet_cidrs(cls):
client = cls.os_admin.network_client
ext_nets = client.list_networks(
**{"router:external": True})['networks']
for ext_net in ext_nets:
ext_subnets = client.list_subnets(
network_id=ext_net['id'])['subnets']
for ext_subnet in ext_subnets:
cls.reserve_subnet_cidr(ext_subnet['cidr'])
@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 ndp proxy
for ndp_proxy in cls.ndp_proxies:
cls._try_delete_resource(cls.delete_ndp_proxy, ndp_proxy)
# 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 Local IP Associations
for association in cls.local_ip_associations:
cls._try_delete_resource(cls.delete_local_ip_association,
association)
# Clean up Local IPs
for local_ip in cls.local_ips:
cls._try_delete_resource(cls.delete_local_ip,
local_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:
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 security_group in cls.security_groups:
cls._try_delete_resource(cls.delete_security_group,
security_group)
# Clean up admin security groups
for security_group in cls.admin_security_groups:
cls._try_delete_resource(cls.delete_security_group,
security_group,
client=cls.admin_client)
# Clean up security group rule templates
for sg_rule_template in cls.sg_rule_templates:
cls._try_delete_resource(
cls.admin_client.delete_default_security_group_rule,
sg_rule_template['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)
# Clean up network_segment_ranges
for network_segment_range in cls.network_segment_ranges:
cls._try_delete_resource(
cls.admin_client.delete_network_segment_range,
network_segment_range['id'])
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.project_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_subnet(cls, network, gateway='', cidr=None, mask_bits=None,
ip_version=None, client=None, reserve_cidr=True,
allocation_pool_size=None, **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 allocation_pool_size: if the CIDR is not defined, this method
will assign one in ``get_subnet_cidrs``. Once done, the allocation pool
will be defined reserving the number of IP addresses requested,
starting from the end of the assigned CIDR.
:param **kwargs: optional parameters to be forwarded to wrapped method
[1] http://netaddr.readthedocs.io/en/latest/tutorial_01.html#supernets-and-subnets # noqa
"""
def allocation_pool(cidr, pool_size):
start = str(netaddr.IPAddress(cidr.last) - pool_size)
end = str(netaddr.IPAddress(cidr.last) - 1)
return {'start': start, 'end': end}
# 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))
else:
kwargs['gateway_ip'] = None
if allocation_pool_size:
kwargs['allocation_pools'] = [
allocation_pool(subnet_cidr, allocation_pool_size)]
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's 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: {0!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 CONF.network.project_network_mask_bits
cidr = netaddr.IPNetwork(CONF.network.project_network_cidr)
elif ip_version == const.IP_VERSION_6:
mask_bits = CONF.network.project_network_v6_mask_bits
cidr = netaddr.IPNetwork(CONF.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
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'],
**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=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
cls.external_network_id)
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,
**kwargs)['floatingip']
# save client to be used later in cls.delete_floatingip
# for final cleanup
fip['client'] = client
cls.floating_ips.append(fip)
return fip
@classmethod
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
client.delete_floatingip(floating_ip['id'])
@classmethod
def create_port_forwarding(cls, fip_id, internal_port_id,
internal_port, external_port,
internal_ip_address=None, protocol="tcp",
client=None):
"""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
forwarding.
: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
cls.port_forwardings.append(pf)
return pf
@classmethod
def update_port_forwarding(cls, fip_id, pf_id, client=None, **kwargs):
"""Wrapper utility for update_port_forwarding."""
client = client or cls.client
return client.update_port_forwarding(fip_id, pf_id, **kwargs)
@classmethod
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_local_ip(cls, network_id=None,
client=None, **kwargs):
"""Creates a Local IP.
Create a Local IP and schedule it for later deletion.
If a client is passed, then it is used for deleting the IP too.
:param network_id: network ID where to create
By default this is 'CONF.network.public_network_id'.
:param client: network client to be used for creating and cleaning up
the Local IP.
:param **kwargs: additional creation parameters to be forwarded to
networking server.
"""
client = client or cls.client
network_id = (network_id or
cls.external_network_id)
local_ip = client.create_local_ip(network_id,
**kwargs)['local_ip']
# save client to be used later in cls.delete_local_ip
# for final cleanup
local_ip['client'] = client
cls.local_ips.append(local_ip)
return local_ip
@classmethod
def delete_local_ip(cls, local_ip, client=None):
"""Delete Local IP
:param client: Client to be used
If client is not given it will use the client used to create
the Local IP, or cls.client if unknown.
"""
client = client or local_ip.get('client') or cls.client
client.delete_local_ip(local_ip['id'])
@classmethod
def create_local_ip_association(cls, local_ip_id, fixed_port_id,
fixed_ip_address=None, client=None):
"""Creates a Local IP association.
Create a Local IP Association and schedule it for later deletion.
If a client is passed, then it is used for deleting the association
too.
:param local_ip_id: The ID of the Local IP.
:param fixed_port_id: The ID of the Neutron port
to be associated with the Local IP
:param fixed_ip_address: The fixed IPv4 address of the Neutron
port to be associated with the Local IP
:param client: network client to be used for creating and cleaning up
the Local IP Association.
"""
client = client or cls.client
association = client.create_local_ip_association(
local_ip_id, fixed_port_id,
fixed_ip_address)['port_association']
# save ID of Local IP for final cleanup
association['local_ip_id'] = local_ip_id
# save client to be used later in
# cls.delete_local_ip_association for final cleanup
association['client'] = client
cls.local_ip_associations.append(association)
return association
@classmethod
def delete_local_ip_association(cls, association, client=None):
"""Delete Local IP Association
:param client: Client to be used
If client is not given it will use the client used to create
the local IP association, or cls.client if unknown.
"""
client = client or association.get('client') or cls.client
client.delete_local_ip_association(association['local_ip_id'],
association['fixed_port_id'])
@classmethod
def create_router_interface(cls, router_id, subnet_id, client=None):
"""Wrapper utility that returns a router interface."""
client = client or cls.client
interface = client.add_router_interface_with_subnet_id(
router_id, subnet_id)
return interface
@classmethod
def add_extra_routes_atomic(cls, *args, **kwargs):
return cls.client.add_extra_routes_atomic(*args, **kwargs)
@classmethod
def remove_extra_routes_atomic(cls, *args, **kwargs):
return cls.client.remove_extra_routes_atomic(*args, **kwargs)
@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,
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']
cls.qos_policies.append(qos_policy)
return qos_policy
@classmethod
def create_qos_dscp_marking_rule(cls, policy_id, dscp_mark):
"""Wrapper utility that creates and returns a QoS dscp rule."""
body = cls.admin_client.create_dscp_marking_rule(
policy_id, dscp_mark)
qos_rule = body['dscp_marking_rule']
cls.qos_rules.append(qos_rule)
return qos_rule
@classmethod
def delete_router(cls, router, client=None):
client = client or cls.client
if 'routes' in router:
client.remove_router_extra_routes(router['id'])
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, 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)
cls.admin_subnetpools.append(body['subnetpool'])
else:
body = client.create_subnetpool(name, **kwargs)
cls.subnetpools.append(body['subnetpool'])
return body['subnetpool']
@classmethod
def create_address_group(cls, name, is_admin=False, **kwargs):
if is_admin:
body = cls.admin_client.create_address_group(name=name, **kwargs)
cls.admin_address_groups.append(body['address_group'])
else:
body = cls.client.create_address_group(name=name, **kwargs)
cls.address_groups.append(body['address_group'])
return body['address_group']
@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)
# Create a project will create a default security group.
sgs_list = cls.admin_client.list_security_groups(
tenant_id=project['id'])['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
cls.security_groups.append(security_group)
return project
@classmethod
def create_security_group(cls, name=None, project=None, client=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')
else:
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']
security_group['client'] = client
cls.security_groups.append(security_group)
return security_group
@classmethod
def delete_security_group(cls, security_group, client=None):
client = client or security_group.get('client') or cls.client
client.delete_security_group(security_group['id'])
@classmethod
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))
@classmethod
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
cls.get_security_group(client=client))
if security_group:
client = client or security_group.get('client')
security_group_id = kwargs.setdefault('security_group_id',
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 = (
constants.DEFAULT_SECURITY_GROUP_RULE_PARAMS[ip_version])
if (('remote_address_group_id' in kwargs or
'remote_group_id' in kwargs) and
'remote_ip_prefix' in default_params):
default_params.pop('remote_ip_prefix')
for key, value in default_params.items():
kwargs.setdefault(key, value)
client = client or cls.client
return client.create_security_group_rule(**kwargs)[
'security_group_rule']
@classmethod
def create_default_security_group_rule(cls, **kwargs):
body = cls.admin_client.create_default_security_group_rule(**kwargs)
default_sg_rule = body['default_security_group_rule']
cls.sg_rule_templates.append(default_sg_rule)
return default_sg_rule
@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, 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
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 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.
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'])
@classmethod
def create_conntrack_helper(cls, router_id, helper, protocol, port,
client=None):
"""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
helper.
: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,
port)['conntrack_helper']
# 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
cls.conntrack_helpers.append(cth)
return cth
@classmethod
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'])
@classmethod
def create_ndp_proxy(cls, router_id, port_id, client=None, **kwargs):
"""Creates a ndp proxy.
Create a ndp proxy and schedule it for later deletion.
If a client is passed, then it is used for deleting the NDP proxy too.
:param router_id: router ID where to create the ndp proxy.
:param port_id: port ID which the ndp proxy associate with
:param client: network client to be used for creating and cleaning up
the ndp proxy.
:param **kwargs: additional creation parameters to be forwarded to
networking server.
"""
client = client or cls.client
data = {'router_id': router_id, 'port_id': port_id}
if kwargs:
data.update(kwargs)
ndp_proxy = client.create_ndp_proxy(**data)['ndp_proxy']
# save client to be used later in cls.delete_ndp_proxy
# for final cleanup
ndp_proxy['client'] = client
cls.ndp_proxies.append(ndp_proxy)
return ndp_proxy
@classmethod
def delete_ndp_proxy(cls, ndp_proxy, client=None):
"""Delete ndp proxy
:param client: Client to be used
If client is not given it will use the client used to create
the ndp proxy, or cls.client if unknown.
"""
client = client or ndp_proxy.get('client') or cls.client
client.delete_ndp_proxy(ndp_proxy['id'])
@classmethod
def get_loaded_network_extensions(cls):
"""Return the network service loaded extensions
:return: list of strings with the alias of the network service loaded
extensions.
"""
body = cls.client.list_extensions()
return [net_ext['alias'] for net_ext in body['extensions']]
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_network_segment_range(cls, name, shared,
project_id, network_type,
physical_network, minimum,
maximum):
"""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_args)
network_segment_range = body['network_segment_range']
cls.network_segment_ranges.append(network_segment_range)
return network_segment_range
@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,
'resource_type': resource_type,
'resource_id': resource_id,
'target_id': target_id,
'event': event,
'enabled': enabled}
if description:
log_args['description'] = description
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)
@classmethod
def create_provider_network(cls, physnet_name, start_segmentation_id,
max_attempts=30, external=False):
segmentation_id = start_segmentation_id
for attempts in range(max_attempts):
try:
return cls.create_network(
name=data_utils.rand_name('test_net'),
shared=not external,
external=external,
provider_network_type='vlan',
provider_physical_network=physnet_name,
provider_segmentation_id=segmentation_id)
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!")
time.sleep(CONF.network.build_interval)
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):
@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
base_url_normalized = utils.normalize_url(base_url)
url_normalized = utils.normalize_url(url)
self.assertTrue(url_normalized.startswith(base_url_normalized))
return url_normalized[len(base_url_normalized):]
@classmethod
def _extract_resources(cls, body):
return body[cls.plural_name]
@classmethod
def _test_resources(cls, resources):
return [res for res in resources if res["name"] in cls.resource_names]
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]
# sort without taking into account whether the network is named with
# a capital letter or not
expected = sorted(retrieved_names, key=lambda v: v.upper())
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)
total_resources = self._extract_resources(body)
expected_resources = self._test_resources(total_resources)
self.assertNotEmpty(expected_resources)
resources = lister(
len(total_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)
# Empty resource list can be returned when any concurrent
# tests delete them
self.assertGreaterEqual(1, len(resources_))
resources.extend(resources_)
return self._test_resources(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)
# Empty resource list can be returned when any concurrent
# tests delete them
self.assertGreaterEqual(1, len(resources_))
resources.extend(self._test_resources(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)
# Empty resource list can be returned when any concurrent
# tests delete them
self.assertGreaterEqual(1, len(resources_))
resources2.extend(self._test_resources(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)
total_resources = self._extract_resources(body)
expected_resources = self._test_resources(total_resources)
page_size = 2
pagination_args['limit'] = page_size
prev_links = {}
resources = []
num_resources = len(total_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._test_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))
@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,
**validation_args)
else:
body = self.list_method(**validation_args)
resources = self._extract_resources(body)
for resource in resources:
self.assertIn(resource['name'], self.resource_names)
View Git Blame Copy Permalink