lib.pm.*.install_packages(): Add functions

Currently, jw-pkg install is sometimes left to the high-level package
manager, which is allowed to install dependencies along, and that's
sometimes unwanted, but there's no backend in the low-level package manager
module lib.pm to allow specifying that. This commit adds it:
lib.pm.*.install_packages().

The functions install a list of local package files using the low-level
package managers rpm / dpkg, bypassing the high-level package managers yum
/ apt / zypper.

pm.rpm runs rpm -U and pm.dpkg runs dpkg -i. Every file is classified
against the installed state of its package, queried by name: a file
for a package that is not installed is installed fresh, a file with
another version installed is upgraded, and a file whose version is already
installed is reinstalled when force is set and skipped otherwise. On rpm,
plain -U refuses a package that is already installed, and --reinstall
aborts the transaction when a file of the incoming package differs from
the installed one, so the upgrades run before the reinstalls and the
same-version files need force.

With only_update, a file for a package that is not installed raises before
anything is installed.

The dpkg route is gated on an apt simulation, because dpkg -i leaves a
half-configured package when a dependency is missing: the simulation runs
first and the install is refused when it would install or remove anything
but the given files' own packages.

Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.4.0
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-10-27 14:51:56 +01:00
commit 36978c55e2
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
2 changed files with 659 additions and 2 deletions

View file

@ -5,7 +5,7 @@ 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.pm.dpkg import install_packages, list_files
from jw.pkg.lib.Result import Result
class CannedLocal(Local):
@ -79,3 +79,261 @@ assert cl2.cmds == [
cl3 = CannedTwoStage('', '')
assert asyncio.run(list_files([], ec = cl3)) == []
assert cl3.cmds == []
# -- install_packages --
class InstallLocal(Local):
"""A canned execution context that answers dpkg queries and
the apt simulation"""
def __init__(
self,
files: dict[str, str],
installed: dict[str, str],
simulate: str,
simulate_status: int = 0,
) -> None:
super().__init__('local')
self.files = files
self.installed = installed
self.simulate = simulate
self.simulate_status = simulate_status
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 '-I' in cmd:
return Result(stdout = self.files.get(cmd[-1], '').encode(), status = 0)
if cmd[0].endswith('dpkg-query'):
name = cmd[-1]
return Result(
stdout = self.installed.get(name, '').encode(),
status = 0 if name in self.installed else 1,
)
if '/usr/bin/apt-get' in cmd:
return Result(
stdout = self.simulate.encode(), status = self.simulate_status
)
return Result(stdout = b'', status = 0)
def dpkg_args(cmd: list[str]) -> list[str]:
"""The dpkg command without the sudo prefix and its options"""
return cmd[cmd.index('/usr/bin/dpkg'):]
# -- Without force or only_update, all files go through a single
# plain -i, gated on the apt simulation
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-41'},
simulate = 'Inst jw-base [1.0.0-42]\n',
)
asyncio.run(install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il))
sim = [cmd for cmd in il.cmds if '/usr/bin/apt-get' in cmd]
assert len(sim) == 1
args = sim[0][sim[0].index('/usr/bin/apt-get'):]
assert '-s' in args
assert '--no-install-recommends' in args
assert args[-1] == '/a/jw-base-run-1.0.0-42.deb'
installs = [dpkg_args(cmd) for cmd in il.cmds if '-i' in cmd]
assert installs == [['/usr/bin/dpkg', '-i', '/a/jw-base-run-1.0.0-42.deb']]
# -- Without force, a file whose version is already installed is
# skipped
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-42'},
simulate = '',
)
asyncio.run(install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il))
assert not [cmd for cmd in il.cmds if '-i' in cmd]
assert not [cmd for cmd in il.cmds if '/usr/bin/apt-get' in cmd]
# -- An empty selection runs no commands
il = InstallLocal(files = {}, installed = {}, simulate = '')
assert asyncio.run(install_packages([], ec = il)) is None
assert il.cmds == []
# -- force reinstalls a file whose version is already installed
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-42'},
simulate = '',
)
asyncio.run(install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True))
assert dpkg_args(il.cmds[-1]) == [
'/usr/bin/dpkg',
'-i',
'--force-reinstall',
'/a/jw-base-run-1.0.0-42.deb',
]
# -- force upgrades a file whose package is installed in another
# version with plain -i
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-41'},
simulate = 'Inst jw-base [1.0.0-42]\n',
)
asyncio.run(install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True))
assert dpkg_args(il.cmds[-1]) == ['/usr/bin/dpkg', '-i', '/a/jw-base-run-1.0.0-42.deb']
# -- force installs a file whose package is not installed with
# plain -i
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {},
simulate = 'Inst jw-base [1.0.0-42]\n',
)
asyncio.run(install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True))
assert dpkg_args(il.cmds[-1]) == ['/usr/bin/dpkg', '-i', '/a/jw-base-run-1.0.0-42.deb']
# -- A file that cannot be queried raises before the simulation
# or the install
il = InstallLocal(
files = {},
installed = {},
simulate = '',
)
try:
asyncio.run(install_packages(['/a/broken.deb'], ec = il, force = True))
except Exception as e:
assert 'package name and version' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in il.cmds if '-i' in cmd]
assert not [cmd for cmd in il.cmds if '/usr/bin/apt-get' in cmd]
# -- only_update raises on files whose package is not installed,
# before simulating or installing anything
il = InstallLocal(
files = {
'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42',
'/a/jw-foo-run-1.0.0-42.deb': 'Package: jw-foo\nVersion: 1.0.0-42',
},
installed = {'jw-base': 'jw-base|||||1.0.0-42'},
simulate = '',
)
try:
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.deb', '/a/jw-foo-run-1.0.0-42.deb'],
ec = il,
force = True,
only_update = True,
)
)
except Exception as e:
assert 'only_update' in str(e)
assert '/a/jw-foo-run-1.0.0-42.deb' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in il.cmds if '-i' in cmd]
assert not [cmd for cmd in il.cmds if '/usr/bin/apt-get' in cmd]
# -- only_update installs an update of an installed package
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-41'},
simulate = 'Inst jw-base [1.0.0-42]\n',
)
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True, only_update = True
)
)
assert dpkg_args(il.cmds[-1]) == ['/usr/bin/dpkg', '-i', '/a/jw-base-run-1.0.0-42.deb']
# -- force runs two installs for a mixed selection, the upgrades
# first
il = InstallLocal(
files = {
'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42',
'/a/jw-base-run-1.0.0-41.deb': 'Package: jw-base\nVersion: 1.0.0-41',
},
installed = {'jw-base': 'jw-base|||||1.0.0-42'},
simulate = 'Inst jw-base [1.0.0-41]\n',
)
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.deb', '/a/jw-base-run-1.0.0-41.deb'],
ec = il,
force = True,
)
)
installs = [dpkg_args(cmd) for cmd in il.cmds if '-i' in cmd]
assert installs == [
['/usr/bin/dpkg', '-i', '/a/jw-base-run-1.0.0-41.deb'],
[
'/usr/bin/dpkg',
'-i',
'--force-reinstall',
'/a/jw-base-run-1.0.0-42.deb',
],
]
# -- The simulation refuses the install when it would pull a
# dependency
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {},
simulate = 'Inst jw-base [1.0.0-42]\nInst jw-foo [2.0]\n',
)
try:
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True)
)
except Exception as e:
assert 'would install' in str(e)
assert 'jw-foo' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in il.cmds if '-i' in cmd]
# -- The simulation refuses the install when it would remove a
# package
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {},
simulate = 'Inst jw-base [1.0.0-42]\nRemv jw-old [1.0]\n',
)
try:
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True)
)
except Exception as e:
assert 'remove' in str(e)
assert 'jw-old' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in il.cmds if '-i' in cmd]
# -- A failed simulation, for instance unmet dependencies, aborts
# with its output
il = InstallLocal(
files = {'/a/jw-base-run-1.0.0-42.deb': 'Package: jw-base\nVersion: 1.0.0-42'},
installed = {},
simulate = (
'The following packages have unmet dependencies:\n'
' jw-base : Depends: jw-foo but it is not going to be installed\n'
),
simulate_status = 100,
)
try:
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.deb'], ec = il, force = True)
)
except Exception as e:
assert 'unmet dependencies' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in il.cmds if '-i' in cmd]

View file

@ -6,7 +6,7 @@ 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.pm.rpm import _parse_verify_output, install_packages, list_files
from jw.pkg.lib.Result import Result
# -- _parse_verify_output --
@ -101,3 +101,234 @@ assert cl.cmds[-1] == ['/usr/bin/rpm', '-qal']
cl = CannedLocal('')
assert asyncio.run(list_files([], ec = cl)) == []
assert cl.cmds == []
# -- install_packages --
def rpm_args(cmd: list[str]) -> list[str]:
"""The rpm command without the sudo prefix and its options"""
return cmd[cmd.index('/usr/bin/rpm'):]
class DispatchLocal(Local):
"""A canned execution context that answers rpm queries"""
def __init__(self, files: dict[str, str], installed: dict[str, str]) -> None:
super().__init__('local')
self.files = files
self.installed = installed
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 '-qp' in cmd:
return Result(stdout = self.files.get(cmd[-1], '').encode(), status = 0)
if '-q' in cmd:
name = cmd[-1]
return Result(
stdout = self.installed.get(name, '').encode(),
status = 0 if name in self.installed else 1,
)
return Result(stdout = b'', status = 0)
# -- Without force, a file whose version is already installed is
# skipped, because plain -U refuses it
dl = DispatchLocal(
files = {'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-42\n'},
)
asyncio.run(install_packages(['/a/jw-base-run-1.0.0-42.x86_64.rpm'], ec = dl))
assert not [cmd for cmd in dl.cmds if '-U' in cmd]
# -- Without force, fresh and update files go through a single
# plain -U
dl = DispatchLocal(
files = {
'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42',
'/b/jw-foo-run-1.0.0-42.x86_64.rpm': 'jw-foo\t1.0.0-42',
},
installed = {'jw-base': 'jw-base|||||1.0.0-41\n'},
)
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.x86_64.rpm', '/b/jw-foo-run-1.0.0-42.x86_64.rpm'],
ec = dl,
)
)
installs = [cmd for cmd in dl.cmds if '-U' in cmd]
assert len(installs) == 1
assert installs[0][0] == '/usr/bin/sudo'
assert rpm_args(installs[0]) == [
'/usr/bin/rpm',
'-U',
'/a/jw-base-run-1.0.0-42.x86_64.rpm',
'/b/jw-foo-run-1.0.0-42.x86_64.rpm',
]
# -- An empty selection installs nothing and runs no command
dl = DispatchLocal(files = {}, installed = {})
asyncio.run(install_packages([], ec = dl))
assert dl.cmds == []
# -- With force, a file whose version is already installed is reinstalled
dl = DispatchLocal(
files = {'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-42\n'},
)
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.x86_64.rpm'], ec = dl, force = True)
)
assert rpm_args(dl.cmds[-1]) == [
'/usr/bin/rpm',
'-U',
'--reinstall',
'/a/jw-base-run-1.0.0-42.x86_64.rpm',
]
# -- With force, a file with another version installed is a plain upgrade
dl = DispatchLocal(
files = {'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-41\n'},
)
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.x86_64.rpm'], ec = dl, force = True)
)
assert rpm_args(dl.cmds[-1]) == [
'/usr/bin/rpm',
'-U',
'/a/jw-base-run-1.0.0-42.x86_64.rpm',
]
# -- With force, a file for a package that is not installed is a plain
# install
dl = DispatchLocal(
files = {'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42'},
installed = {},
)
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.x86_64.rpm'], ec = dl, force = True)
)
assert rpm_args(dl.cmds[-1]) == [
'/usr/bin/rpm',
'-U',
'/a/jw-base-run-1.0.0-42.x86_64.rpm',
]
# -- An unqueryable file raises before anything is installed
dl = DispatchLocal(files = {}, installed = {})
try:
asyncio.run(
install_packages(['/a/jw-base-run-1.0.0-42.x86_64.rpm'], ec = dl, force = True)
)
except Exception:
pass
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in dl.cmds if '-U' in cmd]
# -- With force, a mixed transaction runs two installs, the upgrades first
dl = DispatchLocal(
files = {
'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42',
'/a/jw-base-run-1.0.0-41.x86_64.rpm': 'jw-base\t1.0.0-41',
},
installed = {'jw-base': 'jw-base|||||1.0.0-42\n'},
)
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.x86_64.rpm', '/a/jw-base-run-1.0.0-41.x86_64.rpm'],
ec = dl,
force = True,
)
)
installs = [rpm_args(cmd) for cmd in dl.cmds if '-U' in cmd]
assert installs == [
['/usr/bin/rpm', '-U', '/a/jw-base-run-1.0.0-41.x86_64.rpm'],
[
'/usr/bin/rpm',
'-U',
'--reinstall',
'/a/jw-base-run-1.0.0-42.x86_64.rpm',
],
]
# -- With only_update, a file whose package is not installed raises
# before anything is installed
dl = DispatchLocal(
files = {
'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42',
'/a/jw-foo-run-1.0.0-42.x86_64.rpm': 'jw-foo\t1.0.0-42',
},
installed = {'jw-base': 'jw-base|||||1.0.0-42\n'},
)
try:
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.x86_64.rpm', '/a/jw-foo-run-1.0.0-42.x86_64.rpm'],
ec = dl,
force = True,
only_update = True,
)
)
except Exception as e:
assert 'only_update' in str(e)
assert '/a/jw-foo-run-1.0.0-42.x86_64.rpm' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in dl.cmds if '-U' in cmd]
# -- With only_update, an update of an installed package is installed
dl = DispatchLocal(
files = {'/a/jw-base-run-1.0.0-43.x86_64.rpm': 'jw-base\t1.0.0-43'},
installed = {'jw-base': 'jw-base|||||1.0.0-42\n'},
)
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-43.x86_64.rpm'],
ec = dl,
force = True,
only_update = True,
)
)
installs = [rpm_args(cmd) for cmd in dl.cmds if '-U' in cmd]
assert installs == [
['/usr/bin/rpm', '-U', '/a/jw-base-run-1.0.0-43.x86_64.rpm'],
]
# -- With only_update and force off, a file whose version is
# already installed is skipped
dl = DispatchLocal(
files = {'/a/jw-base-run-1.0.0-42.x86_64.rpm': 'jw-base\t1.0.0-42'},
installed = {'jw-base': 'jw-base|||||1.0.0-42\n'},
)
asyncio.run(
install_packages(
['/a/jw-base-run-1.0.0-42.x86_64.rpm'], ec = dl, only_update = True
)
)
assert not [cmd for cmd in dl.cmds if '-U' in cmd]
# -- With only_update, all files absent raises without installing
dl = DispatchLocal(
files = {'/a/jw-foo-run-1.0.0-42.x86_64.rpm': 'jw-foo\t1.0.0-42'},
installed = {},
)
try:
asyncio.run(
install_packages(
['/a/jw-foo-run-1.0.0-42.x86_64.rpm'], ec = dl, only_update = True
)
)
except Exception as e:
assert 'only_update' in str(e)
else:
raise AssertionError('expected an exception')
assert not [cmd for cmd in dl.cmds if '-U' in cmd]