openstack/neutron-tempest-plugin
Code Issues Proposed changes
aa22c9e1bb
neutron-tempest-plugin/neutron_tempest_plugin/api/base.py
Slawek Kaplonski aa22c9e1bb New basic API tests for the default SG rules templates CRUDs 
This patch adds some basic API tests for the new API for default SG
rules templates. Those new tests are checking if by default SG rules are
set in the same way as legacy rules which were there since "forever".
Second test checks basic lifecycle of the SG rule template.

Depends-On: https://review.opendev.org/c/openstack/neutron/+/883246/

Related-Bug: #1983053
Change-Id: I458f54ff6b73e277fe9506e90fa6af44d9c51101
2023-10-06 06:00:57 +00:00

1633 lines
63 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
@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,
**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))
else:
kwargs['gateway_ip'] = None
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: {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 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
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):
"""Wrapper utility that returns a router interface."""
interface = cls.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'])
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):
segmentation_id = start_segmentation_id
for attempts in range(max_attempts):
try:
return cls.create_network(
name=data_utils.rand_name('test_net'),
shared=True,
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]
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))
@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