import fnmatch import subprocess import shutil import zipfile import tarfile import stat from ast import Not from packaging.specifiers import SpecifierSet from packaging.version import Version, parse, InvalidVersion from pkg_resources import Requirement import io from ci_tools.variables import discover_repo_root, DEV_BUILD_IDENTIFIER, str_to_bool from ci_tools.parsing import ParsedSetup, get_config_setting, get_pyproject from pypi_tools.pypi import PyPIClient import os, sys, platform, glob, re, logging from typing import List, Any, Optional INACTIVE_CLASSIFIER = "Development Status :: 7 - Inactive" MANAGEMENT_PACKAGE_IDENTIFIERS = [ "mgmt", "nspkg", "azure-cognitiveservices", "azure-servicefabric", "azure-keyvault", "azure-synapse", "azure-ai-anomalydetector", ] NO_TESTS_ALLOWED = [] META_PACKAGES = ["azure", "azure-mgmt", "azure-keyvault"] REGRESSION_EXCLUDED_PACKAGES = [ "azure-common", ] MANAGEMENT_PACKAGES_FILTER_EXCLUSIONS = [ "azure-mgmt-core", ] TEST_COMPATIBILITY_MAP = { "azure-ai-ml": ">=3.7", "azure-ai-evaluation": ">=3.9, !=3.13.*" } TEST_PYTHON_DISTRO_INCOMPATIBILITY_MAP = { "azure-storage-blob": "pypy", "azure-storage-queue": "pypy", "azure-storage-file-datalake": "pypy", "azure-storage-file-share": "pypy", "azure-eventhub": "pypy", "azure-servicebus": "pypy", "azure-ai-projects": "pypy", "azure-ai-agents": "pypy", } omit_regression = ( lambda x: "nspkg" not in x and "mgmt" not in x and os.path.basename(x) not in MANAGEMENT_PACKAGE_IDENTIFIERS and os.path.basename(x) not in META_PACKAGES and str_to_bool(get_config_setting(x, "regression", True)) ) omit_docs = lambda x: "nspkg" not in x and os.path.basename(x) not in META_PACKAGES omit_build = lambda x: x # Dummy lambda to match omit type lambda_filter_azure_pkg = lambda x: x.startswith("azure") and "-nspkg" not in x omit_mgmt = lambda x: "mgmt" not in x or os.path.basename(x) in MANAGEMENT_PACKAGES_FILTER_EXCLUSIONS # dict of filter type and filter function omit_function_dict = { "Build": omit_build, "Docs": omit_docs, "Regression": omit_regression, "Omit_management": omit_mgmt, } def unzip_file_to_directory(path_to_zip_file: str, extract_location: str) -> str: if path_to_zip_file.endswith(".zip"): with zipfile.ZipFile(path_to_zip_file, "r") as zip_ref: zip_ref.extractall(extract_location) extracted_dir = os.path.basename(os.path.splitext(path_to_zip_file)[0]) return os.path.join(extract_location, extracted_dir) else: with tarfile.open(path_to_zip_file) as tar_ref: tar_ref.extractall(extract_location) extracted_dir = os.path.basename(path_to_zip_file).replace(".tar.gz", "") return os.path.join(extract_location, extracted_dir) def apply_compatibility_filter(package_set: List[str]) -> List[str]: """ This function takes in a set of paths to python packages. It returns the set filtered by compatibility with the currently running python executable. If a package is unsupported by the executable, it will be omitted from the returned list. A manual override list of these compatibilities is also supported, but requires code change to enable. Check TEST_COMPATIBILITY_MAP in this same file. :param List[str] package_set: a list of paths to python packages. Each item can either be a package folder or a direct path to setup.py. """ collected_packages = [] v = sys.version_info running_major_version = Version(".".join([str(v[0]), str(v[1]), str(v[2])])) for pkg in package_set: try: spec_set = SpecifierSet(ParsedSetup.from_path(pkg).python_requires) except RuntimeError as e: logging.error(f"Unable to parse metadata for package {pkg}, omitting from build.") continue pkg_specs_override = TEST_COMPATIBILITY_MAP.get(os.path.basename(pkg), None) if pkg_specs_override: spec_set = SpecifierSet(pkg_specs_override) distro_compat = True distro_incompat = TEST_PYTHON_DISTRO_INCOMPATIBILITY_MAP.get(os.path.basename(pkg), None) if distro_incompat and distro_incompat in platform.python_implementation().lower(): distro_compat = False if running_major_version in spec_set and distro_compat: collected_packages.append(pkg) logging.debug("Target packages after applying compatibility filter: {}".format(collected_packages)) logging.debug( "Package(s) omitted by compatibility filter: {}".format(generate_difference(package_set, collected_packages)) ) return collected_packages def compare_python_version(version_spec: str) -> bool: """ Compares the current running platform version of python against a version spec. Sanitizes the running platform version to just the major version. """ platform_version = platform.python_version() parsed_version = re.match(r"[0-9\.]+", platform_version, re.IGNORECASE) # we want to be loud if we can't parse out a major version from the version string, not silently # fail and skip running samples on a platform we really should be if parsed_version is None: raise InvalidVersion(f'Unable to parse the platform version. Unparsed value was "{platform_version}".') else: current_sys_version = parse(parsed_version[0]) spec_set = SpecifierSet(version_spec) return current_sys_version in spec_set def generate_difference(original_packages: List[str], filtered_packages: List[str]): return list(set(original_packages) - set(filtered_packages)) def glob_packages(glob_string: str, target_root_dir: str) -> List[str]: if glob_string: individual_globs = [glob_string.strip() for glob_string in glob_string.split(",")] else: individual_globs = "azure-*" collected_top_level_directories = [] for glob_string in individual_globs: globbed = glob.glob(os.path.join(target_root_dir, glob_string, "setup.py")) + glob.glob( os.path.join(target_root_dir, "sdk/*/", glob_string, "setup.py") ) collected_top_level_directories.extend([os.path.dirname(p) for p in globbed]) # handle pyproject.toml separately, as we need to filter them by the presence of a `[project]` section for glob_string in individual_globs: globbed = glob.glob(os.path.join(target_root_dir, glob_string, "pyproject.toml")) + glob.glob( os.path.join(target_root_dir, "sdk/*/", glob_string, "pyproject.toml") ) for p in globbed: if get_pyproject(os.path.dirname(p)): collected_top_level_directories.append(os.path.dirname(p)) # deduplicate, in case we have double coverage from the glob strings. Example: "azure-mgmt-keyvault,azure-mgmt-*" return list(set(collected_top_level_directories)) def apply_business_filter(collected_packages: List[str], filter_type: str) -> List[str]: pkg_set_ci_filtered = list(filter(omit_function_dict.get(filter_type, omit_build), collected_packages)) logging.debug("Target packages after applying business filter: {}".format(pkg_set_ci_filtered)) logging.debug( "Package(s) omitted by business filter: {}".format(generate_difference(collected_packages, pkg_set_ci_filtered)) ) return pkg_set_ci_filtered def discover_targeted_packages( glob_string: str, target_root_dir: str, additional_contains_filter: str = "", filter_type: str = "Build", compatibility_filter: bool = True, include_inactive: bool = False, ) -> List[str]: """ During build and test, the set of targeted packages may expand or contract depending on the needs of the invocation. This function centralizes business and material requirements and outputs the set of packages that should be targeted. :param str glob_string: The basic glob used to query packages within the repo. Defaults to "azure-*" :param str target_root_dir: The root directory in which globbing will begin. :param str additional_contains_filter: Additional filter option. Used when needing to provide one-off filtration that doesn't merit an additional filter_type. Defaults to empty string. :param str filter_type: One a string representing a filter function as a set of options. Options [ "Build", "Docs", "Regression", "Omit_management" ] Defaults to "Build". :param bool compatibility_filter: Enables or disables compatibility filtering of found packages. If the invoking python executable does not match a found package's specifiers, the package will be omitted. Defaults to True. """ # glob the starting package set collected_packages = glob_packages(glob_string, target_root_dir) logging.info( f'Results for glob_string "{glob_string}" and root directory "{target_root_dir}" are: {collected_packages}' ) # apply the additional contains filter collected_packages = [pkg for pkg in collected_packages if additional_contains_filter in pkg] logging.info(f'Results after additional contains filter: "{additional_contains_filter}" {collected_packages}') # filter for compatibility, this means excluding a package that doesn't support py36 when we are running a py36 executable if compatibility_filter: collected_packages = apply_compatibility_filter(collected_packages) logging.info(f"Results after compatibility filter: {collected_packages}") if not include_inactive: collected_packages = apply_inactive_filter(collected_packages) # Apply filter based on filter type. for e.g. Docs, Regression, Management collected_packages = apply_business_filter(collected_packages, filter_type) logging.info(f"Results after business filter: {collected_packages}") return sorted(collected_packages) def is_package_active(package_path: str): disabled = INACTIVE_CLASSIFIER in ParsedSetup.from_path(package_path).classifiers override_value = os.getenv(f"ENABLE_{os.path.basename(package_path).upper().replace('-', '_')}", None) if override_value: return str_to_bool(override_value) else: return not disabled def apply_inactive_filter(collected_packages: List[str]) -> List[str]: packages = [pkg for pkg in collected_packages if is_package_active(pkg)] logging.debug("Target packages after applying inactive filter: {}".format(collected_packages)) logging.debug("Package(s) omitted by inactive filter: {}".format(generate_difference(collected_packages, packages))) return packages def update_requires(setup_path, requires_dict): # This method changes package requirement by overriding the specifier contents = [] with open(setup_path, "r") as setup_file: contents = setup_file.readlines() # find and replace all existing package requirement with new requirement for i in range(0, len(contents) - 1): keys = [k for k in requires_dict.keys() if k in contents[i]] for key in keys: contents[i] = contents[i].replace(key, requires_dict[key]) with open(setup_path, "w") as setup_file: setup_file.writelines(contents) def is_required_version_on_pypi(package_name, spec): client = PyPIClient() try: pypi_results = client.get_ordered_versions(package_name) except: pypi_results = [] versions = [str(v) for v in pypi_results if str(v) in spec] return versions def get_package_from_repo(pkg_name: str, repo_root: Optional[str] = None) -> Optional[ParsedSetup]: root_dir = discover_repo_root(repo_root) glob_path = os.path.join(root_dir, "sdk", "*", pkg_name, "setup.py") paths = glob.glob(glob_path) if paths: setup_py_path = paths[0] parsed_setup = ParsedSetup.from_path(setup_py_path) return parsed_setup return None def get_package_from_repo_or_folder(req: str, prebuilt_wheel_dir: Optional[str] = None) -> Optional[str]: """Takes a package name and a possible prebuilt wheel directory. Attempts to resolve a wheel that matches the package name, and if it can't, attempts to find the package within the repo to install directly from path on disk. During a CI build, it is preferred that the package is installed from a prebuilt wheel directory, as multiple CI environments attempting to install the relative req can cause inconsistent installation issues during parallel tox environment execution. """ local_package = get_package_from_repo(req) if prebuilt_wheel_dir and os.path.exists(prebuilt_wheel_dir) and local_package: prebuilt_package = discover_prebuilt_package(prebuilt_wheel_dir, local_package.setup_filename, "wheel") if prebuilt_package: # return the first package found, there should only be a single one matching given that our prebuilt wheel directory # is populated by the replacement of dev_reqs.txt with the prebuilt wheels # ref tox_harness replace_dev_reqs() calls return os.path.join(prebuilt_wheel_dir, prebuilt_package[0]) if local_package: return local_package.folder else: return None def get_version_from_repo(pkg_name: str, repo_root: Optional[str] = None) -> str: pkg_info = get_package_from_repo(pkg_name, repo_root) if pkg_info: # Remove dev build part if version for this package is already updated to dev build # When building package with dev build version, version for packages in same service is updated to dev build # and other packages will not have dev build number # strip dev build number so we can check if package exists in PyPI and replace version_obj = Version(pkg_info.version) if version_obj.pre: if version_obj.pre[0] == DEV_BUILD_IDENTIFIER: return version_obj.base_version return str(version_obj) else: logging.error("setup.py is not found for package {} to identify current version".format(pkg_name)) exit(1) def get_base_version(pkg_name: str) -> str: root_dir = discover_repo_root() # find version for the package from source. This logic should be revisited to find version from devops feed glob_path = os.path.join(root_dir, "sdk", "*", pkg_name, "setup.py") paths = glob.glob(glob_path) if paths: setup_py_path = paths[0] parsed_setup = ParsedSetup.from_path(setup_py_path) version_obj = Version(parsed_setup.version) return version_obj.base_version else: logging.error("setup.py is not found for package {} to identify current version".format(pkg_name)) exit(1) def process_requires(setup_py_path: str, is_dev_build: bool = False): """ This method processes a setup.py's package requirements to verify if all required packages are available on PyPI. If any azure sdk package is not available on PyPI then requirement will be updated to refer to the sdk "dev_identifier". Examples: azure-storage-blob >= 1.0.1b1 update to require 1.0.1a1 to allow previously published dev versions to be allowed. """ pkg_details = ParsedSetup.from_path(setup_py_path) azure_requirements = [Requirement.parse(r) for r in pkg_details.requires if r.startswith("azure")] # Find package requirements that are not available on PyPI requirement_to_update = {} for req in azure_requirements: pkg_name = req.key spec = SpecifierSet(str(req).replace(pkg_name, "")) if not is_required_version_on_pypi(pkg_name, spec) or is_dev_build: old_req = str(req) version = get_version_from_repo(pkg_name) base_version = get_base_version(pkg_name) logging.info("Updating version {0} in requirement {1} to dev build version".format(version, old_req)) # {} = # ( # ( # (a|b|rc) # \d+ # )? # ( # \.post # \d+ # )? # )? rx = r"{}(((a|b|rc)\d+)?(\.post\d+)?)?".format(base_version) new_req_format = f"{base_version}{DEV_BUILD_IDENTIFIER}1,<{base_version}b0" new_req = re.sub(rx, new_req_format, str(req), flags=re.IGNORECASE) logging.info("New requirement for package {0}: {1}".format(pkg_name, new_req)) requirement_to_update[old_req] = new_req if not requirement_to_update: logging.info("All required packages are available on PyPI") else: logging.info("Packages not available on PyPI:{}".format(requirement_to_update)) update_requires(setup_py_path, requirement_to_update) logging.info("Package requirement is updated in setup.py") def find_sdist(dist_dir: str, pkg_name: str, pkg_version: str) -> Optional[str]: """This function attempts to look within a directory (and all subdirs therein) and find a source distribution for the targeted package and version.""" # This function will find a sdist for given package name if not os.path.exists(dist_dir): logging.error("dist_dir is incorrect") return if pkg_name is None: logging.error("Package name cannot be empty to find sdist") return else: # ensure package name matches cananonicalized package name pkg_name = pkg_name.replace("-", "[-_]") pkg_format = f"{pkg_name}-{pkg_version}.tar.gz" packages = [] for root, dirnames, filenames in os.walk(dist_dir): for filename in fnmatch.filter(filenames, pkg_format): packages.append(os.path.join(root, filename)) packages = [os.path.relpath(w, dist_dir) for w in packages] if not packages: logging.error(f"No sdist is found in directory {dist_dir} with package name format {pkg_format}.") return return packages[0] def pip_install( requirements: List[str], include_dependencies: bool = True, python_executable: Optional[str] = None ) -> bool: """ Attempts to invoke an install operation using the invoking python's pip. Empty requirements are auto-success. """ exe = python_executable or sys.executable command = [exe, "-m", "pip", "install"] if requirements: command.extend([req.strip() for req in requirements]) else: return True try: subprocess.check_call(command) except subprocess.CalledProcessError as f: return False return True def pip_uninstall(requirements: List[str], python_executable: str) -> bool: """ Attempts to invoke an install operation using the invoking python's pip. Empty requirements are auto-success. """ exe = python_executable or sys.executable command = [exe, "-m", "pip", "uninstall", "-y"] if requirements: command.extend([req.strip() for req in requirements]) else: return True try: result = subprocess.check_call(command) return True except subprocess.CalledProcessError as f: return False def pip_install_requirements_file(requirements_file: str, python_executable: Optional[str] = None) -> bool: return pip_install(["-r", requirements_file], True, python_executable) def get_pip_list_output(python_executable: Optional[str] = None): """Uses the invoking python executable to get the output from pip list.""" exe = python_executable or sys.executable out = subprocess.Popen( [exe, "-m", "pip", "list", "--disable-pip-version-check", "--format", "freeze"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) stdout, stderr = out.communicate() collected_output = {} if stdout and (stderr is None): # this should be compatible with py27 https://docs.python.org/2.7/library/stdtypes.html#str.decode for line in stdout.decode("utf-8").split(os.linesep)[2:]: if line: package, version = re.split("==", line) collected_output[package] = version else: raise Exception(stderr) return collected_output def pytest(args: list, cwd: Optional[str] = None, python_executable: Optional[str] = None) -> bool: """ Invokes a set of tests, returns true if successful, false otherwise. """ exe = python_executable or sys.executable commands = [ exe, "-m", "pytest", ] commands.extend(args) logging.info(commands) if cwd: result = subprocess.run(commands, cwd=cwd) else: result = subprocess.run(commands) return result.returncode == 0 def get_interpreter_compatible_tags() -> List[str]: """ This function invokes pip from the invoking interpreter and discovers which tags the interpreter is compatible with. """ commands = [sys.executable, "-m", "pip", "debug", "--verbose"] output = subprocess.run( commands, check=True, capture_output=True, ).stdout.decode(encoding="utf-8") tag_strings = output.split(os.linesep) index = 0 for index, value in enumerate(tag_strings): if "Compatible tags" in value: break tags = tag_strings[index + 1 :] return [tag.strip() for tag in tags if tag] def check_whl_against_tags(whl_name: str, tags: List[str]) -> bool: for tag in tags: if tag in whl_name: return True return False def find_whl(whl_dir: str, pkg_name: str, pkg_version: str) -> Optional[str]: """This function attempts to look within a directory (and all subdirs therein) and find a wheel that matches our targeted name and version AND whose compilation is compatible with the invoking interpreter.""" if not os.path.exists(whl_dir): logging.error("whl_dir is incorrect") return if pkg_name is None: logging.error("Package name cannot be empty to find whl") return pkg_name_format = f"{pkg_name.replace('-', '_')}-{pkg_version}*.whl" whls = [] # todo: replace with glob, we aren't using py2 anymore! for root, dirnames, filenames in os.walk(whl_dir): for filename in fnmatch.filter(filenames, pkg_name_format): whls.append(os.path.join(root, filename)) whls = [os.path.relpath(w, whl_dir) for w in whls] if not whls: logging.info(f"No whl is found in directory {whl_dir} with package name format {pkg_name_format}") logging.info(f"List of whls in directory: {glob.glob(os.path.join(whl_dir, '*.whl'))}") return compatible_tags = get_interpreter_compatible_tags() logging.debug("Dumping visible tags and whls") logging.debug(compatible_tags) logging.debug(whls) if whls: # grab the first whl that matches a tag from our compatible_tags list for whl in whls: if check_whl_against_tags(whl, compatible_tags): logging.info(f"Found whl {whl}") return whl # if whl is platform independent then there should only be one whl in filtered list if len(whls) > 1: # if we have reached here, that means we have whl specific to platform as well. # for now we are failing the test if platform specific wheels are found. Todo: enhance to find platform specific whl logging.error(f"We were unable to locate a compatible wheel for {pkg_name}") sys.exit(1) return None def error_handler_git_access(func, path, exc): """ This function exists because the git idx file is written with strange permissions that prevent it from being deleted. Due to this, we need to register an error handler that attempts to fix the file permissions before re-attempting the delete operations. """ if not os.access(path, os.W_OK): os.chmod(path, stat.S_IWUSR) func(path) else: raise def cleanup_directory(target_directory: str) -> None: """Invokes a directory delete. Specifically handles the case where bad permissions on a git .idx file prevent cleanup of the directory with a generic error. """ if os.path.exists(target_directory): shutil.rmtree(target_directory, ignore_errors=False, onerror=error_handler_git_access) def discover_prebuilt_package(dist_directory: str, setup_path: str, package_type: str) -> List[str]: """Discovers a prebuild wheel or sdist for a given setup path.""" packages = [] pkg = ParsedSetup.from_path(setup_path) if package_type == "wheel": prebuilt_package = find_whl(dist_directory, pkg.name, pkg.version) else: prebuilt_package = find_sdist(dist_directory, pkg.name, pkg.version) if prebuilt_package is not None: packages.append(prebuilt_package) return packages def is_package_compatible( package_name: str, package_requirements: List[Requirement], immutable_requirements: List[Requirement], should_log: bool = True, ) -> bool: """ This function accepts a set of requirements for a package, and ensures that the package is compatible with the immutable_requirements. It is factored this way because we retrieve requirements differently according to the source of the package. If published, we can get requires() from PyPI If locally built wheel, we can get requires() from the metadata of the package If local relative requirement, we can get requires() from a ParsedSetup of the setup.py for the package :param List[Requirement] package_requirements: The dependencies of a dev_requirement file. This is the set of requirements that we are checking compatibility for. :param List[Requirement] immutable_requirements: A list of requirements that the other packages must be compatible with. """ for immutable_requirement in immutable_requirements: for package_requirement in package_requirements: if package_requirement.key == immutable_requirement.key: # if the dev_req line has a requirement that conflicts with the immutable requirement, # we need to resolve it. We KNOW that the immutable requirement will be of form package==version, # so we can reliably pull out the version and check it against the specifier of the dev_req line. try: immutable_version = next(iter(immutable_requirement.specifier)).version # we have no specifier set, so we don't even need to check this except StopIteration: continue if not package_requirement.specifier.contains(immutable_version): if should_log: logging.info( f"Dev req dependency {package_name}'s requirement specifier of {package_requirement}" f"is not compatible with immutable requirement {immutable_requirement}." ) return False return True def get_total_coverage(coverage_file: str, coverage_config_file: str, package_name: str, repo_root: Optional[str] = None) -> Optional[float]: try: import coverage from coverage.exceptions import NoDataError except ImportError: logging.error("Coverage is not installed.") return None cov = coverage.Coverage(data_file=coverage_file, config_file=coverage_config_file) cov.load() original = os.getcwd() output = io.StringIO() old_stdout = sys.stdout sys.stdout = output report = 0.0 try: if repo_root: os.chdir(repo_root) logging.info(f"Running coverage report against \"{coverage_file}\" with \"{coverage_config_file}\" from \"{os.getcwd()}\".") report = cov.report() except NoDataError as e: logging.info(f"Package {package_name} did not generate any coverage output: {e}") except Exception as e: logging.error(f"An error occurred while generating the coverage report for {package_name}: {e}") finally: if repo_root: os.chdir(original) sys.stdout = old_stdout logging.info(f"Total coverage {report} for package {package_name}") return report def resolve_compatible_package(package_name: str, immutable_requirements: List[Requirement]) -> Optional[str]: """ This function attempts to resolve a compatible package version for whatever set of immutable_requirements that the package must be compatible with. It should only be utilized when a package is found to be incompatible with the immutable_requirements. It will attempt to resolve the incompatibility by walking backwards through different versions of until a compatible version is found that works with the immutable_requirements. """ pypi = PyPIClient() immovable_pkgs = {req.key: req for req in immutable_requirements} # Let's use a real use-case to walk through this function. We're going to use the azure-ai-language-conversations # package as an example. # immovable_pkgs = the selected mindependency for azure-ai-language-conversations # -> "azure-core==1.28.0", # -> "isodate==0.6.1", # -> "typing-extensions==4.0.1", # we have the following dev_reqs for azure-ai-language-conversations # -> ../azure-sdk-tools # -> ../azure-identity # -> ../azure-core # as we walk each of the dev reqs, we check for compatibility with the immovable_packages. # (this happens in is_package_compatible) if the dev req is incompatible, we need to resolve it. # THIS function is what resolves it! # since we already know that package_name is incompatible with the immovable_pkgs, we need to walk backwards # through the versions of package_name checking to ensure that each version is compatible with the immovable_pkgs. # if we find one that is, we will return a new requirement string for that package which will replace this dev_req line. for pkg in immovable_pkgs: required_package = immovable_pkgs[pkg].name try: required_pkg_version = next(iter(immovable_pkgs[pkg].specifier)).version except StopIteration: required_pkg_version = None versions = pypi.get_ordered_versions(package_name, True) versions.reverse() # only allow prerelease versions if the dev_req we're targeting is also prerelease if required_pkg_version: if not Version(required_pkg_version).is_prerelease: versions = [v for v in versions if not v.is_prerelease] for version in versions: version_release = pypi.project_release(package_name, version).get("info", {}).get("requires_dist", []) if version_release: requirements_for_dev_req = [Requirement(r) for r in version_release] compatible = is_package_compatible( required_package, requirements_for_dev_req, immutable_requirements, should_log=False ) if compatible: # we have found a compatible version. We can return this as the new requirement line for the dev_req file. return f"{package_name}=={version}" # no changes necessary return None def handle_incompatible_minimum_dev_reqs( setup_path: str, filtered_requirement_list: List[str], packages_for_install: List[Requirement] ) -> List[str]: """ This function is used to handle the case where a dev requirement is incompatible with the current set of packages being installed. This is used to update or remove dev_requirements that are incompatible with a targeted set of packages. :param str setup_path: The path to the setup.py file whose dev_requirements are being filtered. :param List[str] filtered_requirement_list: A filtered copy of the dev_requirements.txt for the targeted package. This list will be modified in place to remove any requirements incompatible with the packages_for_install. :param List[Requirement] packages_for_install: A list of packages that dev_requirements MUST be compatible with. """ cleansed_reqs = [] for dev_requirement_line in filtered_requirement_list: cleansed_dev_requirement_line = dev_requirement_line.strip().replace("-e ", "").split("#")[0].split(";")[0] if cleansed_dev_requirement_line: dev_req_package = None dev_req_version = None requirements_for_dev_req = [] # this is a locally built wheel file, ise pkginfo to get the metadata if os.path.exists(cleansed_dev_requirement_line) and os.path.isfile(cleansed_dev_requirement_line): logging.info( f"We are processing a replaced relative requirement built into a wheel: {cleansed_dev_requirement_line}" ) import pkginfo try: local_package_metadata = pkginfo.get_metadata(cleansed_dev_requirement_line) if local_package_metadata: dev_req_package = local_package_metadata.name dev_req_version = local_package_metadata.version requirements_for_dev_req = [Requirement(r) for r in local_package_metadata.requires_dist] else: logging.error( f"Error while processing locally built requirement {cleansed_dev_requirement_line}. Unable to resolve metadata." ) cleansed_reqs.append(cleansed_dev_requirement_line) except Exception as e: logging.error( f"Unable to determine metadata for locally built requirement {cleansed_dev_requirement_line}: {e}" ) cleansed_reqs.append(cleansed_dev_requirement_line) continue # this is a relative requirement to a package path in the repo, use our ParsedSetup class to get data from setup.py or pyproject.toml elif cleansed_dev_requirement_line.startswith("."): logging.info(f"We are processing a relative requirement: {cleansed_dev_requirement_line}") try: local_package = ParsedSetup.from_path(os.path.join(setup_path, cleansed_dev_requirement_line)) if local_package: dev_req_package = local_package.name dev_req_version = local_package.version requirements_for_dev_req = [Requirement(r) for r in local_package.requires] else: logging.error( f"Error while processing relative requirement {cleansed_dev_requirement_line}. Unable to resolve metadata." ) cleansed_reqs.append(cleansed_dev_requirement_line) except Exception as e: logging.error( f'Unable to determine metadata for relative requirement "{cleansed_dev_requirement_line}", not modifying: {e}' ) cleansed_reqs.append(cleansed_dev_requirement_line) continue # If we got here, this has to be a standard requirement, attempt to parse it as a specifier and if unable to do so, # simply add it to the list as a last fallback. we will log so that we can implement a fix for the edge case later. else: logging.info(f"We are processing a standard requirement: {cleansed_dev_requirement_line}") cleansed_reqs.append(dev_requirement_line) # we understand how to parse it, so we should handle it if dev_req_package: if not is_package_compatible(dev_req_package, requirements_for_dev_req, packages_for_install): new_req = resolve_compatible_package(dev_req_package, packages_for_install) if new_req: cleansed_reqs.append(new_req) else: logging.error( f'Found incompatible dev requirement {dev_req_package}, but unable to locate a compatible version. Not modifying the line: "{dev_requirement_line}".' ) cleansed_reqs.append(cleansed_dev_requirement_line) else: cleansed_reqs.append(cleansed_dev_requirement_line) return cleansed_reqs