neutron_tempest_plugin/bgpvpn/scenario/manager.py

# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
# 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 os
import subprocess

import netaddr

from oslo_log import log
from oslo_utils import netutils

from tempest.common import compute
from tempest.common import utils
from tempest.common.utils.linux import remote_client
from tempest.common.utils import net_utils
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import exceptions as lib_exc
import tempest.test


CONF = config.CONF

LOG = log.getLogger(__name__)

NET_A = 'A'
NET_A_BIS = 'A-Bis'
NET_B = 'B'
NET_C = 'C'

if "SUBNETPOOL_PREFIX_V4" in os.environ:
    subnet_base = netaddr.IPNetwork(os.environ['SUBNETPOOL_PREFIX_V4'])
    if subnet_base.prefixlen > 21:
        raise Exception("if SUBNETPOOL_PREFIX_V4 is set, it needs to offer "
                        "space for at least 8 /24 subnets")
else:
    subnet_base = netaddr.IPNetwork("10.100.0.0/16")


def assign_24(idx):
    # how many addresses in a /24:
    range_size = 2 ** (32 - 24)
    return netaddr.cidr_merge(
        subnet_base[range_size * idx:range_size * (idx + 1)])[0]


S1A = assign_24(1)
S2A = assign_24(2)
S1B = assign_24(4)
S2B = assign_24(6)
S1C = assign_24(6)
NET_A_S1 = str(S1A)
NET_A_S2 = str(S2A)
NET_B_S1 = str(S1B)
NET_B_S2 = str(S2B)
NET_C_S1 = str(S1C)
IP_A_S1_1 = str(S1A[10])
IP_B_S1_1 = str(S1B[20])
IP_C_S1_1 = str(S1C[30])
IP_A_S1_2 = str(S1A[30])
IP_B_S1_2 = str(S1B[40])
IP_A_S1_3 = str(S1A[50])
IP_B_S1_3 = str(S1B[60])
IP_A_S2_1 = str(S2A[50])
IP_B_S2_1 = str(S2B[60])
IP_A_BIS_S1_1 = IP_A_S1_1
IP_A_BIS_S1_2 = IP_A_S1_2
IP_A_BIS_S1_3 = IP_A_S1_3
IP_A_BIS_S2_1 = IP_A_S2_1


class ScenarioTest(tempest.test.BaseTestCase):
    """Base class for scenario tests. Uses tempest own clients. """

    credentials = ['primary']

    @classmethod
    def setup_clients(cls):
        super(ScenarioTest, cls).setup_clients()
        # Clients (in alphabetical order)
        cls.keypairs_client = cls.os_primary.keypairs_client
        cls.servers_client = cls.os_primary.servers_client
        # Neutron network client
        cls.networks_client = cls.os_primary.networks_client
        cls.ports_client = cls.os_primary.ports_client
        cls.routers_client = cls.os_primary.routers_client
        cls.subnets_client = cls.os_primary.subnets_client
        cls.floating_ips_client = cls.os_primary.floating_ips_client
        cls.security_groups_client = cls.os_primary.security_groups_client
        cls.security_group_rules_client = (
            cls.os_primary.security_group_rules_client)

    # ## Test functions library
    #
    # The create_[resource] functions only return body and discard the
    # resp part which is not used in scenario tests

    def _create_port(self, network_id, client=None, namestart='port-quotatest',
                     **kwargs):
        if not client:
            client = self.ports_client
        name = data_utils.rand_name(namestart)
        result = client.create_port(
            name=name,
            network_id=network_id,
            **kwargs)
        self.assertIsNotNone(result, 'Unable to allocate port')
        port = result['port']
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        client.delete_port, port['id'])
        return port

    def create_keypair(self, client=None):
        if not client:
            client = self.keypairs_client
        name = data_utils.rand_name(self.__class__.__name__)
        # We don't need to create a keypair by pubkey in scenario
        body = client.create_keypair(name=name)
        self.addCleanup(client.delete_keypair, name)
        return body['keypair']

    def create_server(self, name=None, image_id=None, flavor=None,
                      validatable=False, wait_until='ACTIVE',
                      clients=None, **kwargs):
        """Wrapper utility that returns a test server.

        This wrapper utility calls the common create test server and
        returns a test server. The purpose of this wrapper is to minimize
        the impact on the code of the tests already using this
        function.
        """

        # NOTE(jlanoux): As a first step, ssh checks in the scenario
        # tests need to be run regardless of the run_validation and
        # validatable parameters and thus until the ssh validation job
        # becomes voting in CI. The test resources management and IP
        # association are taken care of in the scenario tests.
        # Therefore, the validatable parameter is set to false in all
        # those tests. In this way create_server just return a standard
        # server and the scenario tests always perform ssh checks.

        # Needed for the cross_tenant_traffic test:
        if clients is None:
            clients = self.os_primary

        if name is None:
            name = data_utils.rand_name(self.__class__.__name__ + "-server")

        vnic_type = CONF.network.port_vnic_type

        # If vnic_type is configured create port for
        # every network
        if vnic_type:
            ports = []

            create_port_body = {'binding:vnic_type': vnic_type,
                                'namestart': 'port-smoke'}
            if kwargs:
                # Convert security group names to security group ids
                # to pass to create_port
                if 'security_groups' in kwargs:
                    security_groups = \
                        clients.security_groups_client.list_security_groups(
                        ).get('security_groups')
                    sec_dict = dict([(s['name'], s['id'])
                                    for s in security_groups])

                    sec_groups_names = [s['name'] for s in kwargs.pop(
                        'security_groups')]
                    security_groups_ids = [sec_dict[s]
                                           for s in sec_groups_names]

                    if security_groups_ids:
                        create_port_body[
                            'security_groups'] = security_groups_ids
                networks = kwargs.pop('networks', [])
            else:
                networks = []

            # If there are no networks passed to us we look up
            # for the project's private networks and create a port.
            # The same behaviour as we would expect when passing
            # the call to the clients with no networks
            if not networks:
                networks = clients.networks_client.list_networks(
                    **{'router:external': False, 'fields': 'id'})['networks']

            # It's net['uuid'] if networks come from kwargs
            # and net['id'] if they come from
            # clients.networks_client.list_networks
            for net in networks:
                net_id = net.get('uuid', net.get('id'))
                if 'port' not in net:
                    port = self._create_port(network_id=net_id,
                                             client=clients.ports_client,
                                             **create_port_body)
                    ports.append({'port': port['id']})
                else:
                    ports.append({'port': net['port']})
            if ports:
                kwargs['networks'] = ports
            self.ports = ports

        tenant_network = self.get_tenant_network()

        body, servers = compute.create_test_server(
            clients,
            tenant_network=tenant_network,
            wait_until=wait_until,
            name=name, flavor=flavor,
            image_id=image_id, **kwargs)

        self.addCleanup(waiters.wait_for_server_termination,
                        clients.servers_client, body['id'])
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        clients.servers_client.delete_server, body['id'])
        server = clients.servers_client.show_server(body['id'])['server']
        return server

    def get_remote_client(self, ip_address, username=None, private_key=None):
        """Get a SSH client to a remote server

        @param ip_address the server floating or fixed IP address to use
                          for ssh validation
        @param username name of the Linux account on the remote server
        @param private_key the SSH private key to use
        @return a RemoteClient object
        """

        if username is None:
            username = CONF.validation.image_ssh_user
        # Set this with 'keypair' or others to log in with keypair or
        # username/password.
        if CONF.validation.auth_method == 'keypair':
            password = None
            if private_key is None:
                private_key = self.keypair['private_key']
        else:
            password = CONF.validation.image_ssh_password
            private_key = None
        linux_client = remote_client.RemoteClient(ip_address, username,
                                                  pkey=private_key,
                                                  password=password)
        try:
            linux_client.validate_authentication()
        except Exception as e:
            message = ('Initializing SSH connection to %(ip)s failed. '
                       'Error: %(error)s' % {'ip': ip_address,
                                             'error': e})
            caller = test_utils.find_test_caller()
            if caller:
                message = '(%s) %s' % (caller, message)
            LOG.exception(message)
            self._log_console_output()
            raise

        return linux_client

    def _log_console_output(self, servers=None):
        if not CONF.compute_feature_enabled.console_output:
            LOG.debug('Console output not supported, cannot log')
            return
        if not servers:
            servers = self.servers_client.list_servers()
            servers = servers['servers']
        for server in servers:
            try:
                console_output = self.servers_client.get_console_output(
                    server['id'])['output']
                LOG.debug('Console output for %s\nbody=\n%s',
                          server['id'], console_output)
            except lib_exc.NotFound:
                LOG.debug("Server %s disappeared(deleted) while looking "
                          "for the console log", server['id'])

    def _log_net_info(self, exc):
        # network debug is called as part of ssh init
        if not isinstance(exc, lib_exc.SSHTimeout):
            LOG.debug('Network information on a devstack host')

    def ping_ip_address(self, ip_address, should_succeed=True,
                        ping_timeout=None, mtu=None):
        timeout = ping_timeout or CONF.validation.ping_timeout
        cmd = ['ping', '-c1', '-w1']

        if mtu:
            cmd += [
                # don't fragment
                '-M', 'do',
                # ping receives just the size of ICMP payload
                '-s', str(net_utils.get_ping_payload_size(mtu, 4))
            ]
        cmd.append(ip_address)

        def ping():
            proc = subprocess.Popen(cmd,
                                    stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE)
            proc.communicate()

            return (proc.returncode == 0) == should_succeed

        caller = test_utils.find_test_caller()
        LOG.debug('%(caller)s begins to ping %(ip)s in %(timeout)s sec and the'
                  ' expected result is %(should_succeed)s', {
                      'caller': caller, 'ip': ip_address, 'timeout': timeout,
                      'should_succeed':
                      'reachable' if should_succeed else 'unreachable'
                  })
        result = test_utils.call_until_true(ping, timeout, 1)
        LOG.debug('%(caller)s finishes ping %(ip)s in %(timeout)s sec and the '
                  'ping result is %(result)s', {
                      'caller': caller, 'ip': ip_address, 'timeout': timeout,
                      'result': 'expected' if result else 'unexpected'
                  })
        return result

    def check_vm_connectivity(self, ip_address,
                              username=None,
                              private_key=None,
                              should_connect=True,
                              mtu=None):
        """Check server connectivity

        :param ip_address: server to test against
        :param username: server's ssh username
        :param private_key: server's ssh private key to be used
        :param should_connect: True/False indicates positive/negative test
            positive - attempt ping and ssh
            negative - attempt ping and fail if succeed
        :param mtu: network MTU to use for connectivity validation

        :raises: AssertError if the result of the connectivity check does
            not match the value of the should_connect param
        """
        if should_connect:
            msg = "Timed out waiting for %s to become reachable" % ip_address
        else:
            msg = "ip address %s is reachable" % ip_address
        self.assertTrue(self.ping_ip_address(ip_address,
                                             should_succeed=should_connect,
                                             mtu=mtu),
                        msg=msg)
        if should_connect:
            # no need to check ssh for negative connectivity
            self.get_remote_client(ip_address, username, private_key)

    def check_public_network_connectivity(self, ip_address, username,
                                          private_key, should_connect=True,
                                          msg=None, servers=None, mtu=None):
        # The target login is assumed to have been configured for
        # key-based authentication by cloud-init.
        LOG.debug('checking network connections to IP %s with user: %s',
                  ip_address, username)
        try:
            self.check_vm_connectivity(ip_address,
                                       username,
                                       private_key,
                                       should_connect=should_connect,
                                       mtu=mtu)
        except Exception:
            ex_msg = 'Public network connectivity check failed'
            if msg:
                ex_msg += ": " + msg
            LOG.exception(ex_msg)
            self._log_console_output(servers)
            raise


class NetworkScenarioTest(ScenarioTest):
    """Base class for network scenario tests.

    This class provide helpers for network scenario tests, using the neutron
    API. Helpers from ancestor which use the nova network API are overridden
    with the neutron API.

    This Class also enforces using Neutron instead of novanetwork.
    Subclassed tests will be skipped if Neutron is not enabled

    """

    credentials = ['primary', 'admin']

    @classmethod
    def skip_checks(cls):
        super(NetworkScenarioTest, cls).skip_checks()
        if not CONF.service_available.neutron:
            raise cls.skipException('Neutron not available')
        if not utils.is_extension_enabled('bgpvpn', 'network'):
            msg = "Bgpvpn extension not enabled."
            raise cls.skipException(msg)

    def _create_network(self, networks_client=None,
                        tenant_id=None,
                        namestart='network-smoke-',
                        port_security_enabled=True):
        if not networks_client:
            networks_client = self.networks_client
        if not tenant_id:
            tenant_id = networks_client.tenant_id
        name = data_utils.rand_name(namestart)
        network_kwargs = dict(name=name, tenant_id=tenant_id)
        # Neutron disables port security by default so we have to check the
        # config before trying to create the network with port_security_enabled
        if CONF.network_feature_enabled.port_security:
            network_kwargs['port_security_enabled'] = port_security_enabled
        result = networks_client.create_network(**network_kwargs)
        network = result['network']

        self.assertEqual(network['name'], name)
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        networks_client.delete_network,
                        network['id'])
        return network

    def _create_subnet(self, network, subnets_client=None,
                       routers_client=None, namestart='subnet-smoke',
                       **kwargs):
        """Create a subnet for the given network

        within the cidr block configured for tenant networks.
        """
        if not subnets_client:
            subnets_client = self.subnets_client
        if not routers_client:
            routers_client = self.routers_client

        def cidr_in_use(cidr, tenant_id):
            """Check cidr existence

            :returns: True if subnet with cidr already exist in tenant
                  False else
            """
            cidr_in_use = self.os_admin.subnets_client.list_subnets(
                tenant_id=tenant_id, cidr=cidr)['subnets']
            return len(cidr_in_use) != 0

        ip_version = kwargs.pop('ip_version', 4)

        if ip_version == 6:
            tenant_cidr = netaddr.IPNetwork(
                CONF.network.project_network_v6_cidr)
            num_bits = CONF.network.project_network_v6_mask_bits
        else:
            tenant_cidr = netaddr.IPNetwork(CONF.network.project_network_cidr)
            num_bits = CONF.network.project_network_mask_bits

        result = None
        str_cidr = None
        # Repeatedly attempt subnet creation with sequential cidr
        # blocks until an unallocated block is found.
        for subnet_cidr in tenant_cidr.subnet(num_bits):
            str_cidr = str(subnet_cidr)
            if cidr_in_use(str_cidr, tenant_id=network['tenant_id']):
                continue

            subnet = dict(
                name=data_utils.rand_name(namestart),
                network_id=network['id'],
                tenant_id=network['tenant_id'],
                cidr=str_cidr,
                ip_version=ip_version,
                **kwargs
            )
            try:
                result = subnets_client.create_subnet(**subnet)
                break
            except lib_exc.Conflict as e:
                is_overlapping_cidr = 'overlaps with another subnet' in str(e)
                if not is_overlapping_cidr:
                    raise
        self.assertIsNotNone(result, 'Unable to allocate tenant network')

        subnet = result['subnet']
        self.assertEqual(subnet['cidr'], str_cidr)

        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        subnets_client.delete_subnet, subnet['id'])

        return subnet

    def _get_server_port_id_and_ip4(self, server, ip_addr=None):
        ports = self.os_admin.ports_client.list_ports(
            device_id=server['id'], fixed_ip=ip_addr)['ports']
        # A port can have more than one IP address in some cases.
        # If the network is dual-stack (IPv4 + IPv6), this port is associated
        # with 2 subnets
        p_status = ['ACTIVE']
        # NOTE(vsaienko) With Ironic, instances live on separate hardware
        # servers. Neutron does not bind ports for Ironic instances, as a
        # result the port remains in the DOWN state.
        # TODO(vsaienko) remove once bug: #1599836 is resolved.
        if getattr(CONF.service_available, 'ironic', False):
            p_status.append('DOWN')
        port_map = [(p["id"], fxip["ip_address"])
                    for p in ports
                    for fxip in p["fixed_ips"]
                    if netutils.is_valid_ipv4(fxip["ip_address"]) and
                    p['status'] in p_status]
        inactive = [p for p in ports if p['status'] != 'ACTIVE']
        if inactive:
            LOG.warning("Instance has ports that are not ACTIVE: %s", inactive)

        self.assertNotEqual(0, len(port_map),
                            "No IPv4 addresses found in: %s" % ports)
        self.assertEqual(len(port_map), 1,
                         "Found multiple IPv4 addresses: %s. "
                         "Unable to determine which port to target."
                         % port_map)
        return port_map[0]

    def _get_network_by_name(self, network_name):
        net = self.os_admin.networks_client.list_networks(
            name=network_name)['networks']
        self.assertNotEqual(len(net), 0,
                            "Unable to get network by name: %s" % network_name)
        return net[0]

    def create_floating_ip(self, thing, external_network_id=None,
                           port_id=None, client=None):
        """Create a floating IP and associates to a resource/port on Neutron"""
        if not external_network_id:
            external_network_id = CONF.network.public_network_id
        if not client:
            client = self.floating_ips_client
        if not port_id:
            port_id, ip4 = self._get_server_port_id_and_ip4(thing)
        else:
            ip4 = None
        result = client.create_floatingip(
            floating_network_id=external_network_id,
            port_id=port_id,
            tenant_id=thing['tenant_id'],
            fixed_ip_address=ip4
        )
        floating_ip = result['floatingip']
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        client.delete_floatingip,
                        floating_ip['id'])
        return floating_ip

    def _associate_floating_ip(self, floating_ip, server):
        port_id, _ = self._get_server_port_id_and_ip4(server)
        kwargs = dict(port_id=port_id)
        floating_ip = self.floating_ips_client.update_floatingip(
            floating_ip['id'], **kwargs)['floatingip']
        self.assertEqual(port_id, floating_ip['port_id'])
        return floating_ip

    def _disassociate_floating_ip(self, floating_ip):
        """:param floating_ip: floating_ips_client.create_floatingip"""
        kwargs = dict(port_id=None)
        floating_ip = self.floating_ips_client.update_floatingip(
            floating_ip['id'], **kwargs)['floatingip']
        self.assertIsNone(floating_ip['port_id'])
        return floating_ip

    def check_floating_ip_status(self, floating_ip, status):
        """Verifies floatingip reaches the given status

        :param dict floating_ip: floating IP dict to check status
        :param status: target status
        :raises: AssertionError if status doesn't match
        """
        floatingip_id = floating_ip['id']

        def refresh():
            result = (self.floating_ips_client.
                      show_floatingip(floatingip_id)['floatingip'])
            return status == result['status']

        test_utils.call_until_true(refresh,
                                   CONF.network.build_timeout,
                                   CONF.network.build_interval)
        floating_ip = self.floating_ips_client.show_floatingip(
            floatingip_id)['floatingip']
        self.assertEqual(status, floating_ip['status'],
                         message="FloatingIP: {fp} is at status: {cst}. "
                                 "failed  to reach status: {st}"
                         .format(fp=floating_ip, cst=floating_ip['status'],
                                 st=status))
        LOG.info("FloatingIP: {fp} is at status: {st}"
                 .format(fp=floating_ip, st=status))

    def _check_tenant_network_connectivity(self, server,
                                           username,
                                           private_key,
                                           should_connect=True,
                                           servers_for_debug=None):
        if not CONF.network.project_networks_reachable:
            msg = 'Tenant networks not configured to be reachable.'
            LOG.info(msg)
            return
        # The target login is assumed to have been configured for
        # key-based authentication by cloud-init.
        try:
            for net_name, ip_addresses in server['addresses'].items():
                for ip_address in ip_addresses:
                    self.check_vm_connectivity(ip_address['addr'],
                                               username,
                                               private_key,
                                               should_connect=should_connect)
        except Exception as e:
            LOG.exception('Tenant network connectivity check failed')
            self._log_console_output(servers_for_debug)
            self._log_net_info(e)
            raise

    def _check_remote_connectivity(self, source, dest, should_succeed=True,
                                   nic=None):
        """check ping server via source ssh connection

        :param source: RemoteClient: an ssh connection from which to ping
        :param dest: and IP to ping against
        :param should_succeed: boolean should ping succeed or not
        :param nic: specific network interface to ping from
        :returns: boolean -- should_succeed == ping
        :returns: ping is false if ping failed
        """
        def ping_remote():
            try:
                source.ping_host(dest, nic=nic)
            except lib_exc.SSHExecCommandFailed:
                LOG.warning('Failed to ping IP: %s via a ssh connection '
                            'from: %s.', dest, source.ssh_client.host)
                return not should_succeed
            return should_succeed

        return test_utils.call_until_true(ping_remote,
                                          CONF.validation.ping_timeout,
                                          1)

    def _create_security_group(self, security_group_rules_client=None,
                               tenant_id=None,
                               namestart='secgroup-smoke',
                               security_groups_client=None):
        if security_group_rules_client is None:
            security_group_rules_client = self.security_group_rules_client
        if security_groups_client is None:
            security_groups_client = self.security_groups_client
        if tenant_id is None:
            tenant_id = security_groups_client.tenant_id
        secgroup = self._create_empty_security_group(
            namestart=namestart, client=security_groups_client,
            tenant_id=tenant_id)

        # Add rules to the security group
        rules = self._create_loginable_secgroup_rule(
            security_group_rules_client=security_group_rules_client,
            secgroup=secgroup,
            security_groups_client=security_groups_client)
        for rule in rules:
            self.assertEqual(tenant_id, rule['tenant_id'])
            self.assertEqual(secgroup['id'], rule['security_group_id'])
        return secgroup

    def _create_empty_security_group(self, client=None, tenant_id=None,
                                     namestart='secgroup-smoke'):
        """Create a security group without rules.

        Default rules will be created:
         - IPv4 egress to any
         - IPv6 egress to any

        :param tenant_id: secgroup will be created in this tenant
        :returns: the created security group
        """
        if client is None:
            client = self.security_groups_client
        if not tenant_id:
            tenant_id = client.tenant_id
        sg_name = data_utils.rand_name(namestart)
        sg_desc = sg_name + " description"
        sg_dict = dict(name=sg_name,
                       description=sg_desc)
        sg_dict['tenant_id'] = tenant_id
        result = client.create_security_group(**sg_dict)

        secgroup = result['security_group']
        self.assertEqual(secgroup['name'], sg_name)
        self.assertEqual(tenant_id, secgroup['tenant_id'])
        self.assertEqual(secgroup['description'], sg_desc)

        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        client.delete_security_group, secgroup['id'])
        return secgroup

    def _default_security_group(self, client=None, tenant_id=None):
        """Get default secgroup for given tenant_id.

        :returns: default secgroup for given tenant
        """
        if client is None:
            client = self.security_groups_client
        if not tenant_id:
            tenant_id = client.tenant_id
        sgs = [
            sg for sg in list(client.list_security_groups().values())[0]
            if sg['tenant_id'] == tenant_id and sg['name'] == 'default'
        ]
        msg = "No default security group for tenant %s." % (tenant_id)
        self.assertGreater(len(sgs), 0, msg)
        return sgs[0]

    def _create_security_group_rule(self, secgroup=None,
                                    sec_group_rules_client=None,
                                    tenant_id=None,
                                    security_groups_client=None, **kwargs):
        """Create a rule from a dictionary of rule parameters.

        Create a rule in a secgroup. if secgroup not defined will search for
        default secgroup in tenant_id.

        :param secgroup: the security group.
        :param tenant_id: if secgroup not passed -- the tenant in which to
            search for default secgroup
        :param kwargs: a dictionary containing rule parameters:
            for example, to allow incoming ssh:
            rule = {
                    direction: 'ingress'
                    protocol:'tcp',
                    port_range_min: 22,
                    port_range_max: 22
                    }
        """
        if sec_group_rules_client is None:
            sec_group_rules_client = self.security_group_rules_client
        if security_groups_client is None:
            security_groups_client = self.security_groups_client
        if not tenant_id:
            tenant_id = security_groups_client.tenant_id
        if secgroup is None:
            secgroup = self._default_security_group(
                client=security_groups_client, tenant_id=tenant_id)

        ruleset = dict(security_group_id=secgroup['id'],
                       tenant_id=secgroup['tenant_id'])
        ruleset.update(kwargs)

        sg_rule = sec_group_rules_client.create_security_group_rule(**ruleset)
        sg_rule = sg_rule['security_group_rule']

        self.assertEqual(secgroup['tenant_id'], sg_rule['tenant_id'])
        self.assertEqual(secgroup['id'], sg_rule['security_group_id'])

        return sg_rule

    def _create_loginable_secgroup_rule(self, security_group_rules_client=None,
                                        secgroup=None,
                                        security_groups_client=None):
        """Create loginable security group rule

        This function will create:
        1. egress and ingress tcp port 22 allow rule in order to allow ssh
        access for ipv4.
        2. egress and ingress tcp port 80 allow rule in order to allow http
        access for ipv4.
        3. egress and ingress ipv6 icmp allow rule, in order to allow icmpv6.
        4. egress and ingress ipv4 icmp allow rule, in order to allow icmpv4.
        """

        if security_group_rules_client is None:
            security_group_rules_client = self.security_group_rules_client
        if security_groups_client is None:
            security_groups_client = self.security_groups_client
        rules = []
        rulesets = [
            dict(
                # ssh
                protocol='tcp',
                port_range_min=22,
                port_range_max=22,
            ),
            dict(
                # http
                protocol='tcp',
                port_range_min=80,
                port_range_max=80,
            ),
            dict(
                # ping
                protocol='icmp',
            ),
            dict(
                # ipv6-icmp for ping6
                protocol='icmp',
                ethertype='IPv6',
            )
        ]
        sec_group_rules_client = security_group_rules_client
        for ruleset in rulesets:
            for r_direction in ['ingress', 'egress']:
                ruleset['direction'] = r_direction
                try:
                    sg_rule = self._create_security_group_rule(
                        sec_group_rules_client=sec_group_rules_client,
                        secgroup=secgroup,
                        security_groups_client=security_groups_client,
                        **ruleset)
                except lib_exc.Conflict as ex:
                    # if rule already exist - skip rule and continue
                    msg = 'Security group rule already exists'
                    if msg not in ex._error_string:
                        raise ex
                else:
                    self.assertEqual(r_direction, sg_rule['direction'])
                    rules.append(sg_rule)

        return rules

    def _get_router(self, client=None, tenant_id=None):
        """Retrieve a router for the given tenant id.

        If a public router has been configured, it will be returned.

        If a public router has not been configured, but a public
        network has, a tenant router will be created and returned that
        routes traffic to the public network.
        """
        if not client:
            client = self.routers_client
        if not tenant_id:
            tenant_id = client.tenant_id
        router_id = CONF.network.public_router_id
        network_id = CONF.network.public_network_id
        if router_id:
            body = client.show_router(router_id)
            return body['router']
        elif network_id:
            router = self._create_router(client, tenant_id)
            kwargs = {'external_gateway_info': dict(network_id=network_id)}
            router = client.update_router(router['id'], **kwargs)['router']
            return router
        else:
            raise Exception("Neither of 'public_router_id' or "
                            "'public_network_id' has been defined.")

    def _create_router(self, client=None, tenant_id=None,
                       namestart='router-smoke'):
        if not client:
            client = self.routers_client
        if not tenant_id:
            tenant_id = client.tenant_id
        name = data_utils.rand_name(namestart)
        result = client.create_router(name=name,
                                      admin_state_up=True,
                                      tenant_id=tenant_id)
        router = result['router']
        self.assertEqual(router['name'], name)
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        client.delete_router,
                        router['id'])
        return router

    def _update_router_admin_state(self, router, admin_state_up):
        kwargs = dict(admin_state_up=admin_state_up)
        router = self.routers_client.update_router(
            router['id'], **kwargs)['router']
        self.assertEqual(admin_state_up, router['admin_state_up'])

    def create_networks(self, networks_client=None,
                        routers_client=None, subnets_client=None,
                        tenant_id=None, dns_nameservers=None,
                        port_security_enabled=True):
        """Create a network with a subnet connected to a router.

        The baremetal driver is a special case since all nodes are
        on the same shared network.

        :param tenant_id: id of tenant to create resources in.
        :param dns_nameservers: list of dns servers to send to subnet.
        :returns: network, subnet, router
        """
        if CONF.network.shared_physical_network:
            # NOTE(Shrews): This exception is for environments where tenant
            # credential isolation is available, but network separation is
            # not (the current baremetal case). Likely can be removed when
            # test account mgmt is reworked:
            # https://blueprints.launchpad.net/tempest/+spec/test-accounts
            if not CONF.compute.fixed_network_name:
                m = 'fixed_network_name must be specified in config'
                raise lib_exc.InvalidConfiguration(m)
            network = self._get_network_by_name(
                CONF.compute.fixed_network_name)
            router = None
            subnet = None
        else:
            network = self._create_network(
                networks_client=networks_client,
                tenant_id=tenant_id,
                port_security_enabled=port_security_enabled)
            router = self._get_router(client=routers_client,
                                      tenant_id=tenant_id)
            subnet_kwargs = dict(network=network,
                                 subnets_client=subnets_client,
                                 routers_client=routers_client)
            # use explicit check because empty list is a valid option
            if dns_nameservers is not None:
                subnet_kwargs['dns_nameservers'] = dns_nameservers
            subnet = self._create_subnet(**subnet_kwargs)
            if not routers_client:
                routers_client = self.routers_client
            router_id = router['id']
            routers_client.add_router_interface(router_id,
                                                subnet_id=subnet['id'])

            # save a cleanup job to remove this association between
            # router and subnet
            self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                            routers_client.remove_router_interface, router_id,
                            subnet_id=subnet['id'])
        return network, subnet, router

    def _create_security_group_for_test(self):
        self.security_group = self._create_security_group(
            tenant_id=self.bgpvpn_client.tenant_id)

    def _create_subnet_with_cidr(self, network, subnets_client=None,
                                 namestart='subnet-smoke', **kwargs):
        if not subnets_client:
            subnets_client = self.subnets_client
        tenant_cidr = kwargs.get('cidr')
        subnet = dict(
            name=data_utils.rand_name(namestart),
            network_id=network['id'],
            tenant_id=network['tenant_id'],
            **kwargs)
        result = subnets_client.create_subnet(**subnet)
        self.assertIsNotNone(result, 'Unable to allocate tenant network')
        subnet = result['subnet']
        self.assertEqual(subnet['cidr'], tenant_cidr)
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        subnets_client.delete_subnet, subnet['id'])
        return subnet

    def _create_fip_router(self, client=None, public_network_id=None,
                           subnet_id=None):
        router = self._create_router(client, namestart='router-')
        router_id = router['id']
        if public_network_id is None:
            public_network_id = CONF.network.public_network_id
        if client is None:
            client = self.routers_client
        kwargs = {'external_gateway_info': {'network_id': public_network_id}}
        router = client.update_router(router_id, **kwargs)['router']
        if subnet_id is not None:
            client.add_router_interface(router_id, subnet_id=subnet_id)
            self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                            client.remove_router_interface, router_id,
                            subnet_id=subnet_id)
        return router

    def _associate_fip(self, server_index):
        server = self.servers[server_index]
        fip = self.create_floating_ip(
            server, external_network_id=CONF.network.public_network_id,
            port_id=self.ports[server['id']]['id'])
        self.server_fips[server['id']] = fip
        return fip

    def _create_router_and_associate_fip(self, server_index, subnet):
        router = self._create_fip_router(subnet_id=subnet['id'])
        self._associate_fip(server_index)
        return router

    def _create_server(self, name, keypair, network, ip_address,
                       security_group_ids, clients, port_security):
        security_groups = []
        if port_security:
            security_groups = security_group_ids
        create_port_body = {'fixed_ips': [{'ip_address': ip_address}],
                            'namestart': 'port-smoke',
                            'security_groups': security_groups}
        port = self._create_port(network_id=network['id'],
                                 client=clients.ports_client,
                                 **create_port_body)
        create_server_kwargs = {
            'key_name': keypair['name'],
            'networks': [{'uuid': network['id'], 'port': port['id']}]
        }
        body, servers = compute.create_test_server(
            clients, wait_until='ACTIVE', name=name, **create_server_kwargs)
        self.addCleanup(waiters.wait_for_server_termination,
                        clients.servers_client, body['id'])
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        clients.servers_client.delete_server, body['id'])
        server = clients.servers_client.show_server(body['id'])['server']
        LOG.debug('Created server: %s with status: %s', server['id'],
                  server['status'])
        self.ports[server['id']] = port
        return server

    def _create_servers(self, ports_config=None, port_security=True):
        keypair = self.create_keypair()
        security_group_ids = [self.security_group['id']]
        if ports_config is None:
            ports_config = [[self.networks[NET_A], IP_A_S1_1],
                            [self.networks[NET_B], IP_B_S1_1]]
        for (i, port_config) in enumerate(ports_config):
            network = port_config[0]
            server = self._create_server(
                'server-' + str(i + 1), keypair, network, port_config[1],
                security_group_ids, self.os_primary, port_security)
            self.servers.append(server)
            self.servers_keypairs[server['id']] = keypair
            self.server_fixed_ips[server['id']] = (
                server['addresses'][network['name']][0]['addr'])
            self.assertTrue(self.servers_keypairs)

    def _associate_all_nets_to_bgpvpn(self, bgpvpn=None):
        bgpvpn = bgpvpn or self.bgpvpn
        for network in self.networks.values():
            self.bgpvpn_client.create_network_association(
                bgpvpn['id'], network['id'])
        LOG.debug('BGPVPN network associations completed')

    def _create_networks_and_subnets(self, names=None, subnet_cidrs=None,
                                     port_security=True):
        if not names:
            names = [NET_A, NET_B, NET_C]
        if not subnet_cidrs:
            subnet_cidrs = [[NET_A_S1], [NET_B_S1], [NET_C_S1]]
        for (name, subnet_cidrs) in zip(names, subnet_cidrs):
            network = self._create_network(
                namestart=name, port_security_enabled=port_security)
            self.networks[name] = network
            self.subnets[name] = []
            for (j, cidr) in enumerate(subnet_cidrs):
                sub_name = "subnet-%s-%d" % (name, j + 1)
                subnet = self._create_subnet_with_cidr(network,
                                                       namestart=sub_name,
                                                       cidr=cidr,
                                                       ip_version=4)
                self.subnets[name].append(subnet)

    def _setup_ssh_client(self, server):
        server_fip = self.server_fips[server['id']][
            'floating_ip_address']
        private_key = self.servers_keypairs[server['id']][
            'private_key']
        ssh_client = self.get_remote_client(server_fip,
                                            private_key=private_key)
        return ssh_client

    def _check_l3_bgpvpn(self, from_server=None, to_server=None,
                         should_succeed=True, validate_server=False):
        to_server = to_server or self.servers[1]
        destination_srv = None
        if validate_server:
            destination_srv = '%s:%s' % (to_server['name'], to_server['id'])
        destination_ip = self.server_fixed_ips[to_server['id']]
        self._check_l3_bgpvpn_by_specific_ip(from_server=from_server,
                                             to_server_ip=destination_ip,
                                             should_succeed=should_succeed,
                                             validate_server=destination_srv)

    def _check_l3_bgpvpn_by_specific_ip(self, from_server=None,
                                        to_server_ip=None,
                                        should_succeed=True,
                                        validate_server=None,
                                        repeat_validate_server=10):
        from_server = from_server or self.servers[0]
        from_server_ip = self.server_fips[from_server['id']][
            'floating_ip_address']
        if to_server_ip is None:
            to_server_ip = self.server_fixed_ips[self.servers[1]['id']]
        ssh_client = self._setup_ssh_client(from_server)
        check_reachable = should_succeed or validate_server
        msg = ""
        if check_reachable:
            msg = "Timed out waiting for {ip} to become reachable".format(
                ip=to_server_ip)
        else:
            msg = ("Unexpected ping response from VM with IP address "
                   "{dest} originated from VM with IP address "
                   "{src}").format(dest=to_server_ip, src=from_server_ip)
        try:
            result = self._check_remote_connectivity(ssh_client,
                                                     to_server_ip,
                                                     check_reachable)
            # if a negative connectivity check was unsuccessful (unexpected
            # ping reply) then try to know more:
            if not check_reachable and not result:
                try:
                    content = ssh_client.exec_command(
                        "nc %s 80" % to_server_ip).strip()
                    LOG.warning("Can connect to %s: %s", to_server_ip, content)
                except Exception:
                    LOG.warning("Could ping %s, but no http", to_server_ip)

            self.assertTrue(result, msg)

            if validate_server and result:
                # repeating multiple times gives increased odds of avoiding
                # false positives in the case where the dataplane does
                # equal-cost multipath
                for i in range(repeat_validate_server):
                    real_dest = ssh_client.exec_command(
                        "nc %s 80" % to_server_ip).strip()
                    result = real_dest == validate_server
                    self.assertTrue(
                        should_succeed == result,
                        ("Destination server name is '%s', expected is '%s'" %
                         (real_dest, validate_server)))
                    LOG.info("nc server name check %d successful", i)
        except Exception:
            LOG.exception("Error validating connectivity to %s "
                          "from VM with IP address %s: %s",
                          to_server_ip, from_server_ip, msg)
            raise

    def _associate_fip_and_check_l3_bgpvpn(self, should_succeed=True):
        subnet = self.subnets[NET_A][0]
        self.router = self._create_router_and_associate_fip(0, subnet)
        self._check_l3_bgpvpn(should_succeed=should_succeed)

    def _live_migrate(self, server_id, target_host, state,
                      volume_backed=False):
        # If target_host is None,
        # check whether source host is different with
        # the new host after migration.
        if target_host is None:
            source_host = self.get_host_for_server(server_id)
        self._migrate_server_to(server_id, target_host, volume_backed)
        waiters.wait_for_server_status(self.servers_client, server_id, state)
        migration_list = (self.admin_migration_client.list_migrations()
                          ['migrations'])

        msg = ("Live Migration failed. Migrations list for Instance "
               "%s: [" % server_id)
        for live_migration in migration_list:
            if (live_migration['instance_uuid'] == server_id):
                msg += "\n%s" % live_migration
        msg += "]"
        if target_host is None:
            self.assertNotEqual(source_host,
                                self.get_host_for_server(server_id), msg)
        else:
            self.assertEqual(target_host, self.get_host_for_server(server_id),
                             msg)

    def _migrate_server_to(self, server_id, dest_host, volume_backed=False):
        kwargs = dict()
        block_migration = getattr(self, 'block_migration', None)
        if self.block_migration is None:
            if self.is_requested_microversion_compatible('2.24'):
                kwargs['disk_over_commit'] = False
            block_migration = (CONF.compute_feature_enabled.
                               block_migration_for_live_migration and
                               not volume_backed)
        self.admin_servers_client.live_migrate_server(
            server_id, host=dest_host, block_migration=block_migration,
            **kwargs)

    def _create_l3_bgpvpn(self, name='test-l3-bgpvpn', rts=None,
                          import_rts=None, export_rts=None):
        if rts is None and import_rts is None and export_rts is None:
            rts = [self.RT1]
        import_rts = import_rts or []
        export_rts = export_rts or []
        self.bgpvpn = self.create_bgpvpn(
            self.bgpvpn_admin_client, tenant_id=self.bgpvpn_client.tenant_id,
            name=name, route_targets=rts, export_targets=export_rts,
            import_targets=import_rts)
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        self.bgpvpn_admin_client.delete_bgpvpn,
                        self.bgpvpn['id'])
        return self.bgpvpn

    def _update_l3_bgpvpn(self, rts=None, import_rts=None, export_rts=None,
                          bgpvpn=None):
        bgpvpn = bgpvpn or self.bgpvpn
        if rts is None:
            rts = [self.RT1]
        import_rts = import_rts or []
        export_rts = export_rts or []
        LOG.debug('Updating targets in BGPVPN %s', bgpvpn['id'])
        self.bgpvpn_admin_client.update_bgpvpn(bgpvpn['id'],
                                               route_targets=rts,
                                               export_targets=export_rts,
                                               import_targets=import_rts)

    def _setup_ip_forwarding(self, server_index):
        server = self.servers[server_index]
        ssh_client = self._setup_ssh_client(server)
        ssh_client.exec_command("sudo sysctl -w net.ipv4.ip_forward=1")

    def _setup_ip_address(self, server_index, cidr, device=None):
        self._setup_range_ip_address(server_index, [cidr], device=None)

    def _setup_range_ip_address(self, server_index, cidrs, device=None):
        MAX_CIDRS = 50
        if device is None:
            device = 'lo'
        server = self.servers[server_index]
        ssh_client = self._setup_ssh_client(server)
        for i in range(0, len(cidrs), MAX_CIDRS):
            ips = ' '.join(cidrs[i:i + MAX_CIDRS])
            ssh_client.exec_command(
                ("for ip in {ips}; do sudo ip addr add $ip "
                 "dev {dev}; done").format(ips=ips, dev=device))

    def _setup_http_server(self, server_index):
        server = self.servers[server_index]
        ssh_client = self._setup_ssh_client(server)
        ssh_client.exec_command("sudo nc -kl -p 80 -e echo '%s:%s' &"
                                % (server['name'], server['id']))