From 5a4c2fbac1acf0ca96f0600b937765374290dc5e Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Mon, 7 Aug 2023 10:42:35 +0200 Subject: [PATCH] Add Package.get_packages() to return all installed packages --- test/test_modules.py | 50 ++++++++++++++++---- testinfra/modules/package.py | 91 ++++++++++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 9 deletions(-) diff --git a/test/test_modules.py b/test/test_modules.py index e55a23d1..f23b0107 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -31,21 +31,25 @@ ] ) +release_shortcuts = { + "rockylinux9": ".el9", + "debian_bookworm": None, +} + +all_ssh_versions = { + "rockylinux9": "8.", + "debian_bookworm": "1:9.2", +} + @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] + version = all_ssh_versions[docker_image] assert ssh.is_installed assert ssh.version.startswith(version) - release = { - "rockylinux9": ".el9", - "debian_bookworm": None, - }[docker_image] + release = release_shortcuts[docker_image] if release is None: with pytest.raises(NotImplementedError): ssh.release @@ -53,6 +57,34 @@ def test_package(host, docker_image): assert release in ssh.release +@all_images +def test_get_packages(host, docker_image): + arch = "amd64" + name = "openssh-server" + release = release_shortcuts[docker_image] + + package_ssh = host.package(name) + assert package_ssh.is_installed + + name_arch = f"{name}.{arch}" + all_pkgs = host.package.get_packages() + x = [i for i in all_pkgs.keys() if i.startswith("openssh")] + assert x == ["a"] + assert "zsh" not in all_pkgs + assert name_arch in all_pkgs + pkg = all_pkgs[name_arch] + assert pkg["version"] == package_ssh.version + assert pkg["arch"] == arch + assert pkg["name"] == name + + if release is None: + with pytest.raises(NotImplementedError): + package_ssh.release + else: + assert release in pkg["release"] + assert pkg["release"] == package_ssh.release + + def test_held_package(host): python = host.package("python3") assert python.is_installed @@ -318,7 +350,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) 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