KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _fill_nodes_metadata method in lisa

Best Python code snippet using lisa_python
platform.py
Source:platform.py
...253        self._configure_nodes(environment, log)254        with libvirt.open(self.libvirt_conn_str) as lv_conn:255            try:256                self._create_nodes(environment, log, lv_conn)257                self._fill_nodes_metadata(environment, log, lv_conn)258            except Exception as ex:259                assert environment.platform260                if (261                    environment.platform.runbook.keep_environment262                    == constants.ENVIRONMENT_KEEP_NO263                ):264                    self._delete_nodes(environment, log)265                raise ex266    # Pre-determine all the nodes' properties, including the name of all the resouces267    # to be created. This makes it easier to cleanup everything after the test is268    # finished (or fails).269    def _configure_nodes(self, environment: Environment, log: Logger) -> None:270        # Generate a random name for the VMs.271        test_suffix = "".join(random.choice(string.ascii_uppercase) for _ in range(5))272        vm_name_prefix = f"lisa-{test_suffix}"273        self.vm_disks_dir = os.path.join(274            self.platform_runbook.hosts[0].lisa_working_dir, vm_name_prefix275        )276        assert environment.runbook.nodes_requirement277        for i, node_space in enumerate(environment.runbook.nodes_requirement):278            assert isinstance(279                node_space, schema.NodeSpace280            ), f"actual: {type(node_space)}"281            node_runbook: BaseLibvirtNodeSchema = node_space.get_extended_runbook(282                self.__node_runbook_type(), type_name=type(self).type_name()283            )284            if not os.path.exists(node_runbook.disk_img):285                raise LisaException(f"file does not exist: {node_runbook.disk_img}")286            node = environment.create_node_from_requirement(node_space)287            self._configure_node(288                node,289                i,290                node_space,291                node_runbook,292                vm_name_prefix,293            )294    def _configure_node(295        self,296        node: Node,297        node_idx: int,298        node_space: schema.NodeSpace,299        node_runbook: BaseLibvirtNodeSchema,300        vm_name_prefix: str,301    ) -> None:302        node_context = get_node_context(node)303        if (304            not node_runbook.firmware_type305            or node_runbook.firmware_type == FIRMWARE_TYPE_UEFI306        ):307            node_context.use_bios_firmware = False308        elif node_runbook.firmware_type == FIRMWARE_TYPE_BIOS:309            node_context.use_bios_firmware = True310            if node_runbook.enable_secure_boot:311                raise LisaException("Secure-boot requires UEFI firmware.")312        else:313            raise LisaException(314                f"Unknown node firmware type: {node_runbook.firmware_type}."315                f"Expecting either {FIRMWARE_TYPE_UEFI} or {FIRMWARE_TYPE_BIOS}."316            )317        node_context.machine_type = node_runbook.machine_type or None318        node_context.enable_secure_boot = node_runbook.enable_secure_boot319        node_context.vm_name = f"{vm_name_prefix}-{node_idx}"320        if not node.name:321            node.name = node_context.vm_name322        node_context.cloud_init_file_path = os.path.join(323            self.vm_disks_dir, f"{node_context.vm_name}-cloud-init.iso"324        )325        if self.host_node.is_remote:326            node_context.os_disk_source_file_path = node_runbook.disk_img327            node_context.os_disk_base_file_path = os.path.join(328                self.vm_disks_dir, os.path.basename(node_runbook.disk_img)329            )330        else:331            node_context.os_disk_base_file_path = node_runbook.disk_img332        node_context.os_disk_base_file_fmt = DiskImageFormat(333            node_runbook.disk_img_format334        )335        node_context.os_disk_file_path = os.path.join(336            self.vm_disks_dir, f"{node_context.vm_name}-os.qcow2"337        )338        node_context.console_log_file_path = str(339            node.local_log_path / "qemu-console.log"340        )341        # Read extra cloud-init data.342        extra_user_data = (343            node_runbook.cloud_init and node_runbook.cloud_init.extra_user_data344        )345        if extra_user_data:346            node_context.extra_cloud_init_user_data = []347            if isinstance(extra_user_data, str):348                extra_user_data = [extra_user_data]349            for relative_file_path in extra_user_data:350                if not relative_file_path:351                    continue352                file_path = constants.RUNBOOK_PATH.joinpath(relative_file_path)353                with open(file_path, "r") as file:354                    node_context.extra_cloud_init_user_data.append(yaml.safe_load(file))355        # Configure data disks.356        if node_space.disk:357            assert isinstance(358                node_space.disk.data_disk_count, int359            ), f"actual: {type(node_space.disk.data_disk_count)}"360            assert isinstance(361                node_space.disk.data_disk_size, int362            ), f"actual: {type(node_space.disk.data_disk_size)}"363            for i in range(node_space.disk.data_disk_count):364                data_disk = DataDiskContext()365                data_disk.file_path = os.path.join(366                    self.vm_disks_dir, f"{node_context.vm_name}-data-{i}.qcow2"367                )368                data_disk.size_gib = node_space.disk.data_disk_size369                node_context.data_disks.append(data_disk)370    def _create_domain_and_attach_logger(371        self,372        libvirt_conn: libvirt.virConnect,373        node_context: NodeContext,374    ) -> None:375        # Start the VM in the paused state.376        # This gives the console logger a chance to connect before the VM starts377        # for real.378        assert node_context.domain379        node_context.domain.createWithFlags(libvirt.VIR_DOMAIN_START_PAUSED)380        # Attach the console logger381        node_context.console_logger = QemuConsoleLogger()382        node_context.console_logger.attach(383            libvirt_conn, node_context.domain, node_context.console_log_file_path384        )385        # Start the VM.386        node_context.domain.resume()387    # Create all the VMs.388    def _create_nodes(389        self,390        environment: Environment,391        log: Logger,392        lv_conn: libvirt.virConnect,393    ) -> None:394        self.host_node.shell.mkdir(Path(self.vm_disks_dir), exist_ok=True)395        for node in environment.nodes.list():396            node_context = get_node_context(node)397            self._create_node(398                node,399                node_context,400                environment,401                log,402                lv_conn,403            )404    def _create_node(405        self,406        node: Node,407        node_context: NodeContext,408        environment: Environment,409        log: Logger,410        lv_conn: libvirt.virConnect,411    ) -> None:412        # Create required directories and copy the required files to the host413        # node.414        if node_context.os_disk_source_file_path:415            source_exists = self.host_node.tools[Ls].path_exists(416                path=node_context.os_disk_base_file_path, sudo=True417            )418            if not source_exists:419                self.host_node.shell.copy(420                    Path(node_context.os_disk_source_file_path),421                    Path(node_context.os_disk_base_file_path),422                )423        # Create cloud-init ISO file.424        self._create_node_cloud_init_iso(environment, log, node)425        # Create OS disk from the provided image.426        self._create_node_os_disk(environment, log, node)427        # Create data disks428        self._create_node_data_disks(node)429        # Create libvirt domain (i.e. VM).430        xml = self._create_node_domain_xml(environment, log, node, lv_conn)431        node_context.domain = lv_conn.defineXML(xml)432        self._create_domain_and_attach_logger(433            lv_conn,434            node_context,435        )436    # Delete all the VMs.437    def _delete_nodes(self, environment: Environment, log: Logger) -> None:438        # Delete nodes.439        for node in environment.nodes.list():440            self._delete_node(node, log)441        # Delete VM disks directory.442        try:443            self.host_node.shell.remove(Path(self.vm_disks_dir), True)444        except Exception as ex:445            log.warning(f"Failed to delete VM files directory: {ex}")446    def _delete_node_watchdog_callback(self) -> None:447        print("VM delete watchdog timer fired.\n", file=sys.__stderr__)448        faulthandler.dump_traceback(file=sys.__stderr__, all_threads=True)449        os._exit(1)450    def _delete_node(self, node: Node, log: Logger) -> None:451        node_context = get_node_context(node)452        watchdog = Timer(60.0, self._delete_node_watchdog_callback)453        watchdog.start()454        # Stop the VM.455        if node_context.domain:456            log.debug(f"Stop VM: {node_context.vm_name}")457            try:458                # In the libvirt API, "destroy" means "stop".459                node_context.domain.destroy()460            except libvirt.libvirtError as ex:461                log.warning(f"VM stop failed. {ex}")462        # Wait for console log to close.463        # Note: libvirt can deadlock if you try to undefine the VM while the stream464        # is trying to close.465        if node_context.console_logger:466            log.debug(f"Close VM console log: {node_context.vm_name}")467            node_context.console_logger.close()468            node_context.console_logger = None469        # Undefine the VM.470        if node_context.domain:471            log.debug(f"Delete VM: {node_context.vm_name}")472            try:473                node_context.domain.undefineFlags(self._get_domain_undefine_flags())474            except libvirt.libvirtError as ex:475                log.warning(f"VM delete failed. {ex}")476            node_context.domain = None477        watchdog.cancel()478    def _get_domain_undefine_flags(self) -> int:479        return int(480            libvirt.VIR_DOMAIN_UNDEFINE_MANAGED_SAVE481            | libvirt.VIR_DOMAIN_UNDEFINE_SNAPSHOTS_METADATA482            | libvirt.VIR_DOMAIN_UNDEFINE_NVRAM483            | libvirt.VIR_DOMAIN_UNDEFINE_CHECKPOINTS_METADATA484        )485    def _stop_port_forwarding(self, environment: Environment, log: Logger) -> None:486        log.debug(f"Clearing port forwarding rules for environment {environment.name}")487        environment_context = get_environment_context(environment)488        for (port, address) in environment_context.port_forwarding_list:489            self.host_node.tools[Iptables].stop_forwarding(port, address, 22)490    # Retrieve the VMs' dynamic properties (e.g. IP address).491    def _fill_nodes_metadata(492        self, environment: Environment, log: Logger, lv_conn: libvirt.virConnect493    ) -> None:494        environment_context = get_environment_context(environment)495        # Give all the VMs some time to boot and then acquire an IP address.496        timeout = time.time() + environment_context.network_boot_timeout497        if self.host_node.is_remote:498            remote_node = cast(RemoteNode, self.host_node)499            conn_info = remote_node.connection_info500            address = conn_info[constants.ENVIRONMENTS_NODES_REMOTE_ADDRESS]501        for node in environment.nodes.list():502            assert isinstance(node, RemoteNode)503            # Get the VM's IP address.504            local_address = self._get_node_ip_address(505                environment, log, lv_conn, node, timeout...
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.