How to use __resolve_package_name method in lisa

Best Python code snippet using lisa_python

operating_system.py

Source:operating_system.py

...286        """287        Query if a package/tool is installed on the node.288        Return Value - bool289        """290        package_name = self.__resolve_package_name(package)291        return self._package_exists(package_name)292    def is_package_in_repo(self, package: Union[str, Tool, Type[Tool]]) -> bool:293        """294        Query if a package/tool exists in the repo295        Return Value - bool296        """297        package_name = self.__resolve_package_name(package)298        if self._first_time_installation:299            self._initialize_package_installation()300            self._first_time_installation = False301        return self._is_package_in_repo(package_name)302    def update_packages(303        self,304        packages: Union[str, Tool, Type[Tool], Sequence[Union[str, Tool, Type[Tool]]]],305    ) -> None:306        package_names = self._get_package_list(packages)307        self._update_packages(package_names)308    def capture_system_information(self, saved_path: Path) -> None:309        # avoid to involve node, it's ok if some command doesn't exist.310        self._node.execute("uname -vrmo").save_stdout_to_file(saved_path / "uname.txt")311        self._node.execute(312            "uptime -s || last reboot -F | head -1 | awk '{print $9,$6,$7,$8}'",313            shell=True,314        ).save_stdout_to_file(saved_path / "uptime.txt")315        self._node.execute("modinfo hv_netvsc").save_stdout_to_file(316            saved_path / "modinfo-hv_netvsc.txt"317        )318        try:319            self._node.shell.copy_back(320                self._node.get_pure_path("/etc/os-release"),321                saved_path / "os-release.txt",322            )323        except FileNotFoundError:324            self._log.debug("File /etc/os-release doesn't exist.")325    def get_package_information(326        self, package_name: str, use_cached: bool = True327    ) -> VersionInfo:328        found = self._packages.get(package_name, None)329        if found and use_cached:330            return found331        return self._get_package_information(package_name)332    def get_repositories(self) -> List[RepositoryInfo]:333        raise NotImplementedError("get_repositories is not implemented")334    def _process_extra_package_args(self, extra_args: Optional[List[str]]) -> str:335        if extra_args:336            add_args = " ".join(extra_args)337        else:338            add_args = ""339        return add_args340    def _install_packages(341        self,342        packages: List[str],343        signed: bool = True,344        timeout: int = 600,345        extra_args: Optional[List[str]] = None,346    ) -> None:347        raise NotImplementedError()348    def _update_packages(self, packages: Optional[List[str]] = None) -> None:349        raise NotImplementedError()350    def _package_exists(self, package: str) -> bool:351        raise NotImplementedError()352    def _is_package_in_repo(self, package: str) -> bool:353        raise NotImplementedError()354    def _initialize_package_installation(self) -> None:355        # sub os can override it, but it's optional356        pass357    def _get_package_information(self, package_name: str) -> VersionInfo:358        raise NotImplementedError()359    def _get_version_info_from_named_regex_match(360        self, package_name: str, named_matches: Match[str]361    ) -> VersionInfo:362        essential_matches = ["major", "minor", "build"]363        # verify all essential keys are in our match dict364        assert_that(365            all(map(lambda x: x in named_matches.groupdict().keys(), essential_matches))366        ).described_as(367            "VersionInfo fetch could not identify all required parameters."368        ).is_true()369        # fill in 'patch' version if it's missing370        patch_match = named_matches.group("patch")371        if not patch_match:372            patch_match = "0"373        major_match = named_matches.group("major")374        minor_match = named_matches.group("minor")375        build_match = named_matches.group("build")376        major, minor, patch = map(377            int,378            [major_match, minor_match, patch_match],379        )380        build_match = named_matches.group("build")381        self._node.log.debug(382            f"Found {package_name} version "383            f"{major_match}.{minor_match}.{patch_match}-{build_match}"384        )385        return VersionInfo(major, minor, patch, build=build_match)386    def _cache_and_return_version_info(387        self, package_name: str, info: VersionInfo388    ) -> VersionInfo:389        self._packages[package_name] = info390        return info391    def _get_information(self) -> OsInformation:392        # try to set version info from /etc/os-release.393        cat = self._node.tools[Cat]394        cmd_result = cat.run(395            "/etc/os-release",396            expected_exit_code=0,397            expected_exit_code_failure_message="error on get os information",398        )399        vendor: str = ""400        release: str = ""401        codename: str = ""402        full_version: str = ""403        for row in cmd_result.stdout.splitlines():404            os_release_info = self._os_info_pattern.match(row)405            if not os_release_info:406                continue407            if os_release_info.group("name") == "NAME":408                vendor = os_release_info.group("value")409            elif os_release_info.group("name") == "VERSION_ID":410                release = os_release_info.group("value")411            elif os_release_info.group("name") == "VERSION":412                codename = get_matched_str(413                    os_release_info.group("value"),414                    self._distro_codename_pattern,415                )416            elif os_release_info.group("name") == "PRETTY_NAME":417                full_version = os_release_info.group("value")418        if vendor == "":419            raise LisaException("OS vendor information not found")420        if release == "":421            raise LisaException("OS release information not found")422        information = OsInformation(423            version=self._parse_version(release),424            vendor=vendor,425            release=release,426            codename=codename,427            full_version=full_version,428        )429        return information430    def _get_package_list(431        self,432        packages: Union[433            str,434            Tool,435            Type[Tool],436            Sequence[Union[str, Tool, Type[Tool]]],437        ],438    ) -> List[str]:439        package_names: List[str] = []440        if isinstance(packages, (str, Tool, type)):441            packages = [packages]442        package_names = [self.__resolve_package_name(item) for item in packages]443        if self._first_time_installation:444            self._first_time_installation = False445            self._initialize_package_installation()446        return package_names447    def _install_package_from_url(448        self,449        package_url: str,450        package_name: str = "",451        signed: bool = True,452        timeout: int = 600,453    ) -> None:454        """455        Used if the package to be installed needs to be downloaded from a url first.456        """457        # when package is URL, download the package first at the working path.458        wget_tool = self._node.tools[Wget]459        pkg = wget_tool.get(package_url, str(self._node.working_path), package_name)460        self.install_packages(pkg, signed, timeout)461    def wait_running_process(self, process_name: str, timeout: int = 5) -> None:462        # by default, wait for 5 minutes463        timeout = 60 * timeout464        timer = create_timer()465        while timeout > timer.elapsed(False):466            cmd_result = self._node.execute(f"pidof {process_name}")467            if cmd_result.exit_code == 1:468                # not found dpkg or zypper process, it's ok to exit.469                break470            time.sleep(1)471        if timeout < timer.elapsed():472            raise Exception(f"timeout to wait previous {process_name} process stop.")473    def __resolve_package_name(self, package: Union[str, Tool, Type[Tool]]) -> str:474        """475        A package can be a string or a tool or a type of tool.476        Resolve it to a standard package_name so it can be installed.477        """478        if isinstance(package, str):479            package_name = package480        elif isinstance(package, Tool):481            package_name = package.package_name482        else:483            assert isinstance(package, type), f"actual:{type(package)}"484            # Create a temp object, it doesn't query.485            # So they can be queried together.486            tool = package.create(self._node)487            package_name = tool.package_name...

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.