Refactor CondaPackageHelper: cached_property, better typing, no state

tests/by_image/docker-stacks-foundation/test_outdated.py

@@ -10,13 +10,14 @@ from tests.utils.tracked_container import TrackedContainer
 LOGGER = logging.getLogger(__name__)
 
 
+@pytest.mark.parametrize("requested_only", [True, False])
 @pytest.mark.info
-def test_outdated_packages(
+def test_outdated_packages(container: TrackedContainer, requested_only: bool) -> None:
-    container: TrackedContainer, requested_only: bool = True
-) -> None:
     """Getting the list of updatable packages"""
     LOGGER.info(f"Checking outdated packages in {container.image_name} ...")
     pkg_helper = CondaPackageHelper(container)
-    pkg_helper.check_updatable_packages(requested_only)
+    updatable = pkg_helper.find_updatable_packages(requested_only)
-    LOGGER.info(pkg_helper.get_outdated_summary(requested_only))
+    LOGGER.info(pkg_helper.get_outdated_summary(updatable, requested_only))
-    LOGGER.info(f"\n{pkg_helper.get_outdated_table()}\n")
+    LOGGER.info(
+        f"Outdated packages table:\n{pkg_helper.get_outdated_table(updatable)}\n"
+    )

tests/by_image/docker-stacks-foundation/test_packages.py

@@ -17,11 +17,11 @@ only the requested packages i.e. packages requested by `mamba install` in the `D
 This means that it does not check dependencies.
 This choice is a tradeoff to cover the main requirements while achieving a reasonable test duration.
 However, it could be easily changed (or completed) to cover dependencies as well.
-Use `package_helper.installed_packages()` instead of `package_helper.requested_packages()`.
+Use `package_helper.installed_packages` instead of `package_helper.requested_packages`.
 """
 
 import logging
-from collections.abc import Callable, Iterable
+from collections.abc import Callable
 
 import pytest  # type: ignore
 
@@ -67,18 +67,6 @@ EXCLUDED_PACKAGES = [
 ]
 
 
-@pytest.fixture(scope="function")
-def package_helper(container: TrackedContainer) -> CondaPackageHelper:
-    """Return a package helper object that can be used to perform tests on installed packages"""
-    return CondaPackageHelper(container)
-
-
-@pytest.fixture(scope="function")
-def requested_packages(package_helper: CondaPackageHelper) -> dict[str, set[str]]:
-    """Return the list of requested packages (i.e. packages explicitly installed excluding dependencies)"""
-    return package_helper.requested_packages()
-
-
 def is_r_package(package: str) -> bool:
     """Check if a package is an R package"""
     return package.startswith("r-")
@@ -91,34 +79,28 @@ def get_package_import_name(package: str) -> str:
     return PACKAGE_MAPPING.get(package, package)
 
 
-def check_import_python_package(
+def check_import_python_package(container: TrackedContainer, package: str) -> None:
-    package_helper: CondaPackageHelper, package: str
-) -> None:
     """Try to import a Python package from the command line"""
-    package_helper.container.exec_cmd(f'python -c "import {package}"')
+    container.exec_cmd(f'python -c "import {package}"')
 
 
-def check_import_r_package(package_helper: CondaPackageHelper, package: str) -> None:
+def check_import_r_package(container: TrackedContainer, package: str) -> None:
     """Try to import an R package from the command line"""
-    package_helper.container.exec_cmd(f"R --slave -e library({package})")
+    container.exec_cmd(f"R --slave -e library({package})")
 
 
 def _check_import_packages(
-    package_helper: CondaPackageHelper,
+    container: TrackedContainer,
-    packages_to_check: Iterable[str],
+    packages_to_check: list[str],
-    check_function: Callable[[CondaPackageHelper, str], None],
+    check_function: Callable[[TrackedContainer, str], None],
 ) -> None:
-    """Test if packages can be imported
+    """Test if packages can be imported"""
-
-    Note: using a list of packages instead of a fixture for the list of packages
-    since pytest prevents the use of multiple yields
-    """
     failed_imports = []
     LOGGER.info("Testing the import of packages ...")
     for package in packages_to_check:
         LOGGER.info(f"Trying to import {package}")
         try:
-            check_function(package_helper, package)
+            check_function(container, package)
         except AssertionError as err:
             failed_imports.append(package)
             LOGGER.error(f"Failed to import package: {package}, output:\n  {err}")
@@ -126,36 +108,31 @@ def _check_import_packages(
         pytest.fail(f"following packages are not import-able: {failed_imports}")
 
 
-@pytest.fixture(scope="function")
+def get_r_packages(package_helper: CondaPackageHelper) -> list[str]:
-def r_packages(requested_packages: dict[str, set[str]]) -> Iterable[str]:
+    """Return a list of R packages"""
-    """Return an iterable of R packages"""
+    return [
-    return (
         get_package_import_name(pkg)
-        for pkg in requested_packages
+        for pkg in package_helper.requested_packages
         if is_r_package(pkg) and pkg not in EXCLUDED_PACKAGES
-    )
+    ]
 
 
-def test_r_packages(
+def test_r_packages(container: TrackedContainer) -> None:
-    package_helper: CondaPackageHelper, r_packages: Iterable[str]
-) -> None:
     """Test the import of specified R packages"""
-    _check_import_packages(package_helper, r_packages, check_import_r_package)
+    r_packages = get_r_packages(CondaPackageHelper(container))
+    _check_import_packages(container, r_packages, check_import_r_package)
 
 
-@pytest.fixture(scope="function")
+def get_python_packages(package_helper: CondaPackageHelper) -> list[str]:
-def python_packages(requested_packages: dict[str, set[str]]) -> Iterable[str]:
+    """Return a list of Python packages"""
-    """Return an iterable of Python packages"""
+    return [
-    return (
         get_package_import_name(pkg)
-        for pkg in requested_packages
+        for pkg in package_helper.requested_packages
         if not is_r_package(pkg) and pkg not in EXCLUDED_PACKAGES
-    )
+    ]
 
 
-def test_python_packages(
+def test_python_packages(container: TrackedContainer) -> None:
-    package_helper: CondaPackageHelper,
-    python_packages: Iterable[str],
-) -> None:
     """Test the import of specified python packages"""
-    _check_import_packages(package_helper, python_packages, check_import_python_package)
+    python_packages = get_python_packages(CondaPackageHelper(container))
+    _check_import_packages(container, python_packages, check_import_python_package)

tests/utils/conda_package_helper.py

@@ -26,8 +26,8 @@ import json
 import logging
 import re
 from collections import defaultdict
+from functools import cached_property
 from itertools import chain
-from typing import Any
 
 from tabulate import tabulate
 
@@ -43,28 +43,21 @@ class CondaPackageHelper:
         self.container = container
         self.container.run_detached(command=["sleep", "infinity"])
 
-        self.requested: dict[str, set[str]] | None = None
+    @cached_property
-        self.installed: dict[str, set[str]] | None = None
-        self.available: dict[str, set[str]] | None = None
-        self.comparison: list[dict[str, str]] = []
-
     def installed_packages(self) -> dict[str, set[str]]:
         """Return the installed packages"""
-        if self.installed is None:
+        LOGGER.info("Grabbing the list of installed packages ...")
-            LOGGER.info("Grabbing the list of installed packages ...")
+        env_export = self.container.exec_cmd("mamba env export --no-build --json")
-            env_export = self.container.exec_cmd("mamba env export --no-build --json")
+        return self._parse_package_versions(env_export)
-            self.installed = CondaPackageHelper._parse_package_versions(env_export)
-        return self.installed
 
+    @cached_property
     def requested_packages(self) -> dict[str, set[str]]:
         """Return the requested package (i.e. `mamba install <package>`)"""
-        if self.requested is None:
+        LOGGER.info("Grabbing the list of manually requested packages ...")
-            LOGGER.info("Grabbing the list of manually requested packages ...")
+        env_export = self.container.exec_cmd(
-            env_export = self.container.exec_cmd(
+            "mamba env export --no-build --json --from-history"
-                "mamba env export --no-build --json --from-history"
+        )
-            )
+        return self._parse_package_versions(env_export)
-            self.requested = CondaPackageHelper._parse_package_versions(env_export)
-        return self.requested
 
     @staticmethod
     def _parse_package_versions(env_export: str) -> dict[str, set[str]]:
@@ -91,20 +84,16 @@ class CondaPackageHelper:
             packages_dict[package] = version
         return packages_dict
 
+    @cached_property
     def available_packages(self) -> dict[str, set[str]]:
         """Return the available packages"""
-        if self.available is None:
+        LOGGER.info("Grabbing the list of available packages (can take a while) ...")
-            LOGGER.info(
+        return self._extract_available(
-                "Grabbing the list of available packages (can take a while) ..."
+            self.container.exec_cmd("conda search --outdated --quiet")
-            )
+        )
-            # Keeping command line output since `mamba search --outdated --json` is way too long ...
-            self.available = CondaPackageHelper._extract_available(
-                self.container.exec_cmd("conda search --outdated --quiet")
-            )
-        return self.available
 
     @staticmethod
-    def _extract_available(lines: str) -> dict[str, set[str]]:
+    def _extract_available(lines: str) -> defaultdict[str, set[str]]:
         """Extract packages and versions from the lines returned by the list of packages"""
         ddict = defaultdict(set)
         for line in lines.splitlines()[2:]:
@@ -114,39 +103,28 @@ class CondaPackageHelper:
             ddict[pkg].add(version)
         return ddict
 
-    def check_updatable_packages(
+    def find_updatable_packages(self, requested_only: bool) -> list[dict[str, str]]:
-        self, requested_only: bool = True
-    ) -> list[dict[str, str]]:
         """Check the updatable packages including or not dependencies"""
-        requested = self.requested_packages()
+        updatable = []
-        installed = self.installed_packages()
+        for pkg, inst_vs in self.installed_packages.items():
-        available = self.available_packages()
+            avail_vs = self.available_packages[pkg]
-        self.comparison = []
+            if (requested_only and pkg not in self.requested_packages) or (
-        for pkg, inst_vs in installed.items():
+                not avail_vs
-            if not requested_only or pkg in requested:
+            ):
-                avail_vs = sorted(
+                continue
-                    list(available[pkg]), key=CondaPackageHelper.semantic_cmp
+            newest = sorted(avail_vs, key=CondaPackageHelper.semantic_cmp)[-1]
-                )
+            current = min(inst_vs, key=CondaPackageHelper.semantic_cmp)
-                if not avail_vs:
+            if CondaPackageHelper.semantic_cmp(
-                    continue
+                current
-                current = min(inst_vs, key=CondaPackageHelper.semantic_cmp)
+            ) < CondaPackageHelper.semantic_cmp(newest):
-                newest = avail_vs[-1]
+                updatable.append({"Package": pkg, "Current": current, "Newest": newest})
-                if (
+        return updatable
-                    avail_vs
-                    and current != newest
-                    and CondaPackageHelper.semantic_cmp(current)
-                    < CondaPackageHelper.semantic_cmp(newest)
-                ):
-                    self.comparison.append(
-                        {"Package": pkg, "Current": current, "Newest": newest}
-                    )
-        return self.comparison
 
     @staticmethod
-    def semantic_cmp(version_string: str) -> Any:
+    def semantic_cmp(version_string: str) -> tuple[int, ...]:
         """Manage semantic versioning for comparison"""
 
-        def my_split(string: str) -> list[Any]:
+        def my_split(string: str) -> list[list[str]]:
             def version_substrs(x: str) -> list[str]:
                 return re.findall(r"([A-z]+|\d+)", x)
 
@@ -168,15 +146,18 @@ class CondaPackageHelper:
         mss = list(chain(*my_split(version_string)))
         return tuple(map(try_int, mss))
 
-    def get_outdated_summary(self, requested_only: bool = True) -> str:
+    def get_outdated_summary(
+        self, updatable: list[dict[str, str]], requested_only: bool
+    ) -> str:
         """Return a summary of outdated packages"""
-        packages = self.requested if requested_only else self.installed
+        packages = (
-        assert packages is not None
+            self.requested_packages if requested_only else self.installed_packages
+        )
         nb_packages = len(packages)
-        nb_updatable = len(self.comparison)
+        nb_updatable = len(updatable)
         updatable_ratio = nb_updatable / nb_packages
         return f"{nb_updatable}/{nb_packages} ({updatable_ratio:.0%}) packages could be updated"
 
-    def get_outdated_table(self) -> str:
+    def get_outdated_table(self, updatable: list[dict[str, str]]) -> str:
         """Return a table of outdated packages"""
-        return tabulate(self.comparison, headers="keys")
+        return tabulate(updatable, headers="keys")