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

How to use exists_vm method in lisa

Best Python code snippet using lisa_python

hyperv.py

Source:hyperv.py

...15        return ""16    @property17    def can_install(self) -> bool:18        return True19    def exists_vm(self, name: str) -> bool:20        output = self.node.tools[PowerShell].run_cmdlet(21            f"Get-VM -Name {name}",22            fail_on_error=False,23            force_run=True,24        )25        return output.strip() != ""26    def delete_vm(self, name: str) -> None:27        # check if vm is present28        if not self.exists_vm(name):29            return30        # stop and delete vm31        powershell = self.node.tools[PowerShell]32        powershell.run_cmdlet(33            f"Stop-VM -Name {name} -Force",34            force_run=True,35        )36        powershell.run_cmdlet(37            f"Remove-VM -Name {name} -Force",38            force_run=True,39        )40    def create_vm(41        self,42        name: str,43        guest_image_path: str,44        switch_name: str,45        cores: int = 2,46        memory: int = 2048,47        stop_existing_vm: bool = True,48    ) -> None:49        if stop_existing_vm:50            self.delete_vm(name)51        # create a VM in hyperv52        powershell = self.node.tools[PowerShell]53        powershell.run_cmdlet(54            f"New-VM -Name {name} -Generation 1 "55            f"-MemoryStartupBytes {memory}MB -BootDevice VHD "56            f"-VHDPath {guest_image_path} -SwitchName {switch_name}",57            force_run=True,58        )59        # set cores and memory type60        powershell.run_cmdlet(61            f"Set-VM -Name {name} -ProcessorCount {cores} -StaticMemory "62            "-CheckpointType Disabled",63            force_run=True,64        )65        # add disks66        disk_info = powershell.run_cmdlet(67            "(Get-Disk | Where-Object {$_.OperationalStatus -eq 'offline'}).Number",68            force_run=True,69        )70        matched = re.findall(r"\d+", disk_info)71        disk_numbers = [int(x) for x in matched]72        for disk_number in disk_numbers:73            powershell.run_cmdlet(74                f"Add-VMHardDiskDrive -VMName {name} -DiskNumber {disk_number} "75                "-ControllerType 'SCSI'",76                force_run=True,77            )78        # start vm79        powershell.run_cmdlet(80            f"Start-VM -Name {name}",81            force_run=True,82        )83        # wait for vm start84        timeout_start = time.time()85        is_ready = False86        self._log.debug(f"Waiting for VM {name} to start")87        while time.time() - timeout_start < 600:88            try:89                if self.get_ip_address(name):90                    self._log.debug(f"VM {name} is ready")91                    is_ready = True92                    break93            except LisaException as e:94                self._log.debug(f"VM {name} not ready: {e}")95                time.sleep(10)96        if not is_ready:97            raise LisaException(f"VM {name} did not start")98    def exists_switch(self, name: str) -> bool:99        output = self.node.tools[PowerShell].run_cmdlet(100            f"Get-VMSwitch -Name {name}",101            fail_on_error=False,102            force_run=True,103        )104        return output.strip() != ""105    def delete_switch(self, name: str) -> None:106        if self.exists_switch(name):107            self.node.tools[PowerShell].run_cmdlet(108                f"Remove-VMSwitch -Name {name} -Force",109                force_run=True,110            )111    def create_switch(self, name: str) -> None:112        # remove switch if it exists113        self.delete_switch(name)114        # create a new switch115        self.node.tools[PowerShell].run_cmdlet(116            f"New-VMSwitch -Name {name} -SwitchType Internal",117            force_run=True,118        )119    def get_switch_interface_index(self, switch_name: str) -> int:120        output = self.node.tools[PowerShell].run_cmdlet(121            f"(Get-NetAdapter -Name '*{switch_name}*').ifindex",122            force_run=True,123        )124        return int(output.strip())125    def exists_nat(self, name: str) -> bool:126        output = self.node.tools[PowerShell].run_cmdlet(127            f"Get-NetNat -Name {name}",128            fail_on_error=False,129            force_run=True,130        )131        return output.strip() != ""132    def delete_nat(self, name: str) -> None:133        if self.exists_nat(name):134            self.node.tools[PowerShell].run_cmdlet(135                f"Remove-NetNat -Name {name} -Confirm:$false",136                force_run=True,137            )138    def create_nat(self, name: str, ip_range: str) -> None:139        # delete NAT if it exists140        self.delete_nat(name)141        # create a new NAT142        self.node.tools[PowerShell].run_cmdlet(143            f"New-NetNat -Name {name} -InternalIPInterfaceAddressPrefix '{ip_range}' ",144            force_run=True,145        )146    def delete_nat_networking(self, switch_name: str, nat_name: str) -> None:147        # Delete switch148        self.delete_switch(switch_name)149        # delete NAT150        self.delete_nat(nat_name)151    def setup_nat_networking(self, switch_name: str, nat_name: str) -> None:152        """153        Setup NAT networking154        Reference: https://docs.microsoft.com/en-us/virtualization/hyper-v-on-windows/user-guide/setup-nat-network # noqa155        """156        # create a new switch157        self.create_switch(switch_name)158        # find interface index of the switch159        interface_index = self.get_switch_interface_index(switch_name)160        # set switch interface as gateway for NAT161        self.node.tools[PowerShell].run_cmdlet(162            "New-NetIPAddress -IPAddress 192.168.5.1 "163            f"-InterfaceIndex {interface_index} -PrefixLength 24",164            force_run=True,165        )166        # create a new NAT167        self.create_nat(nat_name, "192.168.5.0/24")168    def get_ip_address(self, name: str) -> str:169        # verify vm is running170        assert_that(self.exists_vm(name)).is_true()171        # get vm ip address172        output = self.node.tools[PowerShell].run_cmdlet(173            f"Get-VM -Name {name} | Select -ExpandProperty networkadapters | select"174            " ipaddresses",175            force_run=True,176        )177        # regex match for ip address178        match = re.search(self.IP_REGEX, output)179        ip_address = match.group(0) if match else ""180        # raise exception if ip address is not found181        if not ip_address:182            raise LisaException(f"Could not find IP address for VM {name}")183        return ip_address184    def exist_port_forwarding(...

test_vm.py

Source:test_vm.py

...190        # although test-hvm64-example doesn't depend on storage backend191        # some test like this would have been good where storage backend doesn't exist192        # and it is trying to restore from vm-1 or vm-0193        # vm-0 should fail but vm-1 should succeed194        # if backend.exists_vm(0) is False:195        #     with pytest.raises(Exception):196        #         test_vm.restore()197        #         assert test_vm.is_running is False198        # should not raise any exceptions if everything is fine199        logging.info("test vm with proper args")200        test_vm.restore(snapshot_path=snapshot_file)201        assert get_vm_state(test_vm.vm_name) != "p"202        destroy_vm(test_vm.vm_name)203        logging.info("test vm with proper args and pause=True")204        test_vm.restore(snapshot_path=snapshot_file, pause=True)205        assert get_vm_state(test_vm.vm_name) == "p"206        logging.info("restoring a restored VM")207        test_vm.restore(snapshot_path=snapshot_file)208        # should get the new state...

group.py

Source:group.py

1import uuid2from plugins.sqlbackend import dbsession_method3from . import models4from core import DependencyModule5from core.exceptions import ExistsException6class GroupUtils(DependencyModule):7    __module_name__ = 'grouputils'8    group_params_key = 'group'9    vm_params_key = 'vm'10    def on_register_app(self, app):11        pass12    @property13    def infomanager(self):14        return self._app.infomanager15    def patch_group(self, group_dict):16        infomanager = self.infomanager17        if infomanager:18            group_dict = infomanager.patch_info(19                self.group_params_key, group_dict)20        return group_dict21    def patch_vm(self, vm_dict):22        infomanager = getattr(self._app, 'infomanager')23        if infomanager:24            vm_dict = self._app.infomanager.patch_info(25                self.vm_params_key, vm_dict)26        return vm_dict27    @dbsession_method28    def count_vms(self, session, group_dict):29        vms = self.dbutils_get_group_vms(session, group_dict)30        return len(vms) if vms else 031    @dbsession_method32    def db_get_groups_ports(self, session, user_id):33        return self.dbutils_get_groups_port(session, user_id)34    @dbsession_method35    def db_get_groups(self, session, user_id):36        group_models = self.dbutils_get_groups(session, user_id)37        if group_models:38            g_d = [g.to_dict() for g in group_models]39            for g in g_d:40                vms = self.dbutils_get_group_vms(session, g['group_id']) or []41                g['instances'] = [v.to_dict() for v in vms]42            return g_d43    @dbsession_method44    def db_create_group(self, session, group_dict):45        group_dict = self.patch_group(group_dict)46        group_id = group_dict.setdefault('group_id', None)47        if not group_id or group_id == 'auto':48            group_dict['group_id'] = str(uuid.uuid4())49        exists_group = self.dbutils_get_group_model(50            session, group_dict=group_dict)51        if exists_group:52            try:53                del group_dict['id']54                del group_dict['created']55                del group_dict['instances']56                del group_dict['proxy_url']57            except:58                pass59            exists_group.parse_dict(group_dict)60            # rl_group_id = exists_group.group_id61        else:62            group = models.Group()63            group.parse_dict(group_dict)64            session.add(group)65        session.commit()66        # return group_dict = group_dict_input append default info67        return group_dict68    @dbsession_method69    def db_get_group(self, session, group_dict):70        group = self.dbutils_get_group_model(session, group_dict=group_dict)71        g = group.to_dict()72        vms = self.dbutils_get_group_vms(session, g['group_id']) or []73        g['instances'] = [v.to_dict() for v in vms]74        return g75    @dbsession_method76    def db_drop_group(self, session, group_dict):77        group = self.dbutils_get_group_model(session, group_dict=group_dict)78        if group:79            # drop all vm80            for inst in group.instances:81                session.delete(inst)82            session.delete(group)83            session.commit()84    @dbsession_method85    def db_drop_vm(self, session, vm_dict):86        self.dbutils_drop_vm_model(session, vm_dict=vm_dict)87        session.commit()88    @dbsession_method89    def db_drop_vms_in_group(self, session, group_id):90        vms = session.query(models.Instance).filter(models.Instance.group_id == group_id).all()91        for vm in vms:92            session.delete(vm)93        session.commit()94    @dbsession_method95    def db_create_vm(self, session, vm_dict):96        vm_dict = self.dbutils_create_vm_model(session, vm_dict)97        session.commit()98        return vm_dict99    @dbsession_method100    def db_update_vm(self, session, vm_dict):101        vm = self.dbutils_get_vm_model(session, vm_dict=vm_dict)102        vm.parse_dict(vm_dict)103        session.commit()104    @dbsession_method105    def db_create_vms_onlynew(self, session, vm_dicts):106        if vm_dicts:107            rl = []108            for vm_dict in vm_dicts:109                vm = self.dbutils_get_vm_model(session, vm_dict=vm_dict)110                if vm:111                    # raise ExistsException('VM is exists.')112                    pass113            for vm_dict in vm_dicts:114                vm_rl = self.dbutils_create_vm_model(session, vm_dict)115                rl.append(vm_rl)116            session.commit()117            return rl118    def dbutils_drop_vm_model(self, session, vm_dict):119        vm = self.dbutils_get_vm_model(session, vm_dict=vm_dict)120        if vm:121            session.delete(vm)122    def dbutils_create_vm_model(self, session, vm_dict):123        vm_dict = self.patch_vm(vm_dict)124        exists_vm = self.dbutils_get_vm_model(125            session, vm_dict=vm_dict)126        if exists_vm:127            exists_vm.parse_dict(vm_dict)128        else:129            vm = models.Instance()130            vm.parse_dict(vm_dict)131            session.add(vm)132        # vm_dict is already patched by info133        return vm_dict134    def dbutils_get_vm_model(self, session, id=None, instance_id=None, vm_dict=None):135        if vm_dict and not id and not instance_id:136            return self.dbutils_get_vm_model(session, id=vm_dict.get('id', None),137                                             instance_id=vm_dict.get('instance_id', None))138        if id:139            instance = session.query(models.Instance).filter(140                models.Instance.id == id).first()141            return instance142        elif instance_id:143            instance = session.query(models.Instance).filter(144                models.Instance.instance_id == instance_id).first()145            return instance146    def dbutils_get_group_model(self, session, id=None, group_id=None, group_dict=None):147        if group_dict and not id and not group_id:148            return self.dbutils_get_group_model(session, id=group_dict.get('id', None),149                                                group_id=group_dict.get('group_id', None))150        if id:151            group = session.query(models.Group).filter(152                models.Group.id == id).first()153            return group154        elif group_id:155            group = session.query(models.Group).filter(156                models.Group.group_id == group_id).first()157            return group158    def dbutils_get_groups(self, session, user_id):159        groups = session.query(models.Group).filter(models.Group.user_id == user_id).all()160        return groups161    def dbutils_get_groups_port(self, session, user_id):162        groups = self.dbutils_get_groups(session, user_id)163        ports = [int(g.proxy_url.split(':')[-1]) for g in groups]164        return ports165    def dbutils_get_group_vms(self, session, group_id):166        vms = session.query(models.Instance).filter(models.Instance.group_id == group_id)...

Automation Testing Tutorials

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