janware/jw-pkg
Watch
1
Code Pull requests Activity Actions

cmds.pkg.CmdVerify: Add command #142

Merged
Jan Lindemann merged 3 commits from jan/feature/20261016-cmds-pkg-cmdverify-add-command into master 2026-10-16 09:53:58 +02:00 AGit
Conversation 0 Commits 3 Files changed 17 +615 -33
17 changed files with 615 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

29
src/python/jw/pkg/cmds/pkg/CmdVerify.py Normal file
View file

@ -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)

42
src/python/jw/pkg/lib/Distro.py
View file

@ -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]:
if not name:
log(WARNING, 'No package specified for inspection')
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)
# -- 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))

21
src/python/jw/pkg/lib/FlaggedFile.py Normal file
View file

@ -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}'

9
src/python/jw/pkg/lib/distros/arch/Distro.py
View file

@ -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'
)

11
src/python/jw/pkg/lib/distros/debian/Distro.py
View file

@ -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]:
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)
@override
async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
return await verify_packages(names, ec = self.ctx)

11
src/python/jw/pkg/lib/distros/redhat/Distro.py
View file

@ -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]:
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)
@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(

11
src/python/jw/pkg/lib/distros/suse/Distro.py
View file

@ -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]:
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)
@override
async def _verify(self, names: Iterable[str]) -> Iterable[FlaggedFile]:
return await verify_packages(names, ec = self.ctx)

76
src/python/jw/pkg/lib/pm/dpkg.py
View file

@ -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)
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 +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]:
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()
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

77
src/python/jw/pkg/lib/pm/rpm.py
View file

@ -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],
throw = True,
sudo = False,
mode = InputMode.NonInteractive,
ec = ec
args, throw = True, sudo = False, mode = InputMode.NonInteractive, ec = ec
)
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
)
ret.extend(_parse_verify_output(name, output))
return ret

10
test/integration/jw_pkg/help/test-expected.txt
View file

@ -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

9
test/integration/jw_pkg/secrets/test.py
View file

@ -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

140
test/unit/python/jw/pkg/lib/Distro/test.py
View file

@ -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')

4
test/unit/python/jw/pkg/lib/pm/Makefile Normal file
View file

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

7
test/unit/python/jw/pkg/lib/pm/dpkg/Makefile Normal file
View file

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

81
test/unit/python/jw/pkg/lib/pm/dpkg/test.py Normal file
View file

@ -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 == []

7
test/unit/python/jw/pkg/lib/pm/rpm/Makefile Normal file
View file

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

103
test/unit/python/jw/pkg/lib/pm/rpm/test.py Normal file
View file

@ -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 == []