Refactored PackageManager classes.

Instead of having a custom `get_installed` callable field, with
replicated code for each package manager, the field has now been
promoted to a class method containing the common logic, and the
instances now expect a `list` field (base command to list the installed
packages using the specified package manager) and a `parse_list_line`
callback field (to extract the base package name given a raw line from
the command above).

Also, we shouldn't run the list command if we're running within a Docker
context - the host and container environments will be different.
This commit is contained in:
Fabio Manganiello 2023-08-28 01:26:19 +02:00
parent 86e5f74645
commit 429658e7c8
Signed by: blacklight
GPG key ID: D90FBA7F76362774

View file

@ -20,6 +20,7 @@ from typing import (
Optional, Optional,
Iterable, Iterable,
Mapping, Mapping,
Sequence,
Set, Set,
Type, Type,
Union, Union,
@ -64,12 +65,46 @@ class PackageManager:
The default distro whose configuration we should use if this package The default distro whose configuration we should use if this package
manager is detected. manager is detected.
""" """
install: Iterable[str] = field(default_factory=tuple) install: Sequence[str] = field(default_factory=tuple)
""" The install command, as a sequence of strings. """ """ The install command, as a sequence of strings. """
uninstall: Iterable[str] = field(default_factory=tuple) uninstall: Sequence[str] = field(default_factory=tuple)
""" The uninstall command, as a sequence of strings. """ """ The uninstall command, as a sequence of strings. """
get_installed: Callable[[], Iterable[str]] = lambda: [] list: Sequence[str] = field(default_factory=tuple)
""" A function that returns the list of installed packages. """ """ The command to list the installed packages. """
parse_list_line: Callable[[str], str] = field(default_factory=lambda: lambda s: s)
"""
Internal package-manager dependent function that parses the base package
name from a line returned by the list command.
"""
def _get_installed(self) -> Sequence[str]:
"""
:return: The install context-aware list of installed packages.
It should only used within the context of :meth:`.get_installed`.
"""
if os.environ.get('DOCKER_CTX'):
# If we're running in a Docker build context, don't run the package
# manager to retrieve the list of installed packages, as the host
# and guest systems have different environments.
return ()
return tuple(
line.strip()
for line in subprocess.Popen( # pylint: disable=consider-using-with
self.list, stdout=subprocess.PIPE
)
.communicate()[0]
.decode()
.split('\n')
if line.strip()
)
def get_installed(self) -> Sequence[str]:
"""
:return: The list of installed packages.
"""
return tuple(self.parse_list_line(line) for line in self._get_installed())
class PackageManagers(Enum): class PackageManagers(Enum):
@ -81,54 +116,27 @@ class PackageManagers(Enum):
executable='apk', executable='apk',
install=('apk', 'add', '--update', '--no-interactive', '--no-cache'), install=('apk', 'add', '--update', '--no-interactive', '--no-cache'),
uninstall=('apk', 'del', '--no-interactive'), uninstall=('apk', 'del', '--no-interactive'),
list=('apk', 'list', '--installed'),
default_os='alpine', default_os='alpine',
get_installed=lambda: { parse_list_line=lambda line: re.sub(r'.*\s*\{(.+?)\}\s*.*', r'\1', line),
re.sub(r'.*\s*\{(.+?)\}\s*.*', r'\1', line)
for line in (
line.strip()
for line in subprocess.Popen( # pylint: disable=consider-using-with
['apk', 'list', '--installed'], stdout=subprocess.PIPE
)
.communicate()[0]
.decode()
.split('\n')
)
if line.strip()
},
) )
APT = PackageManager( APT = PackageManager(
executable='apt', executable='apt',
install=('apt', 'install', '-y'), install=('apt', 'install', '-y'),
uninstall=('apt', 'remove', '-y'), uninstall=('apt', 'remove', '-y'),
list=('apt', 'list', '--installed'),
default_os='debian', default_os='debian',
get_installed=lambda: { parse_list_line=lambda line: line.split('/')[0],
line.strip().split('/')[0]
for line in subprocess.Popen( # pylint: disable=consider-using-with
['apt', 'list', '--installed'], stdout=subprocess.PIPE
)
.communicate()[0]
.decode()
.split('\n')
if line.strip()
},
) )
PACMAN = PackageManager( PACMAN = PackageManager(
executable='pacman', executable='pacman',
install=('pacman', '-S', '--noconfirm', '--needed'), install=('pacman', '-S', '--noconfirm', '--needed'),
uninstall=('pacman', '-R', '--noconfirm'), uninstall=('pacman', '-R', '--noconfirm'),
list=('pacman', '-Q'),
default_os='arch', default_os='arch',
get_installed=lambda: { parse_list_line=lambda line: line.split(' ')[0],
line.strip().split(' ')[0]
for line in subprocess.Popen( # pylint: disable=consider-using-with
['pacman', '-Q'], stdout=subprocess.PIPE
)
.communicate()[0]
.decode()
.split('\n')
if line.strip()
},
) )
@classmethod @classmethod
@ -233,7 +241,7 @@ class Dependencies:
""" """
return ( return (
self.install_context == InstallContext.DOCKER self.install_context == InstallContext.DOCKER
or 'DOCKER_CTX' in os.environ or bool(os.environ.get('DOCKER_CTX'))
or os.path.isfile('/.dockerenv') or os.path.isfile('/.dockerenv')
) )