How to use replace_boot_kernel method in lisa

Best Python code snippet using lisa_python

operating_system.py

Source:operating_system.py Github

copy

Full Screen

...253 return cls.__name__254 @classmethod255 def name_pattern(cls) -> Pattern[str]:256 return re.compile(f"^{cls.type_name()}$")257 def replace_boot_kernel(self, kernel_version: str) -> None:258 raise NotImplementedError("update boot entry is not implemented")259 def get_kernel_information(self, force_run: bool = False) -> KernelInformation:260 uname = self._node.tools[Uname]261 uname_result = uname.get_linux_information(force_run=force_run)262 parts: List[str] = [str(x) for x in uname_result.kernel_version]263 kernel_information = KernelInformation(264 version=uname_result.kernel_version,265 raw_version=uname_result.kernel_version_raw,266 hardware_platform=uname_result.hardware_platform,267 operating_system=uname_result.operating_system,268 version_parts=parts,269 )270 return kernel_information271 def install_packages(272 self,273 packages: Union[274 str,275 Tool,276 Type[Tool],277 Sequence[Union[str, Tool, Type[Tool]]],278 ],279 signed: bool = False,280 timeout: int = 1200,281 extra_args: Optional[List[str]] = None,282 ) -> None:283 package_names = self._get_package_list(packages)284 self._install_packages(package_names, signed, timeout, extra_args)285 def package_exists(self, package: Union[str, Tool, Type[Tool]]) -> bool: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_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}")1000 version_info = self._get_version_info_from_named_regex_match(1001 package_name, matches1002 )1003 return self._cache_and_return_version_info(package_name, version_info)1004 def _install_packages(1005 self,1006 packages: List[str],1007 signed: bool = True,1008 timeout: int = 600,1009 extra_args: Optional[List[str]] = None,1010 ) -> None:1011 add_args = self._process_extra_package_args(extra_args)1012 command = f"{self._dnf_tool()} install {add_args} -y {' '.join(packages)}"1013 if not signed:1014 command += " --nogpgcheck"1015 self._node.execute(1016 command,1017 shell=True,1018 sudo=True,1019 timeout=timeout,1020 expected_exit_code=0,1021 expected_exit_code_failure_message=f"Failed to install {packages}.",1022 )1023 self._log.debug(f"{packages} is/are installed successfully.")1024 def _package_exists(self, package: str) -> bool:1025 command = f"{self._dnf_tool()} list installed {package}"1026 result = self._node.execute(command, sudo=True)1027 if result.exit_code == 0:1028 for row in result.stdout.splitlines():1029 if package in row:1030 return True1031 return False1032 def _is_package_in_repo(self, package: str) -> bool:1033 command = f"{self._dnf_tool()} list {package} -y"1034 result = self._node.execute(command, sudo=True, shell=True)1035 return 0 == result.exit_code1036 def _dnf_tool(self) -> str:1037 return "dnf"1038 def _update_packages(self, packages: Optional[List[str]] = None) -> None:1039 command = f"{self._dnf_tool()} -y --nogpgcheck update "1040 if packages:1041 command += " ".join(packages)1042 self._node.execute(command, sudo=True, timeout=3600)1043class Fedora(RPMDistro):1044 # Red Hat Enterprise Linux Server 7.8 (Maipo) => 7.81045 _fedora_release_pattern_version = re.compile(r"^.*release\s+([0-9\.]+).*$")1046 # 305.40.1.el8_4.x86_641047 # 240.el8.x86_641048 __kernel_version_parts_pattern = re.compile(1049 r"^(?P<part1>\d+)\.(?P<part2>\d+)?\.?(?P<part3>\d+)?\.?"1050 r"(?P<distro>.*?)\.(?P<platform>.*?)$"1051 )1052 @classmethod1053 def name_pattern(cls) -> Pattern[str]:1054 return re.compile("^Fedora|fedora$")1055 def get_kernel_information(self, force_run: bool = False) -> KernelInformation:1056 kernel_information = super().get_kernel_information(force_run)1057 # original parts: version_parts=['4', '18', '0', '305.40.1.el8_4.x86_64', '']1058 # target parts: version_parts=['4', '18', '0', '305', '40', '1', 'el8_4',1059 # 'x86_64']1060 groups = find_group_in_lines(1061 kernel_information.version_parts[3], self.__kernel_version_parts_pattern1062 )1063 new_parts = kernel_information.version_parts[:3]1064 # the default '1' is trying to build a meaningful Redhat version number.1065 new_parts.extend(1066 [1067 groups["part1"],1068 groups["part2"],1069 groups["part3"],1070 groups["distro"],1071 groups["platform"],1072 ]1073 )1074 for index, part in enumerate(new_parts):1075 if part is None:1076 new_parts[index] = ""1077 kernel_information.version_parts = new_parts1078 return kernel_information1079 def install_epel(self) -> None:1080 # Extra Packages for Enterprise Linux (EPEL) is a special interest group1081 # (SIG) from the Fedora Project that provides a set of additional packages1082 # for RHEL (and CentOS, and others) from the Fedora sources.1083 major = self._node.os.information.version.major1084 assert_that(major).described_as(1085 "Fedora/RedHat version must be greater than 7"1086 ).is_greater_than_or_equal_to(7)1087 epel_release_rpm_name = f"epel-release-latest-{major}.noarch.rpm"1088 self.install_packages(1089 f"https://dl.fedoraproject.org/pub/epel/{epel_release_rpm_name}"1090 )1091 # replace $releasever to 8 for 8.x1092 if major == 8:1093 sed = self._node.tools[Sed]1094 sed.substitute("$releasever", "8", "/etc/yum.repos.d/epel*.repo", sudo=True)1095 def _verify_package_result(self, result: ExecutableResult, packages: Any) -> None:1096 # yum returns exit_code=1 if DNF handled an error with installation.1097 # We do not want to fail if exit_code=1, but warn since something may1098 # potentially have gone wrong.1099 if result.exit_code == 1:1100 self._log.debug(f"DNF handled error with installation of {packages}")1101 elif result.exit_code == 0:1102 self._log.debug(f"{packages} is/are installed successfully.")1103 else:1104 raise LisaException(1105 f"Failed to install {packages}. exit_code: {result.exit_code}"1106 )1107 def group_install_packages(self, group_name: str) -> None:1108 # trigger to run _initialize_package_installation1109 self._get_package_list(group_name)1110 result = self._node.execute(f'yum -y groupinstall "{group_name}"', sudo=True)1111 self._verify_package_result(result, group_name)1112 def _get_information(self) -> OsInformation:1113 cmd_result = self._node.execute(1114 # Typical output of 'cat /etc/fedora-release' is -1115 # Fedora release 22 (Twenty Two)1116 cmd="cat /etc/fedora-release",1117 no_error_log=True,1118 expected_exit_code=0,1119 expected_exit_code_failure_message="error on get os information",1120 )1121 full_version = cmd_result.stdout1122 if "Fedora" not in full_version:1123 raise LisaException("OS version information not found")1124 vendor = "Fedora"1125 release = get_matched_str(full_version, self._fedora_release_pattern_version)1126 codename = get_matched_str(full_version, self._distro_codename_pattern)1127 information = OsInformation(1128 version=self._parse_version(release),1129 vendor=vendor,1130 release=release,1131 codename=codename,1132 full_version=full_version,1133 )1134 return information1135class Redhat(Fedora):1136 # Red Hat Enterprise Linux Server release 6.9 (Santiago)1137 # CentOS release 6.9 (Final)1138 # CentOS Linux release 8.3.20111139 __legacy_redhat_information_pattern = re.compile(1140 r"^(?P<vendor>.*?)?(?: Enterprise Linux Server)?(?: Linux)?"1141 r"(?: release)? (?P<version>[0-9\.]+)(?: \((?P<codename>.*).*\))?$"1142 )1143 # Oracle Linux Server1144 # Red Hat Enterprise Linux Server1145 # Red Hat Enterprise Linux1146 __vendor_pattern = re.compile(1147 r"^(?P<vendor>.*?)?(?: Enterprise)?(?: Linux)?(?: Server)?$"1148 )1149 @classmethod1150 def name_pattern(cls) -> Pattern[str]:1151 return re.compile("^rhel|Red|AlmaLinux|Rocky|Scientific|acronis|Actifio$")1152 def replace_boot_kernel(self, kernel_version: str) -> None:1153 # Redhat kernel is replaced when installing RPM. For source code1154 # installation, it's implemented in source code installer.1155 ...1156 def capture_system_information(self, saved_path: Path) -> None:1157 super().capture_system_information(saved_path)1158 self._node.shell.copy_back(1159 self._node.get_pure_path("/etc/redhat-release"),1160 saved_path / "redhat-release.txt",1161 )1162 @retry(tries=10, delay=5)1163 def _initialize_package_installation(self) -> None:1164 information = self._get_information()1165 # We may hit issue when run any yum command, caused by out of date1166 # rhui-microsoft-azure-rhel package....

Full Screen

Full Screen

kernel_installer.py

Source:kernel_installer.py Github

copy

Full Screen

...103 # for ubuntu cvm kernel, there is no menuentry added into grub file104 if hasattr(installer.runbook, "source"):105 if installer.runbook.source != "linux-image-azure-fde":106 posix = cast(Posix, node.os)107 posix.replace_boot_kernel(installed_kernel_version)108 else:109 efi_files = node.execute(110 "ls -t /usr/lib/linux/efi/kernel.efi-*-azure-cvm",111 sudo=True,112 shell=True,113 expected_exit_code=0,114 expected_exit_code_failure_message=(115 "fail to find kernel.efi file for kernel type "116 " linux-image-azure-fde"117 ),118 )119 efi_file = efi_files.stdout.splitlines()[0]120 node.execute(121 (...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful