From 429658e7c88cc97b4f054bbd4c5a6606cc9f8ab6 Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Mon, 28 Aug 2023 01:26:19 +0200
Subject: [PATCH] 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.
---
 platypush/utils/manifest.py | 84 ++++++++++++++++++++-----------------
 1 file changed, 46 insertions(+), 38 deletions(-)

diff --git a/platypush/utils/manifest.py b/platypush/utils/manifest.py
index b6c5a7f7..48a8a0ab 100644
--- a/platypush/utils/manifest.py
+++ b/platypush/utils/manifest.py
@@ -20,6 +20,7 @@ from typing import (
     Optional,
     Iterable,
     Mapping,
+    Sequence,
     Set,
     Type,
     Union,
@@ -64,12 +65,46 @@ class PackageManager:
     The default distro whose configuration we should use if this package
     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. """
-    uninstall: Iterable[str] = field(default_factory=tuple)
+    uninstall: Sequence[str] = field(default_factory=tuple)
     """ The uninstall command, as a sequence of strings. """
-    get_installed: Callable[[], Iterable[str]] = lambda: []
-    """ A function that returns the list of installed packages. """
+    list: Sequence[str] = field(default_factory=tuple)
+    """ 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):
@@ -81,54 +116,27 @@ class PackageManagers(Enum):
         executable='apk',
         install=('apk', 'add', '--update', '--no-interactive', '--no-cache'),
         uninstall=('apk', 'del', '--no-interactive'),
+        list=('apk', 'list', '--installed'),
         default_os='alpine',
-        get_installed=lambda: {
-            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()
-        },
+        parse_list_line=lambda line: re.sub(r'.*\s*\{(.+?)\}\s*.*', r'\1', line),
     )
 
     APT = PackageManager(
         executable='apt',
         install=('apt', 'install', '-y'),
         uninstall=('apt', 'remove', '-y'),
+        list=('apt', 'list', '--installed'),
         default_os='debian',
-        get_installed=lambda: {
-            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()
-        },
+        parse_list_line=lambda line: line.split('/')[0],
     )
 
     PACMAN = PackageManager(
         executable='pacman',
         install=('pacman', '-S', '--noconfirm', '--needed'),
         uninstall=('pacman', '-R', '--noconfirm'),
+        list=('pacman', '-Q'),
         default_os='arch',
-        get_installed=lambda: {
-            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()
-        },
+        parse_list_line=lambda line: line.split(' ')[0],
     )
 
     @classmethod
@@ -233,7 +241,7 @@ class Dependencies:
         """
         return (
             self.install_context == InstallContext.DOCKER
-            or 'DOCKER_CTX' in os.environ
+            or bool(os.environ.get('DOCKER_CTX'))
             or os.path.isfile('/.dockerenv')
         )