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

How to use _get_node_information method in lisa

Best Python code snippet using lisa_python

platform_.py

Source:platform_.py

...516                    log.debug(517                        f"exception on deleting boot diagnostic container:"518                        f" {identifier}"519                    )520    def _get_node_information(self, node: Node) -> Dict[str, str]:521        information: Dict[str, Any] = {}522        node.log.debug("detecting lis version...")523        modinfo = node.tools[Modinfo]524        information["lis_version"] = modinfo.get_version("hv_vmbus")525        node.log.debug("detecting vm generation...")526        information[KEY_VM_GENERATION] = node.tools[VmGeneration].get_generation()527        node.log.debug(f"vm generation: {information[KEY_VM_GENERATION]}")528        return information529    def _get_kernel_version(self, node: Node) -> str:530        result: str = ""531        if not result and hasattr(node, ATTRIBUTE_FEATURES):532            # try to get kernel version in Azure. use it, when uname doesn't work533            node.log.debug("detecting kernel version from serial log...")534            serial_console = node.features[features.SerialConsole]535            result = serial_console.get_matched_str(KERNEL_VERSION_PATTERN)536        return result537    def _get_host_version(self, node: Node) -> str:538        result: str = ""539        try:540            if node.is_connected and node.is_posix:541                node.log.debug("detecting host version from dmesg...")542                dmesg = node.tools[Dmesg]543                result = get_matched_str(544                    dmesg.get_output(), HOST_VERSION_PATTERN, first_match=False545                )546        except Exception as identifier:547            # it happens on some error vms. Those error should be caught earlier in548            # test cases not here. So ignore any error here to collect information only.549            node.log.debug(f"error on run dmesg: {identifier}")550        # if not get, try again from serial console log.551        # skip if node is not initialized.552        if not result and hasattr(node, ATTRIBUTE_FEATURES):553            node.log.debug("detecting host version from serial log...")554            serial_console = node.features[features.SerialConsole]555            result = serial_console.get_matched_str(HOST_VERSION_PATTERN)556        return result557    def _get_wala_version(self, node: Node) -> str:558        result = ""559        try:560            if node.is_connected and node.is_posix:561                node.log.debug("detecting wala version from waagent...")562                waagent = node.tools[Waagent]563                result = waagent.get_version()564        except Exception as identifier:565            # it happens on some error vms. Those error should be caught earlier in566            # test cases not here. So ignore any error here to collect information only.567            node.log.debug(f"error on run waagent: {identifier}")568        if not result and hasattr(node, ATTRIBUTE_FEATURES):569            node.log.debug("detecting wala agent version from serial log...")570            serial_console = node.features[features.SerialConsole]571            result = serial_console.get_matched_str(WALA_VERSION_PATTERN)572        return result573    def _get_wala_distro_version(self, node: Node) -> str:574        result = "Unknown"575        try:576            if node.is_connected and node.is_posix:577                waagent = node.tools[Waagent]578                result = waagent.get_distro_version()579        except Exception as identifier:580            # it happens on some error vms. Those error should be caught earlier in581            # test cases not here. So ignore any error here to collect information only.582            node.log.debug(f"error on get waagent distro version: {identifier}")583        return result584    def _get_platform_information(self, environment: Environment) -> Dict[str, str]:585        result: Dict[str, str] = {}586        azure_runbook: AzurePlatformSchema = self.runbook.get_extended_runbook(587            AzurePlatformSchema588        )589        result[AZURE_RG_NAME_KEY] = get_environment_context(590            environment591        ).resource_group_name592        if azure_runbook.availability_set_properties:593            for (594                property_name,595                property_value,596            ) in azure_runbook.availability_set_properties.items():597                if property_name in [598                    "platformFaultDomainCount",599                    "platformUpdateDomainCount",600                ]:601                    continue602                if isinstance(property_value, dict):603                    for key, value in property_value.items():604                        if value:605                            result[key] = value606        if azure_runbook.availability_set_tags:607            for key, value in azure_runbook.availability_set_tags.items():608                if value:609                    result[key] = value610        if azure_runbook.vm_tags:611            for key, value in azure_runbook.vm_tags.items():612                if value:613                    result[key] = value614        return result615    def _get_environment_information(self, environment: Environment) -> Dict[str, str]:616        information: Dict[str, str] = {}617        node_runbook: Optional[AzureNodeSchema] = None618        if environment.nodes:619            node: Optional[Node] = environment.default_node620        else:621            node = None622        if node:623            node_runbook = node.capability.get_extended_runbook(AzureNodeSchema, AZURE)624            for key, method in self._environment_information_hooks.items():625                node.log.debug(f"detecting {key} ...")626                try:627                    value = method(node)628                    if value:629                        information[key] = value630                except Exception as identifier:631                    node.log.exception(f"error on get {key}.", exc_info=identifier)632            information.update(self._get_platform_information(environment))633            if node.is_connected and node.is_posix:634                information.update(self._get_node_information(node))635        elif environment.capability and environment.capability.nodes:636            # get deployment information, if failed on preparing phase637            node_space = environment.capability.nodes[0]638            node_runbook = node_space.get_extended_runbook(639                AzureNodeSchema, type_name=AZURE640            )641        if node_runbook:642            information["location"] = node_runbook.location643            information["vmsize"] = node_runbook.vm_size644            information["image"] = node_runbook.get_image_name()645        return information646    def _initialize(self, *args: Any, **kwargs: Any) -> None:647        # set needed environment variables for authentication648        azure_runbook: AzurePlatformSchema = self.runbook.get_extended_runbook(...

preprocess.py

Source:preprocess.py

...29        Returns30        -------31        matrix with shape [num_edges, num_edge_features]32        """33        self._get_node_information()34        if self.edge_features is None:35            # Create list of feature probabilities36            self.edge_features = [1] * len(self.edges[0])37            for ind, node in enumerate(self.number_of_edges_per_node):38                probability = 1 / node[1]39                for position in self.indices_of_edges_per_node[ind][1]:40                    self.edge_features[position] *= probability41    def _get_node_information(self):42        """43        Get information which nodes have more than one outgoing edge and the respective44        indices of the position in self.edge_features.45        """46        multi_edge_nodes = self.get_nodes_with_multiple_edges(47            edges=self.edges, outgoing=True48        )49        (50            self.number_of_edges_per_node,51            self.indices_of_edges_per_node,52        ) = self.get_number_of_edges_per_node(53            edges=self.edges, multi_edge_nodes=multi_edge_nodes, outgoing=True54        )55    @staticmethod...

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.