Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use wait_for_ssh_or_ping method in tempest

Best Python code snippet using tempest_python

compute.py

Source:compute.py

...90        fip_client = clients.compute_floating_ips_client91        fip_client.associate_floating_ip_to_server(92            floating_ip=validation_resources['floating_ip']['ip'],93            server_id=server['id'])94def wait_for_ssh_or_ping(server, clients, tenant_network,95                         validatable, validation_resources, wait_until,96                         set_floatingip):97    """Wait for the server for SSH or Ping as requested.98    :param server: The server dict as returned by the API99    :param clients: Client manager which provides OpenStack Tempest clients.100    :param tenant_network: Tenant network to be used for creating a server.101    :param validatable: Whether the server will be pingable or sshable.102    :param validation_resources: Resources created for the connection to the103        server. Include a keypair, a security group and an IP.104    :param wait_until: Server status to wait for the server to reach.105        It can be PINGABLE and SSHABLE states when the server is both106        validatable and has the required validation_resources provided.107    :param set_floatingip: If FIP needs to be associated to server108    """109    if set_floatingip and CONF.validation.connect_method == 'floating':110        _setup_validation_fip(111            server, clients, tenant_network, validation_resources)112    server_ip = get_server_ip(113        server, validation_resources=validation_resources)114    if wait_until == 'PINGABLE':115        waiters.wait_for_ping(116            server_ip,117            clients.servers_client.build_timeout,118            clients.servers_client.build_interval119        )120    if wait_until == 'SSHABLE':121        pkey = validation_resources['keypair']['private_key']122        ssh_client = remote_client.RemoteClient(123            server_ip,124            CONF.validation.image_ssh_user,125            pkey=pkey,126            server=server,127            servers_client=clients.servers_client128        )129        waiters.wait_for_ssh(130            ssh_client,131            clients.servers_client.build_timeout132        )133def create_test_server(clients, validatable=False, validation_resources=None,134                       tenant_network=None, wait_until=None,135                       volume_backed=False, name=None, flavor=None,136                       image_id=None, **kwargs):137    """Common wrapper utility returning a test server.138    This method is a common wrapper returning a test server that can be139    pingable or sshable.140    :param clients: Client manager which provides OpenStack Tempest clients.141    :param validatable: Whether the server will be pingable or sshable.142    :param validation_resources: Resources created for the connection to the143        server. Include a keypair, a security group and an IP.144    :param tenant_network: Tenant network to be used for creating a server.145    :param wait_until: Server status to wait for the server to reach after146        its creation. Additionally PINGABLE and SSHABLE states are also147        accepted when the server is both validatable and has the required148        validation_resources provided.149    :param volume_backed: Whether the server is volume backed or not.150        If this is true, a volume will be created and create server will be151        requested with 'block_device_mapping_v2' populated with below values:152        .. code-block:: python153            bd_map_v2 = [{154                'uuid': volume['volume']['id'],155                'source_type': 'volume',156                'destination_type': 'volume',157                'boot_index': 0,158                'delete_on_termination': True}]159            kwargs['block_device_mapping_v2'] = bd_map_v2160        If server needs to be booted from volume with other combination of bdm161        inputs than mentioned above, then pass the bdm inputs explicitly as162        kwargs and image_id as empty string ('').163    :param name: Name of the server to be provisioned. If not defined a random164        string ending with '-instance' will be generated.165    :param flavor: Flavor of the server to be provisioned. If not defined,166        CONF.compute.flavor_ref will be used instead.167    :param image_id: ID of the image to be used to provision the server. If not168        defined, CONF.compute.image_ref will be used instead.169    :returns: a tuple170    """171    if name is None:172        name = data_utils.rand_name(__name__ + "-instance")173    if flavor is None:174        flavor = CONF.compute.flavor_ref175    if image_id is None:176        image_id = CONF.compute.image_ref177    kwargs = fixed_network.set_networks_kwarg(178        tenant_network, kwargs) or {}179    multiple_create_request = (max(kwargs.get('min_count', 0),180                                   kwargs.get('max_count', 0)) > 1)181    if CONF.validation.run_validation and validatable:182        # As a first implementation, multiple pingable or sshable servers will183        # not be supported184        if multiple_create_request:185            msg = ("Multiple pingable or sshable servers not supported at "186                   "this stage.")187            raise ValueError(msg)188        LOG.debug("Provisioning test server with validation resources %s",189                  validation_resources)190        if 'security_groups' in kwargs:191            kwargs['security_groups'].append(192                {'name': validation_resources['security_group']['name']})193        else:194            try:195                kwargs['security_groups'] = [196                    {'name': validation_resources['security_group']['name']}]197            except KeyError:198                LOG.debug("No security group provided.")199        if 'key_name' not in kwargs:200            try:201                kwargs['key_name'] = validation_resources['keypair']['name']202            except KeyError:203                LOG.debug("No key provided.")204        if CONF.validation.connect_method == 'floating':205            if wait_until is None:206                wait_until = 'ACTIVE'207        if 'user_data' not in kwargs:208            # If nothing overrides the default user data script then run209            # a simple script on the host to print networking info. This is210            # to aid in debugging ssh failures.211            script = '''212                     #!/bin/sh213                     echo "Printing {user} user authorized keys"214                     cat ~{user}/.ssh/authorized_keys || true215                     '''.format(user=CONF.validation.image_ssh_user)216            script_clean = textwrap.dedent(script).lstrip().encode('utf8')217            script_b64 = base64.b64encode(script_clean)218            kwargs['user_data'] = script_b64219    if volume_backed:220        volume_name = data_utils.rand_name(__name__ + '-volume')221        volumes_client = clients.volumes_client_latest222        params = {'name': volume_name,223                  'imageRef': image_id,224                  'size': CONF.volume.volume_size}225        if CONF.compute.compute_volume_common_az:226            params.setdefault('availability_zone',227                              CONF.compute.compute_volume_common_az)228        volume = volumes_client.create_volume(**params)229        try:230            waiters.wait_for_volume_resource_status(volumes_client,231                                                    volume['volume']['id'],232                                                    'available')233        except Exception:234            with excutils.save_and_reraise_exception():235                try:236                    volumes_client.delete_volume(volume['volume']['id'])237                    volumes_client.wait_for_resource_deletion(238                        volume['volume']['id'])239                except Exception as exc:240                    LOG.exception("Deleting volume %s failed, exception %s",241                                  volume['volume']['id'], exc)242        bd_map_v2 = [{243            'uuid': volume['volume']['id'],244            'source_type': 'volume',245            'destination_type': 'volume',246            'boot_index': 0,247            'delete_on_termination': True}]248        kwargs['block_device_mapping_v2'] = bd_map_v2249        # Since this is boot from volume an image does not need250        # to be specified.251        image_id = ''252    if CONF.compute.compute_volume_common_az:253        kwargs.setdefault('availability_zone',254                          CONF.compute.compute_volume_common_az)255    body = clients.servers_client.create_server(name=name, imageRef=image_id,256                                                flavorRef=flavor,257                                                **kwargs)258    request_id = body.response['x-openstack-request-id']259    # handle the case of multiple servers260    if multiple_create_request:261        # Get servers created which name match with name param.262        body_servers = clients.servers_client.list_servers()263        servers = \264            [s for s in body_servers['servers'] if s['name'].startswith(name)]265    else:266        body = rest_client.ResponseBody(body.response, body['server'])267        servers = [body]268    if wait_until:269        # NOTE(lyarwood): PINGABLE and SSHABLE both require the instance to270        # go ACTIVE initially before we can setup the fip(s) etc so stash271        # this additional wait state for later use.272        wait_until_extra = None273        if wait_until in ['PINGABLE', 'SSHABLE']:274            wait_until_extra = wait_until275            wait_until = 'ACTIVE'276        for server in servers:277            try:278                waiters.wait_for_server_status(279                    clients.servers_client, server['id'], wait_until,280                    request_id=request_id)281                if CONF.validation.run_validation and validatable:282                    if CONF.validation.connect_method == 'floating':283                        _setup_validation_fip(284                            server, clients, tenant_network,285                            validation_resources)286                    if wait_until_extra:287                        wait_for_ssh_or_ping(288                            server, clients, tenant_network,289                            validatable, validation_resources,290                            wait_until_extra, False)291            except Exception:292                with excutils.save_and_reraise_exception():293                    for server in servers:294                        try:295                            clients.servers_client.delete_server(296                                server['id'])297                        except Exception:298                            LOG.exception('Deleting server %s failed',299                                          server['id'])300                    for server in servers:301                        # NOTE(artom) If the servers were booted with volumes...

test_server_rescue.py

Source:test_server_rescue.py

...144        # method) so to make sure server is ready before detach operation, we145        # need to perform ssh on it, more details are in bug#1960346.146        if validation_resources and CONF.validation.run_validation:147            tenant_network = self.get_tenant_network()148            compute.wait_for_ssh_or_ping(149                server, self.os_primary, tenant_network,150                True, validation_resources, "SSHABLE", True)151        else:152            waiters.wait_for_server_status(153                self.servers_client, server['id'], 'ACTIVE')154class ServerStableDeviceRescueTestIDE(BaseServerStableDeviceRescueTest):155    """Test rescuing server using an IDE device for the rescue disk"""156    @classmethod157    def skip_checks(cls):158        super().skip_checks()159        if not CONF.compute_feature_enabled.ide_bus:160            raise cls.skipException("IDE bus not available.")161    @decorators.idempotent_id('947004c3-e8ef-47d9-9f00-97b74f9eaf96')162    @testtools.skipIf("aarch64" in CONF.scenario.img_file,...

test_server_rescue_negative.py

Source:test_server_rescue_negative.py

1# Copyright 2013 Hewlett-Packard Development Company, L.P.2# Copyright 2014 NEC Corporation.  All rights reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15import testtools16from tempest.api.compute import base17from tempest.common import compute18from tempest.common import utils19from tempest.common import waiters20from tempest import config21from tempest.lib.common.utils import data_utils22from tempest.lib import decorators23from tempest.lib import exceptions as lib_exc24CONF = config.CONF25class ServerRescueNegativeTestJSON(base.BaseV2ComputeTest):26    """Negative tests of server rescue"""27    @classmethod28    def skip_checks(cls):29        super(ServerRescueNegativeTestJSON, cls).skip_checks()30        if not CONF.compute_feature_enabled.rescue:31            msg = "Server rescue not available."32            raise cls.skipException(msg)33    @classmethod34    def setup_credentials(cls):35        cls.set_network_resources(network=True, subnet=True, router=True,36                                  dhcp=True)37        super(ServerRescueNegativeTestJSON, cls).setup_credentials()38    @classmethod39    def resource_setup(cls):40        super(ServerRescueNegativeTestJSON, cls).resource_setup()41        cls.password = data_utils.rand_password()42        rescue_password = data_utils.rand_password()43        # Server for negative tests44        server = cls.create_test_server(adminPass=cls.password,45                                        wait_until='BUILD')46        resc_server = cls.create_test_server(adminPass=rescue_password,47                                             wait_until='ACTIVE')48        cls.server_id = server['id']49        cls.rescue_id = resc_server['id']50        cls.servers_client.rescue_server(51            cls.rescue_id, adminPass=rescue_password)52        waiters.wait_for_server_status(cls.servers_client,53                                       cls.rescue_id, 'RESCUE')54        waiters.wait_for_server_status(cls.servers_client,55                                       cls.server_id, 'ACTIVE')56    def _unrescue(self, server_id):57        self.servers_client.unrescue_server(server_id)58        waiters.wait_for_server_status(self.servers_client,59                                       server_id, 'ACTIVE')60    def _unpause(self, server_id):61        self.servers_client.unpause_server(server_id)62        waiters.wait_for_server_status(self.servers_client,63                                       server_id, 'ACTIVE')64    @decorators.idempotent_id('cc3a883f-43c0-4fb6-a9bb-5579d64984ed')65    @testtools.skipUnless(CONF.compute_feature_enabled.pause,66                          'Pause is not available.')67    @decorators.attr(type=['negative'])68    def test_rescue_paused_instance(self):69        """Test rescuing a paused server should fail"""70        self.servers_client.pause_server(self.server_id)71        self.addCleanup(self._unpause, self.server_id)72        waiters.wait_for_server_status(self.servers_client,73                                       self.server_id, 'PAUSED')74        self.assertRaises(lib_exc.Conflict,75                          self.servers_client.rescue_server,76                          self.server_id)77    @decorators.attr(type=['negative'])78    @decorators.idempotent_id('db22b618-f157-4566-a317-1b6d467a8094')79    def test_rescued_vm_reboot(self):80        """Test rebooing a rescued server should fail"""81        self.assertRaises(lib_exc.Conflict, self.servers_client.reboot_server,82                          self.rescue_id, type='HARD')83    @decorators.attr(type=['negative'])84    @decorators.idempotent_id('6dfc0a55-3a77-4564-a144-1587b7971dde')85    def test_rescue_non_existent_server(self):86        """Test rescuing a non-existing server should fail"""87        non_existent_server = data_utils.rand_uuid()88        self.assertRaises(lib_exc.NotFound,89                          self.servers_client.rescue_server,90                          non_existent_server)91    @decorators.attr(type=['negative'])92    @decorators.idempotent_id('70cdb8a1-89f8-437d-9448-8844fd82bf46')93    def test_rescued_vm_rebuild(self):94        """Test rebuilding a rescued server should fail"""95        self.assertRaises(lib_exc.Conflict,96                          self.servers_client.rebuild_server,97                          self.rescue_id,98                          self.image_ref_alt)99    @decorators.idempotent_id('d0ccac79-0091-4cf4-a1ce-26162d0cc55f')100    @utils.services('volume')101    @decorators.attr(type=['negative'])102    def test_rescued_vm_attach_volume(self):103        """Test attaching volume to a rescued server should fail"""104        volume = self.create_volume()105        # Rescue the server106        self.servers_client.rescue_server(self.server_id,107                                          adminPass=self.password)108        waiters.wait_for_server_status(self.servers_client,109                                       self.server_id, 'RESCUE')110        self.addCleanup(self._unrescue, self.server_id)111        # Attach the volume to the server112        self.assertRaises(lib_exc.Conflict,113                          self.servers_client.attach_volume,114                          self.server_id,115                          volumeId=volume['id'])116    @decorators.idempotent_id('f56e465b-fe10-48bf-b75d-646cda3a8bc9')117    @utils.services('volume')118    @decorators.attr(type=['negative'])119    def test_rescued_vm_detach_volume(self):120        """Test detaching volume from a rescued server should fail"""121        volume = self.create_volume()122        # This test just check detach fail and does not123        # perfom the detach operation but in cleanup from124        # self.attach_volume() it will try to detach the server125        # after unrescue the server. Due to that we need to make126        # server SSHable before it try to detach, more details are127        # in bug#1960346128        validation_resources = self.get_class_validation_resources(129            self.os_primary)130        server = self.create_test_server(131            adminPass=self.password,132            wait_until="SSHABLE",133            validatable=True,134            validation_resources=validation_resources)135        # Attach the volume to the server136        self.attach_volume(server, volume)137        # Rescue the server138        self.servers_client.rescue_server(server['id'],139                                          adminPass=self.password)140        waiters.wait_for_server_status(self.servers_client,141                                       server['id'], 'RESCUE')142        # NOTE(gmann) In next addCleanup, server unrescue is called before the143        # detach volume is called in cleanup (added by self.attach_volume()144        # method) so to make sure server is ready before detach operation, we145        # need to perform ssh on it, more details are in bug#1960346.146        if CONF.validation.run_validation:147            tenant_network = self.get_tenant_network()148            self.addCleanup(compute.wait_for_ssh_or_ping,149                            server, self.os_primary, tenant_network,150                            True, validation_resources, "SSHABLE", True)151        # addCleanup is a LIFO queue152        self.addCleanup(self._unrescue, server['id'])153        # Detach the volume from the server expecting failure154        self.assertRaises(lib_exc.Conflict,155                          self.servers_client.detach_volume,156                          server['id'],...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.