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

How to use wait_operation method in lisa

Best Python code snippet using lisa_python

gce.py

Source:gce.py

...81                zone=ig.zone,82                instanceGroup=ig.name83            ).execute()84            destroyed.append(str(ig.pk))85            self.wait_operation(86                zone=ig.zone,87                operation=ig_del.get('name')88            )89            ig.delete()90        return destroyed91    def _create_instance_group(self, vip, equipments):92        '''create one group to each zone'''93        groups = []94        for eq in equipments:95            zone = eq.get("zone", None)96            group_name = self.__get_instance_group_name(97                vip.group,98                zone99            )100            if group_name not in groups:101                add_ig = self.get_or_none_resource(102                    self.client.instanceGroups,103                    project=self.credential.project,104                    zone=zone,105                    instanceGroup=group_name106                )107                if add_ig is None:108                    conf = {"name": group_name}109                    add_ig = self.client.instanceGroups().insert(110                        project=self.credential.project,111                        zone=zone,112                        body=conf113                    ).execute()114                    self.wait_operation(115                        zone=zone,116                        operation=add_ig.get('name')117                    )118                ig = InstanceGroup.objects(119                    name=group_name,120                    zone=zone121                )122                groups.append(ig)123        vip.vip_id = vip.group124        return groups125    def _add_instance_in_group(self, equipments, vip):126        for eq in equipments:127            zone = eq.get('zone')128            instance_uri = "projects/{}/zones/{}/instances/{}".format(129                self.credential.project,130                zone,131                eq.get('name')132            )133            instances = {134                "instances": [135                    {'instance': instance_uri},136                ]137            }138            add_inst_req = self.client.instanceGroups().addInstances(139                project=self.credential.project,140                zone=zone,141                instanceGroup=self.__get_instance_group_name(142                    eq.get('group'),143                    zone144                ),145                body=instances146            )147            # safe fail when try to re-add instances148            try:149                add_inst = add_inst_req.execute()150            except HttpError as ex:151                if (ex.resp.status == 400 and152                   json.loads(ex.content)["error"]["errors"]153                   [0]["reason"] == "memberAlreadyExists"):154                    continue155                raise ex156            self.wait_operation(157                zone=zone,158                operation=add_inst.get('name')159            )160        return True161    def _destroy_healthcheck(self, vip):162        hc = self.get_or_none_resource(163            self.client.regionHealthChecks,164            project=self.credential.project,165            region=self.credential.region,166            healthCheck=vip.healthcheck167        )168        if hc is None:169            return True170        hc = self.client.regionHealthChecks().delete(171            project=self.credential.project,172            region=self.credential.region,173            healthCheck=vip.healthcheck174        ).execute()175        self.wait_operation(176            region=self.credential.region,177            operation=hc.get('name')178        )179        return True180    def _create_healthcheck(self, vip):181        hc_name = "hc-%s" % vip.group182        hc = self.get_or_none_resource(183            self.client.regionHealthChecks,184            project=self.credential.project,185            region=self.credential.region,186            healthCheck=hc_name187        )188        if hc is not None:189            return hc_name190        conf = {191            "checkIntervalSec": 5,192            "description": "",193            "healthyThreshold": 2,194            "httpHealthCheck": {195                "host": "",196                "port": 80,197                "proxyHeader": "NONE",198                "requestPath": "/health-check/foxha/",199                "response": "WORKING"200            },201            "logConfig": {202                "enable": False203            },204            "name": hc_name,205            "timeoutSec": 5,206            "region": 'southamerica-east1',207            "type": "HTTP",208            "unhealthyThreshold": 2209        }210        hc = self.client.regionHealthChecks().insert(211            project=self.credential.project,212            region=self.credential.region,213            body=conf214        ).execute()215        self.wait_operation(216            region=self.credential.region,217            operation=hc.get('name')218        )219        return hc_name220    def _destroy_backend_service(self, vip):221        bs = self.get_or_none_resource(222            self.client.regionBackendServices,223            project=self.credential.project,224            region=self.credential.region,225            backendService=vip.backend_service226        )227        if bs is None:228            return True229        bs = self.client.regionBackendServices().delete(230            project=self.credential.project,231            region=self.credential.region,232            backendService=vip.backend_service233        ).execute()234        self.wait_operation(235            region=self.credential.region,236            operation=bs.get('name')237        )238        return True239    def _update_backend_service(self, vip, instance_groups):240        conf = {}241        instance_group_uri = []242        for ig in instance_groups:243            uri = "zones/%s/instanceGroups/%s" % (244                ig.zone,245                ig.name246            )247            instance_group_uri.append(uri)248        conf = {"backends": [249                {'group': x,250                 "failover": i > 0} for i, x in enumerate(instance_group_uri)]}251        bs = self.client.regionBackendServices().patch(252            project=self.credential.project,253            region=self.credential.region,254            backendService=vip.backend_service,255            body=conf256        ).execute()257        self.wait_operation(258            region=self.credential.region,259            operation=bs.get('name')260        )261        return True262    def _create_backend_service(self, vip):263        bs_name = "bs-%s" % vip.group264        bs = self.get_or_none_resource(265            self.client.regionBackendServices,266            project=self.credential.project,267            region=self.credential.region,268            backendService=bs_name269        )270        if bs is not None:271            return bs_name272        healthcheck_uri = "regions/%s/healthChecks/%s" % (273            self.credential.region,274            vip.healthcheck275        )276        instance_group_uri = []277        for ig in InstanceGroup.objects.filter(vip=vip):278            uri = "zones/%s/instanceGroups/%s" % (279                ig.zone,280                ig.name281            )282            instance_group_uri.append(uri)283        conf = {284            "name": bs_name,285            "backends": [286                {'group': x,287                 "failover": i > 0} for i, x in enumerate(instance_group_uri)],288            "loadBalancingScheme": "INTERNAL",289            "healthChecks": [healthcheck_uri],290            "protocol": "TCP",291            "failoverPolicy": {292                "disableConnectionDrainOnFailover": True,293                "dropTrafficIfUnhealthy": True,294                "failoverRatio": 0295            }296        }297        bs = self.client.regionBackendServices().insert(298            project=self.credential.project,299            region=self.credential.region,300            body=conf301        ).execute()302        self.wait_operation(303            region=self.credential.region,304            operation=bs.get('name')305        )306        return bs_name307    def _destroy_forwarding_rule(self, vip):308        fr = self.get_or_none_resource(309            self.client.forwardingRules,310            project=self.credential.project,311            region=self.credential.region,312            forwardingRule=vip.forwarding_rule313        )314        if fr is None:315            return True316        fr = self.client.forwardingRules().delete(317            project=self.credential.project,318            region=self.credential.region,319            forwardingRule=vip.forwarding_rule320        ).execute()321        self.wait_operation(322            region=self.credential.region,323            operation=fr.get('name')324        )325        return True326    def _add_tags_in_forwarding_rules(self, vip, **kwargs):327        # add labels with patch328        # forwardint rule does not329        # support add tags on create330        # only add tags on resource update331        team_name = kwargs.get("team_name", None)332        engine_name = kwargs.get("engine_name", None)333        infra_name = kwargs.get("infra_name", None)334        database_name = kwargs.get("database_name", None)335        team_client = TeamClient(336            api_url=TEAM_API_URL, team_name=team_name)337        labels = team_client.make_labels(338            engine_name=engine_name,339            infra_name=infra_name,340            database_name=database_name341        )342        label_fingerprint = self.client\343            .forwardingRules().get(344                project=self.credential.project,345                region=self.credential.region,346                forwardingRule=vip.forwarding_rule347            ).execute().get('labelFingerprint')348        conf = {349            "labelFingerprint": label_fingerprint,350            "labels": labels351        }352        lbl_update = self.client.forwardingRules().setLabels(353            project=self.credential.project,354            region=self.credential.region,355            resource=vip.forwarding_rule,356            body=conf357        ).execute()358        return self.wait_operation(359            region=self.credential.region,360            operation=lbl_update.get('name')361        )362    def _create_forwarding_rule(self, vip, **kwargs):363        fr_name = "fr-%s" % vip.group364        fr = self.get_or_none_resource(365            self.client.forwardingRules,366            project=self.credential.project,367            region=self.credential.region,368            forwardingRule=fr_name369        )370        if fr is not None:371            return fr_name372        backend_service_uri = "regions/%s/backendServices/%s" % (373            self.credential.region,374            vip.backend_service375        )376        ip_uri = "regions/%s/addresses/%s" % (377            self.credential.region,378            vip.vip_ip_name379        )380        conf = {381            "name": fr_name,382            "loadBalancingScheme": "INTERNAL",383            "IPProtocol": "TCP",384            "ports": ["3306"],385            "IPAddress": ip_uri,386            'subnetwork': self.credential.subnetwork,387            "networkTier": "PREMIUM",388            "backendService": backend_service_uri,389            "allowGlobalAccess": True390        }391        fr = self.client.forwardingRules().insert(392            project=self.credential.project,393            region=self.credential.region,394            body=conf395        ).execute()396        self.wait_operation(397            region=self.credential.region,398            operation=fr.get('name')399        )400        return fr_name401    def _destroy_allocate_ip(self, vip):402        ip = self.get_or_none_resource(403            self.client.addresses,404            project=self.credential.project,405            region=self.credential.region,406            address=vip.vip_ip_name407        )408        if ip is None:409            return True410        ip = self.client.addresses().delete(411            project=self.credential.project,412            region=self.credential.region,413            address=vip.vip_ip_name414        ).execute()415        self.wait_operation(416            operation=ip.get('name'),417            region=self.credential.region418        )419        return True420    def _allocate_ip(self, vip):421        ip_name = "%s-lbip" % vip.group422        address = self.get_or_none_resource(423            self.client.addresses,424            project=self.credential.project,425            region=self.credential.region,426            address=ip_name427        )428        if address is None:429            conf = {430                'subnetwork': self.credential.subnetwork,431                'addressType': 'INTERNAL',432                'name': ip_name433            }434            address = self.client.addresses().insert(435                project=self.credential.project,436                region=self.credential.region,437                body=conf438            ).execute()439            self.wait_operation(440                operation=address.get('name'),441                region=self.credential.region442            )443        ip_metadata = self.get_internal_static_ip(ip_name)444        return {'name': ip_name, 'address': ip_metadata.get('address')}445    def get_internal_static_ip(self, ip_name):446        return self.client.addresses().get(447            project=self.credential.project,448            region=self.credential.region,449            address=ip_name450        ).execute()451    def _create_vip(self, vip):...

gdistcc.py

Source:gdistcc.py

...29from googleapiclient import discovery30from oauth2client.client import GoogleCredentials31from six.moves import input32# [START wait_operation]33def wait_operation(operation):34    # NOT thread safe35    credentials = GoogleCredentials.get_application_default()36    compute = discovery.build('compute', 'v1', credentials=credentials)37    # Wait for confirmation that the instance is created38    while True:39        result = compute.zoneOperations().get(40            project=project,41            zone=zone,42            operation=operation['name']).execute()43        if result['status'] == 'DONE':44            return False if ('error' in result) else True45        sys.stdout.write(".")46        sys.stdout.flush()47        time.sleep(2)48# [END wait_operation]49# [START list_instances]50def list_instances(project, zone, globalinstances, distro, includeterm):51    # NOT thread safe52    credentials = GoogleCredentials.get_application_default()53    compute = discovery.build('compute', 'v1', credentials=credentials)54    result = compute.instances().list(project=project, zone=zone).execute()55    if ('items' in result):56        print('%s instances in zone %s:' % (project, zone))57        instancenames = []58        name = prefix + '-' + distro59        if not globalinstances:60            name += '-' + format(str(uuid.getnode())[:8:-1])61        for instance in result['items']:62            if name in instance['name']:63                print(' - ' + instance['name'] + ' - ' + instance['status'])64                if (instance['status'] == 'RUNNING' or includeterm): 65                    instancenames.append(instance['name'])66        return instancenames if (len(instancenames) > 0) else False67    return False68# [END list_instances]69# [START check_gceproject]70def check_gceproject(distro, settingsFile):71    with open(settingsFile) as distros_file:72      distros = json.load(distros_file)['distros']73    for distrol in distros:74      if distrol['name'] == distro:75        return distrol['gceproject']76    print("ERROR: distro compute image family not found")77    exit(-1)78# [END check_gceproject]79# [START create_instance]80def create_instance(project, zone, name, source_disk_image, mtype, distro, number):81    # Unfortunatly this is NOT thread safe82    credentials = GoogleCredentials.get_application_default()83    compute = discovery.build('compute', 'v1', credentials=credentials)84    # Configure the machine85    machine_type = "zones/%s/machineTypes/%s" % (zone, mtype)86    startup_script = open(87        os.path.join(88            os.path.dirname(__file__), 'startup-scripts/%s.sh' % distro), 'r').read()89 90    config = {91        'name': name + "-" + str(number),92        'machineType': machine_type,93        # Specify the boot disk and the image to use as a source.94        'disks': [95            {96                'boot': True,97                'autoDelete': True,98                'initializeParams': {99                    'sourceImage': source_disk_image,100                }101            }102        ],103        # Specify a network interface with NAT to access the public104        # internet.105        'networkInterfaces': [{106            'network': 'global/networks/default',107            'accessConfigs': [108                {'type': 'ONE_TO_ONE_NAT', 'name': 'External NAT'}109            ]110        }],111        # Allow the instance to access cloud storage and logging.112        'serviceAccounts': [{113            'email': 'default',114            'scopes': [115                'https://www.googleapis.com/auth/logging.write'116            ]117        }],118        # Set the instance to preemptible.119        'scheduling': {120            'preemptible': 'true'121        },122        # Metadata is readable from the instance and allows you to123        # pass configuration from deployment scripts to instances.124        'metadata': {125            'items': [{126                # Startup script is automatically executed by the127                # instance upon startup.128                'key': 'startup-script',129                'value': startup_script130            }]131        }132    }133 134    # Create the instance135    operation = compute.instances().insert(136        project=project,137        zone=zone,138        body=config).execute()139    return wait_operation(operation)140# [END create_instance]141# [START check_instance]142def check_instance_ssh(project, zone, name):143    for i in xrange(40):144        145        sys.stdout.write(".")146        sys.stdout.flush()147        cicmd = 'gcloud compute ssh ' + name + \148               ' --zone ' + zone + \149               ' --project ' + project + \150               ' --command "cat /tmp/gdistcc_ready" | grep GDISTCC_READY || true'151 152        result = subprocess.check_output(cicmd, shell=True, stderr=open(os.devnull, 'w'))153                154        if "GDISTCC_READY" in result:155            print('\r - ' + name + ' - ready')156            return True 157        time.sleep(10)158     159    print('WARNING: ' + name + ' did not complete setup in a reasonable time.')160    return False161# [END check_instance]162# [START delete_instance]163def delete_instance(project, zone, name):164    # NOT thread safe165    credentials = GoogleCredentials.get_application_default()166    compute = discovery.build('compute', 'v1', credentials=credentials)167    # Delete the instance168    operation = compute.instances().delete(169        project=project,170        zone=zone,171        instance=name).execute()172    return wait_operation(operation)173# [END delete_instance]174# [START check_distro]175def check_distro(settingsFile):176    with open(settingsFile) as distros_file:177      distros = json.load(distros_file)['distros']178    pydistro = platform.linux_distribution()[0]179    pyversion = platform.linux_distribution()[1].split('.')180    pyversion = pyversion[0] + '.' + pyversion[1]181    for distrol in distros:182      if distrol['pydistro'] == pydistro and distrol['pyversion'] == pyversion:183        return distrol['name']184    print('ERROR: supported distro not detected')185    exit(-1)186# [END check_distro]...

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.