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

How to use APIClient method in Django Test Plus

Best Python code snippet using django-test-plus_python

base.py

Source:base.py

1# -*- encoding: utf-8 -*-2# Licensed to the Apache Software Foundation (ASF) under one3# or more contributor license agreements.  See the NOTICE file4# distributed with this work for additional information5# regarding copyright ownership.  The ASF licenses this file6# to you under the Apache License, Version 2.0 (the7# "License"); you may not use this file except in compliance8# with the License.  You may obtain a copy of the License at9# 10#   http://www.apache.org/licenses/LICENSE-2.011# 12# Unless required by applicable law or agreed to in writing,13# software distributed under the License is distributed on an14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY15# KIND, either express or implied.  See the License for the16# specific language governing permissions and limitations17# under the License.18""" Base class for all Cloudstack resources19    -Virtual machine, Volume, Snapshot etc20"""21from utils import is_server_ssh_ready, random_gen22from marvin.cloudstackAPI import *23#Import System modules24import time25import hashlib26import base6427import types28class Domain:29    """ Domain Life Cycle """30    def __init__(self, items):31        self.__dict__.update(items)32    @classmethod33    def create(cls, apiclient, services, name=None, networkdomain=None,34               parentdomainid=None):35        """Creates an domain"""36        cmd = createDomain.createDomainCmd()37        if name:38            cmd.name = "-".join([name, random_gen()])39        elif "name" in services:40            cmd.name = "-".join([services["name"], random_gen()])41        if networkdomain:42            cmd.networkdomain = networkdomain43        elif "networkdomain" in services:44            cmd.networkdomain = services["networkdomain"]45        if parentdomainid:46            cmd.parentdomainid = parentdomainid47        elif "parentdomainid" in services:48            cmd.parentdomainid = services["parentdomainid"]49        return Domain(apiclient.createDomain(cmd).__dict__)50    def delete(self, apiclient, cleanup=None):51        """Delete an domain"""52        cmd = deleteDomain.deleteDomainCmd()53        cmd.id = self.id54        if cleanup:55            cmd.cleanup = cleanup56        apiclient.deleteDomain(cmd)57    @classmethod58    def list(cls, apiclient, **kwargs):59        """Lists domains"""60        cmd = listDomains.listDomainsCmd()61        [setattr(cmd, k, v) for k, v in kwargs.items()]62        return(apiclient.listDomains(cmd))63class Account:64    """ Account Life Cycle """65    def __init__(self, items):66        self.__dict__.update(items)67    @classmethod68    def create(cls, apiclient, services, admin=False, domainid=None):69        """Creates an account"""70        cmd = createAccount.createAccountCmd()71        #0 - User, 1 - Root Admin, 2 - Domain Admin72        cmd.accounttype = 2 if (admin and domainid) else int(admin)73        cmd.email = services["email"]74        cmd.firstname = services["firstname"]75        cmd.lastname = services["lastname"]76        # Password Encoding77        mdf = hashlib.md5()78        mdf.update(services["password"])79        cmd.password = mdf.hexdigest()80        cmd.username = "-".join([services["username"], random_gen()])81        if domainid:82            cmd.domainid = domainid83        account = apiclient.createAccount(cmd)84        return Account(account.__dict__)85    def delete(self, apiclient):86        """Delete an account"""87        cmd = deleteAccount.deleteAccountCmd()88        cmd.id = self.account.id89        apiclient.deleteAccount(cmd)90    @classmethod91    def list(cls, apiclient, **kwargs):92        """Lists accounts and provides detailed account information for93        listed accounts"""94        cmd = listAccounts.listAccountsCmd()95        [setattr(cmd, k, v) for k, v in kwargs.items()]96        return(apiclient.listAccounts(cmd))97class User:98    """ User Life Cycle """99    def __init__(self, items):100        self.__dict__.update(items)101    @classmethod102    def create(cls, apiclient, services, account, domainid):103        cmd = createUser.createUserCmd()104        """Creates an user"""105        cmd.account = account106        cmd.domainid = domainid107        cmd.email = services["email"]108        cmd.firstname = services["firstname"]109        cmd.lastname = services["lastname"]110        # Password Encoding111        mdf = hashlib.md5()112        mdf.update(services["password"])113        cmd.password = mdf.hexdigest()114        cmd.username = "-".join([services["username"], random_gen()])115        user = apiclient.createUser(cmd)116        return User(user.__dict__)117    def delete(self, apiclient):118        """Delete an account"""119        cmd = deleteUser.deleteUserCmd()120        cmd.id = self.id121        apiclient.deleteUser(cmd)122    @classmethod123    def list(cls, apiclient, **kwargs):124        """Lists users and provides detailed account information for125        listed users"""126        cmd = listUsers.listUsersCmd()127        [setattr(cmd, k, v) for k, v in kwargs.items()]128        return(apiclient.listUsers(cmd))129class VirtualMachine:130    """Manage virtual machine lifecycle"""131    def __init__(self, items, services):132        self.__dict__.update(items)133        self.username = services["username"]134        self.password = services["password"]135        self.ssh_port = services["ssh_port"]136        self.ssh_client = None137        #extract out the ipaddress138        self.ipaddress = self.nic[0].ipaddress139    @classmethod140    def create(cls, apiclient, services, templateid=None, accountid=None,141                    domainid=None, networkids=None, serviceofferingid=None,142                    securitygroupids=None, mode='basic'):143        """Create the instance"""144        145        cmd = deployVirtualMachine.deployVirtualMachineCmd()146        147        if serviceofferingid:148            cmd.serviceofferingid = serviceofferingid149        elif "serviceoffering" in services:150            cmd.serviceofferingid = services["serviceoffering"]151        cmd.zoneid = services["zoneid"]152        cmd.hypervisor = services["hypervisor"]153        if accountid:154            cmd.account = accountid155        elif "account" in services:156            cmd.account = services["account"]157        if domainid:158            cmd.domainid = domainid159        elif "domainid" in services:160            cmd.domainid = services["domainid"]161        if networkids:162            cmd.networkids = networkids163        elif "networkids" in services:164            cmd.networkids = services["networkids"]165        if templateid:166            cmd.templateid = templateid167        elif "template" in services:168            cmd.templateid = services["template"]169        if "diskoffering" in services:170            cmd.diskofferingid = services["diskoffering"]171        172        if securitygroupids:173            cmd.securitygroupids = [str(sg_id) for sg_id in securitygroupids]174        175        if "userdata" in services:176            cmd.userdata = base64.b64encode(services["userdata"])177        178        virtual_machine = apiclient.deployVirtualMachine(cmd)179        180        # VM should be in Running state after deploy181        timeout = 10182        while True:    183            vm_status = VirtualMachine.list(184                                            apiclient,185                                            id=virtual_machine.id186                                            )187            if isinstance(vm_status, list):188                if vm_status[0].state == 'Running':189                    break190                elif timeout == 0:191                    raise Exception(192                            "TimeOutException: Failed to start VM (ID: %s)" % 193                                                        virtual_machine.id)194            195            time.sleep(10)196            timeout = timeout -1197        return VirtualMachine(virtual_machine.__dict__, services)198    def start(self, apiclient):199        """Start the instance"""200        cmd = startVirtualMachine.startVirtualMachineCmd()201        cmd.id = self.id202        apiclient.startVirtualMachine(cmd)203    def stop(self, apiclient):204        """Stop the instance"""205        cmd = stopVirtualMachine.stopVirtualMachineCmd()206        cmd.id = self.id207        apiclient.stopVirtualMachine(cmd)208    def reboot(self, apiclient):209        """Reboot the instance"""210        cmd = rebootVirtualMachine.rebootVirtualMachineCmd()211        cmd.id = self.id212        apiclient.rebootVirtualMachine(cmd)213    def delete(self, apiclient):214        """Destroy an Instance"""215        cmd = destroyVirtualMachine.destroyVirtualMachineCmd()216        cmd.id = self.id217        apiclient.destroyVirtualMachine(cmd)218    def attach_volume(self, apiclient, volume):219        """Attach volume to instance"""220        cmd = attachVolume.attachVolumeCmd()221        cmd.id = volume.id222        cmd.virtualmachineid = self.id223        return apiclient.attachVolume(cmd)224    def detach_volume(self, apiclient, volume):225        """Detach volume to instance"""226        cmd = detachVolume.detachVolumeCmd()227        cmd.id = volume.id228        return apiclient.detachVolume(cmd)229    @classmethod230    def list(cls, apiclient, **kwargs):231        """List all VMs matching criteria"""232        cmd = listVirtualMachines.listVirtualMachinesCmd()233        [setattr(cmd, k, v) for k, v in kwargs.items()]234        return(apiclient.listVirtualMachines(cmd))235class Volume:236    """Manage Volume Lifecycle237    """238    def __init__(self, items):239        self.__dict__.update(items)240    @classmethod241    def create(cls, apiclient, services, zoneid=None, account=None, domainid=None,242               diskofferingid=None):243        """Create Volume"""244        cmd = createVolume.createVolumeCmd()245        cmd.name = services["diskname"]246        if diskofferingid:247            cmd.diskofferingid = diskofferingid248        elif "diskofferingid" in services:249            cmd.diskofferingid = services["diskofferingid"]250        if zoneid:251            cmd.zoneid = zoneid252        elif "zoneid" in services:253            cmd.zoneid = services["zoneid"]254        if account:255            cmd.account = account256        elif "account" in services:257            cmd.account = services["account"]258        if domainid:259            cmd.domainid = domainid260        elif "domainid" in services:261            cmd.domainid = services["domainid"]262        return Volume(apiclient.createVolume(cmd).__dict__)263    @classmethod264    def create_custom_disk(cls, apiclient, services, account=None, domainid=None):265        """Create Volume from Custom disk offering"""266        cmd = createVolume.createVolumeCmd()267        cmd.name = services["diskname"]268        cmd.diskofferingid = services["customdiskofferingid"]269        cmd.size = services["customdisksize"]270        cmd.zoneid = services["zoneid"]271        272        if account:273            cmd.account = account274        else:275            cmd.account = services["account"]276        277        if domainid:278            cmd.domainid = domainid279        else:280            cmd.domainid = services["domainid"]281            282        return Volume(apiclient.createVolume(cmd).__dict__)283    @classmethod284    def create_from_snapshot(cls, apiclient, snapshot_id, services,285                             account=None, domainid=None):286        """Create Volume from snapshot"""287        cmd = createVolume.createVolumeCmd()288        cmd.name = "-".join([services["diskname"], random_gen()])289        cmd.snapshotid = snapshot_id290        cmd.zoneid = services["zoneid"]291        cmd.size = services["size"]292        if account:293            cmd.account = account294        else:295            cmd.account = services["account"]296        if domainid:297            cmd.domainid = domainid298        else:299            cmd.domainid = services["domainid"]300        return Volume(apiclient.createVolume(cmd).__dict__)301    def delete(self, apiclient):302        """Delete Volume"""303        cmd = deleteVolume.deleteVolumeCmd()304        cmd.id = self.id305        apiclient.deleteVolume(cmd)306    @classmethod307    def list(cls, apiclient, **kwargs):308        """List all volumes matching criteria"""309        cmd = listVolumes.listVolumesCmd()310        [setattr(cmd, k, v) for k, v in kwargs.items()]311        return(apiclient.listVolumes(cmd))312class Snapshot:313    """Manage Snapshot Lifecycle314    """315    def __init__(self, items):316        self.__dict__.update(items)317    @classmethod318    def create(cls, apiclient, volume_id, account=None, domainid=None):319        """Create Snapshot"""320        cmd = createSnapshot.createSnapshotCmd()321        cmd.volumeid = volume_id322        if account:323            cmd.account = account324        if domainid:325            cmd.domainid = domainid326        return Snapshot(apiclient.createSnapshot(cmd).__dict__)327    def delete(self, apiclient):328        """Delete Snapshot"""329        cmd = deleteSnapshot.deleteSnapshotCmd()330        cmd.id = self.id331        apiclient.deleteSnapshot(cmd)332    @classmethod333    def list(cls, apiclient, **kwargs):334        """List all snapshots matching criteria"""335        cmd = listSnapshots.listSnapshotsCmd()336        [setattr(cmd, k, v) for k, v in kwargs.items()]337        return(apiclient.listSnapshots(cmd))338class Template:339    """Manage template life cycle"""340    def __init__(self, items):341        self.__dict__.update(items)342    @classmethod343    def create(cls, apiclient, services, volumeid=None,344               account=None, domainid=None):345        """Create template from Volume"""346        #Create template from Virtual machine and Volume ID347        cmd = createTemplate.createTemplateCmd()348        cmd.displaytext = services["displaytext"]349        cmd.name = "-".join([services["name"], random_gen()])350        cmd.ostypeid = services["ostypeid"]351        cmd.isfeatured = services["isfeatured"] if "isfeatured" in services else False352        cmd.ispublic = services["ispublic"] if "ispublic" in services else False353        cmd.isextractable = services["isextractable"] if "isextractable" in services else False354        cmd.passwordenabled = services["passwordenabled"] if "passwordenabled" in services else False355        356        if volumeid:357            cmd.volumeid = volumeid358        if account:359            cmd.account = account360        if domainid:361            cmd.domainid = domainid362        return Template(apiclient.createTemplate(cmd).__dict__)363    @classmethod364    def register(cls, apiclient, services, zoneid=None, account=None, domainid=None):365        """Create template from URL"""366        367        #Create template from Virtual machine and Volume ID368        cmd = registerTemplate.registerTemplateCmd()369        cmd.displaytext = services["displaytext"]370        cmd.name = "-".join([services["name"], random_gen()])371        cmd.format = services["format"]372        cmd.hypervisor = services["hypervisor"]373        cmd.ostypeid = services["ostypeid"]374        cmd.url = services["url"]375        376        if zoneid:377            cmd.zoneid = zoneid378        else:379            cmd.zoneid = services["zoneid"]380        cmd.isfeatured = services["isfeatured"] if "isfeatured" in services else False381        cmd.ispublic = services["ispublic"] if "ispublic" in services else False382        cmd.isextractable = services["isextractable"] if "isextractable" in services else False383        if account:384            cmd.account = account385        if domainid:386            cmd.domainid = domainid387        388        # Register Template389        template = apiclient.registerTemplate(cmd)390    391        if isinstance(template, list):392            return Template(template[0].__dict__)393    @classmethod394    def create_from_snapshot(cls, apiclient, snapshot, services, random_name=True):395        """Create Template from snapshot"""396        #Create template from Virtual machine and Snapshot ID397        cmd = createTemplate.createTemplateCmd()398        cmd.displaytext = services["displaytext"]399        cmd.name = "-".join([400                             services["name"], 401                             random_gen()402                            ]) if random_name else services["name"]403        cmd.ostypeid = services["ostypeid"]404        cmd.snapshotid = snapshot.id405        return Template(apiclient.createTemplate(cmd).__dict__)406    def delete(self, apiclient):407        """Delete Template"""408        409        cmd = deleteTemplate.deleteTemplateCmd()410        cmd.id = self.id411        apiclient.deleteTemplate(cmd)412    def download(self, apiclient, timeout=5, interval=60):413        """Download Template"""414        #Sleep to ensure template is in proper state before download415        time.sleep(interval)416        417        while True:418            template_response = Template.list(419                                    apiclient,420                                    id=self.id,421                                    zoneid=self.zoneid,422                                    templatefilter='self'423                                    )424            if isinstance(template_response, list):425                426                template = template_response[0]427                # If template is ready,428                # template.status = Download Complete429                # Downloading - x% Downloaded430                # Error - Any other string 431                if template.status == 'Download Complete':432                    break433                434                elif 'Downloaded' in template.status:435                    time.sleep(interval)436                elif 'Installing' not in template.status:437                    raise Exception("ErrorInDownload")438            elif timeout == 0:439                break440            441            else:442                time.sleep(interval)443                timeout = timeout - 1444        return445                446    @classmethod447    def list(cls, apiclient, **kwargs):448        """List all templates matching criteria"""449        cmd = listTemplates.listTemplatesCmd()450        [setattr(cmd, k, v) for k, v in kwargs.items()]451        return(apiclient.listTemplates(cmd))452class Iso:453    """Manage ISO life cycle"""454    def __init__(self, items):455        self.__dict__.update(items)456    @classmethod457    def create(cls, apiclient, services, account=None, domainid=None):458        """Create an ISO"""459        #Create ISO from URL460        cmd = registerIso.registerIsoCmd()461        cmd.displaytext = services["displaytext"]462        cmd.name = services["name"]463        cmd.ostypeid = services["ostypeid"]464        cmd.url = services["url"]465        cmd.zoneid = services["zoneid"]466        467        if "isextractable" in services:468            cmd.isextractable = services["isextractable"]469        if "isfeatured" in services:470            cmd.isfeatured = services["isfeatured"]471        if "ispublic" in services:472            cmd.ispublic = services["ispublic"]473        if account:474            cmd.account = account475        if domainid:476            cmd.domainid = domainid477        # Register ISO478        iso = apiclient.registerIso(cmd)479        480        if iso:481            return Iso(iso[0].__dict__)482    def delete(self, apiclient):483        """Delete an ISO"""484        cmd = deleteIso.deleteIsoCmd()485        cmd.id = self.id486        apiclient.deleteIso(cmd)487        return488    def download(self, apiclient, timeout=5, interval=60):489        """Download an ISO"""490        #Ensuring ISO is successfully downloaded491        while True:492            time.sleep(interval)493            cmd = listIsos.listIsosCmd()494            cmd.id = self.id495            iso_response = apiclient.listIsos(cmd)496            497            if isinstance(iso_response, list):     498                response = iso_response[0]499                # Again initialize timeout to avoid listISO failure500                timeout = 5501                # Check whether download is in progress(for Ex:10% Downloaded)502                # or ISO is 'Successfully Installed'503                if response.status == 'Successfully Installed':504                    return505                elif 'Downloaded' in response.status:506                    time.sleep(interval)507                elif 'Installing' not in response.status:508                    raise Exception("ErrorInDownload")509            elif timeout == 0:510                raise Exception("TimeoutException")511            else:512                timeout = timeout - 1513        return514    @classmethod515    def list(cls, apiclient, **kwargs):516        """Lists all available ISO files."""517        cmd = listIsos.listIsosCmd()518        [setattr(cmd, k, v) for k, v in kwargs.items()]519        return(apiclient.listIsos(cmd))520class PublicIPAddress:521    """Manage Public IP Addresses"""522    def __init__(self, items):523        self.__dict__.update(items)524    @classmethod525    def create(cls, apiclient, accountid, zoneid=None, domainid=None,526               services=None, networkid=None):527        """Associate Public IP address"""528        cmd = associateIpAddress.associateIpAddressCmd()529        cmd.account = accountid530        cmd.zoneid = zoneid or services["zoneid"]531        cmd.domainid = domainid or services["domainid"]532        if networkid:533            cmd.networkid = networkid534        return PublicIPAddress(apiclient.associateIpAddress(cmd).__dict__)535    def delete(self, apiclient):536        """Dissociate Public IP address"""537        cmd = disassociateIpAddress.disassociateIpAddressCmd()538        cmd.id = self.ipaddress.id539        apiclient.disassociateIpAddress(cmd)540        return541    @classmethod542    def list(cls, apiclient, **kwargs):543        """List all Public IPs matching criteria"""544        cmd = listPublicIpAddresses.listPublicIpAddressesCmd()545        [setattr(cmd, k, v) for k, v in kwargs.items()]546        return(apiclient.listPublicIpAddresses(cmd))547class NATRule:548    """Manage port forwarding rule"""549    def __init__(self, items):550        self.__dict__.update(items)551    @classmethod552    def create(cls, apiclient, virtual_machine, services, ipaddressid=None):553        """Create Port forwarding rule"""554        cmd = createPortForwardingRule.createPortForwardingRuleCmd()555        if ipaddressid:556            cmd.ipaddressid = ipaddressid557        elif "ipaddressid" in services:558            cmd.ipaddressid = services["ipaddressid"]559        cmd.privateport = services["privateport"]560        cmd.publicport = services["publicport"]561        cmd.protocol = services["protocol"]562        cmd.virtualmachineid = virtual_machine.id563        564        return NATRule(apiclient.createPortForwardingRule(cmd).__dict__)565    def delete(self, apiclient):566        """Delete port forwarding"""567        cmd = deletePortForwardingRule.deletePortForwardingRuleCmd()568        cmd.id = self.id569        apiclient.deletePortForwardingRule(cmd)570        return571    @classmethod572    def list(cls, apiclient, **kwargs):573        """List all NAT rules matching criteria"""574        cmd = listPortForwardingRules.listPortForwardingRulesCmd()575        [setattr(cmd, k, v) for k, v in kwargs.items()]576        return(apiclient.listPortForwardingRules(cmd))577    578class StaticNATRule:579    """Manage Static NAT rule"""580    def __init__(self, items):581        self.__dict__.update(items)582    @classmethod583    def create(cls, apiclient, services, ipaddressid=None):584        """Creates static ip forwarding rule"""585        586        cmd = createIpForwardingRule.createIpForwardingRuleCmd()587        cmd.protocol = services["protocol"]588        cmd.startport = services["startport"]589        590        if "endport" in services:591            cmd.endport = services["endport"]592            593        if "cidrlist" in services:594            cmd.cidrlist = services["cidrlist"]595        596        if ipaddressid:597            cmd.ipaddressid = ipaddressid598        elif "ipaddressid" in services:599            cmd.ipaddressid = services["ipaddressid"]600            601        return StaticNATRule(apiclient.createIpForwardingRule(cmd).__dict__)602    def delete(self, apiclient):603        """Delete IP forwarding rule"""604        cmd = deleteIpForwardingRule.deleteIpForwardingRuleCmd()605        cmd.id = self.id606        apiclient.deleteIpForwardingRule(cmd)607        return608    @classmethod609    def list(cls, apiclient, **kwargs):610        """List all IP forwarding rules matching criteria"""611        cmd = listIpForwardingRules.listIpForwardingRulesCmd()612        [setattr(cmd, k, v) for k, v in kwargs.items()]613        return(apiclient.listIpForwardingRules(cmd))614    615    @classmethod616    def enable(cls, apiclient, ipaddressid, virtualmachineid):617        """Enables Static NAT rule"""618        619        cmd = enableStaticNat.enableStaticNatCmd()620        cmd.ipaddressid = ipaddressid621        cmd.virtualmachineid = virtualmachineid622        apiclient.enableStaticNat(cmd)623        return624        625    @classmethod626    def disable(cls, apiclient, ipaddressid, virtualmachineid):627        """Disables Static NAT rule"""628        629        cmd = disableStaticNat.disableStaticNatCmd()630        cmd.ipaddressid = ipaddressid631        apiclient.disableStaticNat(cmd)632        return633    634class FireWallRule:635    """Manage Firewall rule"""636    def __init__(self, items):637        self.__dict__.update(items)638    @classmethod639    def create(cls, apiclient, ipaddressid, protocol, cidrlist=None,640               startport=None, endport=None):641        """Create Firewall Rule"""642        cmd = createFirewallRule.createFirewallRuleCmd()643        cmd.ipaddressid = ipaddressid644        cmd.protocol = protocol645        if cidrlist:646            cmd.cidrlist = cidrlist647        if startport:648            cmd.startport = startport649        if endport:650            cmd.endport = endport651        return FireWallRule(apiclient.createFirewallRule(cmd).__dict__)652    def delete(self, apiclient):653        """Delete Firewall rule"""654        cmd = deleteFirewallRule.deleteFirewallRuleCmd()655        cmd.id = self.id656        apiclient.deleteFirewallRule(cmd)657        return658    @classmethod659    def list(cls, apiclient, **kwargs):660        """List all Firewall Rules matching criteria"""661        cmd = listFirewallRules.listFirewallRulesCmd()662        [setattr(cmd, k, v) for k, v in kwargs.items()]663        return(apiclient.listFirewallRules(cmd))664class ServiceOffering:665    """Manage service offerings cycle"""666    def __init__(self, items):667        self.__dict__.update(items)668    @classmethod669    def create(cls, apiclient, services, domainid=None):670        """Create Service offering"""671        cmd = createServiceOffering.createServiceOfferingCmd()672        cmd.cpunumber = services["cpunumber"]673        cmd.cpuspeed = services["cpuspeed"]674        cmd.displaytext = services["displaytext"]675        cmd.memory = services["memory"]676        cmd.name = services["name"]677        # Service Offering private to that domain678        if domainid:679            cmd.domainid = domainid680        return ServiceOffering(apiclient.createServiceOffering(cmd).__dict__)681    def delete(self, apiclient):682        """Delete Service offering"""683        cmd = deleteServiceOffering.deleteServiceOfferingCmd()684        cmd.id = self.id685        apiclient.deleteServiceOffering(cmd)686        return687    @classmethod688    def list(cls, apiclient, **kwargs):689        """Lists all available service offerings."""690        cmd = listServiceOfferings.listServiceOfferingsCmd()691        [setattr(cmd, k, v) for k, v in kwargs.items()]692        return(apiclient.listServiceOfferings(cmd))693class DiskOffering:694    """Manage disk offerings cycle"""695    def __init__(self, items):696        self.__dict__.update(items)697    @classmethod698    def create(cls, apiclient, services, custom=False):699        """Create Disk offering"""700        cmd = createDiskOffering.createDiskOfferingCmd()701        cmd.displaytext = services["displaytext"]702        cmd.name = services["name"]703        if custom:704            cmd.customized = True705        else:706            cmd.disksize = services["disksize"]707        return DiskOffering(apiclient.createDiskOffering(cmd).__dict__)708    def delete(self, apiclient):709        """Delete Disk offering"""710        cmd = deleteDiskOffering.deleteDiskOfferingCmd()711        cmd.id = self.id712        apiclient.deleteDiskOffering(cmd)713        return714    @classmethod715    def list(cls, apiclient, **kwargs):716        """Lists all available disk offerings."""717        cmd = listDiskOfferings.listDiskOfferingsCmd()718        [setattr(cmd, k, v) for k, v in kwargs.items()]719        return(apiclient.listDiskOfferings(cmd))720class SnapshotPolicy:721    """Manage snapshot policies"""722    def __init__(self, items):723        self.__dict__.update(items)724    @classmethod725    def create(cls, apiclient, volumeid, services):726        """Create Snapshot policy"""727        cmd = createSnapshotPolicy.createSnapshotPolicyCmd()728        cmd.intervaltype = services["intervaltype"]729        cmd.maxsnaps = services["maxsnaps"]730        cmd.schedule = services["schedule"]731        cmd.timezone = services["timezone"]732        cmd.volumeid = volumeid733        return SnapshotPolicy(apiclient.createSnapshotPolicy(cmd).__dict__)734    def delete(self, apiclient):735        """Delete Snapshot policy"""736        cmd = deleteSnapshotPolicies.deleteSnapshotPoliciesCmd()737        cmd.id = self.id738        apiclient.deleteSnapshotPolicies(cmd)739        return740    @classmethod741    def list(cls, apiclient, **kwargs):742        """Lists snapshot policies."""743        cmd = listSnapshotPolicies.listSnapshotPoliciesCmd()744        [setattr(cmd, k, v) for k, v in kwargs.items()]745        return(apiclient.listSnapshotPolicies(cmd))746class LoadBalancerRule:747    """Manage Load Balancer rule"""748    def __init__(self, items):749        self.__dict__.update(items)750    @classmethod751    def create(cls, apiclient, services, ipaddressid, accountid=None):752        """Create Load balancing Rule"""753        cmd = createLoadBalancerRule.createLoadBalancerRuleCmd()754        cmd.publicipid = ipaddressid or services["ipaddressid"]755        cmd.account = accountid or services["account"]756        cmd.name = services["name"]757        cmd.algorithm = services["alg"]758        cmd.privateport = services["privateport"]759        cmd.publicport = services["publicport"]760        return LoadBalancerRule(apiclient.createLoadBalancerRule(cmd).__dict__)761    def delete(self, apiclient):762        """Delete load balancing rule"""763        cmd = deleteLoadBalancerRule.deleteLoadBalancerRuleCmd()764        cmd.id = self.id765        apiclient.deleteLoadBalancerRule(cmd)766        return767    def assign(self, apiclient, vms):768        """Assign virtual machines to load balancing rule"""769        cmd = assignToLoadBalancerRule.assignToLoadBalancerRuleCmd()770        cmd.id = self.id771        cmd.virtualmachineids = [str(vm.id) for vm in vms]772        apiclient.assignToLoadBalancerRule(cmd)773        return774    def remove(self, apiclient, vms):775        """Remove virtual machines from load balancing rule"""776        cmd = removeFromLoadBalancerRule.removeFromLoadBalancerRuleCmd()777        cmd.id = self.id778        cmd.virtualmachineids = [str(vm.id) for vm in vms]779        apiclient.removeFromLoadBalancerRule(cmd)780        return781    @classmethod782    def list(cls, apiclient, **kwargs):783        """List all Load balancing rules matching criteria"""784        cmd = listLoadBalancerRules.listLoadBalancerRulesCmd()785        [setattr(cmd, k, v) for k, v in kwargs.items()]786        return(apiclient.listLoadBalancerRules(cmd))787class Cluster:788    """Manage Cluster life cycle"""789    def __init__(self, items):790        self.__dict__.update(items)791    @classmethod792    def create(cls, apiclient, services, zoneid=None, podid=None):793        """Create Cluster"""794        cmd = addCluster.addClusterCmd()795        cmd.clustertype = services["clustertype"]796        cmd.hypervisor = services["hypervisor"]797        798        if zoneid:799            cmd.zoneid = zoneid800        else:801            cmd.zoneid = services["zoneid"]802        803        if podid:804            cmd.podid = podid805        else:806            cmd.podid = services["podid"]807        if "username" in services:808            cmd.username = services["username"]809        if "password" in services:810            cmd.password = services["password"]811        if "url" in services:812            cmd.url = services["url"]813        if "clustername" in services:814            cmd.clustername = services["clustername"]815        return Cluster(apiclient.addCluster(cmd)[0].__dict__)816    def delete(self, apiclient):817        """Delete Cluster"""818        cmd = deleteCluster.deleteClusterCmd()819        cmd.id = self.id820        apiclient.deleteCluster(cmd)821        return822    @classmethod823    def list(cls, apiclient, **kwargs):824        """List all Clusters matching criteria"""825        cmd = listClusters.listClustersCmd()826        [setattr(cmd, k, v) for k, v in kwargs.items()]827        return(apiclient.listClusters(cmd))828class Host:829    """Manage Host life cycle"""830    def __init__(self, items):831        self.__dict__.update(items)832    @classmethod833    def create(cls, apiclient, cluster, services, zoneid=None, podid=None):834        """Create Host in cluster"""835        836        cmd = addHost.addHostCmd()837        cmd.hypervisor = services["hypervisor"]838        cmd.url = services["url"]839        cmd.clusterid = cluster.id840        841        if zoneid:842            cmd.zoneid = zoneid843        else:844            cmd.zoneid = services["zoneid"]845        846        if podid:847            cmd.podid = podid848        else:849            cmd.podid = services["podid"]850        if "clustertype" in services:851            cmd.clustertype = services["clustertype"]852        if "username" in services:853            cmd.username = services["username"]854        if "password" in services:855            cmd.password = services["password"]856        857        # Add host858        host = apiclient.addHost(cmd)859        860        if isinstance(host, list):861            return Host(host[0].__dict__)862    def delete(self, apiclient):863        """Delete Host"""864        # Host must be in maintenance mode before deletion865        cmd = prepareHostForMaintenance.prepareHostForMaintenanceCmd()866        cmd.id = self.id867        apiclient.prepareHostForMaintenance(cmd)868        time.sleep(30)869        cmd = deleteHost.deleteHostCmd()870        cmd.id = self.id871        apiclient.deleteHost(cmd)872        return873    def enableMaintenance(self, apiclient):874        """enables maintainance mode Host"""875        876        cmd = prepareHostForMaintenance.prepareHostForMaintenanceCmd()877        cmd.id = self.id878        return apiclient.prepareHostForMaintenance(cmd)879    @classmethod880    def list(cls, apiclient, **kwargs):881        """List all Hosts matching criteria"""882        cmd = listHosts.listHostsCmd()883        [setattr(cmd, k, v) for k, v in kwargs.items()]884        return(apiclient.listHosts(cmd))885class StoragePool:886    """Manage Storage pools (Primary Storage)"""887    def __init__(self, items):888        self.__dict__.update(items)889    @classmethod890    def create(cls, apiclient, services, clusterid=None, zoneid=None, podid=None):891        """Create Storage pool (Primary Storage)"""892        cmd = createStoragePool.createStoragePoolCmd()893        cmd.name = services["name"]894        895        if podid:896            cmd.podid = podid897        else:898            cmd.podid = services["podid"]899            900        cmd.url = services["url"]901        if clusterid:902            cmd.clusterid = clusterid903        elif "clusterid" in services:904            cmd.clusterid = services["clusterid"]905        906        if zoneid:907            cmd.zoneid = zoneid908        else:909            cmd.zoneid = services["zoneid"]910        return StoragePool(apiclient.createStoragePool(cmd).__dict__)911    def delete(self, apiclient):912        """Delete Storage pool (Primary Storage)"""913        # Storage pool must be in maintenance mode before deletion914        cmd = enableStorageMaintenance.enableStorageMaintenanceCmd()915        cmd.id = self.id916        apiclient.enableStorageMaintenance(cmd)917        time.sleep(30)918        cmd = deleteStoragePool.deleteStoragePoolCmd()919        cmd.id = self.id920        apiclient.deleteStoragePool(cmd)921        return922    def enableMaintenance(self, apiclient):923        """enables maintainance mode Storage pool"""924        925        cmd = enableStorageMaintenance.enableStorageMaintenanceCmd()926        cmd.id = self.id927        return apiclient.enableStorageMaintenance(cmd)928    @classmethod929    def list(cls, apiclient, **kwargs):930        """List all storage pools matching criteria"""931        cmd = listStoragePools.listStoragePoolsCmd()932        [setattr(cmd, k, v) for k, v in kwargs.items()]933        return(apiclient.listStoragePools(cmd))934class Network:935    """Manage Network pools"""936    def __init__(self, items):937        self.__dict__.update(items)938    @classmethod939    def create(cls, apiclient, services, accountid=None, domainid=None):940        """Create Network for account"""941        cmd = createNetwork.createNetworkCmd()942        cmd.name = services["name"]943        cmd.displaytext = services["displaytext"]944        cmd.networkofferingid = services["networkoffering"]945        cmd.zoneid = services["zoneid"]946        if accountid:947            cmd.account = accountid948        if domainid:949            cmd.domainid = domainid950        return Network(apiclient.createNetwork(cmd).__dict__)951    def delete(self, apiclient):952        """Delete Account"""953        cmd = deleteNetwork.deleteNetworkCmd()954        cmd.id = self.id955        apiclient.deleteNetwork(cmd)956    @classmethod957    def list(cls, apiclient, **kwargs):958        """List all Networks matching criteria"""959        cmd = listNetworks.listNetworksCmd()960        [setattr(cmd, k, v) for k, v in kwargs.items()]961        return(apiclient.listNetworks(cmd))962class Vpn:963    """Manage VPN life cycle"""964    def __init__(self, items):965        self.__dict__.update(items)966    @classmethod967    def create(cls, apiclient, publicipid, account=None, domainid=None):968        """Create VPN for Public IP address"""969        cmd = createRemoteAccessVpn.createRemoteAccessVpnCmd()970        cmd.publicipid = publicipid971        if account:972            cmd.account = account973        if domainid:974            cmd.domainid = domainid975        return Vpn(apiclient.createRemoteAccessVpn(cmd).__dict__)976    def delete(self, apiclient):977        """Delete remote VPN access"""978        cmd = deleteRemoteAccessVpn.deleteRemoteAccessVpnCmd()979        cmd.publicipid = self.publicipid980        apiclient.deleteRemoteAccessVpn(cmd)981class VpnUser:982    """Manage VPN user"""983    def __init__(self, items):984        self.__dict__.update(items)985    @classmethod986    def create(cls, apiclient, username, password, account=None, domainid=None):987        """Create VPN user"""988        cmd = addVpnUser.addVpnUserCmd()989        cmd.username = username990        cmd.password = password991        if account:992            cmd.account = account993        if domainid:994            cmd.domainid = domainid995        return VpnUser(apiclient.addVpnUser(cmd).__dict__)996    def delete(self, apiclient):997        """Remove VPN user"""998        cmd = removeVpnUser.removeVpnUserCmd()999        cmd.username = self.username1000        cmd.account = self.account1001        cmd.domainid = self.domainid1002        apiclient.removeVpnUser(cmd)1003class Zone:1004    """Manage Zone"""1005    def __init__(self, items):1006        self.__dict__.update(items)1007    @classmethod1008    def create(cls, apiclient, services, domainid=None):1009        """Create zone"""1010        cmd = createZone.createZoneCmd()1011        cmd.dns1 = services["dns1"]1012        cmd.internaldns1 = services["internaldns1"]1013        cmd.name = services["name"]1014        cmd.networktype = services["networktype"]1015        if "dns2" in services:1016            cmd.dns2 = services["dns2"]1017        if "internaldns2" in services:1018            cmd.internaldns2 = services["internaldns2"]1019        if domainid:1020            cmd.domainid = domainid1021        return Zone(apiclient.createZone(cmd).__dict__)1022    def delete(self, apiclient):1023        """Delete Zone"""1024        cmd = deleteZone.deleteZoneCmd()1025        cmd.id = self.id1026        apiclient.deleteZone(cmd)1027    def update(self, apiclient, **kwargs):1028        """Update the zone"""1029        1030        cmd = updateZone.updateZoneCmd()1031        cmd.id = self.id1032        [setattr(cmd, k, v) for k, v in kwargs.items()]1033        return apiclient.updateZone(cmd)1034        1035        1036    @classmethod1037    def list(cls, apiclient, **kwargs):1038        """List all Zones matching criteria"""1039        cmd = listZones.listZonesCmd()1040        [setattr(cmd, k, v) for k, v in kwargs.items()]1041        return(apiclient.listZones(cmd))1042class Pod:1043    """Manage Pod"""1044    def __init__(self, items):1045        self.__dict__.update(items)1046    @classmethod1047    def create(cls, apiclient, services):1048        """Create Pod"""1049        cmd = createPod.createPodCmd()1050        cmd.gateway = services["gateway"]1051        cmd.netmask = services["netmask"]1052        cmd.name = services["name"]1053        cmd.startip = services["startip"]1054        cmd.endip = services["endip"]1055        cmd.zoneid = services["zoneid"]1056        return Pod(apiclient.createPod(cmd).__dict__)1057    def delete(self, apiclient):1058        """Delete Pod"""1059        cmd = deletePod.deletePodCmd()1060        cmd.id = self.id1061        apiclient.deletePod(cmd)1062    @classmethod1063    def list(cls, apiclient, **kwargs):1064        "Returns a default pod for specified zone"1065        cmd = listPods.listPodsCmd()1066        [setattr(cmd, k, v) for k, v in kwargs.items()]1067        return apiclient.listPods(cmd)1068class PublicIpRange:1069    """Manage VlanIpRange"""1070    def __init__(self, items):1071        self.__dict__.update(items)1072    @classmethod1073    def create(cls, apiclient, services):1074        """Create VlanIpRange"""1075        1076        cmd = createVlanIpRange.createVlanIpRangeCmd()1077        cmd.gateway = services["gateway"]1078        cmd.netmask = services["netmask"]1079        cmd.forvirtualnetwork = services["forvirtualnetwork"]1080        cmd.startip = services["startip"]1081        cmd.endip = services["endip"]1082        cmd.zoneid = services["zoneid"]1083        cmd.podid = services["podid"]1084        cmd.vlan = services["vlan"]1085        return PublicIpRange(apiclient.createVlanIpRange(cmd).__dict__)1086    def delete(self, apiclient):1087        """Delete VlanIpRange"""1088        cmd = deleteVlanIpRange.deleteVlanIpRangeCmd()1089        cmd.id = self.id1090        apiclient.deleteVlanIpRange(cmd)1091    @classmethod1092    def list(cls, apiclient, **kwargs):1093        """Lists all VLAN IP ranges."""1094        cmd = listVlanIpRanges.listVlanIpRangesCmd()1095        [setattr(cmd, k, v) for k, v in kwargs.items()]1096        return(apiclient.listVlanIpRanges(cmd))1097class SecondaryStorage:1098    """Manage Secondary storage"""1099    def __init__(self, items):1100        self.__dict__.update(items)1101    @classmethod1102    def create(cls, apiclient, services):1103        """Create Secondary Storage"""1104        cmd = addSecondaryStorage.addSecondaryStorageCmd()1105        cmd.url = services["url"]1106        if "zoneid" in services:1107            cmd.zoneid = services["zoneid"]1108        return SecondaryStorage(apiclient.addSecondaryStorage(cmd).__dict__)1109    def delete(self, apiclient):1110        """Delete Secondary Storage"""1111        cmd = deleteHost.deleteHostCmd()1112        cmd.id = self.id1113        apiclient.deleteHost(cmd)1114        1115class SecurityGroup:1116    """Manage Security Groups"""1117    def __init__(self, items):1118        self.__dict__.update(items)1119    @classmethod1120    def create(cls, apiclient, services, account=None, domainid=None,1121               description=None):1122        """Create security group"""1123        cmd = createSecurityGroup.createSecurityGroupCmd()1124        cmd.name = services["name"]1125        if account:1126            cmd.account = account1127        if domainid:1128            cmd.domainid=domainid1129        if description:1130            cmd.description=description1131            1132        return SecurityGroup(apiclient.createSecurityGroup(cmd).__dict__)1133    def delete(self, apiclient):1134        """Delete Security Group"""1135        cmd = deleteSecurityGroup.deleteSecurityGroupCmd()1136        cmd.id = self.id1137        apiclient.deleteSecurityGroup(cmd)1138        1139    def authorize(self, apiclient, services,1140                  account=None, domainid=None):1141        """Authorize Ingress Rule"""1142        1143        cmd=authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd()1144        1145        if domainid:1146            cmd.domainid = domainid1147        if account:1148            cmd.account = account1149                1150        cmd.securitygroupid=self.id1151        cmd.protocol=services["protocol"]1152        1153        if services["protocol"] == 'ICMP':1154            cmd.icmptype = -11155            cmd.icmpcode = -11156        else:1157            cmd.startport = services["startport"]1158            cmd.endport = services["endport"]1159        1160        cmd.cidrlist = services["cidrlist"]1161        return (apiclient.authorizeSecurityGroupIngress(cmd).__dict__)1162    1163    def revoke(self, apiclient, id):1164        """Revoke ingress rule"""1165        1166        cmd=revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd()1167        cmd.id=id1168        return apiclient.revokeSecurityGroupIngress(cmd)1169    1170    @classmethod1171    def list(cls, apiclient, **kwargs):1172        """Lists all security groups."""1173        cmd = listSecurityGroups.listSecurityGroupsCmd()1174        [setattr(cmd, k, v) for k, v in kwargs.items()]...

common.py

Source:common.py

1# Licensed to the Apache Software Foundation (ASF) under one2# or more contributor license agreements.  See the NOTICE file3# distributed with this work for additional information4# regarding copyright ownership.  The ASF licenses this file5# to you under the Apache License, Version 2.0 (the6# "License"); you may not use this file except in compliance7# with the License.  You may obtain a copy of the License at8# 9#   http://www.apache.org/licenses/LICENSE-2.010# 11# Unless required by applicable law or agreed to in writing,12# software distributed under the License is distributed on an13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY14# KIND, either express or implied.  See the License for the15# specific language governing permissions and limitations16# under the License.17"""Common functions18"""19#Import Local Modules20import marvin21from marvin.cloudstackTestCase import *22from marvin.cloudstackAPI import *23from marvin.remoteSSHClient import remoteSSHClient24from utils import *25from base import *26#Import System modules27import time28def wait_for_cleanup(apiclient, configs=None):29    """Sleeps till the cleanup configs passed"""30    # Configs list consists of the list of global configs31    if not isinstance(configs, list):32        return33    for config in configs:34        cmd = listConfigurations.listConfigurationsCmd()35        cmd.name = config36        cmd.listall = True37        try:38            config_descs = apiclient.listConfigurations(cmd)39        except Exception as e:40            raise Exception("Failed to fetch configurations: %s" % e)41        if not isinstance(config_descs, list):42            raise Exception("List configs didn't returned a valid data")43        config_desc = config_descs[0]44        # Sleep for the config_desc.value time45        time.sleep(int(config_desc.value))46    return47def get_domain(apiclient, services=None):48    "Returns a default domain"49    cmd = listDomains.listDomainsCmd()50    if services:51        if "domainid" in services:52            cmd.id = services["domainid"]53    domains = apiclient.listDomains(cmd)54    if isinstance(domains, list):55        assert len(domains) > 056        return domains[0]57    else:58        raise Exception("Failed to find specified domain.")59def get_zone(apiclient, services=None):60    "Returns a default zone"61    cmd = listZones.listZonesCmd()62    if services:63        if "zoneid" in services:64            cmd.id = services["zoneid"]65    zones = apiclient.listZones(cmd)66    if isinstance(zones, list):67        assert len(zones) > 0, "There are no available zones in the deployment"68        return zones[0]69    else:70        raise Exception("Failed to find specified zone.")71def get_pod(apiclient, zoneid, services=None):72    "Returns a default pod for specified zone"73    cmd = listPods.listPodsCmd()74    cmd.zoneid = zoneid75    if services:76        if "podid" in services:77            cmd.id = services["podid"]78    pods = apiclient.listPods(cmd)79    if isinstance(pods, list):80        assert len(pods) > 0, "No pods found for zone %s"%zoneid81        return pods[0]82    else:83        raise Exception("Exception: Failed to find specified pod.")84def get_template(apiclient, zoneid, ostype, services=None):85    "Returns a template"86    cmd = listOsTypes.listOsTypesCmd()87    cmd.description = ostype88    ostypes = apiclient.listOsTypes(cmd)89    if isinstance(ostypes, list):90        ostypeid = ostypes[0].id91    else:92        raise Exception(93            "Failed to find OS type with description: %s" % ostype)94    cmd = listTemplates.listTemplatesCmd()95    cmd.templatefilter = 'featured'96    cmd.zoneid = zoneid97    if services:98        if "template" in services:99            cmd.id = services["template"]100    list_templates = apiclient.listTemplates(cmd)101    if isinstance(list_templates, list):102        assert len(list_templates) > 0, "received empty response on template of type %s"%ostype103        for template in list_templates:104            if template.ostypeid == ostypeid:105                return template106    raise Exception("Exception: Failed to find template with OSTypeID: %s" %107                                                                    ostypeid)108    return109def download_systemplates_sec_storage(server, services):110    """Download System templates on sec storage"""111    try:112        # Login to management server113        ssh = remoteSSHClient(114                                          server["ipaddress"],115                                          server["port"],116                                          server["username"],117                                          server["password"]118                             )119    except Exception:120        raise Exception("SSH access failted for server with IP address: %s" %121                                                            server["ipaddess"])122    # Mount Secondary Storage on Management Server123    cmds = [124            "mkdir -p %s" % services["mnt_dir"],125            "mount -t nfs %s:/%s %s" % (126                                        services["sec_storage"],127                                        services["path"],128                                        services["mnt_dir"]129                                        ),130            "%s -m %s -u %s -h %s -F" % (131                                         services["command"],132                                         services["mnt_dir"],133                                         services["download_url"],134                                         services["hypervisor"]135                                        )136            ]137    for c in cmds:138        result = ssh.execute(c)139    res = str(result)140    # Unmount the Secondary storage141    ssh.execute("umount %s" % (services["mnt_dir"]))142    if res.count("Successfully installed system VM template") == 1:143        return144    else:145        raise Exception("Failed to download System Templates on Sec Storage")146    return147def wait_for_ssvms(apiclient, zoneid, podid, interval=60):148    """After setup wait for SSVMs to come Up"""149    time.sleep(interval)150    timeout = 40151    while True:152            list_ssvm_response = list_ssvms(153                                        apiclient,154                                        systemvmtype='secondarystoragevm',155                                        zoneid=zoneid,156                                        podid=podid157                                        )158            ssvm = list_ssvm_response[0]159            if ssvm.state != 'Running':160                # Sleep to ensure SSVMs are Up and Running161                time.sleep(interval)162                timeout = timeout - 1163            elif ssvm.state == 'Running':164                break165            elif timeout == 0:166                raise Exception("SSVM failed to come up")167                break168    timeout = 40169    while True:170            list_ssvm_response = list_ssvms(171                                        apiclient,172                                        systemvmtype='consoleproxy',173                                        zoneid=zoneid,174                                        podid=podid175                                        )176            cpvm = list_ssvm_response[0]177            if cpvm.state != 'Running':178                # Sleep to ensure SSVMs are Up and Running179                time.sleep(interval)180                timeout = timeout - 1181            elif cpvm.state == 'Running':182                break183            elif timeout == 0:184                raise Exception("CPVM failed to come up")185                break186    return187def download_builtin_templates(apiclient, zoneid, hypervisor, host,188                                                linklocalip, interval=60):189    """After setup wait till builtin templates are downloaded"""190    # Change IPTABLES Rules191    get_process_status(192                        host["ipaddress"],193                        host["port"],194                        host["username"],195                        host["password"],196                        linklocalip,197                        "iptables -P INPUT ACCEPT"198                    )199    time.sleep(interval)200    # Find the BUILTIN Templates for given Zone, Hypervisor201    list_template_response = list_templates(202                                    apiclient,203                                    hypervisor=hypervisor,204                                    zoneid=zoneid,205                                    templatefilter='self'206                                    )207    if not isinstance(list_template_response, list):208        raise Exception("Failed to download BUILTIN templates")209    # Ensure all BUILTIN templates are downloaded210    templateid = None211    for template in list_template_response:212        if template.templatetype == "BUILTIN":213                templateid = template.id214    # Sleep to ensure that template is in downloading state after adding215    # Sec storage216    time.sleep(interval)217    while True:218        template_response = list_templates(219                                    apiclient,220                                    id=templateid,221                                    zoneid=zoneid,222                                    templatefilter='self'223                                    )224        template = template_response[0]225        # If template is ready,226        # template.status = Download Complete227        # Downloading - x% Downloaded228        # Error - Any other string229        if template.status == 'Download Complete':230            break231        elif 'Downloaded' in template.status:232            time.sleep(interval)233        elif 'Installing' not in template.status:234            raise Exception("ErrorInDownload")235    return236def update_resource_limit(apiclient, resourcetype, account=None,237                                    domainid=None, max=None, projectid=None):238    """Updates the resource limit to 'max' for given account"""239    cmd = updateResourceLimit.updateResourceLimitCmd()240    cmd.resourcetype = resourcetype241    if account:242        cmd.account = account243    if domainid:244        cmd.domainid = domainid245    if max:246        cmd.max = max247    if projectid:248        cmd.projectid = projectid249    apiclient.updateResourceLimit(cmd)250    return251def list_os_types(apiclient, **kwargs):252    """List all os types matching criteria"""253    cmd = listOsTypes.listOsTypesCmd()254    [setattr(cmd, k, v) for k, v in kwargs.items()]255    return(apiclient.listOsTypes(cmd))256def list_routers(apiclient, **kwargs):257    """List all Routers matching criteria"""258    cmd = listRouters.listRoutersCmd()259    [setattr(cmd, k, v) for k, v in kwargs.items()]260    return(apiclient.listRouters(cmd))261def list_zones(apiclient, **kwargs):262    """List all Zones matching criteria"""263    cmd = listZones.listZonesCmd()264    [setattr(cmd, k, v) for k, v in kwargs.items()]265    return(apiclient.listZones(cmd))266def list_networks(apiclient, **kwargs):267    """List all Networks matching criteria"""268    cmd = listNetworks.listNetworksCmd()269    [setattr(cmd, k, v) for k, v in kwargs.items()]270    return(apiclient.listNetworks(cmd))271def list_clusters(apiclient, **kwargs):272    """List all Clusters matching criteria"""273    cmd = listClusters.listClustersCmd()274    [setattr(cmd, k, v) for k, v in kwargs.items()]275    return(apiclient.listClusters(cmd))276def list_ssvms(apiclient, **kwargs):277    """List all SSVMs matching criteria"""278    cmd = listSystemVms.listSystemVmsCmd()279    [setattr(cmd, k, v) for k, v in kwargs.items()]280    return(apiclient.listSystemVms(cmd))281def list_storage_pools(apiclient, **kwargs):282    """List all storage pools matching criteria"""283    cmd = listStoragePools.listStoragePoolsCmd()284    [setattr(cmd, k, v) for k, v in kwargs.items()]285    return(apiclient.listStoragePools(cmd))286def list_virtual_machines(apiclient, **kwargs):287    """List all VMs matching criteria"""288    cmd = listVirtualMachines.listVirtualMachinesCmd()289    [setattr(cmd, k, v) for k, v in kwargs.items()]290    return(apiclient.listVirtualMachines(cmd))291def list_hosts(apiclient, **kwargs):292    """List all Hosts matching criteria"""293    cmd = listHosts.listHostsCmd()294    [setattr(cmd, k, v) for k, v in kwargs.items()]295    return(apiclient.listHosts(cmd))296def list_configurations(apiclient, **kwargs):297    """List configuration with specified name"""298    cmd = listConfigurations.listConfigurationsCmd()299    [setattr(cmd, k, v) for k, v in kwargs.items()]300    return(apiclient.listConfigurations(cmd))301def list_publicIP(apiclient, **kwargs):302    """List all Public IPs matching criteria"""303    cmd = listPublicIpAddresses.listPublicIpAddressesCmd()304    [setattr(cmd, k, v) for k, v in kwargs.items()]305    return(apiclient.listPublicIpAddresses(cmd))306def list_nat_rules(apiclient, **kwargs):307    """List all NAT rules matching criteria"""308    cmd = listPortForwardingRules.listPortForwardingRulesCmd()309    [setattr(cmd, k, v) for k, v in kwargs.items()]310    return(apiclient.listPortForwardingRules(cmd))311def list_lb_rules(apiclient, **kwargs):312    """List all Load balancing rules matching criteria"""313    cmd = listLoadBalancerRules.listLoadBalancerRulesCmd()314    [setattr(cmd, k, v) for k, v in kwargs.items()]315    return(apiclient.listLoadBalancerRules(cmd))316def list_lb_instances(apiclient, **kwargs):317    """List all Load balancing instances matching criteria"""318    cmd = listLoadBalancerRuleInstances.listLoadBalancerRuleInstancesCmd()319    [setattr(cmd, k, v) for k, v in kwargs.items()]320    return(apiclient.listLoadBalancerRuleInstances(cmd))321def list_firewall_rules(apiclient, **kwargs):322    """List all Firewall Rules matching criteria"""323    cmd = listFirewallRules.listFirewallRulesCmd()324    [setattr(cmd, k, v) for k, v in kwargs.items()]325    return(apiclient.listFirewallRules(cmd))326def list_volumes(apiclient, **kwargs):327    """List all volumes matching criteria"""328    cmd = listVolumes.listVolumesCmd()329    [setattr(cmd, k, v) for k, v in kwargs.items()]330    return(apiclient.listVolumes(cmd))331def list_isos(apiclient, **kwargs):332    """Lists all available ISO files."""333    cmd = listIsos.listIsosCmd()334    [setattr(cmd, k, v) for k, v in kwargs.items()]335    return(apiclient.listIsos(cmd))336def list_snapshots(apiclient, **kwargs):337    """List all snapshots matching criteria"""338    cmd = listSnapshots.listSnapshotsCmd()339    [setattr(cmd, k, v) for k, v in kwargs.items()]340    return(apiclient.listSnapshots(cmd))341def list_templates(apiclient, **kwargs):342    """List all templates matching criteria"""343    cmd = listTemplates.listTemplatesCmd()344    [setattr(cmd, k, v) for k, v in kwargs.items()]345    return(apiclient.listTemplates(cmd))346def list_domains(apiclient, **kwargs):347    """Lists domains"""348    cmd = listDomains.listDomainsCmd()349    [setattr(cmd, k, v) for k, v in kwargs.items()]350    return(apiclient.listDomains(cmd))351def list_accounts(apiclient, **kwargs):352    """Lists accounts and provides detailed account information for353    listed accounts"""354    cmd = listAccounts.listAccountsCmd()355    [setattr(cmd, k, v) for k, v in kwargs.items()]356    return(apiclient.listAccounts(cmd))357def list_users(apiclient, **kwargs):358    """Lists users and provides detailed account information for359    listed users"""360    cmd = listUsers.listUsersCmd()361    [setattr(cmd, k, v) for k, v in kwargs.items()]362    return(apiclient.listUsers(cmd))363def list_snapshot_policy(apiclient, **kwargs):364    """Lists snapshot policies."""365    cmd = listSnapshotPolicies.listSnapshotPoliciesCmd()366    [setattr(cmd, k, v) for k, v in kwargs.items()]367    return(apiclient.listSnapshotPolicies(cmd))368def list_events(apiclient, **kwargs):369    """Lists events"""370    cmd = listEvents.listEventsCmd()371    [setattr(cmd, k, v) for k, v in kwargs.items()]372    return(apiclient.listEvents(cmd))373def list_disk_offering(apiclient, **kwargs):374    """Lists all available disk offerings."""375    cmd = listDiskOfferings.listDiskOfferingsCmd()376    [setattr(cmd, k, v) for k, v in kwargs.items()]377    return(apiclient.listDiskOfferings(cmd))378def list_service_offering(apiclient, **kwargs):379    """Lists all available service offerings."""380    cmd = listServiceOfferings.listServiceOfferingsCmd()381    [setattr(cmd, k, v) for k, v in kwargs.items()]382    return(apiclient.listServiceOfferings(cmd))383def list_vlan_ipranges(apiclient, **kwargs):384    """Lists all VLAN IP ranges."""385    cmd = listVlanIpRanges.listVlanIpRangesCmd()386    [setattr(cmd, k, v) for k, v in kwargs.items()]387    return(apiclient.listVlanIpRanges(cmd))388def list_usage_records(apiclient, **kwargs):389    """Lists usage records for accounts"""390    cmd = listUsageRecords.listUsageRecordsCmd()391    [setattr(cmd, k, v) for k, v in kwargs.items()]392    return(apiclient.listUsageRecords(cmd))393def list_nw_service_prividers(apiclient, **kwargs):394    """Lists Network service providers"""395    cmd = listNetworkServiceProviders.listNetworkServiceProvidersCmd()396    [setattr(cmd, k, v) for k, v in kwargs.items()]397    return(apiclient.listNetworkServiceProviders(cmd))398def list_virtual_router_elements(apiclient, **kwargs):399    """Lists Virtual Router elements"""400    cmd = listVirtualRouterElements.listVirtualRouterElementsCmd()401    [setattr(cmd, k, v) for k, v in kwargs.items()]402    return(apiclient.listVirtualRouterElements(cmd))403def list_network_offerings(apiclient, **kwargs):404    """Lists network offerings"""405    cmd = listNetworkOfferings.listNetworkOfferingsCmd()406    [setattr(cmd, k, v) for k, v in kwargs.items()]407    return(apiclient.listNetworkOfferings(cmd))408def list_resource_limits(apiclient, **kwargs):409    """Lists resource limits"""410    cmd = listResourceLimits.listResourceLimitsCmd()411    [setattr(cmd, k, v) for k, v in kwargs.items()]412    return(apiclient.listResourceLimits(cmd))413def list_vpc_offerings(apiclient, **kwargs):414    """ Lists VPC offerings """415    cmd = listVPCOfferings.listVPCOfferingsCmd()416    [setattr(cmd, k, v) for k, v in kwargs.items()]...

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.