Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _check_host_capabilities method in lisa

Best Python code snippet using lisa_python

platform.py

Source:platform.py

...164            # Rectify the general node capabilities with this node's specific165            # requirements.166            node_requirement = node_space.generate_min_capability(nodes_capabilities)167            nodes_requirement.append(node_requirement)168        if not self._check_host_capabilities(nodes_requirement, host_capabilities, log):169            return False170        environment.runbook.nodes_requirement = nodes_requirement171        return True172    def _get_host_capabilities(173        self, lv_conn: libvirt.virConnect, log: Logger174    ) -> _HostCapabilities:175        host_capabilities = _HostCapabilities()176        capabilities_xml_str = lv_conn.getCapabilities()177        capabilities_xml = ET.fromstring(capabilities_xml_str)178        host_xml = capabilities_xml.find("host")179        assert host_xml180        topology_xml = host_xml.find("topology")181        assert topology_xml182        cells_xml = topology_xml.find("cells")183        assert cells_xml184        for cell in cells_xml.findall("cell"):185            cpus_xml = cell.find("cpus")186            assert cpus_xml187            host_capabilities.core_count += int(cpus_xml.attrib["num"])188        # Get free memory.189        # Include the disk cache size, as it will be freed if memory becomes limited.190        memory_stats = lv_conn.getMemoryStats(libvirt.VIR_NODE_MEMORY_STATS_ALL_CELLS)191        host_capabilities.free_memory_kib = (192            memory_stats[libvirt.VIR_NODE_MEMORY_STATS_FREE]193            + memory_stats[libvirt.VIR_NODE_MEMORY_STATS_CACHED]194        )195        log.debug(196            f"QEMU host: "197            f"CPU Cores = {host_capabilities.core_count}, "198            f"Free Memory = {host_capabilities.free_memory_kib} KiB"199        )200        return host_capabilities201    # Create the set of capabilities that are generally supported on QEMU nodes.202    def _create_node_capabilities(203        self, host_capabilities: _HostCapabilities204    ) -> schema.NodeSpace:205        node_capabilities = schema.NodeSpace()206        node_capabilities.name = "QEMU"207        node_capabilities.node_count = 1208        node_capabilities.core_count = search_space.IntRange(209            min=1, max=host_capabilities.core_count210        )211        node_capabilities.disk = schema.DiskOptionSettings(212            data_disk_count=search_space.IntRange(min=0),213            data_disk_size=search_space.IntRange(min=1),214        )215        node_capabilities.network_interface = schema.NetworkInterfaceOptionSettings()216        node_capabilities.network_interface.max_nic_count = 1217        node_capabilities.network_interface.nic_count = 1218        node_capabilities.gpu_count = 0219        node_capabilities.features = search_space.SetSpace[schema.FeatureSettings](220            is_allow_set=True,221            items=[222                schema.FeatureSettings.create(SerialConsole.name()),223            ],224        )225        return node_capabilities226    # Check that the VM requirements can be fulfilled by the host.227    def _check_host_capabilities(228        self,229        nodes_requirements: List[schema.NodeSpace],230        host_capabilities: _HostCapabilities,231        log: Logger,232    ) -> bool:233        total_required_memory_mib = 0234        for node_requirements in nodes_requirements:235            # Calculate the total amount of memory required for all the VMs.236            assert isinstance(node_requirements.memory_mb, int)237            total_required_memory_mib += node_requirements.memory_mb238        # Ensure host has enough memory for all the VMs.239        total_required_memory_kib = total_required_memory_mib * 1024240        if total_required_memory_kib > host_capabilities.free_memory_kib:241            log.error(...

localhostcloud.py

Source:localhostcloud.py

...158            self.log.error("Could not create configuration drive for metadata: %s", error)159            return -1160        #Check that host can run requested vm161        instance = None162        host_cap = self._check_host_capabilities(conn)163        self.log.debug(host_cap)164        if job.request_cpus > (host_cap['vcpus'] - host_cap['activeCpus']):165            self.log.error("Host doesn't have available vcpus to boot VM")166            return -1167        if job.request_ram > host_cap['freeMemory']:168            self.log.error("Host doesn't have available ram to boot VM")169            return -1170        mem = job.request_ram + 1000171        #create domain def and launch domain172        virt_call = "virt-install --name "+hostname+" --network="+network.name()+ \173                    " --print-xml --dry-run -r "+str(mem)+" --disk path="+path+ \174                    ",sparse=true --disk path="+config_tmp+\175                    "/config.iso,device=cdrom --import --serial file,path="+config_tmp+\176                    "/boot-log --vcpus "+str(job.request_cpus)177        self.log.debug(virt_call)178        image_xml = subprocess.check_output(virt_call, shell=True)179        self.log.debug(type(image_xml))180        instance = conn.createXML(image_xml.decode(), 0)181        if instance is None:182            self.log.error("Failed to create domain from xml definition")183            return -1184        else:185            self.log.debug("New Image request successful.")186        """187        if instance:188#           new_vm = VM(vmid=instance.ID(), hostname=hostname)189#           self.vms[instance.ID()] = new_vm190#           engine = self._get_db_engine()191#           Base = automap_base()192#           Base.prepare(engine, reflect=True)193#           db_session = Session(engine)194#           vmobj = Base.classes.csv2_vms195            VM = self.config.db_map.classes.csv2_vms196            vm_dict = {197                'vmid': instance.ID(),198                'hostname': hostname,199                'status': 'New',200                'last_updated': int(time.time())201            }202            new_vm = VM(**vm_dict)203            self.config.db_open()204            self.config.db_session.merge(new_vm)205            self.config.db_close(commit=True)206        """207        self.log.debug('vm create')208        conn.close()209    def vm_destroy(self, vm):210        """211        Destroy VM on cloud.212        :param vm: ID of VM to destroy.213        """214        self.log.debug('vm destroy')215        conn = libvirt.open('qemu:///system')216        if conn is None:217            self.log.error("Failed to open connection with hypervisor")218            return -1219        try:220            instance = conn.lookupByName(vm.hostname)221            instance.destroy()222            del self.vms[vm.vmid]223        except libvirt.libvirtError as error:224            self.log.exception("VM %s not found on %s: Removing from CS: %s",225                               vm.hostname, self.name, error)226            del self.vms[vm.vmid]227        except Exception as ex:228            self.log.exception("Unhandled Exception trying to destroy VM: %s: %s",229                               vm.hostname, ex)230        conn.close()231        #Cleanup tmp dir and image copy232        try:233            subprocess.call('rm -f '+self.default_imagerepo+'/'+vm.image, shell=True)234        except Exception as ex:235            self.log.exception("Exception is deleting VM %s image %s: %s", vm.name, vm.image, ex)236        try:237            pipe = subprocess.Popen(['ls', '/tmp'], stdout=subprocess.PIPE)238            tmp = pipe.communicate()[0]239            for tmp_file in tmp.split():240                tmp_file = tmp_file.rstrip()241                if tmp_file.enswith(vm.hostname):242                    tmp_dir = '/tmp/'+tmp_file243                    subprocess.call('rm -rf '+tmp_dir, shell=True)244        except Exception as ex:245            self.log.exception("Could not remove the tmp directory for VM %s: %s", vm.hostname, ex)246    def vm_update(self):247        """I don't think this will be needed at all."""248        self.log.debug('vm update')249        conn = libvirt.open('qemu:///system')250        if conn is None:251            self.log.error("Failed to open connection to hypervisor")252            return -1253        try:254            listvms = conn.listAllDomains()255            for ovm in listvms:256                if owm.ID() == -1:257                    #domain not running258                    pass259                try:260                    self.vms[ovm.name()].status = ovm.state()261                except KeyError:262                    pass  # Will need to deal with unexpected vm still there by263                    # checking hostname and rebuilding vm obj if its a CS booted264                    # vm - probably have this as a config option since we265                    # sometimes remove  VMs intentionally266        except Exception as ex:267            self.log.exception(ex)268    def _find_network(self, netname):269        """270        Find network on OpenStack given a network name.271        :param netname: name of network to look for.272        :return: network object.273        """274        conn = libvirt.open('qemu:///system')275        if conn is None:276            self.log.error("Failed to open connection with hypervisor")277            return -1278        network = None279        try:280            network = conn.networkLookupByName(netname)281        except Exception as ex:282            self.log.exception("Unable to list networks for %s: Exception: %s", self.name, ex)283        return network284#   def _get_db_engine(self):285#       """286#       Get a connection to the database.287#       :return: db connection object.288#       """289#       return create_engine("mysql://" + csconfig.config.db_user + ":" +290#                            csconfig.config.db_password + "@" +291#                            csconfig.config.db_host + ":" +292#                            str(csconfig.config.db_port) + "/" +293#                            csconfig.config.db_name)294    def _generate_meta(self, name):295        instance_id = name296        host = name297        (fd, file_path) = tempfile.mkstemp(text=True)298        meta_info = {'instance-id':instance_id, 'local-hostname':host}299        with open(file_path, 'w') as yaml_file:300            yaml.dump(meta_info, yaml_file, default_flow_style=False)301        return file_path302    def _check_host_capabilities(self, conn):303        vcpus = conn.getMaxVcpus(None)304        node_info = conn.getInfo()305        mem = conn.getFreeMemory()306        cap = {'vcpus': vcpus, 'activeCpus': node_info[2], 'memory': node_info[1], 'freeMemory': mem}...

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.