Code indexing in gitaly is broken and leads to code not being visible to the user. We work on the issue with highest priority.

Skip to content
Snippets Groups Projects

Draft: feat: add package tests to for night runs to test min, latest and random dep. packages

Closed appel_c requested to merge feat/run_ci_against_various_packages_nox into main
Files
4
+ 187
0
""" Module for handling package versions from pyproject.toml file. """
import enum
import os
import random
import warnings
from copy import deepcopy
from typing import DefaultDict, Literal
import tomlkit.api
from packaging.requirements import Requirement
from packaging.version import Version
from piptools.repositories.pypi import PyPIRepository
from tomlkit.toml_document import TOMLDocument
class Operation(str, enum.Enum):
"""Operator class."""
MIN = "min"
MAX = "max"
RANDOM = "random"
class PackageVersionHandler:
"""PackageVersionHandler class."""
def __init__(self, toml_filepath: str, include_optional_dep: bool = True):
"""Initialize the PackageVersionHandler class.
Ths class should be used as a context manager
This class is used to load available package versions from PyPI
based on the dependencies in the pyproject.toml file.
It can provide the minimum required versions for each package,
the maximum required versions for each package or a random
distribution of versions for each package used for CI pipelines.
Args:
toml_filepath (str): Path to the pyproject.toml file.
include_optional_dep (bool): Include optional dependencies. Defaults to True.
Example:
>>> with PackageVersionHandler(toml_file="./pyproject.toml") as handler:
>>> toml_min = handler.get_toml_min_req()
>>> toml_latest = handler.get_toml_latest_req()
>>> toml_random = handler.get_toml_random_req()
>>> handler.write_toml_file("requirements_min.txt", min_req)
"""
self.repo = PyPIRepository(pip_args=[""], cache_dir="")
self._toml_filepath = toml_filepath
self._project_toml_save_copy = None
self._available_candidates = {}
self._pkg_dep = {}
self._include_optional_dep = include_optional_dep
def __enter__(self):
self._project_toml_save_copy = self.load_toml_file(self._toml_filepath)
self.run()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.write_toml_file(self._toml_filepath, self._project_toml_save_copy)
def get_project_toml(self):
"""Create copy from the save_copy of the project toml file."""
return deepcopy(self._project_toml_save_copy)
@property
def available_candidates(self) -> dict:
"""Returns dictionary with available candidates for each package."""
return self._available_candidates
def get_toml(self, operation: Operation) -> TOMLDocument:
"""Returns the minimum required versions for each package.
Args:
operation (Operation): Option to pick "min", "max" or "random" versions.
"""
toml = self.get_project_toml()
dependencies = self._select_versions(operation=operation)
for name, dep_list in dependencies.items():
if name == "core_dep":
toml["project"]["dependencies"] = dependencies["core_dep"]
continue
toml["project"]["optional-dependencies"][name] = dependencies[name]
return toml
def _select_versions(self, operation: Operation) -> dict:
"""Select versions based on the input argument.
Args:
operation (Literal["min", "max", "random"]): Operation to perform on the available candidates.
"""
if operation == Operation.MIN:
op = min
elif operation == Operation.MAX:
op = max
elif operation == Operation.RANDOM:
op = random.choice
else:
raise ValueError(
f"Invalid operation: {operation}. Choose from 'min', 'max' or 'random'."
)
dependencies = DefaultDict(lambda: [])
for name, dep in self._available_candidates.items():
for dep_name, dep_version in dep.items():
version = op(dep_version)
dependencies[name].extend([f"{dep_name}=={version.public}"])
return dependencies
def load_toml_file(self, filepath: str) -> TOMLDocument:
"""load toml file and return the content as a dictionary."""
filepath = os.path.abspath(filepath)
with open(filepath) as f:
return tomlkit.api.load(f)
def write_toml_file(self, filepath: str, toml_dict: TOMLDocument) -> None:
"""Save toml file to disk."""
filepath = os.path.abspath(filepath)
with open(filepath, "w") as f:
tomlkit.api.dump(toml_dict, f)
def get_package_dependencies_from_toml(self, include_optional_dep: bool = True) -> dict:
"""Gets the package dependencies from the pyproject.toml file.
Args:
filepath (str): Path to the pyproject.toml file.
include_optional_dep (bool): Include optional dependencies. Defaults to True.
"""
project_toml = self.get_project_toml()
self._pkg_dep["core_dep"] = project_toml["project"]["dependencies"]
if include_optional_dep:
if "optional-dependencies" not in project_toml["project"]:
return self._pkg_dep
for name, dep in project_toml["project"]["optional-dependencies"].items():
self._pkg_dep[name] = dep
return self._pkg_dep
def get_available_candidates(self, pkg_dep: dict) -> dict:
"""Update available candidates from dependency list.
Args:
dependency_list (list[str]): List of dependencies as parsed from pyproject.toml file.
"""
rtr = {}
for name, dep in pkg_dep.items():
rtr[name] = {}
for req_str in dep:
rtr[name].update(self.update_candidate(req_str))
return rtr
def update_candidate(self, req_str: str) -> dict:
"""Update candidate information in self._available_candidates.
Args:
req_str (str): Requirement string in the format from toml.
"""
ireq = Requirement(req_str)
all_candidates = set(
[
Version(candidate.version.public)
for candidate in self.repo.find_all_candidates(ireq.name)
]
)
return {ireq.name: sorted(list(ireq.specifier.filter(all_candidates, prereleases=False)))}
def run(self) -> None:
"""Run the package version handler."""
self._pkg_dep = self.get_package_dependencies_from_toml(
include_optional_dep=self._include_optional_dep
)
self._available_candidates = self.get_available_candidates(self._pkg_dep)
if __name__ == "__main__":
with PackageVersionHandler(toml_filepath="./pyproject.toml") as handler:
toml_min = handler.get_toml(Operation.MIN)
toml_latest = handler.get_toml(Operation.MAX)
toml_random = handler.get_toml(Operation.RANDOM)
handler.write_toml_file("./pyproject.toml", toml_min)
handler.write_toml_file("./requirements_latest.toml", toml_latest)
handler.write_toml_file("./requirements_random.toml", toml_random)
print("Done!")
Loading