diff --git a/tests/by_image/docker-stacks-foundation/test_outdated.py b/tests/by_image/docker-stacks-foundation/test_outdated.py index 9c9a3419..423f48ac 100644 --- a/tests/by_image/docker-stacks-foundation/test_outdated.py +++ b/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( - container: TrackedContainer, requested_only: bool = True -) -> None: +def test_outdated_packages(container: TrackedContainer, requested_only: bool) -> 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) - LOGGER.info(pkg_helper.get_outdated_summary(requested_only)) - LOGGER.info(f"\n{pkg_helper.get_outdated_table()}\n") + updatable = pkg_helper.find_updatable_packages(requested_only) + LOGGER.info(pkg_helper.get_outdated_summary(updatable, requested_only)) + LOGGER.info( + f"Outdated packages table:\n{pkg_helper.get_outdated_table(updatable)}\n" + ) diff --git a/tests/by_image/docker-stacks-foundation/test_packages.py b/tests/by_image/docker-stacks-foundation/test_packages.py index dc15d530..f3e07d96 100644 --- a/tests/by_image/docker-stacks-foundation/test_packages.py +++ b/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( - package_helper: CondaPackageHelper, package: str -) -> None: +def check_import_python_package(container: TrackedContainer, 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, - packages_to_check: Iterable[str], - check_function: Callable[[CondaPackageHelper, str], None], + container: TrackedContainer, + packages_to_check: list[str], + check_function: Callable[[TrackedContainer, str], None], ) -> None: - """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 - """ + """Test if packages can be imported""" 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 r_packages(requested_packages: dict[str, set[str]]) -> Iterable[str]: - """Return an iterable of R packages""" - return ( +def get_r_packages(package_helper: CondaPackageHelper) -> list[str]: + """Return a list of R packages""" + 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( - package_helper: CondaPackageHelper, r_packages: Iterable[str] -) -> None: +def test_r_packages(container: TrackedContainer) -> 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 python_packages(requested_packages: dict[str, set[str]]) -> Iterable[str]: - """Return an iterable of Python packages""" - return ( +def get_python_packages(package_helper: CondaPackageHelper) -> list[str]: + """Return a list of Python packages""" + 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( - package_helper: CondaPackageHelper, - python_packages: Iterable[str], -) -> None: +def test_python_packages(container: TrackedContainer) -> 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) diff --git a/tests/utils/conda_package_helper.py b/tests/utils/conda_package_helper.py index b6cd8306..818749f1 100644 --- a/tests/utils/conda_package_helper.py +++ b/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 - self.installed: dict[str, set[str]] | None = None - self.available: dict[str, set[str]] | None = None - self.comparison: list[dict[str, str]] = [] - + @cached_property 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 ...") - env_export = self.container.exec_cmd("mamba env export --no-build --json") - self.installed = CondaPackageHelper._parse_package_versions(env_export) - return self.installed + LOGGER.info("Grabbing the list of installed packages ...") + env_export = self.container.exec_cmd("mamba env export --no-build --json") + return self._parse_package_versions(env_export) + @cached_property def requested_packages(self) -> dict[str, set[str]]: """Return the requested package (i.e. `mamba install `)""" - if self.requested is None: - LOGGER.info("Grabbing the list of manually requested packages ...") - env_export = self.container.exec_cmd( - "mamba env export --no-build --json --from-history" - ) - self.requested = CondaPackageHelper._parse_package_versions(env_export) - return self.requested + LOGGER.info("Grabbing the list of manually requested packages ...") + env_export = self.container.exec_cmd( + "mamba env export --no-build --json --from-history" + ) + return self._parse_package_versions(env_export) @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) ..." - ) - # 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 + LOGGER.info("Grabbing the list of available packages (can take a while) ...") + return self._extract_available( + self.container.exec_cmd("conda search --outdated --quiet") + ) @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( - self, requested_only: bool = True - ) -> list[dict[str, str]]: + def find_updatable_packages(self, requested_only: bool) -> list[dict[str, str]]: """Check the updatable packages including or not dependencies""" - requested = self.requested_packages() - installed = self.installed_packages() - available = self.available_packages() - self.comparison = [] - for pkg, inst_vs in installed.items(): - if not requested_only or pkg in requested: - avail_vs = sorted( - list(available[pkg]), key=CondaPackageHelper.semantic_cmp - ) - if not avail_vs: - continue - current = min(inst_vs, key=CondaPackageHelper.semantic_cmp) - newest = avail_vs[-1] - if ( - 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 + updatable = [] + for pkg, inst_vs in self.installed_packages.items(): + avail_vs = self.available_packages[pkg] + if (requested_only and pkg not in self.requested_packages) or ( + not avail_vs + ): + continue + newest = sorted(avail_vs, key=CondaPackageHelper.semantic_cmp)[-1] + current = min(inst_vs, key=CondaPackageHelper.semantic_cmp) + if CondaPackageHelper.semantic_cmp( + current + ) < CondaPackageHelper.semantic_cmp(newest): + updatable.append({"Package": pkg, "Current": current, "Newest": newest}) + return updatable @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 - assert packages is not None + packages = ( + 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")