KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _setup_validation_fip method in tempest

Best Python code snippet using tempest_python
compute.py
Source:compute.py
...68                return address['addr']69        raise exceptions.ServerUnreachable(server_id=server['id'])70    else:71        raise lib_exc.InvalidConfiguration()72def _setup_validation_fip(73        server, clients, tenant_network, validation_resources):74    if CONF.service_available.neutron:75        ifaces = clients.interfaces_client.list_interfaces(server['id'])76        validation_port = None77        for iface in ifaces['interfaceAttachments']:78            if iface['net_id'] == tenant_network['id']:79                validation_port = iface['port_id']80                break81        if not validation_port:82            # NOTE(artom) This will get caught by the catch-all clause in83            # the wait_until loop below84            raise ValueError('Unable to setup floating IP for validation: '85                             'port not found on tenant network')86        clients.floating_ips_client.update_floatingip(87            validation_resources['floating_ip']['id'],88            port_id=validation_port)89    else: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:...
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.