src/python/jw/pkg/cmds/pkg/CmdVerify.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, override
+
+from .Cmd import Cmd, Parent
+
+if TYPE_CHECKING:
+    from argparse import ArgumentParser, Namespace
+
+class CmdVerify(Cmd):  # export
+
+    def __init__(self, parent: Parent) -> None:
+        super().__init__(
+            parent,
+            'verify',
+            help = (
+                'Verify the integrity of installed packages: report files'
+                ' modified or deleted outside of the package manager'
+            )
+        )
+
+    @override
+    def add_arguments(self, parser: ArgumentParser) -> None:
+        super().add_arguments(parser)
+
+    @override
+    async def _run(self, args: Namespace) -> None:
+        for file in await self.distro.verify():
+            print(file)

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

@@ -17,6 +17,7 @@ if TYPE_CHECKING:
 
     from .base import Result
     from .ExecContext import ExecContext
+    from .FlaggedFile import FlaggedFile
     from .Package import Package
     from .PackageFilter import PackageFilter
 
@@ -471,11 +472,42 @@ 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]:
+    async def pkg_files(self, names: str | list[str] | None = None) -> Iterable[str]:
-        if not name:
+        # -- The file paths of the packages selected by the
-            log(WARNING, 'No package specified for inspection')
+        #    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)
+
+    # -- verify
+
+    @abc.abstractmethod
+    async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
+        pass
+
+    async def verify(
+        self,
+        names: Collection[str] = [],
+        filter: PackageFilter | None = None,
+    ) -> list[FlaggedFile]:
+        """Verify the integrity of installed packages: report files that
+        were modified or deleted outside of the package manager.
+
+        Without names, all packages selected by filter (or the default
+        package filter, if any) are verified.
+        """
+        if not names:
+            names = [p.name for p in await self.select([], filter = filter)]
+        if not names:
+            log(WARNING, 'No packages to verify')
             return []
-        return await self._pkg_files(name)
+        return list(await self._verify(names))

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

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from typing import override
+
+class FlaggedFile:  # export
+    """A file of an installed package that failed the package manager's
+    integrity verification (Distro.verify()).
+
+    flags is the verify status line of the package manager for this
+    file, or the string "missing" for a file the package manager
+    reports as missing.
+    """
+
+    def __init__(self, pkg: str, file: str, flags: str) -> None:
+        self.pkg = pkg
+        self.file = file
+        self.flags = flags
+
+    @override
+    def __repr__(self) -> str:
+        return f'{self.pkg} {self.file} {self.flags}'

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

@@ -8,6 +8,7 @@ if TYPE_CHECKING:
     from typing import Iterable
 
     from ...base import Result
+    from ...FlaggedFile import FlaggedFile
     from ...Package import Package
 
 class Distro(Base):
@@ -66,7 +67,13 @@ 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'
         )
+
+    @override
+    async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
+        raise NotImplementedError(
+            'distro verify is not yet implemented for Arch-like distributions'
+        )

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

@@ -6,12 +6,13 @@ from typing import TYPE_CHECKING, Any, override
 
 from ...Distro import Distro as Base
 from ...log import NOTICE, log
-from ...pm.dpkg import list_files, query_packages, run_dpkg
+from ...pm.dpkg import list_files, query_packages, run_dpkg, verify_packages
 
 if TYPE_CHECKING:
     from typing import Iterable
 
     from ...base import Result
+    from ...FlaggedFile import FlaggedFile
     from ...Package import Package
 
 class Distro(Base):
@@ -85,5 +86,9 @@ class Distro(Base):
         await self.dpkg(['-P', *names], sudo = True)
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
-        return await list_files(name, ec = self.ctx)
+        return await list_files(names, ec = self.ctx)
+
+    @override
+    async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
+        return await verify_packages(names, ec = self.ctx)

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

@@ -3,13 +3,14 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, override
 
 from ...Distro import Distro as Base
-from ...pm.rpm import list_files, query_packages, run_rpm
+from ...pm.rpm import list_files, query_packages, run_rpm, verify_packages
 
 if TYPE_CHECKING:
     from typing import Any, Collection, Iterable
 
     from ...base import Result
     from ...ExecContext import ExecContext
+    from ...FlaggedFile import FlaggedFile
     from ...Package import Package
 
 class Distro(Base):
@@ -70,8 +71,12 @@ class Distro(Base):
         await self.rpm(['-e', *names], sudo = True)
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
-        return await list_files(name, ec = self.ctx)
+        return await list_files(names, ec = self.ctx)
+
+    @override
+    async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
+        return await verify_packages(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

@@ -3,7 +3,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Any, override
 
 from ...Distro import Distro as Base
-from ...pm.rpm import list_files, query_packages, run_rpm
+from ...pm.rpm import list_files, query_packages, run_rpm, verify_packages
 
 if TYPE_CHECKING:
     from collections.abc import Collection
@@ -11,6 +11,7 @@ if TYPE_CHECKING:
 
     from ...base import Result
     from ...ExecContext import ExecContext
+    from ...FlaggedFile import FlaggedFile
     from ...Package import Package
 
 class Distro(Base):
@@ -95,5 +96,9 @@ class Distro(Base):
         await self.rpm(['-e', *names], sudo = True)
 
     @override
-    async def _pkg_files(self, name: str) -> Iterable[str]:
+    async def _pkg_files(self, names: list[str] | None) -> Iterable[str]:
-        return await list_files(name, ec = self.ctx)
+        return await list_files(names, ec = self.ctx)
+
+    @override
+    async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
+        return await verify_packages(names, ec = self.ctx)

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

@@ -1,11 +1,12 @@
 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
 
 from ..base import InputMode
+from ..FlaggedFile import FlaggedFile
 from ..Package import Package
 from ..util import run_cmd, run_sudo
 
@@ -26,21 +27,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)
+        await run_sudo(cmd) if sudo else await
-        if sudo else await run_cmd(cmd, ec = ec, cmd_input = InputMode.NonInteractive)
+        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 +66,61 @@ 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]:
+async def list_files(
-    file_list_str = await run_dpkg(['-L', pkg], sudo = False, ec = ec)
+    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()
+
+def _parse_verify_output(
+    pkg: str,
+    output: str,
+) -> list[FlaggedFile]:
+    """Parse the output of `dpkg -V <pkg>` into a list of flagged
+    files.
+
+    Reports md5sum (P) mismatches and missing files (M), the cases
+    that indicate a package was modified outside of the package
+    manager. The remaining dpkg verify flags (access permissions,
+    configuration file rdev, user, group, size, mtime, and
+    executable bit) are noise on dev machines and are ignored.
+    """
+    ret: list[FlaggedFile] = []
+    for line in output.splitlines():
+        parts = line.split(None, 1)
+        if len(parts) != 2:
+            continue
+        flags, file = parts
+        if 'P' in flags or 'M' in flags:
+            ret.append(FlaggedFile(pkg, file, flags))
+    return ret
+
+async def verify_packages(
+    names: Iterable[str],
+    ec: ExecContext | None = None,
+) -> list[FlaggedFile]:
+    ret: list[FlaggedFile] = []
+    for name in names:
+        # dpkg -V reports its findings on standard output; the exit
+        # status is not informative, so it is ignored
+        output = await run_dpkg(['-V', name], sudo = False, throw = False, ec = ec)
+        ret.extend(_parse_verify_output(name, output))
+    return ret

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

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import re
+
 from typing import TYPE_CHECKING, Any, Collection, Iterable, cast
 
 from ..base import InputMode
+from ..FlaggedFile import FlaggedFile
 from ..Package import Package
 from ..util import run_cmd, run_sudo
 
@@ -60,12 +63,74 @@ 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],
+        args, throw = True, sudo = False, mode = InputMode.NonInteractive, ec = ec
-        throw = True,
+    )
+    return stdout.splitlines()
+
+def _parse_verify_output(
+    pkg: str,
+    output: str,
+) -> list[FlaggedFile]:
+    """Parse the output of `rpm -qV <pkg>` into a list of flagged
+    files.
+
+    Reports content (md5) mismatches and missing files, the cases
+    that indicate a package was modified outside of the package
+    manager. Mismatches in size, mode, mtime, user, and group are
+    noise on dev machines and are ignored.
+    """
+    ret: list[FlaggedFile] = []
+    for line in output.splitlines():
+        # rpm appends a parenthesized annotation, for instance
+        # "(Permission denied)", to a line whose file could not be
+        # accessed; it is not part of the file name
+        line = re.sub(r'\s*\([^)]*\)$', '', line)
+        parts = line.split()
+        if not parts:
+            continue
+        # rpm reports a missing file with the literal word "missing"
+        # in the flag field, the file name in the usual place
+        if parts[0] == 'missing':
+            if len(parts) >= 2:
+                ret.append(FlaggedFile(pkg, parts[-1], 'missing'))
+            continue
+        flags = parts[0]
+        if len(flags) not in (8, 9) or len(parts) < 2:
+            continue
+        if not all(ch in 'SM5DLUGTP.?' for ch in flags):
+            continue
+        if flags[2] == '5':
+            ret.append(FlaggedFile(pkg, parts[-1], flags))
+    return ret
+
+async def verify_packages(
+    names: Iterable[str],
+    ec: ExecContext | None = None,
+) -> list[FlaggedFile]:
+    ret: list[FlaggedFile] = []
+    for name in names:
+        # rpm exits with 1 both when differences are found and when
+        # the package is not installed, so only the output is
+        # informative
+        output = await run_rpm(
+            ['-qV', name],
+            throw = False,
             sudo = False,
             mode = InputMode.NonInteractive,
             ec = ec
         )
-    return stdout.splitlines()
+        ret.extend(_parse_verify_output(name, output))
+    return ret

test/integration/jw_pkg/help/test-expected.txt

@@ -66,6 +66,8 @@ Available subcommands of packages:
     reboot-required  Check whether the machine needs rebooting
     refresh          Refresh the distribution's notion of available packages
     select           Select packages by filter
+    verify           Verify the integrity of installed packages: report files
+                     modified or deleted outside of the package manager
 ============= Running: jw-pkg.py -t ../../../.. --log-level info packages delete --help
 usage: jw-pkg.py packages delete [-h] [names ...]
 
@@ -140,6 +142,14 @@ Select packages by filter
 positional arguments:
   filter      Package filter string
 
+options:
+  -h, --help  show this help message and exit
+============= Running: jw-pkg.py -t ../../../.. --log-level info packages verify --help
+usage: jw-pkg.py packages verify [-h]
+
+Verify the integrity of installed packages: report files modified or deleted
+outside of the package manager
+
 options:
   -h, --help  show this help message and exit
 ============= Running: jw-pkg.py -t ../../../.. --log-level info platform --help

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]:
+    async def _pkg_files(self, names: list[str] | None) -> list[str]:
-        return list(self.__files.get(name, []))
+        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

@@ -1,12 +1,19 @@
 from __future__ import annotations
 
+import asyncio
+
 from typing import TYPE_CHECKING, Iterable, cast, override
 
 from jw.pkg.lib.Distro import Distro
+from jw.pkg.lib.distros.suse.Distro import Distro as SuseDistro
+from jw.pkg.lib.ec.Local import Local
+from jw.pkg.lib.Package import Package
+from jw.pkg.lib.PackageFilter import PackageFilterString
+from jw.pkg.lib.Result import Result
 
 if TYPE_CHECKING:
     from jw.pkg.lib.ExecContext import ExecContext
-    from jw.pkg.lib.Package import Package
+    from jw.pkg.lib.FlaggedFile import FlaggedFile
 
 OS_RELEASE_UBUNTU = """\
 NAME="Ubuntu"
@@ -56,9 +63,79 @@ 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 []
 
+    @override
+    async def _verify(self, names: Iterable[str]) -> list[FlaggedFile]:
+        return []
+
+# -- verify() --
+
+class VerifyDistro(DummyDistro):
+
+    def __init__(
+        self, ec: ExecContext, id: str, os_release_str: str | None = None
+    ) -> None:
+        super().__init__(ec, id, os_release_str)
+        self.verified: list[str] | None = None
+
+    @override
+    async def _verify(self, names: Iterable[str]) -> list[FlaggedFile]:
+        self.verified = list(names)
+        return []
+
+vd = VerifyDistro(ec = cast('ExecContext', object()), id = 'suse')
+
+# -- Given names are passed to the distro backend untouched
+assert asyncio.run(vd.verify(names = ['a', 'b'])) == []
+assert vd.verified == ['a', 'b']
+
+# -- Without names, the (empty) selection is verified, so the
+# backend is not invoked
+vd.verified = None
+assert asyncio.run(vd.verify()) == []
+assert vd.verified is None
+
+# -- pkg_files --
+
+# -- Without a filter, the selection is passed through as the
+#    whole system; the paths come back unfiltered
+
+class TmplDistro(DummyDistro):
+
+    def __init__(self) -> None:
+        super().__init__(ec = cast('ExecContext', object()), id = 'suse')
+        self.files: dict[str, list[str]] = {}
+
+    @override
+    async def _select_by_name(self, names: Iterable[str]) -> list[Package]:
+        return [Package(n) for n in self.files]
+
+    @override
+    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[n])
+        return ret
+
+t = TmplDistro()
+t.files = {
+    'a': ['/usr/share/a/a.conf.jw-tmpl', '/usr/bin/a'],
+    'b': ['/etc/b/b.conf.jw-tmpl'],
+}
+assert sorted(asyncio.run(t.pkg_files())) == [
+    '/etc/b/b.conf.jw-tmpl', '/usr/bin/a', '/usr/share/a/a.conf.jw-tmpl'
+]
+
+# -- A single package name as a str is accepted for compatibility
+assert sorted(asyncio.run(t.pkg_files('a'))) == [
+    '/usr/bin/a',
+    '/usr/share/a/a.conf.jw-tmpl',
+]
+
 def distro(os_release_str: str) -> Distro:
     # -- The execution context is stored but not used by the properties
     # under test, so a dummy object is sufficient
@@ -150,4 +227,63 @@ for macro in Distro.macros():
     # argparse help escaping has been undone
     d.expand_macros(macro.replace('%%', '%'))
 
+# -- The rpm distro's batched enumeration --
+
+# -- Without a package filter, the selection is the whole system,
+#    which is listed without package names; with one, the selected
+#    names are listed in one command, and a selection without
+#    matches issues no listing at all
+class CannedRpmLocal(Local):
+
+    def __init__(self) -> None:
+        super().__init__('local')
+        self.cmds: list[list[str]] = []
+
+    @override
+    async def _run(
+        self,
+        cmd: list[str],
+        wd: str | None,
+        verbose: bool,
+        cmd_input: bytes | None,
+        mod_env: dict[str, str] | None,
+        interactive: bool,
+        log_prefix: str,
+    ) -> Result:
+        self.cmds.append(cmd)
+        if '-qal' in cmd:
+            out = '/usr/bin/x\n/etc/a/a.conf.jw-tmpl\n'
+        elif '-ql' in cmd:
+            out = '/etc/a/a.conf.jw-tmpl\n/etc/b/b.conf.jw-tmpl\n'
+        else:
+            out = 'a|v|p|https://jw.dev/a|m\nb|v|p|https://jw.dev/b|m\n'
+        return Result(stdout = out.encode(), status = 0)
+
+cl = CannedRpmLocal()
+sd = SuseDistro(cl, id = 'suse', os_release_str = OS_RELEASE_SUSE)
+assert asyncio.run(sd.pkg_files()) == [
+    '/usr/bin/x',
+    '/etc/a/a.conf.jw-tmpl',
+]
+assert cl.cmds[-1] == ['/usr/bin/rpm', '-qal']
+
+sdf = SuseDistro(
+    cl,
+    id = 'suse',
+    os_release_str = OS_RELEASE_SUSE,
+    default_pkg_filter = PackageFilterString('url=~jw.dev')
+)
+assert asyncio.run(sdf.pkg_files()) == \
+    ['/etc/a/a.conf.jw-tmpl', '/etc/b/b.conf.jw-tmpl']
+assert cl.cmds[-1] == ['/usr/bin/rpm', '-ql', 'a', 'b']
+
+sde = SuseDistro(
+    cl,
+    id = 'suse',
+    os_release_str = OS_RELEASE_SUSE,
+    default_pkg_filter = PackageFilterString('url=~zzz-nomatch')
+)
+assert asyncio.run(sde.pkg_files()) == []
+assert '--queryformat' in cl.cmds[-1]
+
 print('All Distro tests passed')

test/unit/python/jw/pkg/lib/pm/Makefile

@@ -0,0 +1,4 @@
+TOPDIR = ../../../../../../..
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/dirs.mk

test/unit/python/jw/pkg/lib/pm/dpkg/Makefile

@@ -0,0 +1,7 @@
+TOPDIR = ../../../../../../../..
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/py-run.mk
+
+all:
+test: run

test/unit/python/jw/pkg/lib/pm/dpkg/test.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import asyncio
+
+from typing import override
+
+from jw.pkg.lib.ec.Local import Local
+from jw.pkg.lib.pm.dpkg import list_files
+from jw.pkg.lib.Result import Result
+
+class CannedLocal(Local):
+
+    def __init__(self, stdout: str) -> None:
+        super().__init__('local')
+        self.stdout = stdout
+        self.cmds: list[list[str]] = []
+
+    @override
+    async def _run(
+        self,
+        cmd: list[str],
+        wd: str | None,
+        verbose: bool,
+        cmd_input: bytes | None,
+        mod_env: dict[str, str] | None,
+        interactive: bool,
+        log_prefix: str,
+    ) -> Result:
+        self.cmds.append(cmd)
+        return Result(stdout = self.stdout.encode(), status = 0)
+
+# -- One listing over several packages
+cl = CannedLocal('/usr/bin/b\n/etc/b/b.conf.jw-tmpl\n')
+assert asyncio.run(list_files(['a', 'b'], ec = cl)) == \
+    ['/usr/bin/b', '/etc/b/b.conf.jw-tmpl']
+assert cl.cmds[-1] == [
+    '/usr/bin/dpkg',
+    '-L',
+    'a',
+    'b',
+]
+
+# -- The unfiltered selection is 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
+class CannedTwoStage(Local):
+
+    def __init__(self, query: str, listing: str) -> None:
+        super().__init__('local')
+        self.query = query
+        self.listing = listing
+        self.cmds: list[list[str]] = []
+
+    @override
+    async def _run(
+        self,
+        cmd: list[str],
+        wd: str | None,
+        verbose: bool,
+        cmd_input: bytes | None,
+        mod_env: dict[str, str] | None,
+        interactive: bool,
+        log_prefix: str,
+    ) -> Result:
+        self.cmds.append(cmd)
+        out = self.query if '-W' in cmd else self.listing
+        return Result(stdout = out.encode(), status = 0)
+
+cl2 = CannedTwoStage('a\nb\n', 'a:\n/usr/bin/a\n\nb:\n/etc/b/b.conf.jw-tmpl\n')
+assert asyncio.run(list_files(None, ec = cl2)) == \
+    ['/usr/bin/a', '/etc/b/b.conf.jw-tmpl']
+assert cl2.cmds == [
+    ['/usr/bin/dpkg-query', '-W', '-f=${?installed:binary:Package}\n'],
+    ['/usr/bin/dpkg-query', '-L', 'a', 'b'],
+]
+
+# -- An empty selection lists nothing and runs no query
+cl3 = CannedTwoStage('', '')
+assert asyncio.run(list_files([], ec = cl3)) == []
+assert cl3.cmds == []

test/unit/python/jw/pkg/lib/pm/rpm/Makefile

@@ -0,0 +1,7 @@
+TOPDIR = ../../../../../../../..
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/py-run.mk
+
+all:
+test: run

test/unit/python/jw/pkg/lib/pm/rpm/test.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import asyncio
+
+from typing import override
+
+from jw.pkg.lib.ec.Local import Local
+from jw.pkg.lib.FlaggedFile import FlaggedFile
+from jw.pkg.lib.pm.rpm import _parse_verify_output, list_files
+from jw.pkg.lib.Result import Result
+
+# -- _parse_verify_output --
+
+# -- A clean line is not reported
+assert _parse_verify_output('pkg', '.....UG..    /usr/bin/foo\n') == []
+
+# -- Mismatches in size, mode, mtime, user, and group are noise
+assert _parse_verify_output('pkg', 'SM...UGT.  c /etc/foo\n') == []
+
+# -- A content (md5) mismatch is reported, with the flags preserved
+ret = _parse_verify_output('pkg', 'SM5....T.  c /etc/foo\n')
+assert len(ret) == 1
+assert ret[0].pkg == 'pkg'
+assert ret[0].file == '/etc/foo'
+assert ret[0].flags == 'SM5....T.'
+
+# -- The 8-character flag layout of older rpm versions is understood
+ret = _parse_verify_output('pkg', 'SM5....T /etc/foo\n')
+assert len(ret) == 1
+assert ret[0].flags == 'SM5....T'
+
+# -- A missing file is reported, config files included
+ret = _parse_verify_output('pkg', 'missing     /etc/opt/jw-ws/pi\n')
+assert len(ret) == 1
+assert ret[0].file == '/etc/opt/jw-ws/pi'
+assert ret[0].flags == 'missing'
+ret = _parse_verify_output('pkg', 'missing   c /etc/opt/jw-ws/pi/models.json\n')
+assert len(ret) == 1
+assert ret[0].file == '/etc/opt/jw-ws/pi/models.json'
+
+# -- A trailing parenthesized annotation is not part of the file
+# name
+ret = _parse_verify_output(
+    'pkg', 'missing   c /etc/phpMyAdmin/config.inc.php.jw-tmpl (Permission denied)\n'
+)
+assert len(ret) == 1
+assert ret[0].file == '/etc/phpMyAdmin/config.inc.php.jw-tmpl'
+
+# -- Lines without a file name and foreign output are ignored
+assert _parse_verify_output('pkg', '.....UG..\n') == []
+assert _parse_verify_output('pkg', 'package foo is not installed\n') == []
+assert _parse_verify_output('pkg', 'error: rpm verify failed\n') == []
+
+# -- The output of several files is accumulated
+ret = _parse_verify_output(
+    'pkg', 'SM5....T.  c /etc/foo\nmissing   c /etc/bar\n.....UG..  c /etc/baz\n'
+)
+assert [f.file for f in ret] == ['/etc/foo', '/etc/bar']
+
+# -- FlaggedFile renders as pkg file flags
+f = FlaggedFile('pkg', '/etc/foo', 'SM5....T.')
+assert repr(f) == 'pkg /etc/foo SM5....T.'
+
+print('All rpm pm tests passed')
+
+# -- list_files --
+
+class CannedLocal(Local):
+
+    def __init__(self, stdout: str) -> None:
+        super().__init__('local')
+        self.stdout = stdout
+        self.cmds: list[list[str]] = []
+
+    @override
+    async def _run(
+        self,
+        cmd: list[str],
+        wd: str | None,
+        verbose: bool,
+        cmd_input: bytes | None,
+        mod_env: dict[str, str] | None,
+        interactive: bool,
+        log_prefix: str,
+    ) -> Result:
+        self.cmds.append(cmd)
+        return Result(stdout = self.stdout.encode(), status = 0)
+
+# -- One query over several packages, the output split into lines
+cl = CannedLocal('/usr/bin/a\n/etc/a/a.conf.jw-tmpl\n')
+assert asyncio.run(list_files(['a', 'b'], ec = cl)) == \
+    ['/usr/bin/a', '/etc/a/a.conf.jw-tmpl']
+assert cl.cmds[-1] == ['/usr/bin/rpm', '-ql', 'a', 'b']
+
+# -- The unfiltered selection is one query without package names
+cl = CannedLocal('/usr/bin/a\n/etc/a/a.conf.jw-tmpl\n')
+assert asyncio.run(list_files(None, ec = cl)) == ['/usr/bin/a', '/etc/a/a.conf.jw-tmpl']
+assert cl.cmds[-1] == ['/usr/bin/rpm', '-qal']
+
+# -- An empty selection lists nothing and runs no query
+cl = CannedLocal('')
+assert asyncio.run(list_files([], ec = cl)) == []
+assert cl.cmds == []