# tempest.lib.common.validation_resources

# Source code for tempest.lib.common.validation_resources

    # Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
    # Copyright (c) 2017 IBM Corp.
    #    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 fixtures
    from oslo_log import log as logging
    from oslo_utils import excutils

    from tempest.lib.common.utils import data_utils
    from tempest.lib import exceptions as lib_exc

    LOG = logging.getLogger(__name__)

    def _network_service(clients, use_neutron):
        # Internal helper to select the right network clients
        if use_neutron:
            return clients.network
        else:
            return clients.compute

    def create_ssh_security_group(clients, add_rule=False, ethertype='IPv4',
                                  use_neutron=True):
        """Create a security group for ping/ssh testing

        Create a security group to be attached to a VM using the nova or neutron
        clients. If rules are added, the group can be attached to a VM to enable
        connectivity validation over ICMP and further testing over SSH.

        :param clients: Instance of `tempest.lib.services.clients.ServiceClients`
            or of a subclass of it. Resources are provisioned using clients from
            `clients`.
        :param add_rule: Whether security group rules are provisioned or not.
            Defaults to `False`.
        :param ethertype: 'IPv4' or 'IPv6'. Honoured only in case neutron is used.
        :param use_neutron: When True resources are provisioned via neutron, when
            False resources are provisioned via nova.
        :returns: A dictionary with the security group as returned by the API.

        Examples::

            from tempest.common import validation_resources as vr
            from tempest.lib import auth
            from tempest.lib.services import clients

            creds = auth.get_credentials('http://mycloud/identity/v3',
                                         username='me', project_name='me',
                                         password='secret', domain_name='Default')
            osclients = clients.ServiceClients(creds, 'http://mycloud/identity/v3')
            # Security group for IPv4 tests
            sg4 = vr.create_ssh_security_group(osclients, add_rule=True)
            # Security group for IPv6 tests
            sg6 = vr.create_ssh_security_group(osclients, ethertype='IPv6',
                                               add_rule=True)
        """
        network_service = _network_service(clients, use_neutron)
        security_groups_client = network_service.SecurityGroupsClient()
        security_group_rules_client = network_service.SecurityGroupRulesClient()
        # Security Group clients for nova and neutron behave the same
        sg_name = data_utils.rand_name('securitygroup-')
        sg_description = data_utils.rand_name('description-')
        security_group = security_groups_client.create_security_group(
            name=sg_name, description=sg_description)['security_group']
        # Security Group Rules clients require different parameters depending on
        # the network service in use
        if add_rule:
            try:
                if use_neutron:
                    security_group_rules_client.create_security_group_rule(
                        security_group_id=security_group['id'],
                        protocol='tcp',
                        ethertype=ethertype,
                        port_range_min=22,
                        port_range_max=22,
                        direction='ingress')
                    security_group_rules_client.create_security_group_rule(
                        security_group_id=security_group['id'],
                        protocol='icmp',
                        ethertype=ethertype,
                        direction='ingress')
                else:
                    security_group_rules_client.create_security_group_rule(
                        parent_group_id=security_group['id'], ip_protocol='tcp',
                        from_port=22, to_port=22)
                    security_group_rules_client.create_security_group_rule(
                        parent_group_id=security_group['id'], ip_protocol='icmp',
                        from_port=-1, to_port=-1)
            except Exception as sgc_exc:
                # If adding security group rules fails, we cleanup the SG before
                # re-raising the failure up
                with excutils.save_and_reraise_exception():
                    try:
                        msg = ('Error while provisioning security group rules in '
                               'security group %s. Trying to cleanup.')
                        # The exceptions logging is already handled, so using
                        # debug here just to provide more context
                        LOG.debug(msg, sgc_exc)
                        clear_validation_resources(
                            clients, keypair=None, floating_ip=None,
                            security_group=security_group,
                            use_neutron=use_neutron)
                    except Exception as cleanup_exc:
                        msg = ('Error during cleanup of a security group. '
                               'The cleanup was triggered by an exception during '
                               'the provisioning of security group rules.\n'
                               'Provisioning exception: %s\n'
                               'First cleanup exception: %s')
                        LOG.exception(msg, sgc_exc, cleanup_exc)
        LOG.debug("SSH Validation resource security group with tcp and icmp "
                  "rules %s created", sg_name)
        return security_group

    def create_validation_resources(clients, keypair=False, floating_ip=False,
                                    security_group=False,
                                    security_group_rules=False,
                                    ethertype='IPv4', use_neutron=True,
                                    floating_network_id=None,
                                    floating_network_name=None):
        """Provision resources for VM ping/ssh testing

        Create resources required to be able to ping / ssh a virtual machine:
        keypair, security group, security group rules and a floating IP.
        Which of those resources are required may depend on the cloud setup and on
        the specific test and it can be controlled via the corresponding
        arguments.

        Provisioned resources are returned in a dictionary.

        :param clients: Instance of `tempest.lib.services.clients.ServiceClients`
            or of a subclass of it. Resources are provisioned using clients from
            `clients`.
        :param keypair: Whether to provision a keypair. Defaults to False.
        :param floating_ip: Whether to provision a floating IP. Defaults to False.
        :param security_group: Whether to provision a security group. Defaults to
            False.
        :param security_group_rules: Whether to provision security group rules.
            Defaults to False.
        :param ethertype: 'IPv4' or 'IPv6'. Honoured only in case neutron is used.
        :param use_neutron: When True resources are provisioned via neutron, when
            False resources are provisioned via nova.
        :param floating_network_id: The id of the network used to provision a
            floating IP. Only used if a floating IP is requested and with neutron.
        :param floating_network_name: The name of the floating IP pool used to
            provision the floating IP. Only used if a floating IP is requested and
            with nova-net.
        :returns: A dictionary with the resources in the format they are returned
            by the API. Valid keys are 'keypair', 'floating_ip' and
            'security_group'.

        Examples::

            from tempest.common import validation_resources as vr
            from tempest.lib import auth
            from tempest.lib.services import clients

            creds = auth.get_credentials('http://mycloud/identity/v3',
                                         username='me', project_name='me',
                                         password='secret', domain_name='Default')
            osclients = clients.ServiceClients(creds, 'http://mycloud/identity/v3')
            # Request keypair and floating IP
            resources = dict(keypair=True, security_group=False,
                             security_group_rules=False, floating_ip=True)
            resources = vr.create_validation_resources(
                osclients, use_neutron=True,
                floating_network_id='4240E68E-23DA-4C82-AC34-9FEFAA24521C',
                **resources)

            # The floating IP to be attached to the VM
            floating_ip = resources['floating_ip']['ip']
        """
        # Create and Return the validation resources required to validate a VM
        msg = ('Requested validation resources keypair %s, floating IP %s, '
               'security group %s')
        LOG.debug(msg, keypair, floating_ip, security_group)
        validation_data = {}
        try:
            if keypair:
                keypair_name = data_utils.rand_name('keypair')
                validation_data.update(
                    clients.compute.KeyPairsClient().create_keypair(
                        name=keypair_name))
                LOG.debug("Validation resource key %s created", keypair_name)
            if security_group:
                validation_data['security_group'] = create_ssh_security_group(
                    clients, add_rule=security_group_rules,
                    use_neutron=use_neutron, ethertype=ethertype)
            if floating_ip:
                floating_ip_client = _network_service(
                    clients, use_neutron).FloatingIPsClient()
                if use_neutron:
                    floatingip = floating_ip_client.create_floatingip(
                        floating_network_id=floating_network_id)
                    # validation_resources['floating_ip'] has historically looked
                    # like a compute API POST /os-floating-ips response, so we need
                    # to mangle it a bit for a Neutron response with different
                    # fields.
                    validation_data['floating_ip'] = floatingip['floatingip']
                    validation_data['floating_ip']['ip'] = (
                        floatingip['floatingip']['floating_ip_address'])
                else:
                    # NOTE(mriedem): The os-floating-ips compute API was deprecated
                    # in the 2.36 microversion. Any tests for CRUD operations on
                    # floating IPs using the compute API should be capped at 2.35.
                    validation_data.update(floating_ip_client.create_floating_ip(
                        pool=floating_network_name))
                LOG.debug("Validation resource floating IP %s created",
                          validation_data['floating_ip'])
        except Exception as prov_exc:
            # If something goes wrong, cleanup as much as possible before we
            # re-raise the exception
            with excutils.save_and_reraise_exception():
                if validation_data:
                    # Cleanup may fail as well
                    try:
                        msg = ('Error while provisioning validation resources %s. '
                               'Trying to cleanup what we provisioned so far: %s')
                        # The exceptions logging is already handled, so using
                        # debug here just to provide more context
                        LOG.debug(msg, prov_exc, str(validation_data))
                        clear_validation_resources(
                            clients,
                            keypair=validation_data.get('keypair', None),
                            floating_ip=validation_data.get('floating_ip', None),
                            security_group=validation_data.get('security_group',
                                                               None),
                            use_neutron=use_neutron)
                    except Exception as cleanup_exc:
                        msg = ('Error during cleanup of validation resources. '
                               'The cleanup was triggered by an exception during '
                               'the provisioning step.\n'
                               'Provisioning exception: %s\n'
                               'First cleanup exception: %s')
                        LOG.exception(msg, prov_exc, cleanup_exc)
        return validation_data

    def clear_validation_resources(clients, keypair=None, floating_ip=None,
                                   security_group=None, use_neutron=True):
        """Cleanup resources for VM ping/ssh testing

        Cleanup a set of resources provisioned via `create_validation_resources`.
        In case of errors during cleanup, the exception is logged and the cleanup
        process is continued. The first exception that was raised is re-raised
        after the cleanup is complete.

        :param clients: Instance of `tempest.lib.services.clients.ServiceClients`
            or of a subclass of it. Resources are provisioned using clients from
            `clients`.
        :param keypair: A dictionary with the keypair to be deleted. Defaults to
            None.
        :param floating_ip: A dictionary with the floating_ip to be deleted.
            Defaults to None.
        :param security_group: A dictionary with the security_group to be deleted.
            Defaults to None.
        :param use_neutron: When True resources are provisioned via neutron, when
            False resources are provisioned via nova.

        Examples::

            from tempest.common import validation_resources as vr
            from tempest.lib import auth
            from tempest.lib.services import clients

            creds = auth.get_credentials('http://mycloud/identity/v3',
                                         username='me', project_name='me',
                                         password='secret', domain_name='Default')
            osclients = clients.ServiceClients(creds, 'http://mycloud/identity/v3')
            # Request keypair and floating IP
            resources = dict(keypair=True, security_group=False,
                             security_group_rules=False, floating_ip=True)
            resources = vr.create_validation_resources(
                osclients, validation_resources=resources, use_neutron=True,
                floating_network_id='4240E68E-23DA-4C82-AC34-9FEFAA24521C')

            # Now cleanup the resources
            try:
                vr.clear_validation_resources(osclients, use_neutron=True,
                                              **resources)
            except Exception as e:
                LOG.exception('Something went wrong during cleanup, ignoring')
        """
        has_exception = None
        if keypair:
            keypair_client = clients.compute.KeyPairsClient()
            keypair_name = keypair['name']
            try:
                keypair_client.delete_keypair(keypair_name)
            except lib_exc.NotFound:
                LOG.warning(
                    "Keypair %s is not found when attempting to delete",
                    keypair_name
                )
            except Exception as exc:
                LOG.exception('Exception raised while deleting key %s',
                              keypair_name)
                if not has_exception:
                    has_exception = exc
        network_service = _network_service(clients, use_neutron)
        if security_group:
            security_group_client = network_service.SecurityGroupsClient()
            sec_id = security_group['id']
            try:
                security_group_client.delete_security_group(sec_id)
                security_group_client.wait_for_resource_deletion(sec_id)
            except lib_exc.NotFound:
                LOG.warning("Security group %s is not found when attempting "
                            "to delete", sec_id)
            except lib_exc.Conflict as exc:
                LOG.exception('Conflict while deleting security '
                              'group %s VM might not be deleted', sec_id)
                if not has_exception:
                    has_exception = exc
            except Exception as exc:
                LOG.exception('Exception raised while deleting security '
                              'group %s', sec_id)
                if not has_exception:
                    has_exception = exc
        if floating_ip:
            floating_ip_client = network_service.FloatingIPsClient()
            fip_id = floating_ip['id']
            try:
                if use_neutron:
                    floating_ip_client.delete_floatingip(fip_id)
                else:
                    floating_ip_client.delete_floating_ip(fip_id)
            except lib_exc.NotFound:
                LOG.warning('Floating ip %s not found while attempting to '
                            'delete', fip_id)
            except Exception as exc:
                LOG.exception('Exception raised while deleting ip %s', fip_id)
                if not has_exception:
                    has_exception = exc
        if has_exception:
            raise has_exception

    class ValidationResourcesFixture(fixtures.Fixture):
        """Fixture to provision and cleanup validation resources"""

        DICT_KEYS = ['keypair', 'security_group', 'floating_ip']

        def __init__(self, clients, keypair=False, floating_ip=False,
                     security_group=False, security_group_rules=False,
                     ethertype='IPv4', use_neutron=True, floating_network_id=None,
                     floating_network_name=None):
            """Create a ValidationResourcesFixture

            Create a ValidationResourcesFixture fixtures, which provisions the
            resources required to be able to ping / ssh a virtual machine upon
            setUp and clears them out upon cleanup. Resources are  keypair,
            security group, security group rules and a floating IP - depending
            on the params.

            The fixture exposes a dictionary that includes provisioned resources.

            :param clients: `tempest.lib.services.clients.ServiceClients` or of a
                subclass of it. Resources are provisioned using clients from
                `clients`.
            :param keypair: Whether to provision a keypair. Defaults to False.
            :param floating_ip: Whether to provision a floating IP.
                Defaults to False.
            :param security_group: Whether to provision a security group.
                Defaults to False.
            :param security_group_rules: Whether to provision security group rules.
                Defaults to False.
            :param ethertype: 'IPv4' or 'IPv6'. Honoured only if neutron is used.
            :param use_neutron: When True resources are provisioned via neutron,
                when False resources are provisioned via nova.
            :param floating_network_id: The id of the network used to provision a
                floating IP. Only used if a floating IP is requested in case
                neutron is used.
            :param floating_network_name: The name of the floating IP pool used to
                provision the floating IP. Only used if a floating IP is requested
                and with nova-net.
            :returns: A dictionary with the same keys as the input
                `validation_resources` and the resources for values in the format
                 they are returned by the API.

            Examples::

                from tempest.common import validation_resources as vr
                from tempest.lib import auth
                from tempest.lib.services import clients
                import testtools

                class TestWithVR(testtools.TestCase):

                    def setUp(self):
                        creds = auth.get_credentials(
                            'http://mycloud/identity/v3',
                             username='me', project_name='me',
                             password='secret', domain_name='Default')

                        osclients = clients.ServiceClients(
                            creds, 'http://mycloud/identity/v3')
                        # Request keypair and floating IP
                        resources = dict(keypair=True, security_group=False,
                                         security_group_rules=False,
                                         floating_ip=True)
                        network_id = '4240E68E-23DA-4C82-AC34-9FEFAA24521C'
                        self.vr = self.useFixture(vr.ValidationResourcesFixture(
                            osclients, use_neutron=True,
                            floating_network_id=network_id,
                            **resources)

                    def test_use_ip(self):
                        # The floating IP to be attached to the VM
                        floating_ip = self.vr['floating_ip']['ip']
            """
            self._clients = clients
            self._keypair = keypair
            self._floating_ip = floating_ip
            self._security_group = security_group
            self._security_group_rules = security_group_rules
            self._ethertype = ethertype
            self._use_neutron = use_neutron
            self._floating_network_id = floating_network_id
            self._floating_network_name = floating_network_name
            self._validation_resources = None

        def _setUp(self):
            msg = ('Requested setup of ValidationResources keypair %s, floating '
                   'IP %s, security group %s')
            LOG.debug(msg, self._keypair, self._floating_ip, self._security_group)
            self._validation_resources = create_validation_resources(
                self._clients, keypair=self._keypair,
                floating_ip=self._floating_ip,
                security_group=self._security_group,
                security_group_rules=self._security_group_rules,
                ethertype=self._ethertype, use_neutron=self._use_neutron,
                floating_network_id=self._floating_network_id,
                floating_network_name=self._floating_network_name)
            # If provisioning raises an exception we won't have anything to
            # cleanup here, so we don't need a try-finally around provisioning
            vr = self._validation_resources
            self.addCleanup(clear_validation_resources, self._clients,
                            keypair=vr.get('keypair', None),
                            floating_ip=vr.get('floating_ip', None),
                            security_group=vr.get('security_group', None),
                            use_neutron=self._use_neutron)

        @property
        def resources(self):
            return self._validation_resources

this page last updated: 2026-08-12 11:42:01

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug