Add Package.get_packages() to return all installed packages

test/test_modules.py

@@ -31,26 +31,57 @@
     ]
 )
 
+# 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
+
+
+@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):
@@ -89,14 +120,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 +345,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)

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