Merge pull request #1580 from mathbunnyru/asalikhov/fix_typing

Fix all typing issues
This commit is contained in:
Ayaz Salikhov
2022-01-24 19:14:11 +03:00
committed by GitHub
25 changed files with 184 additions and 129 deletions

View File

@@ -27,7 +27,8 @@ from collections import defaultdict
from itertools import chain
import logging
import json
from typing import Optional
from typing import Any, Optional
from docker.models.containers import Container
from tabulate import tabulate
@@ -40,14 +41,16 @@ class CondaPackageHelper:
"""Conda package helper permitting to get information about packages"""
def __init__(self, container: TrackedContainer):
self.running_container = CondaPackageHelper.start_container(container)
self.running_container: Container = CondaPackageHelper.start_container(
container
)
self.requested: Optional[dict[str, set[str]]] = None
self.installed: Optional[dict[str, set[str]]] = None
self.available: Optional[dict[str, set[str]]] = None
self.comparison: list[dict[str, str]] = []
@staticmethod
def start_container(container: TrackedContainer):
def start_container(container: TrackedContainer) -> Container:
"""Start the TrackedContainer and return an instance of a running container"""
LOGGER.info(f"Starting container {container.image_name} ...")
return container.run_detached(
@@ -85,13 +88,13 @@ class CondaPackageHelper:
)
return self.requested
def _execute_command(self, command):
def _execute_command(self, command: list[str]) -> str:
"""Execute a command on a running container"""
rc = self.running_container.exec_run(command)
return rc.output.decode("utf-8")
return rc.output.decode("utf-8") # type: ignore
@staticmethod
def _packages_from_json(env_export) -> dict[str, set[str]]:
def _packages_from_json(env_export: str) -> dict[str, set[str]]:
"""Extract packages and versions from the lines returned by the list of specifications"""
# dependencies = filter(lambda x: isinstance(x, str), json.loads(env_export).get("dependencies"))
dependencies = json.loads(env_export).get("dependencies")
@@ -114,7 +117,7 @@ class CondaPackageHelper:
packages_dict[package] = version
return packages_dict
def available_packages(self):
def available_packages(self) -> dict[str, set[str]]:
"""Return the available packages"""
if self.available is None:
LOGGER.info("Grabing the list of available packages (can take a while) ...")
@@ -125,11 +128,13 @@ class CondaPackageHelper:
return self.available
@staticmethod
def _extract_available(lines):
def _extract_available(lines: str) -> dict[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:]:
pkg, version = re.match(r"^(\S+)\s+(\S+)", line, re.MULTILINE).groups()
match = re.match(r"^(\S+)\s+(\S+)", line, re.MULTILINE)
assert match is not None
pkg, version = match.groups()
ddict[pkg].add(version)
return ddict
@@ -162,11 +167,11 @@ class CondaPackageHelper:
return self.comparison
@staticmethod
def semantic_cmp(version_string: str):
def semantic_cmp(version_string: str) -> Any:
"""Manage semantic versioning for comparison"""
def mysplit(string):
def version_substrs(x):
def mysplit(string: str) -> list[Any]:
def version_substrs(x: str) -> list[str]:
return re.findall(r"([A-z]+|\d+)", x)
return list(chain(map(version_substrs, string.split("."))))
@@ -189,7 +194,9 @@ class CondaPackageHelper:
def get_outdated_summary(self, requested_only: bool = True) -> str:
"""Return a summary of outdated packages"""
nb_packages = len(self.requested if requested_only else self.installed)
packages = self.requested if requested_only else self.installed
assert packages is not None
nb_packages = len(packages)
nb_updatable = len(self.comparison)
updatable_ratio = nb_updatable / nb_packages
return f"{nb_updatable}/{nb_packages} ({updatable_ratio:.0%}) packages could be updated"

View File

@@ -3,7 +3,7 @@
import logging
import pytest
import pytest # type: ignore
from conftest import TrackedContainer
from package_helper import CondaPackageHelper
@@ -12,7 +12,9 @@ LOGGER = logging.getLogger(__name__)
@pytest.mark.info
def test_outdated_packages(container: TrackedContainer, requested_only: bool = True):
def test_outdated_packages(
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)

View File

@@ -37,8 +37,9 @@ Example:
import logging
import pytest
import pytest # type: ignore
from conftest import TrackedContainer
from typing import Callable, Iterable
from package_helper import CondaPackageHelper
@@ -86,7 +87,7 @@ def packages(package_helper: CondaPackageHelper) -> dict[str, set[str]]:
return package_helper.requested_packages()
def package_map(package: str) -> str:
def get_package_import_name(package: str) -> str:
"""Perform a mapping between the python package name and the name used for the import"""
return PACKAGE_MAPPING.get(package, package)
@@ -112,7 +113,7 @@ def _check_import_package(
"""Generic function executing a command"""
LOGGER.debug(f"Trying to import a package with [{command}] ...")
rc = package_helper.running_container.exec_run(command)
return rc.exit_code
return rc.exit_code # type: ignore
def check_import_python_package(
@@ -129,10 +130,10 @@ def check_import_r_package(package_helper: CondaPackageHelper, package: str) ->
)
def _import_packages(
def _check_import_packages(
package_helper: CondaPackageHelper,
filtered_packages: dict[str, set[str]],
check_function,
filtered_packages: Iterable[str],
check_function: Callable[[CondaPackageHelper, str], int],
max_failures: int,
) -> None:
"""Test if packages can be imported
@@ -156,33 +157,36 @@ def _import_packages(
@pytest.fixture(scope="function")
def r_packages(packages: dict[str, set[str]]):
def r_packages(packages: dict[str, set[str]]) -> Iterable[str]:
"""Return an iterable of R packages"""
# package[2:] is to remove the leading "r-" appended on R packages
return map(
lambda package: package_map(package[2:]), filter(r_package_predicate, packages)
lambda package: get_package_import_name(package[2:]),
filter(r_package_predicate, packages),
)
def test_r_packages(
package_helper: CondaPackageHelper, r_packages, max_failures: int = 0
):
package_helper: CondaPackageHelper, r_packages: Iterable[str], max_failures: int = 0
) -> None:
"""Test the import of specified R packages"""
return _import_packages(
_check_import_packages(
package_helper, r_packages, check_import_r_package, max_failures
)
@pytest.fixture(scope="function")
def python_packages(packages: dict[str, set[str]]):
def python_packages(packages: dict[str, set[str]]) -> Iterable[str]:
"""Return an iterable of Python packages"""
return map(package_map, filter(python_package_predicate, packages))
return map(get_package_import_name, filter(python_package_predicate, packages))
def test_python_packages(
package_helper: CondaPackageHelper, python_packages, max_failures: int = 0
):
package_helper: CondaPackageHelper,
python_packages: Iterable[str],
max_failures: int = 0,
) -> None:
"""Test the import of specified python packages"""
return _import_packages(
_check_import_packages(
package_helper, python_packages, check_import_python_package, max_failures
)