lib.Distro.pkg_files(): Take list[str] | None

Distro.pkg_files() takes a package name and returns a list of files in that
package. This commit generalizes that - pass None to it and get the list of
all packaged files in the system, and pass a list of strings to retrieve
all files belonging to the named packages.

The same arguments are propagated to _pkg_files(), which in turn propagates
them to list_files().

This allows faster generation of file lists than before. Having to retrieve
a list of matching packages and query each package for its contained files
can be collapsed into one or two backend calls.

Signed-off-by: Jan Lindemann <jan@janware.com>

src/python/jw/pkg/lib/Distro.py

@@ -471,11 +471,18 @@ class Distro(abc.ABC):
     # -- pkg_files
 
     @abc.abstractmethod
-    async def _pkg_files(self, name: str) -> Iterable[str]:
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
         pass
 
-    async def pkg_files(self, name: str) -> Iterable[str]:
-        if not name:
-            log(WARNING, 'No package specified for inspection')
-            return []
-        return await self._pkg_files(name)
+    async def pkg_files(self, names: str | list[str] | None = None) -> Iterable[str]:
+        # -- The file paths of the packages selected by the
+        #    argument: None, the default selection, filtered to the
+        #    app's packages when a filter is set, all installed
+        #    packages otherwise; [], no packages; [names], the
+        #    listed packages; a str names one package, accepted for
+        #    compatibility with the single-package call
+        if isinstance(names, str):
+            names = [names]
+        if names is None and self.default_pkg_filter is not None:
+            names = [p.name for p in await self.select()]
+        return await self._pkg_files(names)

src/python/jw/pkg/lib/distros/arch/Distro.py

@@ -66,7 +66,7 @@ class Distro(Base):
         )
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
         raise NotImplementedError(
             'distro pkg ls yet implemented for Arch-like distributions'
         )

src/python/jw/pkg/lib/distros/debian/Distro.py

@@ -85,5 +85,5 @@ class Distro(Base):
         await self.dpkg(['-P', *names], sudo = True)
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
-        return await list_files(name, ec = self.ctx)
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
+        return await list_files(names, ec = self.ctx)

src/python/jw/pkg/lib/distros/redhat/Distro.py

@@ -70,8 +70,8 @@ class Distro(Base):
         await self.rpm(['-e', *names], sudo = True)
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
-        return await list_files(name, ec = self.ctx)
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
+        return await list_files(names, ec = self.ctx)
 
     async def _download(self, name: str, tmp_dir: str) -> str:
         raise NotImplementedError(

src/python/jw/pkg/lib/distros/suse/Distro.py

@@ -95,5 +95,5 @@ class Distro(Base):
         await self.rpm(['-e', *names], sudo = True)
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
-        return await list_files(name, ec = self.ctx)
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
+        return await list_files(names, ec = self.ctx)

src/python/jw/pkg/lib/pm/dpkg.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Iterable, cast
+from typing import TYPE_CHECKING, Any, Iterable, cast
 
 if TYPE_CHECKING:
     from ..ExecContext import ExecContext
@@ -26,21 +26,25 @@ def meta_map() -> dict[str, str]:
     return _meta_map
 
 async def _run(
-    cmd: list[str], sudo: bool = False, ec: ExecContext | None = None
+    cmd: list[str],
+    sudo: bool = False,
+    ec: ExecContext | None = None,
+    **kwargs: Any,
 ) -> str:
     return (
-        await run_sudo(cmd)
-        if sudo else await run_cmd(cmd, ec = ec, cmd_input = InputMode.NonInteractive)
+        await run_sudo(cmd) if sudo else await
+        run_cmd(cmd, ec = ec, cmd_input = InputMode.NonInteractive, **kwargs)
     ).stdout_str
 
 async def run_dpkg(  # export
     args: list[str],
     sudo: bool = False,
-    ec: ExecContext | None = None
+    ec: ExecContext | None = None,
+    **kwargs: Any,
 ) -> str:
     cmd = ['/usr/bin/dpkg']
     cmd.extend(args)
-    return await _run(cmd, sudo, ec)
+    return await _run(cmd, sudo, ec, **kwargs)
 
 async def run_dpkg_query(  # export
     args: list[str],
@@ -61,6 +65,26 @@ async def query_packages(names: Iterable[str] = [],
     specs = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo = False, ec = ec)
     return cast('Iterable[Package]', Package.parse_specs_str(specs))
 
-async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]:
-    file_list_str = await run_dpkg(['-L', pkg], sudo = False, ec = ec)
+async def list_files(
+    names: list[str] | None = None,
+    ec: ExecContext | None = None,
+) -> list[str]:
+    # -- None: all installed packages. The standard interface, two
+    #    queries: select the installed packages, then list their
+    #    files in one command; the list output carries a
+    #    package-name header and a blank line per package, both
+    #    dropped. []: no packages, no command
+    if names is None:
+        names = (
+            await run_dpkg_query(
+                ['-W', '-f=${?installed:binary:Package}\n'], sudo = False, ec = ec
+            )
+        ).split()
+        if not names:
+            return []
+        stdout = await run_dpkg_query(['-L', *names], sudo = False, ec = ec)
+        return [line for line in stdout.splitlines() if line and not line.endswith(':')]
+    if not names:
+        return []
+    file_list_str = await run_dpkg(['-L', *names], sudo = False, ec = ec)
     return file_list_str.splitlines()

src/python/jw/pkg/lib/pm/rpm.py

@@ -60,12 +60,19 @@ async def query_packages(  # export
     )
     return cast('Iterable[Package]', Package.parse_specs_str(specs))
 
-async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]:
+async def list_files(
+    names: list[str] | None = None,
+    ec: ExecContext | None = None,
+) -> list[str]:
+    # -- None: all installed packages, one query without package
+    #    names. []: no packages, no command
+    if names is None:
+        args = ['-qal']
+    elif not names:
+        return []
+    else:
+        args = ['-ql', *names]
     stdout = await run_rpm(
-        ['-ql', pkg],
-        throw = True,
-        sudo = False,
-        mode = InputMode.NonInteractive,
-        ec = ec
+        args, throw = True, sudo = False, mode = InputMode.NonInteractive, ec = ec
     )
     return stdout.splitlines()

test/integration/jw_pkg/secrets/test.py

@@ -71,8 +71,13 @@ class FakeDistro(SuseDistro):
         return [Package(name) for name in names if name in self.__files]
 
     @override
-    async def _pkg_files(self, name: str) -> list[str]:
-        return list(self.__files.get(name, []))
+    async def _pkg_files(self, names: list[str] | None) -> list[str]:
+        if names is None:
+            names = list(self.__files)
+        ret: list[str] = []
+        for n in names:
+            ret.extend(self.__files.get(n, []))
+        return ret
 
 def run_cli(distro: FakeDistro, argv: list[str]) -> tuple[list[str], list[str]]:
     """Run the real jw-pkg CLI in-process, return stdout and stderr

test/unit/python/jw/pkg/lib/Distro/test.py

@@ -56,7 +56,7 @@ class DummyDistro(Distro):
         pass
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
         return []
 
 def distro(os_release_str: str) -> Distro: