KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use associate_floating_ip method in tempest

Best Python code snippet using tempest_python

multi-provision.py

Source:multi-provision.py

1#!/usr/bin/env python2"""3multi-provision is a Python script which allows you to provision multiple4virtual machines on any provider supported by mist.io5This script has the following capabilities:6    * Deploy a specified amount of virtual machines7    * Print out information of the main network interface (mac and ip)8    * Run a post-processing script on each of the machines9--- Requirements ---10pip install -U mist11--- Usage ---12Run 'multi-provision.py -h' for an overview13--- Author ---14Dimitris Moraits15--- License ---16MIT17"""18import argparse19import getpass20import logging21import sys22import json23from mistclient import MistClient24from mistcommand.helpers.login import parse_config25def get_args():26    """27    Supports the command-line arguments listed below.28    """29    parser = argparse.ArgumentParser(30        description="Provision multiple VM's through mist.io. You can get "31        "information returned with the name of the virtual machine created "32        "and its main mac and ip address in IPv4 format. A post-script can be "33        "specified for post-processing.")34    parser.add_argument('-b', '--basename', nargs=1, required=True,35                        help='Basename of the newly deployed VMs',36                        dest='basename', type=str)37    parser.add_argument('-d', '--debug', required=False,38                        help='Enable debug output', dest='debug',39                        action='store_true')40    parser.add_argument('-i', '--print-ips', required=False,41                        help='Enable IP output', dest='ips',42                        action='store_true')43    parser.add_argument('-m', '--print-macs', required=False,44                        help='Enable MAC output', dest='macs',45                        action='store_true')46    parser.add_argument('-l', '--log-file', nargs=1, required=False,47                        help='File to log to (default = stdout)',48                        dest='logfile', type=str)49    parser.add_argument('-n', '--number', nargs=1, required=False,50                        help='Amount of VMs to deploy (default = 1)',51                        dest='quantity', type=int, default=[1])52    parser.add_argument('-M', '--monitoring', required=False,53                        help='Enable monitoring on the virtual machines',54                        dest='monitoring', action='store_true')55    parser.add_argument('-B', '--backend-name', required=False,56                        help='The name of the backend to use for provisioning.'57                        ' Defaults to the first available backend',58                        dest='backend_name', type=str)59    parser.add_argument('-I', '--image-id', required=True,60                        help='The image to deploy', dest='image_id')61    parser.add_argument('-S', '--size-id', required=True,62                        help='The id of the size/flavor to use',63                        dest='size_id')64    parser.add_argument('-N', '--networks', required=False, nargs='+',65                        help='The ids of the networks to assign to the VMs',66                        dest='networks')67    parser.add_argument('-s', '--post-script', nargs=1, required=False,68                        help='Script to be called after each VM is created and'69                        ' booted.', dest='post_script', type=str)70    parser.add_argument('-P', '--script-params', nargs=1, required=False,71                        help='Script to be called after each VM is created and'72                        ' booted.', dest='script_params', type=str)73    parser.add_argument('-H', '--host', required=False,74                        help='mist.io instance to connect to', dest='host',75                        type=str, default='https://mist.io')76    parser.add_argument('-u', '--user', nargs=1, required=False,77                        help='email registered to mist.io', dest='username',78                        type=str)79    parser.add_argument('-p', '--password', nargs=1, required=False,80                        help='The password with which to connect to the host. '81                        'If not specified, the user is prompted at runtime for'82                        ' a password', dest='password', type=str)83    parser.add_argument('-v', '--verbose', required=False,84                        help='Enable verbose output', dest='verbose',85                        action='store_true')86    parser.add_argument('-w', '--wait-max', nargs=1, required=False,87                        help='Maximum amount of seconds to wait when gathering'88                        ' information (default = 600)', dest='maxwait',89                        type=int, default=[600])90    parser.add_argument('-f', '--associate-floating-ip', required=False, action='store_true',91                        help='Auto-associates floating ips to vms in Openstack backens',92                        dest='associate_floating_ip',)93    args = parser.parse_args()94    return args95def main():96    # Handling arguments97    args = get_args()98    backend_name = None99    if args.backend_name:100        backend_name = args.backend_name101    basename = args.basename[0]102    image_id = args.image_id103    size_id = args.size_id104    monitoring = args.monitoring105    quantity = args.quantity[0]106    print_ips = args.ips107    print_macs = args.macs108    associate_floating_ip = args.associate_floating_ip109    host = args.host.rstrip('/')110    username = None111    if args.username:112        username = args.username[0]113    password = None114    if args.password:115        password = args.password[0]116    log_file = None117    if args.logfile:118        log_file = args.logfile[0]119    post_script = None120    if args.post_script:121        post_script = args.post_script[0]122    script_params = ''123    if args.script_params:124        script_params = args.script_params[0]125    if args.networks:126        networks = args.networks127    debug = args.debug128    verbose = args.verbose129    maxwait = args.maxwait[0]130    # Logging settings131    if debug:132        log_level = logging.DEBUG133    elif verbose:134        log_level = logging.INFO135    else:136        log_level = logging.WARNING137    if log_file:138        logging.basicConfig(filename=log_file,139                            format='%(asctime)s %(levelname)s %(message)s',140                            level=log_level)141    else:142        logging.basicConfig(filename=log_file,143                            format='%(asctime)s %(levelname)s %(message)s',144                            level=log_level)145    logger = logging.getLogger(__name__)146    # Getting user password147    if username and password is None:148        logger.debug('No command line password received, requesting password '149                     'from user')150        password = getpass.getpass(prompt='Enter mist.io password for user %s:'151                                   % (username))152    try:153        client = None154        try:155            logger.info('Connecting to mist.io with username %s' % (username))156            if username:157                client = MistClient(email=username, password=password,158                                    mist_uri=host, verify=False)159            else:160                credentials = parse_config()161                client = MistClient(email=credentials['email'],162                                    password=credentials['password'],163                                    mist_uri=credentials['mist_uri'])164        except IOError, e:165            pass166        if not client:167            logger.error('Could not connect to mist.io with user %s and '168                         'specified password' % (username))169            return 1170        # Get the requested or first backend171        if backend_name:172            backends = client.backends(name=backend_name)173        else:174            backends = client.backends()175        if not backends:176            logger.critical('Backend unavailable')177            return 1178        else:179            backend = backends[0]180        # Create a new ssh keypair for this deployment181        private = client.generate_key()182        try:183            client.add_key(key_name=basename, private=private)184        except:185            logger.warn('Key %s already exists' % basename)186        res = backend.create_machine(name=basename,187                                     key=client.keys(search=basename)[0],188                                     image_id=image_id,189                                     location_id=backend.locations[0]['id'],190                                     size_id=size_id,191                                     networks=args.networks,192                                     async=True,193                                     fire_and_forget=False,194                                     quantity=quantity,195                                     monitoring=monitoring,196                                     verbose=True,197                                     timeout=maxwait,198                                     associate_floating_ip=associate_floating_ip199                                     )200        if print_ips or post_script:201            try:202                logger.info('Updating VM list from backend')203                backend.update_machines()204            except:  # Retry in case of network glitch205                logger.warn('Backend unavailable. Retrying')206                backend.update_machines()207            probes = [p for p in res['logs']208                      if p['action'] == 'probe' and not p['error']]209            for p in probes:210                try:211                    machine = backend.machines(p['machine_id'])[0]212                except:213                    machine = None214                print '----------'215                if machine:216                    print 'Machine:', machine.name217                    if print_ips:218                        print 'Public ip addresses:'219                        for i in machine.info.get('public_ips', []):220                            print " - ", i221                        print 'Private ip addresses:'222                        for i in machine.info.get('private_ips'):223                            print " - ", i224                else:225                    print 'Machine:', p['machine_id']226                if print_macs:227                    print 'MACS:', json.dumps(p.get('result', {}).get('macs'))228                if post_script:229                    job = client.run_script(backend.id, p['machine_id'],230                                            post_script,231                                            script_params=script_params,232                                            fire_and_forget=True)233                    if job.get('job_id'):234                        print 'Post deployment script queued: %s/jobs/%s' % \235                              (host, job.get('job_id'))236            print '----------'237    except Exception, e:238        logger.critical('Caught exception: %s' % str(e))239        return 1240    logger.info('Finished all tasks')241    return 0242# Start program243if __name__ == "__main__":...

openstack.py

Source:openstack.py

...203                                 self.get_projectid() + "/servers", json=payload, headers=header)204        server = response.json()205        print(response.text)206        server_id = server['server']['id']207        self.associate_floating_ip(server_id)208        time.sleep(10)209        flag = self.get_url_of_instance(vm_name)210        while( flag == None ):211            print(vm_name)212            time.sleep(2)213            self.associate_floating_ip(server_id)214            flag = self.get_url_of_instance(vm_name)215    def associate_floating_ip(self, server_id):216        token = self.get_keystone_authtoken()217        header = {218            "X-Auth-Token": token219        }220        payload = {221            "addFloatingIp" :222                {223                    "address": self.get_free_floatings()224                }}225        response = requests.post("http://" + self.controllerip + ":8774/v2.1/servers/" +226                                 server_id + "/action", json=payload, headers=header)227        if( response.status_code == 200 ):228            return True229        else:...

vpc_fip_fixture.py

Source:vpc_fip_fixture.py

...40            return (floating_ip, fip_allocation_id)41        else:42            return (None, None)43    # end allocate_floating_ip44    def associate_floating_ip(self, fip_allocation_id, instance_id):45        out = self.ec2_base._shell_with_ec2_env(46            'euca-associate-address -a %s %s' % (fip_allocation_id,47                                                 instance_id), True).split('\t')48        if out:49            fip_allocation_id = out[1]50            self.logger.info(51                'Associated Floating IP (Assoc ID : %s) to Instance %s' % (fip_allocation_id,52                                                                           instance_id))53            return True54        else:55            return False56    # end associate_floating_ip57    def create_and_assoc_fip(self, instance_id):58        (fip, fip_alloc_id) = self.allocate_floating_ip()59        if not self.associate_floating_ip(fip_alloc_id, instance_id):60            self.logger.error('Error while applying FIP to instance %s' %61                              (instance_id))62            return (None, None)63        return (fip, fip_alloc_id)64    # end create_and_assoc_fip65    def disassociate_floating_ip(self, fip_allocation_id, fip):66        out = self.ec2_base._shell_with_ec2_env('euca-disassociate-address %s' % (67            fip_allocation_id), True)68        if out == 'True':69            fip_allocation_id.replace('eipassoc', 'eipalloc')70            self.logger.info('Floating IP %s disassociated ' % (fip))71            return True72        else:73            return False74    # end disassociate_floating_ip75    def release_floating_ip(self, fip_allocation_id, fip):76        out = self.ec2_base._shell_with_ec2_env(77            'euca-release-address %s' % fip_allocation_id, True)78        if out:79            self.logger.info('Floating IP (Alloc ID %s) %s released' %80                             (fip, fip_allocation_id))81            return True82        else:83            return False84    # end release_floating_ip85    def disassoc_and_delete_fip(self, fip_allocation_id, fip):86        if not self.disassociate_floating_ip(fip_allocation_id, fip):87            self.logger.error('Disassociation of FIP %s failed' % fip)88        if not self.release_floating_ip(fip_allocation_id, fip):89            self.logger.error('Unable to deallocate FIP %s ' % (fip))90    # end disassoc_and_delete_fip91    def verify_fip(self, floating_ip):92        out = self.ec2_base._shell_with_ec2_env(93            'euca-describe-addresses --filter domain=vpc| grep %s' % (floating_ip), True).split('\n')94        self.logger.debug(out)95        foundIp = False96        for ip in out:97            ip = filter(None, ip.split(' '))98            if len(ip) == 0:99                continue100            if ip[0] == floating_ip:...

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.