Add Package.get_packages() to return all installed packages

test/test_modules.py

@@ -31,28 +31,55 @@
     ]
 )
 
+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
     else:
         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()
+    assert f"zsh.{arch}" 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 +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