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

How to use _get_package_information method in lisa

Best Python code snippet using lisa_python

operating_system.py

Source:operating_system.py

...327    ) -> 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_name488        return package_name489class BSD(Posix):490    ...491class MacOS(Posix):492    @classmethod493    def name_pattern(cls) -> Pattern[str]:494        return re.compile("^Darwin$")495class Linux(Posix):496    ...497class CoreOs(Linux):498    @classmethod499    def name_pattern(cls) -> Pattern[str]:500        return re.compile("^coreos|Flatcar|flatcar$")501@dataclass502# `apt-get update` repolist is of the form `<status>:<id> <uri> <name> <metadata>`503# Example:504# Get:5 http://azure.archive.ubuntu.com/ubuntu focal-updates/main amd64 Packages [1298 kB] # noqa: E501505class DebianRepositoryInfo(RepositoryInfo):506    # status for the repository. Examples: `Hit`, `Get`507    status: str508    # id for the repository. Examples : 1, 2509    id: str510    # uri for the repository. Example: `http://azure.archive.ubuntu.com/ubuntu`511    uri: str512    # metadata for the repository. Example: `amd64 Packages [1298 kB]`513    metadata: str514class Debian(Linux):515    # Get:5 http://azure.archive.ubuntu.com/ubuntu focal-updates/main amd64 Packages [1298 kB] # noqa: E501516    _debian_repository_info_pattern = re.compile(517        r"(?P<status>\S+):(?P<id>\d+)\s+(?P<uri>\S+)\s+(?P<name>\S+)"518        r"\s+(?P<metadata>.*)\s*"519    )520    """ Package: dpdk521        Version: 20.11.3-0ubuntu1~backport20.04-202111041420~ubuntu20.04.1522        Version: 1:2.25.1-1ubuntu3.2523    """524    _debian_package_information_regex = re.compile(525        r"Package: ([a-zA-Z0-9:_\-\.]+)\r?\n"  # package name group526        r"Version: ([a-zA-Z0-9:_\-\.~+]+)\r?\n"  # version number group527    )528    _debian_version_splitter_regex = re.compile(529        r"([0-9]+:)?"  # some examples have a mystery number followed by a ':' (git)530        r"(?P<major>[0-9]+)\."  # major531        r"(?P<minor>[0-9]+)\."  # minor532        r"(?P<patch>[0-9]+)"  # patch533        r"-(?P<build>[a-zA-Z0-9-_\.~+]+)"  # build534    )535    # apt-cache policy git536    # git:537    #   Installed: 1:2.17.1-1ubuntu0.9538    #   Candidate: 1:2.17.1-1ubuntu0.9539    #   Version table:540    #  *** 1:2.17.1-1ubuntu0.9 500541    #         500 http://azure.archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages # noqa: E501542    #         500 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages # noqa: E501543    #         100 /var/lib/dpkg/status544    #      1:2.17.0-1ubuntu1 500545    #         500 http://azure.archive.ubuntu.com/ubuntu bionic/main amd64 Packages546    # apt-cache policy mock547    # mock:548    #   Installed: (none)549    #   Candidate: 1.3.2-2550    #   Version table:551    #      1.3.2-2 500552    #         500 http://azure.archive.ubuntu.com/ubuntu bionic/universe amd64 Packages # noqa: E501553    # apt-cache policy test554    # N: Unable to locate package test555    _package_candidate_pattern = re.compile(556        r"([\w\W]*?)(Candidate: \(none\)|Unable to locate package.*)", re.M557    )558    # E: The repository 'http://azure.archive.ubuntu.com/ubuntu groovy Release' no longer has a Release file. # noqa: E501559    _repo_not_exist_pattern = re.compile("does not have a Release file", re.M)560    @classmethod561    def name_pattern(cls) -> Pattern[str]:562        return re.compile("^debian|Forcepoint|Kali$")563    def get_apt_error(self, stdout: str) -> List[str]:564        error_lines: List[str] = []565        for line in stdout.splitlines(keepends=False):566            if line.startswith("E: "):567                error_lines.append(line)568        return error_lines569    def _get_package_information(self, package_name: str) -> VersionInfo:570        # run update of package info571        apt_info = self._node.execute(572            f"apt show {package_name}",573            expected_exit_code=0,574            expected_exit_code_failure_message=(575                f"Could not find package information for package {package_name}"576            ),577        )578        match = self._debian_package_information_regex.search(apt_info.stdout)579        if not match:580            raise LisaException(581                "Package information parsing could not find regex match "582                f" for {package_name} using regex "583                f"{self._debian_package_information_regex.pattern}"584            )585        version_str = match.group(2)586        match = self._debian_version_splitter_regex.search(version_str)587        if not match:588            raise LisaException(589                f"Could not parse version info: {version_str} "590                "for package {package_name}"591            )592        self._node.log.debug(f"Attempting to parse version string: {version_str}")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"713            + "\n".join(self.get_apt_error(install_result.stdout))714            + "\n",715        )716    def _package_exists(self, package: str) -> bool:717        command = "dpkg --get-selections"718        result = self._node.execute(command, sudo=True, shell=True)719        package_pattern = re.compile(f"{package}([ \t]+)install")720        # Not installed package not shown in the output721        # Uninstall package will show as deinstall722        # vim                                             deinstall723        # vim-common                                      install724        if len(list(filter(package_pattern.match, result.stdout.splitlines()))) == 1:725            return True726        return False727    def _is_package_in_repo(self, package: str) -> bool:728        command = f"apt-cache policy {package}"729        result = self._node.execute(command, sudo=True, shell=True)730        matched = get_matched_str(result.stdout, self._package_candidate_pattern)731        if matched:732            return False733        return True734    def _get_information(self) -> OsInformation:735        # try to set version info from /etc/os-release.736        cat = self._node.tools[Cat]737        cmd_result = cat.run(738            "/etc/os-release",739            expected_exit_code=0,740            expected_exit_code_failure_message="error on get os information",741        )742        vendor: str = ""743        release: str = ""744        codename: str = ""745        full_version: str = ""746        for row in cmd_result.stdout.splitlines():747            os_release_info = super()._os_info_pattern.match(row)748            if not os_release_info:749                continue750            if os_release_info.group("name") == "NAME":751                vendor = os_release_info.group("value")752            elif os_release_info.group("name") == "VERSION":753                codename = get_matched_str(754                    os_release_info.group("value"),755                    super()._distro_codename_pattern,756                )757            elif os_release_info.group("name") == "PRETTY_NAME":758                full_version = os_release_info.group("value")759        # version return from /etc/os-release is integer in debian760        # so get the precise version from /etc/debian_version761        # e.g.762        # marketplace image - credativ debian 9-backports 9.20190313.0763        # version from /etc/os-release is 9764        # version from /etc/debian_version is 9.8765        # marketplace image - debian debian-10 10-backports-gen2 0.20210201.535766        # version from /etc/os-release is 10767        # version from /etc/debian_version is 10.7768        cmd_result = cat.run(769            "/etc/debian_version",770            expected_exit_code=0,771            expected_exit_code_failure_message="error on get debian version",772        )773        release = cmd_result.stdout774        if vendor == "":775            raise LisaException("OS vendor information not found")776        if release == "":777            raise LisaException("OS release information not found")778        information = OsInformation(779            version=self._parse_version(release),780            vendor=vendor,781            release=release,782            codename=codename,783            full_version=full_version,784        )785        return information786    def _update_packages(self, packages: Optional[List[str]] = None) -> None:787        command = (788            "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y "789            '-o Dpkg::Options::="--force-confdef" -o Dpkg::Options::="--force-confold" '790        )791        if packages:792            command += " ".join(packages)793        self._node.execute(command, sudo=True, timeout=3600)794class Ubuntu(Debian):795    __lsb_os_info_pattern = re.compile(796        r"^(?P<name>.*):(\s+)(?P<value>.*?)?$", re.MULTILINE797    )798    # gnulinux-5.11.0-1011-azure-advanced-3fdd2548-1430-450b-b16d-9191404598fb799    # prefix: gnulinux800    # postfix: advanced-3fdd2548-1430-450b-b16d-9191404598fb801    __menu_id_parts_pattern = re.compile(802        r"^(?P<prefix>.*?)-.*-(?P<postfix>.*?-.*?-.*?-.*?-.*?-.*?)?$"803    )804    @classmethod805    def name_pattern(cls) -> Pattern[str]:806        return re.compile("^Ubuntu|ubuntu$")807    def replace_boot_kernel(self, kernel_version: str) -> None:808        # set installed kernel to default809        #810        # get boot entry id811        # positive example:812        #         menuentry 'Ubuntu, with Linux 5.11.0-1011-azure' --class ubuntu813        # --class gnu-linux --class gnu --class os $menuentry_id_option814        # 'gnulinux-5.11.0-1011-azure-advanced-3fdd2548-1430-450b-b16d-9191404598fb' {815        #816        # negative example:817        #         menuentry 'Ubuntu, with Linux 5.11.0-1011-azure (recovery mode)'818        # --class ubuntu --class gnu-linux --class gnu --class os $menuentry_id_option819        # 'gnulinux-5.11.0-1011-azure-recovery-3fdd2548-1430-450b-b16d-9191404598fb' {820        cat = self._node.tools[Cat]821        menu_id_pattern = re.compile(822            r"^.*?menuentry '.*?(?:"823            + kernel_version824            + r"[^ ]*?)(?<! \(recovery mode\))' "825            r".*?\$menuentry_id_option .*?'(?P<menu_id>.*)'.*$",826            re.M,827        )828        result = cat.run("/boot/grub/grub.cfg", sudo=True)829        submenu_id = get_matched_str(result.stdout, menu_id_pattern)830        assert submenu_id, (831            f"cannot find sub menu id from grub config by pattern: "832            f"{menu_id_pattern.pattern}"833        )834        self._log.debug(f"matched submenu_id: {submenu_id}")835        # get first level menu id in boot menu836        # input is the sub menu id like:837        # gnulinux-5.11.0-1011-azure-advanced-3fdd2548-1430-450b-b16d-9191404598fb838        # output is,839        # gnulinux-advanced-3fdd2548-1430-450b-b16d-9191404598fb840        menu_id = self.__menu_id_parts_pattern.sub(841            r"\g<prefix>-\g<postfix>", submenu_id842        )843        assert menu_id, f"cannot composite menu id from {submenu_id}"844        # composite boot menu in grub845        menu_entry = f"{menu_id}>{submenu_id}"846        self._log.debug(f"composited menu_entry: {menu_entry}")847        self._replace_default_entry(menu_entry)848        self._node.execute("update-grub", sudo=True)849        try:850            # install tool packages851            self.install_packages(852                [853                    f"linux-tools-{kernel_version}-azure",854                    f"linux-cloud-tools-{kernel_version}-azure",855                    f"linux-headers-{kernel_version}-azure",856                ]857            )858        except Exception as identifier:859            self._log.debug(860                f"ignorable error on install packages after replaced kernel: "861                f"{identifier}"862            )863    def _get_information(self) -> OsInformation:864        cmd_result = self._node.execute(865            cmd="lsb_release -a",866            shell=True,867            no_error_log=True,868            expected_exit_code=0,869            expected_exit_code_failure_message="error on get os information",870        )871        assert cmd_result.stdout, "not found os information from 'lsb_release -a'"872        for row in cmd_result.stdout.splitlines():873            os_release_info = self.__lsb_os_info_pattern.match(row)874            if os_release_info:875                if os_release_info.group("name") == "Distributor ID":876                    vendor = os_release_info.group("value")877                elif os_release_info.group("name") == "Release":878                    release = os_release_info.group("value")879                elif os_release_info.group("name") == "Codename":880                    codename = os_release_info.group("value")881                elif os_release_info.group("name") == "Description":882                    full_version = os_release_info.group("value")883        if vendor == "":884            raise LisaException("OS vendor information not found")885        if release == "":886            raise LisaException("OS release information not found")887        information = OsInformation(888            version=self._parse_version(release),889            vendor=vendor,890            release=release,891            codename=codename,892            full_version=full_version,893        )894        return information895    def _replace_default_entry(self, entry: str) -> None:896        self._log.debug(f"set boot entry to: {entry}")897        sed = self._node.tools[Sed]898        sed.substitute(899            regexp="GRUB_DEFAULT=.*",900            replacement=f"GRUB_DEFAULT='{entry}'",901            file="/etc/default/grub",902            sudo=True,903        )904        # output to log for troubleshooting905        cat = self._node.tools[Cat]906        cat.run("/etc/default/grub")907class FreeBSD(BSD):908    @retry(tries=10, delay=5)909    def _install_packages(910        self,911        packages: List[str],912        signed: bool = True,913        timeout: int = 600,914        extra_args: Optional[List[str]] = None,915    ) -> None:916        if self._first_time_installation:917            self._initialize_package_installation()918        self._first_time_installation = False919        command = f"env ASSUME_ALWAYS_YES=yes pkg install -y {' '.join(packages)}"920        install_result = self._node.execute(921            command, shell=True, sudo=True, timeout=timeout922        )923        # get error lines.924        install_result.assert_exit_code(925            0,926            f"Failed to install {packages}, "927            f"please check the package name and repo are correct or not.\n",928        )929    @retry(tries=10, delay=5)930    def _initialize_package_installation(self) -> None:931        result = self._node.execute("pkg update", sudo=True)932        result.assert_exit_code(message="fail to run pkg update")933class OpenBSD(BSD):934    ...935@dataclass936# dnf repolist is of the form `<id> <name>`937# Example:938# microsoft-azure-rhel8-eus  Microsoft Azure RPMs for RHEL8 Extended Update Support939class RPMRepositoryInfo(RepositoryInfo):940    # id for the repository, for example: microsoft-azure-rhel8-eus941    id: str942# Linux distros that use RPM.943class RPMDistro(Linux):944    # microsoft-azure-rhel8-eus  Microsoft Azure RPMs for RHEL8 Extended Update Support945    _rpm_repository_info_pattern = re.compile(r"(?P<id>\S+)\s+(?P<name>\S.*\S)\s*")946    # ex: dpdk-20.11-3.el8.x86_64 or dpdk-18.11.8-1.el7_8.x86_64947    _rpm_version_splitter_regex = re.compile(948        r"(?P<package_name>[a-zA-Z0-9\-_]+)-"949        r"(?P<major>[0-9]+)\."950        r"(?P<minor>[0-9]+)\.?"951        r"(?P<patch>[0-9]+)?"952        r"(?P<build>-[a-zA-Z0-9-_\.]+)?"953    )954    def get_repositories(self) -> List[RepositoryInfo]:955        if self._first_time_installation:956            self._initialize_package_installation()957            self._first_time_installation = False958        repo_list_str = self._node.execute(959            f"{self._dnf_tool()} repolist", sudo=True960        ).stdout.splitlines()961        # skip to the first entry in the output962        for index, repo_str in enumerate(repo_list_str):963            if repo_str.startswith("repo id"):964                header_index = index965                break966        repo_list_str = repo_list_str[header_index + 1 :]967        repositories: List[RepositoryInfo] = []968        for line in repo_list_str:969            repo_info = self._rpm_repository_info_pattern.search(line)970            if repo_info:971                repositories.append(972                    RPMRepositoryInfo(973                        name=repo_info.group("name"), id=repo_info.group("id").lower()974                    )975                )976        return repositories977    def add_repository(978        self,979        repo: str,980        no_gpgcheck: bool = True,981        repo_name: Optional[str] = None,982        keys_location: Optional[List[str]] = None,983    ) -> None:984        self._node.tools[YumConfigManager].add_repository(repo, no_gpgcheck)985    def _get_package_information(self, package_name: str) -> VersionInfo:986        rpm_info = self._node.execute(987            f"rpm -q {package_name}",988            expected_exit_code=0,989            expected_exit_code_failure_message=(990                f"Could not find package information for package {package_name}"991            ),992        )993        # rpm package should be of format (package_name)-(version)994        matches = self._rpm_version_splitter_regex.search(rpm_info.stdout)995        if not matches:996            raise LisaException(997                f"Could not parse package version {rpm_info} for {package_name}"998            )999        self._node.log.debug(f"Attempting to parse version string: {rpm_info.stdout}")...

agent.py

Source:agent.py

...188    def _stringify_looseversion(item):189        if isinstance(item, LooseVersion):190            return str(item)191        return item192    def _get_package_information(self):193        versions_dict = {self._client.ip: {}}194        # ALba is always installed with OpenvStorage. The current split however offloads retrieving Alba information to the AlbaNode which is not195        # present for non openvstorage-hc installs. Therefore explicititely request the alba information like this (otherwise update will get jeopardized).196        final_dict = {}197        threads = []198        for fct in Toolbox.fetch_hooks(component='update', sub_component='get_package_info_cluster'):199            thread = Thread(target=fct, args=(self._client, versions_dict))200            thread.start()201            threads.append(thread)202        for thread in threads:203            thread.join()204        for versions in versions_dict[self._client.ip].itervalues():205            for package, version in versions.iteritems():206                if package in final_dict:...

pip.py

Source:pip.py

...25        cmd = [self.pip_path]26        cmd.extend(args)27        result = subprocess.run(cmd, capture_output=True)28        return result.stdout29    def _get_package_information(self, package):30        result = self._execute(['show', package])31        lines = result.splitlines()32        rv = {}33        for line in lines:34            key, value = line.decode().split(':', 1)35            rv[key.strip()] = value.strip()36        return rv37    def _get_current_version(self, package):38        return self._get_package_information(package)['Version']39    def _get_target_version(self, package):40        return getattr(self._config, package).get('version', 'latest')41    def _update_package(self, package):42        _exclusions = ()43        if package in _exclusions:44            logging.info("Skipping '{}' package '{}' for update check."45                         "".format(self._name, package))46            return47        logging.debug("Checking '{}' package '{}' for updates"48                      "".format(self._name, package))49        _starting_version = self._get_current_version(package)50        logging.debug("Current version for '{}' : {}"51                      "".format(package, _starting_version))52        logging.debug("Target version for '{}' : {}"...

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.