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

How to use wait_running_package_process method in lisa

Best Python code snippet using lisa_python

operating_system.py

Source:operating_system.py

...593        version_info = self._get_version_info_from_named_regex_match(594            package_name, match595        )596        return self._cache_and_return_version_info(package_name, version_info)597    def wait_running_package_process(self) -> None:598        is_first_time: bool = True599        # wait for 10 minutes600        timeout = 60 * 10601        timer = create_timer()602        while timeout > timer.elapsed(False):603            # fix the dpkg, in case it's broken.604            dpkg_result = self._node.execute(605                "dpkg --force-all --configure -a", sudo=True606            )607            pidof_result = self._node.execute("pidof dpkg dpkg-deb")608            if dpkg_result.exit_code == 0 and pidof_result.exit_code == 1:609                # not found dpkg process, it's ok to exit.610                break611            if is_first_time:612                is_first_time = False613                self._log.debug("found system dpkg process, waiting it...")614            time.sleep(1)615        if timeout < timer.elapsed():616            raise Exception("timeout to wait previous dpkg process stop.")617    def get_repositories(self) -> List[RepositoryInfo]:618        self._initialize_package_installation()619        repo_list_str = self._node.execute("apt-get update", sudo=True).stdout620        repositories: List[RepositoryInfo] = []621        for line in repo_list_str.splitlines():622            matched = self._debian_repository_info_pattern.search(line)623            if matched:624                repositories.append(625                    DebianRepositoryInfo(626                        name=matched.group("name"),627                        status=matched.group("status"),628                        id=matched.group("id"),629                        uri=matched.group("uri"),630                        metadata=matched.group("metadata"),631                    )632                )633        return repositories634    @retry(tries=10, delay=5)635    def add_repository(636        self,637        repo: str,638        no_gpgcheck: bool = True,639        repo_name: Optional[str] = None,640        keys_location: Optional[List[str]] = None,641    ) -> None:642        if keys_location:643            for key_location in keys_location:644                wget = self._node.tools[Wget]645                key_file_path = wget.get(646                    url=key_location,647                    file_path=str(self._node.working_path),648                    force_run=True,649                )650                self._node.execute(651                    cmd=f"apt-key add {key_file_path}",652                    sudo=True,653                    expected_exit_code=0,654                    expected_exit_code_failure_message="fail to add apt key",655                )656        # This command will trigger apt update too, so it doesn't need to update657        # repos again.658        self._node.execute(659            cmd=f'apt-add-repository -y "{repo}"',660            sudo=True,661            expected_exit_code=0,662            expected_exit_code_failure_message="fail to add repository",663        )664        # apt update will not be triggered on Debian during add repo665        if type(self._node.os) == Debian:666            self._node.execute("apt-get update", sudo=True)667    @retry(tries=10, delay=5)668    def _initialize_package_installation(self) -> None:669        # wait running system package process.670        self.wait_running_package_process()671        result = self._node.execute("apt-get update", sudo=True)672        if self._repo_not_exist_pattern.search(result.stdout):673            raise RepoNotExistException(self._node.os)674        result.assert_exit_code(message="\n".join(self.get_apt_error(result.stdout)))675    @retry(tries=10, delay=5)676    def _install_packages(677        self,678        packages: List[str],679        signed: bool = True,680        timeout: int = 600,681        extra_args: Optional[List[str]] = None,682    ) -> None:683        file_packages = []684        for index, package in enumerate(packages):685            if package.endswith(".deb"):686                # If the package is a .deb file then it would first need to be unpacked.687                # using dpkg command before installing it like other packages.688                file_packages.append(package)689                package = Path(package).stem690                packages[index] = package691        add_args = self._process_extra_package_args(extra_args)692        command = (693            f"DEBIAN_FRONTEND=noninteractive apt-get {add_args} "694            f"-y install {' '.join(packages)}"695        )696        if not signed:697            command += " --allow-unauthenticated"698        self.wait_running_package_process()699        if file_packages:700            self._node.execute(701                f"dpkg -i {' '.join(file_packages)}", sudo=True, timeout=timeout702            )703            # after install package, need update the repo704            self._initialize_package_installation()705        install_result = self._node.execute(706            command, shell=True, sudo=True, timeout=timeout707        )708        # get error lines.709        install_result.assert_exit_code(710            0,711            f"Failed to install {packages}, "712            f"please check the package name and repo are correct or not.\n"...

kernel_installer.py

Source:kernel_installer.py

...174        repo_entry = (175            f"deb {self.repo_url} {version_name} "176            f"restricted main multiverse universe"177        )178        ubuntu.wait_running_package_process()179        result = node.execute(f'add-apt-repository -y "{repo_entry}"', sudo=True)180        result.assert_exit_code(181            0, "failed on add repo\n\n".join(ubuntu.get_apt_error(result.stdout))182        )183        full_package_name = f"{runbook.source}/{version_name}"184        self._log.info(f"installing kernel package: {full_package_name}")185        ubuntu.install_packages(full_package_name)186        kernel_version = self._get_kernel_version(runbook.source, node)187        return kernel_version188    def _get_kernel_version(self, source: str, node: Node) -> str:189        # get kernel version from apt packages190        # linux-azure-edge/focal-proposed,now 5.11.0.1011.11~20.04.10 amd64 [installed]191        # output: 5.11.0.1011192        # linux-image-4.18.0-1025-azure/bionic-updates,bionic-security,now...

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.