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'},
 <redacted>
 'zypper.x86_64': {'arch': 'x86_64',
				   'name': 'zypper',
				   'release': '150200.39.1',
				   'version': '1.14.57'}}

Author: CarstenGrohmann

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

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'},
+         <redacted>
+         '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