From 5e7c46c65aae0f98da33b96c5e88687fb4aac290 Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Mon, 7 Aug 2023 12:40:11 +0200 Subject: [PATCH 1/2] Extract constants from functions to file level --- test/test_modules.py | 37 +++++++++++++++++++------------------ 1 file changed, 19 insertions(+), 18 deletions(-) diff --git a/test/test_modules.py b/test/test_modules.py index e55a23d1..d921ba2b 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -31,26 +31,31 @@ ] ) +# content: ssh version, release shortcut, +ssh_pkg_info = { + "rockylinux9": ("8.", ".el9"), + "debian_bookworm": ("1:9.2", None), +} + +# content: distribution, codename, architecture, release_regex +docker_image_info = { + "rockylinux9": ("rocky", None, "x86_64", r"^9.\d+$"), + "debian_bookworm": ("debian", "bookworm", "amd64", r"^12"), +} + @all_images def test_package(host, docker_image): assert not host.package("zsh").is_installed ssh = host.package("openssh-server") - version = { - "rockylinux9": "8.", - "debian_bookworm": "1:9.2", - }[docker_image] + ssh_version, sshd_release = ssh_pkg_info[docker_image] assert ssh.is_installed - assert ssh.version.startswith(version) - release = { - "rockylinux9": ".el9", - "debian_bookworm": None, - }[docker_image] - if release is None: + assert ssh.version.startswith(ssh_version) + if sshd_release is None: with pytest.raises(NotImplementedError): ssh.release else: - assert release in ssh.release + assert sshd_release in ssh.release def test_held_package(host): @@ -89,14 +94,10 @@ def test_uninstalled_package_version(host): def test_systeminfo(host, docker_image): assert host.system_info.type == "linux" - release, distribution, codename, arch = { - "rockylinux9": (r"^9.\d+$", "rocky", None, "x86_64"), - "debian_bookworm": (r"^12", "debian", "bookworm", "x86_64"), - }[docker_image] - + distribution, codename, unused_arch, release_regex = docker_image_info[docker_image] assert host.system_info.distribution == distribution assert host.system_info.codename == codename - assert re.match(release, host.system_info.release) + assert re.match(release_regex, host.system_info.release) @all_images @@ -318,7 +319,7 @@ def test_file(host): def test_ansible_unavailable(host): - expected = "Ansible module is only available with " "ansible connection backend" + expected = "Ansible module is only available with ansible connection backend" with pytest.raises(RuntimeError) as excinfo: host.ansible("setup") assert expected in str(excinfo.value) From 4ce9dd7804c631c2928c37793bc6fe26120c7825 Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Mon, 7 Aug 2023 12:45:39 +0200 Subject: [PATCH 2/2] Add Package.get_packages() to return all installed packages Get all installed packages with name version number, release number (if available) and architecture: >>> host.package.get_packages() {'acl.x86_64': {'arch': 'x86_64', 'name': 'acl', 'release': '4.3.1', 'version': '2.2.52'}, 'zypper.x86_64': {'arch': 'x86_64', 'name': 'zypper', 'release': '150200.39.1', 'version': '1.14.57'}} --- test/test_modules.py | 26 +++++++++++ testinfra/modules/package.py | 91 ++++++++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+) diff --git a/test/test_modules.py b/test/test_modules.py index d921ba2b..70b6a101 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -58,6 +58,32 @@ def test_package(host, docker_image): assert sshd_release in ssh.release +@all_images +def test_get_packages(host, docker_image): + arch = docker_image_info[docker_image][2] + sshd_release_number = ssh_pkg_info[docker_image][1] + + package_ssh = host.package("openssh-server") + assert package_ssh.is_installed + + all_pkgs = host.package.get_packages() + assert f"zsh.{arch}" not in all_pkgs + + name_arch = f"openssh-server.{arch}" + assert name_arch in all_pkgs + + pkg = all_pkgs[name_arch] + assert pkg["version"] == package_ssh.version + assert pkg["arch"] == arch + assert pkg["name"] == "openssh-server" + if sshd_release_number is None: + with pytest.raises(NotImplementedError): + package_ssh.release + else: + assert sshd_release_number in pkg["release"] + assert pkg["release"] == package_ssh.release + + def test_held_package(host): python = host.package("python3") assert python.is_installed diff --git a/testinfra/modules/package.py b/testinfra/modules/package.py index 6f5610e6..2503812e 100644 --- a/testinfra/modules/package.py +++ b/testinfra/modules/package.py @@ -20,6 +20,23 @@ def __init__(self, name): self.name = name super().__init__() + @classmethod + def get_packages(cls): + """Get all installed packages with name version number, release number (if available) and architecture + + >>> host.package.get_packages() + {'acl.x86_64': {'arch': 'x86_64', + 'name': 'acl', + 'release': '4.3.1', + 'version': '2.2.52'}, + + 'zypper.x86_64': {'arch': 'x86_64', + 'name': 'zypper', + 'release': '150200.39.1', + 'version': '1.14.57'}} + """ + raise NotImplementedError + @property def is_installed(self): """Test if the package is installed @@ -94,6 +111,29 @@ def get_module_class(cls, host): class DebianPackage(Package): + @classmethod + def get_packages(cls): + out = cls.run(r"dpkg-query -f '${Package}|${Version}|${Architecture}\n' -W") + assert not out.stderr + pkgs = {} + for line in out.stdout.splitlines(): + line = line.strip() + if not line: + continue + name, version, arch = line.split("|") + pkg_key = f"{name}.{arch}" + assert pkg_key not in pkgs, ( + f"Package {pkg_key} already added to package list. " + "Check for duplicate package in command output" + ) + pkgs[pkg_key] = { + "name": name, + "version": version, + "release": None, + "arch": arch, + } + return pkgs + @property def is_installed(self): result = self.run_test("dpkg-query -f '${Status}' -W %s", self.name) @@ -155,6 +195,34 @@ def version(self): class RpmPackage(Package): + @classmethod + def get_packages(cls): + out = cls.run( + r'rpm -qa --queryformat "%{NAME}|%{VERSION}|%{RELEASE}|%{ARCH}\n"' + ) + assert not out.stderr + pkgs = {} + for line in out.stdout.splitlines(): + line = line.strip() + if not line: + continue + name, version, release, arch = line.split("|") + # Ignore GPG keys imported with "rpm --import" e.g. "gpg-pubkey|50a3dd1c|50f35137|(none)" + if name == "gpg-pubkey" and arch == "(none)": + continue + pkg_key = f"{name}.{arch}" + assert pkg_key not in pkgs, ( + f"Package {pkg_key} already added to package list. " + "Check for duplicate package in command output" + ) + pkgs[pkg_key] = { + "name": name, + "version": version, + "release": release, + "arch": arch, + } + return pkgs + @property def is_installed(self): return self.run_test("rpm -q %s", self.name).rc == 0 @@ -185,6 +253,29 @@ def release(self): class ArchPackage(Package): + @classmethod + def get_packages(cls): + out = cls.run(r'expac "%n|%v|%a"') + assert not out.stderr + pkgs = {} + for line in out.stdout.splitlines(): + line = line.strip() + if not line: + continue + name, version, arch = line.split("|") + pkg_key = f"{name}.{arch}" + assert pkg_key not in pkgs, ( + f"Package {pkg_key} already added to package list. " + "Check for duplicate package in command output" + ) + pkgs[pkg_key] = { + "name": name, + "version": version, + "release": None, + "arch": arch, + } + return pkgs + @property def is_installed(self): return self.run_test("pacman -Q %s", self.name).rc == 0