diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/pkg_resources')
19 files changed, 6142 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/__init__.py b/.venv/lib/python3.12/site-packages/pkg_resources/__init__.py new file mode 100644 index 00000000..8a2fbfa4 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/__init__.py @@ -0,0 +1,3714 @@ +""" +Package resource API +-------------------- + +A resource is a logical file contained within a package, or a logical +subdirectory thereof. The package resource API expects resource names +to have their path parts separated with ``/``, *not* whatever the local +path separator is. Do not use os.path operations to manipulate resource +names being passed into the API. + +The package resource API is designed to work with normal filesystem packages, +.egg files, and unpacked .egg files. It can also work in a limited way with +.zip files and with custom PEP 302 loaders that support the ``get_data()`` +method. + +This module is deprecated. Users are directed to :mod:`importlib.resources`, +:mod:`importlib.metadata` and :pypi:`packaging` instead. +""" + +from __future__ import annotations + +import sys + +if sys.version_info < (3, 9): # noqa: UP036 # Check for unsupported versions + raise RuntimeError("Python 3.9 or later is required") + +import _imp +import collections +import email.parser +import errno +import functools +import importlib +import importlib.abc +import importlib.machinery +import inspect +import io +import ntpath +import operator +import os +import pkgutil +import platform +import plistlib +import posixpath +import re +import stat +import tempfile +import textwrap +import time +import types +import warnings +import zipfile +import zipimport +from collections.abc import Iterable, Iterator, Mapping, MutableSequence +from pkgutil import get_importer +from typing import ( + TYPE_CHECKING, + Any, + BinaryIO, + Callable, + Literal, + NamedTuple, + NoReturn, + Protocol, + TypeVar, + Union, + overload, +) + +sys.path.extend(((vendor_path := os.path.join(os.path.dirname(os.path.dirname(__file__)), 'setuptools', '_vendor')) not in sys.path) * [vendor_path]) # fmt: skip +# workaround for #4476 +sys.modules.pop('backports', None) + +# capture these to bypass sandboxing +from os import open as os_open, utime # isort: skip +from os.path import isdir, split # isort: skip + +try: + from os import mkdir, rename, unlink + + WRITE_SUPPORT = True +except ImportError: + # no write support, probably under GAE + WRITE_SUPPORT = False + +import packaging.markers +import packaging.requirements +import packaging.specifiers +import packaging.utils +import packaging.version +from jaraco.text import drop_comment, join_continuation, yield_lines +from platformdirs import user_cache_dir as _user_cache_dir + +if TYPE_CHECKING: + from _typeshed import BytesPath, StrOrBytesPath, StrPath + from _typeshed.importlib import LoaderProtocol + from typing_extensions import Self, TypeAlias + +warnings.warn( + "pkg_resources is deprecated as an API. " + "See https://setuptools.pypa.io/en/latest/pkg_resources.html", + DeprecationWarning, + stacklevel=2, +) + +_T = TypeVar("_T") +_DistributionT = TypeVar("_DistributionT", bound="Distribution") +# Type aliases +_NestedStr: TypeAlias = Union[str, Iterable[Union[str, Iterable["_NestedStr"]]]] +_StrictInstallerType: TypeAlias = Callable[["Requirement"], "_DistributionT"] +_InstallerType: TypeAlias = Callable[["Requirement"], Union["Distribution", None]] +_PkgReqType: TypeAlias = Union[str, "Requirement"] +_EPDistType: TypeAlias = Union["Distribution", _PkgReqType] +_MetadataType: TypeAlias = Union["IResourceProvider", None] +_ResolvedEntryPoint: TypeAlias = Any # Can be any attribute in the module +_ResourceStream: TypeAlias = Any # TODO / Incomplete: A readable file-like object +# Any object works, but let's indicate we expect something like a module (optionally has __loader__ or __file__) +_ModuleLike: TypeAlias = Union[object, types.ModuleType] +# Any: Should be _ModuleLike but we end up with issues where _ModuleLike doesn't have _ZipLoaderModule's __loader__ +_ProviderFactoryType: TypeAlias = Callable[[Any], "IResourceProvider"] +_DistFinderType: TypeAlias = Callable[[_T, str, bool], Iterable["Distribution"]] +_NSHandlerType: TypeAlias = Callable[[_T, str, str, types.ModuleType], Union[str, None]] +_AdapterT = TypeVar( + "_AdapterT", _DistFinderType[Any], _ProviderFactoryType, _NSHandlerType[Any] +) + + +class _ZipLoaderModule(Protocol): + __loader__: zipimport.zipimporter + + +_PEP440_FALLBACK = re.compile(r"^v?(?P<safe>(?:[0-9]+!)?[0-9]+(?:\.[0-9]+)*)", re.I) + + +class PEP440Warning(RuntimeWarning): + """ + Used when there is an issue with a version or specifier not complying with + PEP 440. + """ + + +parse_version = packaging.version.Version + +_state_vars: dict[str, str] = {} + + +def _declare_state(vartype: str, varname: str, initial_value: _T) -> _T: + _state_vars[varname] = vartype + return initial_value + + +def __getstate__() -> dict[str, Any]: + state = {} + g = globals() + for k, v in _state_vars.items(): + state[k] = g['_sget_' + v](g[k]) + return state + + +def __setstate__(state: dict[str, Any]) -> dict[str, Any]: + g = globals() + for k, v in state.items(): + g['_sset_' + _state_vars[k]](k, g[k], v) + return state + + +def _sget_dict(val): + return val.copy() + + +def _sset_dict(key, ob, state) -> None: + ob.clear() + ob.update(state) + + +def _sget_object(val): + return val.__getstate__() + + +def _sset_object(key, ob, state) -> None: + ob.__setstate__(state) + + +_sget_none = _sset_none = lambda *args: None + + +def get_supported_platform(): + """Return this platform's maximum compatible version. + + distutils.util.get_platform() normally reports the minimum version + of macOS that would be required to *use* extensions produced by + distutils. But what we want when checking compatibility is to know the + version of macOS that we are *running*. To allow usage of packages that + explicitly require a newer version of macOS, we must also know the + current version of the OS. + + If this condition occurs for any other platform with a version in its + platform strings, this function should be extended accordingly. + """ + plat = get_build_platform() + m = macosVersionString.match(plat) + if m is not None and sys.platform == "darwin": + try: + major_minor = '.'.join(_macos_vers()[:2]) + build = m.group(3) + plat = f'macosx-{major_minor}-{build}' + except ValueError: + # not macOS + pass + return plat + + +__all__ = [ + # Basic resource access and distribution/entry point discovery + 'require', + 'run_script', + 'get_provider', + 'get_distribution', + 'load_entry_point', + 'get_entry_map', + 'get_entry_info', + 'iter_entry_points', + 'resource_string', + 'resource_stream', + 'resource_filename', + 'resource_listdir', + 'resource_exists', + 'resource_isdir', + # Environmental control + 'declare_namespace', + 'working_set', + 'add_activation_listener', + 'find_distributions', + 'set_extraction_path', + 'cleanup_resources', + 'get_default_cache', + # Primary implementation classes + 'Environment', + 'WorkingSet', + 'ResourceManager', + 'Distribution', + 'Requirement', + 'EntryPoint', + # Exceptions + 'ResolutionError', + 'VersionConflict', + 'DistributionNotFound', + 'UnknownExtra', + 'ExtractionError', + # Warnings + 'PEP440Warning', + # Parsing functions and string utilities + 'parse_requirements', + 'parse_version', + 'safe_name', + 'safe_version', + 'get_platform', + 'compatible_platforms', + 'yield_lines', + 'split_sections', + 'safe_extra', + 'to_filename', + 'invalid_marker', + 'evaluate_marker', + # filesystem utilities + 'ensure_directory', + 'normalize_path', + # Distribution "precedence" constants + 'EGG_DIST', + 'BINARY_DIST', + 'SOURCE_DIST', + 'CHECKOUT_DIST', + 'DEVELOP_DIST', + # "Provider" interfaces, implementations, and registration/lookup APIs + 'IMetadataProvider', + 'IResourceProvider', + 'FileMetadata', + 'PathMetadata', + 'EggMetadata', + 'EmptyProvider', + 'empty_provider', + 'NullProvider', + 'EggProvider', + 'DefaultProvider', + 'ZipProvider', + 'register_finder', + 'register_namespace_handler', + 'register_loader_type', + 'fixup_namespace_packages', + 'get_importer', + # Warnings + 'PkgResourcesDeprecationWarning', + # Deprecated/backward compatibility only + 'run_main', + 'AvailableDistributions', +] + + +class ResolutionError(Exception): + """Abstract base for dependency resolution errors""" + + def __repr__(self) -> str: + return self.__class__.__name__ + repr(self.args) + + +class VersionConflict(ResolutionError): + """ + An already-installed version conflicts with the requested version. + + Should be initialized with the installed Distribution and the requested + Requirement. + """ + + _template = "{self.dist} is installed but {self.req} is required" + + @property + def dist(self) -> Distribution: + return self.args[0] + + @property + def req(self) -> Requirement: + return self.args[1] + + def report(self): + return self._template.format(**locals()) + + def with_context( + self, required_by: set[Distribution | str] + ) -> Self | ContextualVersionConflict: + """ + If required_by is non-empty, return a version of self that is a + ContextualVersionConflict. + """ + if not required_by: + return self + args = self.args + (required_by,) + return ContextualVersionConflict(*args) + + +class ContextualVersionConflict(VersionConflict): + """ + A VersionConflict that accepts a third parameter, the set of the + requirements that required the installed Distribution. + """ + + _template = VersionConflict._template + ' by {self.required_by}' + + @property + def required_by(self) -> set[str]: + return self.args[2] + + +class DistributionNotFound(ResolutionError): + """A requested distribution was not found""" + + _template = ( + "The '{self.req}' distribution was not found " + "and is required by {self.requirers_str}" + ) + + @property + def req(self) -> Requirement: + return self.args[0] + + @property + def requirers(self) -> set[str] | None: + return self.args[1] + + @property + def requirers_str(self): + if not self.requirers: + return 'the application' + return ', '.join(self.requirers) + + def report(self): + return self._template.format(**locals()) + + def __str__(self) -> str: + return self.report() + + +class UnknownExtra(ResolutionError): + """Distribution doesn't have an "extra feature" of the given name""" + + +_provider_factories: dict[type[_ModuleLike], _ProviderFactoryType] = {} + +PY_MAJOR = f'{sys.version_info.major}.{sys.version_info.minor}' +EGG_DIST = 3 +BINARY_DIST = 2 +SOURCE_DIST = 1 +CHECKOUT_DIST = 0 +DEVELOP_DIST = -1 + + +def register_loader_type( + loader_type: type[_ModuleLike], provider_factory: _ProviderFactoryType +) -> None: + """Register `provider_factory` to make providers for `loader_type` + + `loader_type` is the type or class of a PEP 302 ``module.__loader__``, + and `provider_factory` is a function that, passed a *module* object, + returns an ``IResourceProvider`` for that module. + """ + _provider_factories[loader_type] = provider_factory + + +@overload +def get_provider(moduleOrReq: str) -> IResourceProvider: ... +@overload +def get_provider(moduleOrReq: Requirement) -> Distribution: ... +def get_provider(moduleOrReq: str | Requirement) -> IResourceProvider | Distribution: + """Return an IResourceProvider for the named module or requirement""" + if isinstance(moduleOrReq, Requirement): + return working_set.find(moduleOrReq) or require(str(moduleOrReq))[0] + try: + module = sys.modules[moduleOrReq] + except KeyError: + __import__(moduleOrReq) + module = sys.modules[moduleOrReq] + loader = getattr(module, '__loader__', None) + return _find_adapter(_provider_factories, loader)(module) + + +@functools.cache +def _macos_vers(): + version = platform.mac_ver()[0] + # fallback for MacPorts + if version == '': + plist = '/System/Library/CoreServices/SystemVersion.plist' + if os.path.exists(plist): + with open(plist, 'rb') as fh: + plist_content = plistlib.load(fh) + if 'ProductVersion' in plist_content: + version = plist_content['ProductVersion'] + return version.split('.') + + +def _macos_arch(machine): + return {'PowerPC': 'ppc', 'Power_Macintosh': 'ppc'}.get(machine, machine) + + +def get_build_platform(): + """Return this platform's string for platform-specific distributions + + XXX Currently this is the same as ``distutils.util.get_platform()``, but it + needs some hacks for Linux and macOS. + """ + from sysconfig import get_platform + + plat = get_platform() + if sys.platform == "darwin" and not plat.startswith('macosx-'): + try: + version = _macos_vers() + machine = _macos_arch(os.uname()[4].replace(" ", "_")) + return f"macosx-{version[0]}.{version[1]}-{machine}" + except ValueError: + # if someone is running a non-Mac darwin system, this will fall + # through to the default implementation + pass + return plat + + +macosVersionString = re.compile(r"macosx-(\d+)\.(\d+)-(.*)") +darwinVersionString = re.compile(r"darwin-(\d+)\.(\d+)\.(\d+)-(.*)") +# XXX backward compat +get_platform = get_build_platform + + +def compatible_platforms(provided: str | None, required: str | None) -> bool: + """Can code for the `provided` platform run on the `required` platform? + + Returns true if either platform is ``None``, or the platforms are equal. + + XXX Needs compatibility checks for Linux and other unixy OSes. + """ + if provided is None or required is None or provided == required: + # easy case + return True + + # macOS special cases + reqMac = macosVersionString.match(required) + if reqMac: + provMac = macosVersionString.match(provided) + + # is this a Mac package? + if not provMac: + # this is backwards compatibility for packages built before + # setuptools 0.6. All packages built after this point will + # use the new macOS designation. + provDarwin = darwinVersionString.match(provided) + if provDarwin: + dversion = int(provDarwin.group(1)) + macosversion = f"{reqMac.group(1)}.{reqMac.group(2)}" + if ( + dversion == 7 + and macosversion >= "10.3" + or dversion == 8 + and macosversion >= "10.4" + ): + return True + # egg isn't macOS or legacy darwin + return False + + # are they the same major version and machine type? + if provMac.group(1) != reqMac.group(1) or provMac.group(3) != reqMac.group(3): + return False + + # is the required OS major update >= the provided one? + if int(provMac.group(2)) > int(reqMac.group(2)): + return False + + return True + + # XXX Linux and other platforms' special cases should go here + return False + + +@overload +def get_distribution(dist: _DistributionT) -> _DistributionT: ... +@overload +def get_distribution(dist: _PkgReqType) -> Distribution: ... +def get_distribution(dist: Distribution | _PkgReqType) -> Distribution: + """Return a current distribution object for a Requirement or string""" + if isinstance(dist, str): + dist = Requirement.parse(dist) + if isinstance(dist, Requirement): + dist = get_provider(dist) + if not isinstance(dist, Distribution): + raise TypeError("Expected str, Requirement, or Distribution", dist) + return dist + + +def load_entry_point(dist: _EPDistType, group: str, name: str) -> _ResolvedEntryPoint: + """Return `name` entry point of `group` for `dist` or raise ImportError""" + return get_distribution(dist).load_entry_point(group, name) + + +@overload +def get_entry_map( + dist: _EPDistType, group: None = None +) -> dict[str, dict[str, EntryPoint]]: ... +@overload +def get_entry_map(dist: _EPDistType, group: str) -> dict[str, EntryPoint]: ... +def get_entry_map(dist: _EPDistType, group: str | None = None): + """Return the entry point map for `group`, or the full entry map""" + return get_distribution(dist).get_entry_map(group) + + +def get_entry_info(dist: _EPDistType, group: str, name: str) -> EntryPoint | None: + """Return the EntryPoint object for `group`+`name`, or ``None``""" + return get_distribution(dist).get_entry_info(group, name) + + +class IMetadataProvider(Protocol): + def has_metadata(self, name: str) -> bool: + """Does the package's distribution contain the named metadata?""" + ... + + def get_metadata(self, name: str) -> str: + """The named metadata resource as a string""" + ... + + def get_metadata_lines(self, name: str) -> Iterator[str]: + """Yield named metadata resource as list of non-blank non-comment lines + + Leading and trailing whitespace is stripped from each line, and lines + with ``#`` as the first non-blank character are omitted.""" + ... + + def metadata_isdir(self, name: str) -> bool: + """Is the named metadata a directory? (like ``os.path.isdir()``)""" + ... + + def metadata_listdir(self, name: str) -> list[str]: + """List of metadata names in the directory (like ``os.listdir()``)""" + ... + + def run_script(self, script_name: str, namespace: dict[str, Any]) -> None: + """Execute the named script in the supplied namespace dictionary""" + ... + + +class IResourceProvider(IMetadataProvider, Protocol): + """An object that provides access to package resources""" + + def get_resource_filename( + self, manager: ResourceManager, resource_name: str + ) -> str: + """Return a true filesystem path for `resource_name` + + `manager` must be a ``ResourceManager``""" + ... + + def get_resource_stream( + self, manager: ResourceManager, resource_name: str + ) -> _ResourceStream: + """Return a readable file-like object for `resource_name` + + `manager` must be a ``ResourceManager``""" + ... + + def get_resource_string( + self, manager: ResourceManager, resource_name: str + ) -> bytes: + """Return the contents of `resource_name` as :obj:`bytes` + + `manager` must be a ``ResourceManager``""" + ... + + def has_resource(self, resource_name: str) -> bool: + """Does the package contain the named resource?""" + ... + + def resource_isdir(self, resource_name: str) -> bool: + """Is the named resource a directory? (like ``os.path.isdir()``)""" + ... + + def resource_listdir(self, resource_name: str) -> list[str]: + """List of resource names in the directory (like ``os.listdir()``)""" + ... + + +class WorkingSet: + """A collection of active distributions on sys.path (or a similar list)""" + + def __init__(self, entries: Iterable[str] | None = None) -> None: + """Create working set from list of path entries (default=sys.path)""" + self.entries: list[str] = [] + self.entry_keys: dict[str | None, list[str]] = {} + self.by_key: dict[str, Distribution] = {} + self.normalized_to_canonical_keys: dict[str, str] = {} + self.callbacks: list[Callable[[Distribution], object]] = [] + + if entries is None: + entries = sys.path + + for entry in entries: + self.add_entry(entry) + + @classmethod + def _build_master(cls): + """ + Prepare the master working set. + """ + ws = cls() + try: + from __main__ import __requires__ + except ImportError: + # The main program does not list any requirements + return ws + + # ensure the requirements are met + try: + ws.require(__requires__) + except VersionConflict: + return cls._build_from_requirements(__requires__) + + return ws + + @classmethod + def _build_from_requirements(cls, req_spec): + """ + Build a working set from a requirement spec. Rewrites sys.path. + """ + # try it without defaults already on sys.path + # by starting with an empty path + ws = cls([]) + reqs = parse_requirements(req_spec) + dists = ws.resolve(reqs, Environment()) + for dist in dists: + ws.add(dist) + + # add any missing entries from sys.path + for entry in sys.path: + if entry not in ws.entries: + ws.add_entry(entry) + + # then copy back to sys.path + sys.path[:] = ws.entries + return ws + + def add_entry(self, entry: str) -> None: + """Add a path item to ``.entries``, finding any distributions on it + + ``find_distributions(entry, True)`` is used to find distributions + corresponding to the path entry, and they are added. `entry` is + always appended to ``.entries``, even if it is already present. + (This is because ``sys.path`` can contain the same value more than + once, and the ``.entries`` of the ``sys.path`` WorkingSet should always + equal ``sys.path``.) + """ + self.entry_keys.setdefault(entry, []) + self.entries.append(entry) + for dist in find_distributions(entry, True): + self.add(dist, entry, False) + + def __contains__(self, dist: Distribution) -> bool: + """True if `dist` is the active distribution for its project""" + return self.by_key.get(dist.key) == dist + + def find(self, req: Requirement) -> Distribution | None: + """Find a distribution matching requirement `req` + + If there is an active distribution for the requested project, this + returns it as long as it meets the version requirement specified by + `req`. But, if there is an active distribution for the project and it + does *not* meet the `req` requirement, ``VersionConflict`` is raised. + If there is no active distribution for the requested project, ``None`` + is returned. + """ + dist: Distribution | None = None + + candidates = ( + req.key, + self.normalized_to_canonical_keys.get(req.key), + safe_name(req.key).replace(".", "-"), + ) + + for candidate in filter(None, candidates): + dist = self.by_key.get(candidate) + if dist: + req.key = candidate + break + + if dist is not None and dist not in req: + # XXX add more info + raise VersionConflict(dist, req) + return dist + + def iter_entry_points( + self, group: str, name: str | None = None + ) -> Iterator[EntryPoint]: + """Yield entry point objects from `group` matching `name` + + If `name` is None, yields all entry points in `group` from all + distributions in the working set, otherwise only ones matching + both `group` and `name` are yielded (in distribution order). + """ + return ( + entry + for dist in self + for entry in dist.get_entry_map(group).values() + if name is None or name == entry.name + ) + + def run_script(self, requires: str, script_name: str) -> None: + """Locate distribution for `requires` and run `script_name` script""" + ns = sys._getframe(1).f_globals + name = ns['__name__'] + ns.clear() + ns['__name__'] = name + self.require(requires)[0].run_script(script_name, ns) + + def __iter__(self) -> Iterator[Distribution]: + """Yield distributions for non-duplicate projects in the working set + + The yield order is the order in which the items' path entries were + added to the working set. + """ + seen = set() + for item in self.entries: + if item not in self.entry_keys: + # workaround a cache issue + continue + + for key in self.entry_keys[item]: + if key not in seen: + seen.add(key) + yield self.by_key[key] + + def add( + self, + dist: Distribution, + entry: str | None = None, + insert: bool = True, + replace: bool = False, + ) -> None: + """Add `dist` to working set, associated with `entry` + + If `entry` is unspecified, it defaults to the ``.location`` of `dist`. + On exit from this routine, `entry` is added to the end of the working + set's ``.entries`` (if it wasn't already present). + + `dist` is only added to the working set if it's for a project that + doesn't already have a distribution in the set, unless `replace=True`. + If it's added, any callbacks registered with the ``subscribe()`` method + will be called. + """ + if insert: + dist.insert_on(self.entries, entry, replace=replace) + + if entry is None: + entry = dist.location + keys = self.entry_keys.setdefault(entry, []) + keys2 = self.entry_keys.setdefault(dist.location, []) + if not replace and dist.key in self.by_key: + # ignore hidden distros + return + + self.by_key[dist.key] = dist + normalized_name = packaging.utils.canonicalize_name(dist.key) + self.normalized_to_canonical_keys[normalized_name] = dist.key + if dist.key not in keys: + keys.append(dist.key) + if dist.key not in keys2: + keys2.append(dist.key) + self._added_new(dist) + + @overload + def resolve( + self, + requirements: Iterable[Requirement], + env: Environment | None, + installer: _StrictInstallerType[_DistributionT], + replace_conflicting: bool = False, + extras: tuple[str, ...] | None = None, + ) -> list[_DistributionT]: ... + @overload + def resolve( + self, + requirements: Iterable[Requirement], + env: Environment | None = None, + *, + installer: _StrictInstallerType[_DistributionT], + replace_conflicting: bool = False, + extras: tuple[str, ...] | None = None, + ) -> list[_DistributionT]: ... + @overload + def resolve( + self, + requirements: Iterable[Requirement], + env: Environment | None = None, + installer: _InstallerType | None = None, + replace_conflicting: bool = False, + extras: tuple[str, ...] | None = None, + ) -> list[Distribution]: ... + def resolve( + self, + requirements: Iterable[Requirement], + env: Environment | None = None, + installer: _InstallerType | None | _StrictInstallerType[_DistributionT] = None, + replace_conflicting: bool = False, + extras: tuple[str, ...] | None = None, + ) -> list[Distribution] | list[_DistributionT]: + """List all distributions needed to (recursively) meet `requirements` + + `requirements` must be a sequence of ``Requirement`` objects. `env`, + if supplied, should be an ``Environment`` instance. If + not supplied, it defaults to all distributions available within any + entry or distribution in the working set. `installer`, if supplied, + will be invoked with each requirement that cannot be met by an + already-installed distribution; it should return a ``Distribution`` or + ``None``. + + Unless `replace_conflicting=True`, raises a VersionConflict exception + if + any requirements are found on the path that have the correct name but + the wrong version. Otherwise, if an `installer` is supplied it will be + invoked to obtain the correct version of the requirement and activate + it. + + `extras` is a list of the extras to be used with these requirements. + This is important because extra requirements may look like `my_req; + extra = "my_extra"`, which would otherwise be interpreted as a purely + optional requirement. Instead, we want to be able to assert that these + requirements are truly required. + """ + + # set up the stack + requirements = list(requirements)[::-1] + # set of processed requirements + processed = set() + # key -> dist + best: dict[str, Distribution] = {} + to_activate: list[Distribution] = [] + + req_extras = _ReqExtras() + + # Mapping of requirement to set of distributions that required it; + # useful for reporting info about conflicts. + required_by = collections.defaultdict[Requirement, set[str]](set) + + while requirements: + # process dependencies breadth-first + req = requirements.pop(0) + if req in processed: + # Ignore cyclic or redundant dependencies + continue + + if not req_extras.markers_pass(req, extras): + continue + + dist = self._resolve_dist( + req, best, replace_conflicting, env, installer, required_by, to_activate + ) + + # push the new requirements onto the stack + new_requirements = dist.requires(req.extras)[::-1] + requirements.extend(new_requirements) + + # Register the new requirements needed by req + for new_requirement in new_requirements: + required_by[new_requirement].add(req.project_name) + req_extras[new_requirement] = req.extras + + processed.add(req) + + # return list of distros to activate + return to_activate + + def _resolve_dist( + self, req, best, replace_conflicting, env, installer, required_by, to_activate + ) -> Distribution: + dist = best.get(req.key) + if dist is None: + # Find the best distribution and add it to the map + dist = self.by_key.get(req.key) + if dist is None or (dist not in req and replace_conflicting): + ws = self + if env is None: + if dist is None: + env = Environment(self.entries) + else: + # Use an empty environment and workingset to avoid + # any further conflicts with the conflicting + # distribution + env = Environment([]) + ws = WorkingSet([]) + dist = best[req.key] = env.best_match( + req, ws, installer, replace_conflicting=replace_conflicting + ) + if dist is None: + requirers = required_by.get(req, None) + raise DistributionNotFound(req, requirers) + to_activate.append(dist) + if dist not in req: + # Oops, the "best" so far conflicts with a dependency + dependent_req = required_by[req] + raise VersionConflict(dist, req).with_context(dependent_req) + return dist + + @overload + def find_plugins( + self, + plugin_env: Environment, + full_env: Environment | None, + installer: _StrictInstallerType[_DistributionT], + fallback: bool = True, + ) -> tuple[list[_DistributionT], dict[Distribution, Exception]]: ... + @overload + def find_plugins( + self, + plugin_env: Environment, + full_env: Environment | None = None, + *, + installer: _StrictInstallerType[_DistributionT], + fallback: bool = True, + ) -> tuple[list[_DistributionT], dict[Distribution, Exception]]: ... + @overload + def find_plugins( + self, + plugin_env: Environment, + full_env: Environment | None = None, + installer: _InstallerType | None = None, + fallback: bool = True, + ) -> tuple[list[Distribution], dict[Distribution, Exception]]: ... + def find_plugins( + self, + plugin_env: Environment, + full_env: Environment | None = None, + installer: _InstallerType | None | _StrictInstallerType[_DistributionT] = None, + fallback: bool = True, + ) -> tuple[ + list[Distribution] | list[_DistributionT], + dict[Distribution, Exception], + ]: + """Find all activatable distributions in `plugin_env` + + Example usage:: + + distributions, errors = working_set.find_plugins( + Environment(plugin_dirlist) + ) + # add plugins+libs to sys.path + map(working_set.add, distributions) + # display errors + print('Could not load', errors) + + The `plugin_env` should be an ``Environment`` instance that contains + only distributions that are in the project's "plugin directory" or + directories. The `full_env`, if supplied, should be an ``Environment`` + contains all currently-available distributions. If `full_env` is not + supplied, one is created automatically from the ``WorkingSet`` this + method is called on, which will typically mean that every directory on + ``sys.path`` will be scanned for distributions. + + `installer` is a standard installer callback as used by the + ``resolve()`` method. The `fallback` flag indicates whether we should + attempt to resolve older versions of a plugin if the newest version + cannot be resolved. + + This method returns a 2-tuple: (`distributions`, `error_info`), where + `distributions` is a list of the distributions found in `plugin_env` + that were loadable, along with any other distributions that are needed + to resolve their dependencies. `error_info` is a dictionary mapping + unloadable plugin distributions to an exception instance describing the + error that occurred. Usually this will be a ``DistributionNotFound`` or + ``VersionConflict`` instance. + """ + + plugin_projects = list(plugin_env) + # scan project names in alphabetic order + plugin_projects.sort() + + error_info: dict[Distribution, Exception] = {} + distributions: dict[Distribution, Exception | None] = {} + + if full_env is None: + env = Environment(self.entries) + env += plugin_env + else: + env = full_env + plugin_env + + shadow_set = self.__class__([]) + # put all our entries in shadow_set + list(map(shadow_set.add, self)) + + for project_name in plugin_projects: + for dist in plugin_env[project_name]: + req = [dist.as_requirement()] + + try: + resolvees = shadow_set.resolve(req, env, installer) + + except ResolutionError as v: + # save error info + error_info[dist] = v + if fallback: + # try the next older version of project + continue + else: + # give up on this project, keep going + break + + else: + list(map(shadow_set.add, resolvees)) + distributions.update(dict.fromkeys(resolvees)) + + # success, no need to try any more versions of this project + break + + sorted_distributions = list(distributions) + sorted_distributions.sort() + + return sorted_distributions, error_info + + def require(self, *requirements: _NestedStr) -> list[Distribution]: + """Ensure that distributions matching `requirements` are activated + + `requirements` must be a string or a (possibly-nested) sequence + thereof, specifying the distributions and versions required. The + return value is a sequence of the distributions that needed to be + activated to fulfill the requirements; all relevant distributions are + included, even if they were already activated in this working set. + """ + needed = self.resolve(parse_requirements(requirements)) + + for dist in needed: + self.add(dist) + + return needed + + def subscribe( + self, callback: Callable[[Distribution], object], existing: bool = True + ) -> None: + """Invoke `callback` for all distributions + + If `existing=True` (default), + call on all existing ones, as well. + """ + if callback in self.callbacks: + return + self.callbacks.append(callback) + if not existing: + return + for dist in self: + callback(dist) + + def _added_new(self, dist) -> None: + for callback in self.callbacks: + callback(dist) + + def __getstate__( + self, + ) -> tuple[ + list[str], + dict[str | None, list[str]], + dict[str, Distribution], + dict[str, str], + list[Callable[[Distribution], object]], + ]: + return ( + self.entries[:], + self.entry_keys.copy(), + self.by_key.copy(), + self.normalized_to_canonical_keys.copy(), + self.callbacks[:], + ) + + def __setstate__(self, e_k_b_n_c) -> None: + entries, keys, by_key, normalized_to_canonical_keys, callbacks = e_k_b_n_c + self.entries = entries[:] + self.entry_keys = keys.copy() + self.by_key = by_key.copy() + self.normalized_to_canonical_keys = normalized_to_canonical_keys.copy() + self.callbacks = callbacks[:] + + +class _ReqExtras(dict["Requirement", tuple[str, ...]]): + """ + Map each requirement to the extras that demanded it. + """ + + def markers_pass(self, req: Requirement, extras: tuple[str, ...] | None = None): + """ + Evaluate markers for req against each extra that + demanded it. + + Return False if the req has a marker and fails + evaluation. Otherwise, return True. + """ + return not req.marker or any( + req.marker.evaluate({'extra': extra}) + for extra in self.get(req, ()) + (extras or ("",)) + ) + + +class Environment: + """Searchable snapshot of distributions on a search path""" + + def __init__( + self, + search_path: Iterable[str] | None = None, + platform: str | None = get_supported_platform(), + python: str | None = PY_MAJOR, + ) -> None: + """Snapshot distributions available on a search path + + Any distributions found on `search_path` are added to the environment. + `search_path` should be a sequence of ``sys.path`` items. If not + supplied, ``sys.path`` is used. + + `platform` is an optional string specifying the name of the platform + that platform-specific distributions must be compatible with. If + unspecified, it defaults to the current platform. `python` is an + optional string naming the desired version of Python (e.g. ``'3.6'``); + it defaults to the current version. + + You may explicitly set `platform` (and/or `python`) to ``None`` if you + wish to map *all* distributions, not just those compatible with the + running platform or Python version. + """ + self._distmap: dict[str, list[Distribution]] = {} + self.platform = platform + self.python = python + self.scan(search_path) + + def can_add(self, dist: Distribution) -> bool: + """Is distribution `dist` acceptable for this environment? + + The distribution must match the platform and python version + requirements specified when this environment was created, or False + is returned. + """ + py_compat = ( + self.python is None + or dist.py_version is None + or dist.py_version == self.python + ) + return py_compat and compatible_platforms(dist.platform, self.platform) + + def remove(self, dist: Distribution) -> None: + """Remove `dist` from the environment""" + self._distmap[dist.key].remove(dist) + + def scan(self, search_path: Iterable[str] | None = None) -> None: + """Scan `search_path` for distributions usable in this environment + + Any distributions found are added to the environment. + `search_path` should be a sequence of ``sys.path`` items. If not + supplied, ``sys.path`` is used. Only distributions conforming to + the platform/python version defined at initialization are added. + """ + if search_path is None: + search_path = sys.path + + for item in search_path: + for dist in find_distributions(item): + self.add(dist) + + def __getitem__(self, project_name: str) -> list[Distribution]: + """Return a newest-to-oldest list of distributions for `project_name` + + Uses case-insensitive `project_name` comparison, assuming all the + project's distributions use their project's name converted to all + lowercase as their key. + + """ + distribution_key = project_name.lower() + return self._distmap.get(distribution_key, []) + + def add(self, dist: Distribution) -> None: + """Add `dist` if we ``can_add()`` it and it has not already been added""" + if self.can_add(dist) and dist.has_version(): + dists = self._distmap.setdefault(dist.key, []) + if dist not in dists: + dists.append(dist) + dists.sort(key=operator.attrgetter('hashcmp'), reverse=True) + + @overload + def best_match( + self, + req: Requirement, + working_set: WorkingSet, + installer: _StrictInstallerType[_DistributionT], + replace_conflicting: bool = False, + ) -> _DistributionT: ... + @overload + def best_match( + self, + req: Requirement, + working_set: WorkingSet, + installer: _InstallerType | None = None, + replace_conflicting: bool = False, + ) -> Distribution | None: ... + def best_match( + self, + req: Requirement, + working_set: WorkingSet, + installer: _InstallerType | None | _StrictInstallerType[_DistributionT] = None, + replace_conflicting: bool = False, + ) -> Distribution | None: + """Find distribution best matching `req` and usable on `working_set` + + This calls the ``find(req)`` method of the `working_set` to see if a + suitable distribution is already active. (This may raise + ``VersionConflict`` if an unsuitable version of the project is already + active in the specified `working_set`.) If a suitable distribution + isn't active, this method returns the newest distribution in the + environment that meets the ``Requirement`` in `req`. If no suitable + distribution is found, and `installer` is supplied, then the result of + calling the environment's ``obtain(req, installer)`` method will be + returned. + """ + try: + dist = working_set.find(req) + except VersionConflict: + if not replace_conflicting: + raise + dist = None + if dist is not None: + return dist + for dist in self[req.key]: + if dist in req: + return dist + # try to download/install + return self.obtain(req, installer) + + @overload + def obtain( + self, + requirement: Requirement, + installer: _StrictInstallerType[_DistributionT], + ) -> _DistributionT: ... + @overload + def obtain( + self, + requirement: Requirement, + installer: Callable[[Requirement], None] | None = None, + ) -> None: ... + @overload + def obtain( + self, + requirement: Requirement, + installer: _InstallerType | None = None, + ) -> Distribution | None: ... + def obtain( + self, + requirement: Requirement, + installer: Callable[[Requirement], None] + | _InstallerType + | None + | _StrictInstallerType[_DistributionT] = None, + ) -> Distribution | None: + """Obtain a distribution matching `requirement` (e.g. via download) + + Obtain a distro that matches requirement (e.g. via download). In the + base ``Environment`` class, this routine just returns + ``installer(requirement)``, unless `installer` is None, in which case + None is returned instead. This method is a hook that allows subclasses + to attempt other ways of obtaining a distribution before falling back + to the `installer` argument.""" + return installer(requirement) if installer else None + + def __iter__(self) -> Iterator[str]: + """Yield the unique project names of the available distributions""" + for key in self._distmap.keys(): + if self[key]: + yield key + + def __iadd__(self, other: Distribution | Environment) -> Self: + """In-place addition of a distribution or environment""" + if isinstance(other, Distribution): + self.add(other) + elif isinstance(other, Environment): + for project in other: + for dist in other[project]: + self.add(dist) + else: + raise TypeError(f"Can't add {other!r} to environment") + return self + + def __add__(self, other: Distribution | Environment) -> Self: + """Add an environment or distribution to an environment""" + new = self.__class__([], platform=None, python=None) + for env in self, other: + new += env + return new + + +# XXX backward compatibility +AvailableDistributions = Environment + + +class ExtractionError(RuntimeError): + """An error occurred extracting a resource + + The following attributes are available from instances of this exception: + + manager + The resource manager that raised this exception + + cache_path + The base directory for resource extraction + + original_error + The exception instance that caused extraction to fail + """ + + manager: ResourceManager + cache_path: str + original_error: BaseException | None + + +class ResourceManager: + """Manage resource extraction and packages""" + + extraction_path: str | None = None + + def __init__(self) -> None: + # acts like a set + self.cached_files: dict[str, Literal[True]] = {} + + def resource_exists( + self, package_or_requirement: _PkgReqType, resource_name: str + ) -> bool: + """Does the named resource exist?""" + return get_provider(package_or_requirement).has_resource(resource_name) + + def resource_isdir( + self, package_or_requirement: _PkgReqType, resource_name: str + ) -> bool: + """Is the named resource an existing directory?""" + return get_provider(package_or_requirement).resource_isdir(resource_name) + + def resource_filename( + self, package_or_requirement: _PkgReqType, resource_name: str + ) -> str: + """Return a true filesystem path for specified resource""" + return get_provider(package_or_requirement).get_resource_filename( + self, resource_name + ) + + def resource_stream( + self, package_or_requirement: _PkgReqType, resource_name: str + ) -> _ResourceStream: + """Return a readable file-like object for specified resource""" + return get_provider(package_or_requirement).get_resource_stream( + self, resource_name + ) + + def resource_string( + self, package_or_requirement: _PkgReqType, resource_name: str + ) -> bytes: + """Return specified resource as :obj:`bytes`""" + return get_provider(package_or_requirement).get_resource_string( + self, resource_name + ) + + def resource_listdir( + self, package_or_requirement: _PkgReqType, resource_name: str + ) -> list[str]: + """List the contents of the named resource directory""" + return get_provider(package_or_requirement).resource_listdir(resource_name) + + def extraction_error(self) -> NoReturn: + """Give an error message for problems extracting file(s)""" + + old_exc = sys.exc_info()[1] + cache_path = self.extraction_path or get_default_cache() + + tmpl = textwrap.dedent( + """ + Can't extract file(s) to egg cache + + The following error occurred while trying to extract file(s) + to the Python egg cache: + + {old_exc} + + The Python egg cache directory is currently set to: + + {cache_path} + + Perhaps your account does not have write access to this directory? + You can change the cache directory by setting the PYTHON_EGG_CACHE + environment variable to point to an accessible directory. + """ + ).lstrip() + err = ExtractionError(tmpl.format(**locals())) + err.manager = self + err.cache_path = cache_path + err.original_error = old_exc + raise err + + def get_cache_path(self, archive_name: str, names: Iterable[StrPath] = ()) -> str: + """Return absolute location in cache for `archive_name` and `names` + + The parent directory of the resulting path will be created if it does + not already exist. `archive_name` should be the base filename of the + enclosing egg (which may not be the name of the enclosing zipfile!), + including its ".egg" extension. `names`, if provided, should be a + sequence of path name parts "under" the egg's extraction location. + + This method should only be called by resource providers that need to + obtain an extraction location, and only for names they intend to + extract, as it tracks the generated names for possible cleanup later. + """ + extract_path = self.extraction_path or get_default_cache() + target_path = os.path.join(extract_path, archive_name + '-tmp', *names) + try: + _bypass_ensure_directory(target_path) + except Exception: + self.extraction_error() + + self._warn_unsafe_extraction_path(extract_path) + + self.cached_files[target_path] = True + return target_path + + @staticmethod + def _warn_unsafe_extraction_path(path) -> None: + """ + If the default extraction path is overridden and set to an insecure + location, such as /tmp, it opens up an opportunity for an attacker to + replace an extracted file with an unauthorized payload. Warn the user + if a known insecure location is used. + + See Distribute #375 for more details. + """ + if os.name == 'nt' and not path.startswith(os.environ['windir']): + # On Windows, permissions are generally restrictive by default + # and temp directories are not writable by other users, so + # bypass the warning. + return + mode = os.stat(path).st_mode + if mode & stat.S_IWOTH or mode & stat.S_IWGRP: + msg = ( + "Extraction path is writable by group/others " + "and vulnerable to attack when " + "used with get_resource_filename ({path}). " + "Consider a more secure " + "location (set with .set_extraction_path or the " + "PYTHON_EGG_CACHE environment variable)." + ).format(**locals()) + warnings.warn(msg, UserWarning) + + def postprocess(self, tempname: StrOrBytesPath, filename: StrOrBytesPath) -> None: + """Perform any platform-specific postprocessing of `tempname` + + This is where Mac header rewrites should be done; other platforms don't + have anything special they should do. + + Resource providers should call this method ONLY after successfully + extracting a compressed resource. They must NOT call it on resources + that are already in the filesystem. + + `tempname` is the current (temporary) name of the file, and `filename` + is the name it will be renamed to by the caller after this routine + returns. + """ + + if os.name == 'posix': + # Make the resource executable + mode = ((os.stat(tempname).st_mode) | 0o555) & 0o7777 + os.chmod(tempname, mode) + + def set_extraction_path(self, path: str) -> None: + """Set the base path where resources will be extracted to, if needed. + + If you do not call this routine before any extractions take place, the + path defaults to the return value of ``get_default_cache()``. (Which + is based on the ``PYTHON_EGG_CACHE`` environment variable, with various + platform-specific fallbacks. See that routine's documentation for more + details.) + + Resources are extracted to subdirectories of this path based upon + information given by the ``IResourceProvider``. You may set this to a + temporary directory, but then you must call ``cleanup_resources()`` to + delete the extracted files when done. There is no guarantee that + ``cleanup_resources()`` will be able to remove all extracted files. + + (Note: you may not change the extraction path for a given resource + manager once resources have been extracted, unless you first call + ``cleanup_resources()``.) + """ + if self.cached_files: + raise ValueError("Can't change extraction path, files already extracted") + + self.extraction_path = path + + def cleanup_resources(self, force: bool = False) -> list[str]: + """ + Delete all extracted resource files and directories, returning a list + of the file and directory names that could not be successfully removed. + This function does not have any concurrency protection, so it should + generally only be called when the extraction path is a temporary + directory exclusive to a single process. This method is not + automatically called; you must call it explicitly or register it as an + ``atexit`` function if you wish to ensure cleanup of a temporary + directory used for extractions. + """ + # XXX + return [] + + +def get_default_cache() -> str: + """ + Return the ``PYTHON_EGG_CACHE`` environment variable + or a platform-relevant user cache dir for an app + named "Python-Eggs". + """ + return os.environ.get('PYTHON_EGG_CACHE') or _user_cache_dir(appname='Python-Eggs') + + +def safe_name(name: str) -> str: + """Convert an arbitrary string to a standard distribution name + + Any runs of non-alphanumeric/. characters are replaced with a single '-'. + """ + return re.sub('[^A-Za-z0-9.]+', '-', name) + + +def safe_version(version: str) -> str: + """ + Convert an arbitrary string to a standard version string + """ + try: + # normalize the version + return str(packaging.version.Version(version)) + except packaging.version.InvalidVersion: + version = version.replace(' ', '.') + return re.sub('[^A-Za-z0-9.]+', '-', version) + + +def _forgiving_version(version) -> str: + """Fallback when ``safe_version`` is not safe enough + >>> parse_version(_forgiving_version('0.23ubuntu1')) + <Version('0.23.dev0+sanitized.ubuntu1')> + >>> parse_version(_forgiving_version('0.23-')) + <Version('0.23.dev0+sanitized')> + >>> parse_version(_forgiving_version('0.-_')) + <Version('0.dev0+sanitized')> + >>> parse_version(_forgiving_version('42.+?1')) + <Version('42.dev0+sanitized.1')> + >>> parse_version(_forgiving_version('hello world')) + <Version('0.dev0+sanitized.hello.world')> + """ + version = version.replace(' ', '.') + match = _PEP440_FALLBACK.search(version) + if match: + safe = match["safe"] + rest = version[len(safe) :] + else: + safe = "0" + rest = version + local = f"sanitized.{_safe_segment(rest)}".strip(".") + return f"{safe}.dev0+{local}" + + +def _safe_segment(segment): + """Convert an arbitrary string into a safe segment""" + segment = re.sub('[^A-Za-z0-9.]+', '-', segment) + segment = re.sub('-[^A-Za-z0-9]+', '-', segment) + return re.sub(r'\.[^A-Za-z0-9]+', '.', segment).strip(".-") + + +def safe_extra(extra: str) -> str: + """Convert an arbitrary string to a standard 'extra' name + + Any runs of non-alphanumeric characters are replaced with a single '_', + and the result is always lowercased. + """ + return re.sub('[^A-Za-z0-9.-]+', '_', extra).lower() + + +def to_filename(name: str) -> str: + """Convert a project or version name to its filename-escaped form + + Any '-' characters are currently replaced with '_'. + """ + return name.replace('-', '_') + + +def invalid_marker(text: str) -> SyntaxError | Literal[False]: + """ + Validate text as a PEP 508 environment marker; return an exception + if invalid or False otherwise. + """ + try: + evaluate_marker(text) + except SyntaxError as e: + e.filename = None + e.lineno = None + return e + return False + + +def evaluate_marker(text: str, extra: str | None = None) -> bool: + """ + Evaluate a PEP 508 environment marker. + Return a boolean indicating the marker result in this environment. + Raise SyntaxError if marker is invalid. + + This implementation uses the 'pyparsing' module. + """ + try: + marker = packaging.markers.Marker(text) + return marker.evaluate() + except packaging.markers.InvalidMarker as e: + raise SyntaxError(e) from e + + +class NullProvider: + """Try to implement resources and metadata for arbitrary PEP 302 loaders""" + + egg_name: str | None = None + egg_info: str | None = None + loader: LoaderProtocol | None = None + + def __init__(self, module: _ModuleLike) -> None: + self.loader = getattr(module, '__loader__', None) + self.module_path = os.path.dirname(getattr(module, '__file__', '')) + + def get_resource_filename( + self, manager: ResourceManager, resource_name: str + ) -> str: + return self._fn(self.module_path, resource_name) + + def get_resource_stream( + self, manager: ResourceManager, resource_name: str + ) -> BinaryIO: + return io.BytesIO(self.get_resource_string(manager, resource_name)) + + def get_resource_string( + self, manager: ResourceManager, resource_name: str + ) -> bytes: + return self._get(self._fn(self.module_path, resource_name)) + + def has_resource(self, resource_name: str) -> bool: + return self._has(self._fn(self.module_path, resource_name)) + + def _get_metadata_path(self, name): + return self._fn(self.egg_info, name) + + def has_metadata(self, name: str) -> bool: + if not self.egg_info: + return False + + path = self._get_metadata_path(name) + return self._has(path) + + def get_metadata(self, name: str) -> str: + if not self.egg_info: + return "" + path = self._get_metadata_path(name) + value = self._get(path) + try: + return value.decode('utf-8') + except UnicodeDecodeError as exc: + # Include the path in the error message to simplify + # troubleshooting, and without changing the exception type. + exc.reason += f' in {name} file at path: {path}' + raise + + def get_metadata_lines(self, name: str) -> Iterator[str]: + return yield_lines(self.get_metadata(name)) + + def resource_isdir(self, resource_name: str) -> bool: + return self._isdir(self._fn(self.module_path, resource_name)) + + def metadata_isdir(self, name: str) -> bool: + return bool(self.egg_info and self._isdir(self._fn(self.egg_info, name))) + + def resource_listdir(self, resource_name: str) -> list[str]: + return self._listdir(self._fn(self.module_path, resource_name)) + + def metadata_listdir(self, name: str) -> list[str]: + if self.egg_info: + return self._listdir(self._fn(self.egg_info, name)) + return [] + + def run_script(self, script_name: str, namespace: dict[str, Any]) -> None: + script = 'scripts/' + script_name + if not self.has_metadata(script): + raise ResolutionError( + "Script {script!r} not found in metadata at {self.egg_info!r}".format( + **locals() + ), + ) + + script_text = self.get_metadata(script).replace('\r\n', '\n') + script_text = script_text.replace('\r', '\n') + script_filename = self._fn(self.egg_info, script) + namespace['__file__'] = script_filename + if os.path.exists(script_filename): + source = _read_utf8_with_fallback(script_filename) + code = compile(source, script_filename, 'exec') + exec(code, namespace, namespace) + else: + from linecache import cache + + cache[script_filename] = ( + len(script_text), + 0, + script_text.split('\n'), + script_filename, + ) + script_code = compile(script_text, script_filename, 'exec') + exec(script_code, namespace, namespace) + + def _has(self, path) -> bool: + raise NotImplementedError( + "Can't perform this operation for unregistered loader type" + ) + + def _isdir(self, path) -> bool: + raise NotImplementedError( + "Can't perform this operation for unregistered loader type" + ) + + def _listdir(self, path) -> list[str]: + raise NotImplementedError( + "Can't perform this operation for unregistered loader type" + ) + + def _fn(self, base: str | None, resource_name: str): + if base is None: + raise TypeError( + "`base` parameter in `_fn` is `None`. Either override this method or check the parameter first." + ) + self._validate_resource_path(resource_name) + if resource_name: + return os.path.join(base, *resource_name.split('/')) + return base + + @staticmethod + def _validate_resource_path(path) -> None: + """ + Validate the resource paths according to the docs. + https://setuptools.pypa.io/en/latest/pkg_resources.html#basic-resource-access + + >>> warned = getfixture('recwarn') + >>> warnings.simplefilter('always') + >>> vrp = NullProvider._validate_resource_path + >>> vrp('foo/bar.txt') + >>> bool(warned) + False + >>> vrp('../foo/bar.txt') + >>> bool(warned) + True + >>> warned.clear() + >>> vrp('/foo/bar.txt') + >>> bool(warned) + True + >>> vrp('foo/../../bar.txt') + >>> bool(warned) + True + >>> warned.clear() + >>> vrp('foo/f../bar.txt') + >>> bool(warned) + False + + Windows path separators are straight-up disallowed. + >>> vrp(r'\\foo/bar.txt') + Traceback (most recent call last): + ... + ValueError: Use of .. or absolute path in a resource path \ +is not allowed. + + >>> vrp(r'C:\\foo/bar.txt') + Traceback (most recent call last): + ... + ValueError: Use of .. or absolute path in a resource path \ +is not allowed. + + Blank values are allowed + + >>> vrp('') + >>> bool(warned) + False + + Non-string values are not. + + >>> vrp(None) + Traceback (most recent call last): + ... + AttributeError: ... + """ + invalid = ( + os.path.pardir in path.split(posixpath.sep) + or posixpath.isabs(path) + or ntpath.isabs(path) + or path.startswith("\\") + ) + if not invalid: + return + + msg = "Use of .. or absolute path in a resource path is not allowed." + + # Aggressively disallow Windows absolute paths + if (path.startswith("\\") or ntpath.isabs(path)) and not posixpath.isabs(path): + raise ValueError(msg) + + # for compatibility, warn; in future + # raise ValueError(msg) + issue_warning( + msg[:-1] + " and will raise exceptions in a future release.", + DeprecationWarning, + ) + + def _get(self, path) -> bytes: + if hasattr(self.loader, 'get_data') and self.loader: + # Already checked get_data exists + return self.loader.get_data(path) # type: ignore[attr-defined] + raise NotImplementedError( + "Can't perform this operation for loaders without 'get_data()'" + ) + + +register_loader_type(object, NullProvider) + + +def _parents(path): + """ + yield all parents of path including path + """ + last = None + while path != last: + yield path + last = path + path, _ = os.path.split(path) + + +class EggProvider(NullProvider): + """Provider based on a virtual filesystem""" + + def __init__(self, module: _ModuleLike) -> None: + super().__init__(module) + self._setup_prefix() + + def _setup_prefix(self): + # Assume that metadata may be nested inside a "basket" + # of multiple eggs and use module_path instead of .archive. + eggs = filter(_is_egg_path, _parents(self.module_path)) + egg = next(eggs, None) + egg and self._set_egg(egg) + + def _set_egg(self, path: str) -> None: + self.egg_name = os.path.basename(path) + self.egg_info = os.path.join(path, 'EGG-INFO') + self.egg_root = path + + +class DefaultProvider(EggProvider): + """Provides access to package resources in the filesystem""" + + def _has(self, path) -> bool: + return os.path.exists(path) + + def _isdir(self, path) -> bool: + return os.path.isdir(path) + + def _listdir(self, path): + return os.listdir(path) + + def get_resource_stream( + self, manager: object, resource_name: str + ) -> io.BufferedReader: + return open(self._fn(self.module_path, resource_name), 'rb') + + def _get(self, path) -> bytes: + with open(path, 'rb') as stream: + return stream.read() + + @classmethod + def _register(cls) -> None: + loader_names = ( + 'SourceFileLoader', + 'SourcelessFileLoader', + ) + for name in loader_names: + loader_cls = getattr(importlib.machinery, name, type(None)) + register_loader_type(loader_cls, cls) + + +DefaultProvider._register() + + +class EmptyProvider(NullProvider): + """Provider that returns nothing for all requests""" + + # A special case, we don't want all Providers inheriting from NullProvider to have a potentially None module_path + module_path: str | None = None # type: ignore[assignment] + + _isdir = _has = lambda self, path: False + + def _get(self, path) -> bytes: + return b'' + + def _listdir(self, path): + return [] + + def __init__(self) -> None: + pass + + +empty_provider = EmptyProvider() + + +class ZipManifests(dict[str, "MemoizedZipManifests.manifest_mod"]): + """ + zip manifest builder + """ + + # `path` could be `StrPath | IO[bytes]` but that violates the LSP for `MemoizedZipManifests.load` + @classmethod + def build(cls, path: str) -> dict[str, zipfile.ZipInfo]: + """ + Build a dictionary similar to the zipimport directory + caches, except instead of tuples, store ZipInfo objects. + + Use a platform-specific path separator (os.sep) for the path keys + for compatibility with pypy on Windows. + """ + with zipfile.ZipFile(path) as zfile: + items = ( + ( + name.replace('/', os.sep), + zfile.getinfo(name), + ) + for name in zfile.namelist() + ) + return dict(items) + + load = build + + +class MemoizedZipManifests(ZipManifests): + """ + Memoized zipfile manifests. + """ + + class manifest_mod(NamedTuple): + manifest: dict[str, zipfile.ZipInfo] + mtime: float + + def load(self, path: str) -> dict[str, zipfile.ZipInfo]: # type: ignore[override] # ZipManifests.load is a classmethod + """ + Load a manifest at path or return a suitable manifest already loaded. + """ + path = os.path.normpath(path) + mtime = os.stat(path).st_mtime + + if path not in self or self[path].mtime != mtime: + manifest = self.build(path) + self[path] = self.manifest_mod(manifest, mtime) + + return self[path].manifest + + +class ZipProvider(EggProvider): + """Resource support for zips and eggs""" + + eagers: list[str] | None = None + _zip_manifests = MemoizedZipManifests() + # ZipProvider's loader should always be a zipimporter or equivalent + loader: zipimport.zipimporter + + def __init__(self, module: _ZipLoaderModule) -> None: + super().__init__(module) + self.zip_pre = self.loader.archive + os.sep + + def _zipinfo_name(self, fspath): + # Convert a virtual filename (full path to file) into a zipfile subpath + # usable with the zipimport directory cache for our target archive + fspath = fspath.rstrip(os.sep) + if fspath == self.loader.archive: + return '' + if fspath.startswith(self.zip_pre): + return fspath[len(self.zip_pre) :] + raise AssertionError(f"{fspath} is not a subpath of {self.zip_pre}") + + def _parts(self, zip_path): + # Convert a zipfile subpath into an egg-relative path part list. + # pseudo-fs path + fspath = self.zip_pre + zip_path + if fspath.startswith(self.egg_root + os.sep): + return fspath[len(self.egg_root) + 1 :].split(os.sep) + raise AssertionError(f"{fspath} is not a subpath of {self.egg_root}") + + @property + def zipinfo(self): + return self._zip_manifests.load(self.loader.archive) + + def get_resource_filename( + self, manager: ResourceManager, resource_name: str + ) -> str: + if not self.egg_name: + raise NotImplementedError( + "resource_filename() only supported for .egg, not .zip" + ) + # no need to lock for extraction, since we use temp names + zip_path = self._resource_to_zip(resource_name) + eagers = self._get_eager_resources() + if '/'.join(self._parts(zip_path)) in eagers: + for name in eagers: + self._extract_resource(manager, self._eager_to_zip(name)) + return self._extract_resource(manager, zip_path) + + @staticmethod + def _get_date_and_size(zip_stat): + size = zip_stat.file_size + # ymdhms+wday, yday, dst + date_time = zip_stat.date_time + (0, 0, -1) + # 1980 offset already done + timestamp = time.mktime(date_time) + return timestamp, size + + # FIXME: 'ZipProvider._extract_resource' is too complex (12) + def _extract_resource(self, manager: ResourceManager, zip_path) -> str: # noqa: C901 + if zip_path in self._index(): + for name in self._index()[zip_path]: + last = self._extract_resource(manager, os.path.join(zip_path, name)) + # return the extracted directory name + return os.path.dirname(last) + + timestamp, _size = self._get_date_and_size(self.zipinfo[zip_path]) + + if not WRITE_SUPPORT: + raise OSError( + '"os.rename" and "os.unlink" are not supported on this platform' + ) + try: + if not self.egg_name: + raise OSError( + '"egg_name" is empty. This likely means no egg could be found from the "module_path".' + ) + real_path = manager.get_cache_path(self.egg_name, self._parts(zip_path)) + + if self._is_current(real_path, zip_path): + return real_path + + outf, tmpnam = _mkstemp( + ".$extract", + dir=os.path.dirname(real_path), + ) + os.write(outf, self.loader.get_data(zip_path)) + os.close(outf) + utime(tmpnam, (timestamp, timestamp)) + manager.postprocess(tmpnam, real_path) + + try: + rename(tmpnam, real_path) + + except OSError: + if os.path.isfile(real_path): + if self._is_current(real_path, zip_path): + # the file became current since it was checked above, + # so proceed. + return real_path + # Windows, del old file and retry + elif os.name == 'nt': + unlink(real_path) + rename(tmpnam, real_path) + return real_path + raise + + except OSError: + # report a user-friendly error + manager.extraction_error() + + return real_path + + def _is_current(self, file_path, zip_path): + """ + Return True if the file_path is current for this zip_path + """ + timestamp, size = self._get_date_and_size(self.zipinfo[zip_path]) + if not os.path.isfile(file_path): + return False + stat = os.stat(file_path) + if stat.st_size != size or stat.st_mtime != timestamp: + return False + # check that the contents match + zip_contents = self.loader.get_data(zip_path) + with open(file_path, 'rb') as f: + file_contents = f.read() + return zip_contents == file_contents + + def _get_eager_resources(self): + if self.eagers is None: + eagers = [] + for name in ('native_libs.txt', 'eager_resources.txt'): + if self.has_metadata(name): + eagers.extend(self.get_metadata_lines(name)) + self.eagers = eagers + return self.eagers + + def _index(self): + try: + return self._dirindex + except AttributeError: + ind = {} + for path in self.zipinfo: + parts = path.split(os.sep) + while parts: + parent = os.sep.join(parts[:-1]) + if parent in ind: + ind[parent].append(parts[-1]) + break + else: + ind[parent] = [parts.pop()] + self._dirindex = ind + return ind + + def _has(self, fspath) -> bool: + zip_path = self._zipinfo_name(fspath) + return zip_path in self.zipinfo or zip_path in self._index() + + def _isdir(self, fspath) -> bool: + return self._zipinfo_name(fspath) in self._index() + + def _listdir(self, fspath): + return list(self._index().get(self._zipinfo_name(fspath), ())) + + def _eager_to_zip(self, resource_name: str): + return self._zipinfo_name(self._fn(self.egg_root, resource_name)) + + def _resource_to_zip(self, resource_name: str): + return self._zipinfo_name(self._fn(self.module_path, resource_name)) + + +register_loader_type(zipimport.zipimporter, ZipProvider) + + +class FileMetadata(EmptyProvider): + """Metadata handler for standalone PKG-INFO files + + Usage:: + + metadata = FileMetadata("/path/to/PKG-INFO") + + This provider rejects all data and metadata requests except for PKG-INFO, + which is treated as existing, and will be the contents of the file at + the provided location. + """ + + def __init__(self, path: StrPath) -> None: + self.path = path + + def _get_metadata_path(self, name): + return self.path + + def has_metadata(self, name: str) -> bool: + return name == 'PKG-INFO' and os.path.isfile(self.path) + + def get_metadata(self, name: str) -> str: + if name != 'PKG-INFO': + raise KeyError("No metadata except PKG-INFO is available") + + with open(self.path, encoding='utf-8', errors="replace") as f: + metadata = f.read() + self._warn_on_replacement(metadata) + return metadata + + def _warn_on_replacement(self, metadata) -> None: + replacement_char = '�' + if replacement_char in metadata: + tmpl = "{self.path} could not be properly decoded in UTF-8" + msg = tmpl.format(**locals()) + warnings.warn(msg) + + def get_metadata_lines(self, name: str) -> Iterator[str]: + return yield_lines(self.get_metadata(name)) + + +class PathMetadata(DefaultProvider): + """Metadata provider for egg directories + + Usage:: + + # Development eggs: + + egg_info = "/path/to/PackageName.egg-info" + base_dir = os.path.dirname(egg_info) + metadata = PathMetadata(base_dir, egg_info) + dist_name = os.path.splitext(os.path.basename(egg_info))[0] + dist = Distribution(basedir, project_name=dist_name, metadata=metadata) + + # Unpacked egg directories: + + egg_path = "/path/to/PackageName-ver-pyver-etc.egg" + metadata = PathMetadata(egg_path, os.path.join(egg_path,'EGG-INFO')) + dist = Distribution.from_filename(egg_path, metadata=metadata) + """ + + def __init__(self, path: str, egg_info: str) -> None: + self.module_path = path + self.egg_info = egg_info + + +class EggMetadata(ZipProvider): + """Metadata provider for .egg files""" + + def __init__(self, importer: zipimport.zipimporter) -> None: + """Create a metadata provider from a zipimporter""" + + self.zip_pre = importer.archive + os.sep + self.loader = importer + if importer.prefix: + self.module_path = os.path.join(importer.archive, importer.prefix) + else: + self.module_path = importer.archive + self._setup_prefix() + + +_distribution_finders: dict[type, _DistFinderType[Any]] = _declare_state( + 'dict', '_distribution_finders', {} +) + + +def register_finder( + importer_type: type[_T], distribution_finder: _DistFinderType[_T] +) -> None: + """Register `distribution_finder` to find distributions in sys.path items + + `importer_type` is the type or class of a PEP 302 "Importer" (sys.path item + handler), and `distribution_finder` is a callable that, passed a path + item and the importer instance, yields ``Distribution`` instances found on + that path item. See ``pkg_resources.find_on_path`` for an example.""" + _distribution_finders[importer_type] = distribution_finder + + +def find_distributions(path_item: str, only: bool = False) -> Iterable[Distribution]: + """Yield distributions accessible via `path_item`""" + importer = get_importer(path_item) + finder = _find_adapter(_distribution_finders, importer) + return finder(importer, path_item, only) + + +def find_eggs_in_zip( + importer: zipimport.zipimporter, path_item: str, only: bool = False +) -> Iterator[Distribution]: + """ + Find eggs in zip files; possibly multiple nested eggs. + """ + if importer.archive.endswith('.whl'): + # wheels are not supported with this finder + # they don't have PKG-INFO metadata, and won't ever contain eggs + return + metadata = EggMetadata(importer) + if metadata.has_metadata('PKG-INFO'): + yield Distribution.from_filename(path_item, metadata=metadata) + if only: + # don't yield nested distros + return + for subitem in metadata.resource_listdir(''): + if _is_egg_path(subitem): + subpath = os.path.join(path_item, subitem) + dists = find_eggs_in_zip(zipimport.zipimporter(subpath), subpath) + yield from dists + elif subitem.lower().endswith(('.dist-info', '.egg-info')): + subpath = os.path.join(path_item, subitem) + submeta = EggMetadata(zipimport.zipimporter(subpath)) + submeta.egg_info = subpath + yield Distribution.from_location(path_item, subitem, submeta) + + +register_finder(zipimport.zipimporter, find_eggs_in_zip) + + +def find_nothing( + importer: object | None, path_item: str | None, only: bool | None = False +): + return () + + +register_finder(object, find_nothing) + + +def find_on_path(importer: object | None, path_item, only=False): + """Yield distributions accessible on a sys.path directory""" + path_item = _normalize_cached(path_item) + + if _is_unpacked_egg(path_item): + yield Distribution.from_filename( + path_item, + metadata=PathMetadata(path_item, os.path.join(path_item, 'EGG-INFO')), + ) + return + + entries = (os.path.join(path_item, child) for child in safe_listdir(path_item)) + + # scan for .egg and .egg-info in directory + for entry in sorted(entries): + fullpath = os.path.join(path_item, entry) + factory = dist_factory(path_item, entry, only) + yield from factory(fullpath) + + +def dist_factory(path_item, entry, only): + """Return a dist_factory for the given entry.""" + lower = entry.lower() + is_egg_info = lower.endswith('.egg-info') + is_dist_info = lower.endswith('.dist-info') and os.path.isdir( + os.path.join(path_item, entry) + ) + is_meta = is_egg_info or is_dist_info + return ( + distributions_from_metadata + if is_meta + else find_distributions + if not only and _is_egg_path(entry) + else resolve_egg_link + if not only and lower.endswith('.egg-link') + else NoDists() + ) + + +class NoDists: + """ + >>> bool(NoDists()) + False + + >>> list(NoDists()('anything')) + [] + """ + + def __bool__(self) -> Literal[False]: + return False + + def __call__(self, fullpath: object): + return iter(()) + + +def safe_listdir(path: StrOrBytesPath): + """ + Attempt to list contents of path, but suppress some exceptions. + """ + try: + return os.listdir(path) + except (PermissionError, NotADirectoryError): + pass + except OSError as e: + # Ignore the directory if does not exist, not a directory or + # permission denied + if e.errno not in (errno.ENOTDIR, errno.EACCES, errno.ENOENT): + raise + return () + + +def distributions_from_metadata(path: str): + root = os.path.dirname(path) + if os.path.isdir(path): + if len(os.listdir(path)) == 0: + # empty metadata dir; skip + return + metadata: _MetadataType = PathMetadata(root, path) + else: + metadata = FileMetadata(path) + entry = os.path.basename(path) + yield Distribution.from_location( + root, + entry, + metadata, + precedence=DEVELOP_DIST, + ) + + +def non_empty_lines(path): + """ + Yield non-empty lines from file at path + """ + for line in _read_utf8_with_fallback(path).splitlines(): + line = line.strip() + if line: + yield line + + +def resolve_egg_link(path): + """ + Given a path to an .egg-link, resolve distributions + present in the referenced path. + """ + referenced_paths = non_empty_lines(path) + resolved_paths = ( + os.path.join(os.path.dirname(path), ref) for ref in referenced_paths + ) + dist_groups = map(find_distributions, resolved_paths) + return next(dist_groups, ()) + + +if hasattr(pkgutil, 'ImpImporter'): + register_finder(pkgutil.ImpImporter, find_on_path) + +register_finder(importlib.machinery.FileFinder, find_on_path) + +_namespace_handlers: dict[type, _NSHandlerType[Any]] = _declare_state( + 'dict', '_namespace_handlers', {} +) +_namespace_packages: dict[str | None, list[str]] = _declare_state( + 'dict', '_namespace_packages', {} +) + + +def register_namespace_handler( + importer_type: type[_T], namespace_handler: _NSHandlerType[_T] +) -> None: + """Register `namespace_handler` to declare namespace packages + + `importer_type` is the type or class of a PEP 302 "Importer" (sys.path item + handler), and `namespace_handler` is a callable like this:: + + def namespace_handler(importer, path_entry, moduleName, module): + # return a path_entry to use for child packages + + Namespace handlers are only called if the importer object has already + agreed that it can handle the relevant path item, and they should only + return a subpath if the module __path__ does not already contain an + equivalent subpath. For an example namespace handler, see + ``pkg_resources.file_ns_handler``. + """ + _namespace_handlers[importer_type] = namespace_handler + + +def _handle_ns(packageName, path_item): + """Ensure that named package includes a subpath of path_item (if needed)""" + + importer = get_importer(path_item) + if importer is None: + return None + + # use find_spec (PEP 451) and fall-back to find_module (PEP 302) + try: + spec = importer.find_spec(packageName) + except AttributeError: + # capture warnings due to #1111 + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + loader = importer.find_module(packageName) + else: + loader = spec.loader if spec else None + + if loader is None: + return None + module = sys.modules.get(packageName) + if module is None: + module = sys.modules[packageName] = types.ModuleType(packageName) + module.__path__ = [] + _set_parent_ns(packageName) + elif not hasattr(module, '__path__'): + raise TypeError("Not a package:", packageName) + handler = _find_adapter(_namespace_handlers, importer) + subpath = handler(importer, path_item, packageName, module) + if subpath is not None: + path = module.__path__ + path.append(subpath) + importlib.import_module(packageName) + _rebuild_mod_path(path, packageName, module) + return subpath + + +def _rebuild_mod_path(orig_path, package_name, module: types.ModuleType) -> None: + """ + Rebuild module.__path__ ensuring that all entries are ordered + corresponding to their sys.path order + """ + sys_path = [_normalize_cached(p) for p in sys.path] + + def safe_sys_path_index(entry): + """ + Workaround for #520 and #513. + """ + try: + return sys_path.index(entry) + except ValueError: + return float('inf') + + def position_in_sys_path(path): + """ + Return the ordinal of the path based on its position in sys.path + """ + path_parts = path.split(os.sep) + module_parts = package_name.count('.') + 1 + parts = path_parts[:-module_parts] + return safe_sys_path_index(_normalize_cached(os.sep.join(parts))) + + new_path = sorted(orig_path, key=position_in_sys_path) + new_path = [_normalize_cached(p) for p in new_path] + + if isinstance(module.__path__, list): + module.__path__[:] = new_path + else: + module.__path__ = new_path + + +def declare_namespace(packageName: str) -> None: + """Declare that package 'packageName' is a namespace package""" + + msg = ( + f"Deprecated call to `pkg_resources.declare_namespace({packageName!r})`.\n" + "Implementing implicit namespace packages (as specified in PEP 420) " + "is preferred to `pkg_resources.declare_namespace`. " + "See https://setuptools.pypa.io/en/latest/references/" + "keywords.html#keyword-namespace-packages" + ) + warnings.warn(msg, DeprecationWarning, stacklevel=2) + + _imp.acquire_lock() + try: + if packageName in _namespace_packages: + return + + path: MutableSequence[str] = sys.path + parent, _, _ = packageName.rpartition('.') + + if parent: + declare_namespace(parent) + if parent not in _namespace_packages: + __import__(parent) + try: + path = sys.modules[parent].__path__ + except AttributeError as e: + raise TypeError("Not a package:", parent) from e + + # Track what packages are namespaces, so when new path items are added, + # they can be updated + _namespace_packages.setdefault(parent or None, []).append(packageName) + _namespace_packages.setdefault(packageName, []) + + for path_item in path: + # Ensure all the parent's path items are reflected in the child, + # if they apply + _handle_ns(packageName, path_item) + + finally: + _imp.release_lock() + + +def fixup_namespace_packages(path_item: str, parent: str | None = None) -> None: + """Ensure that previously-declared namespace packages include path_item""" + _imp.acquire_lock() + try: + for package in _namespace_packages.get(parent, ()): + subpath = _handle_ns(package, path_item) + if subpath: + fixup_namespace_packages(subpath, package) + finally: + _imp.release_lock() + + +def file_ns_handler( + importer: object, + path_item: StrPath, + packageName: str, + module: types.ModuleType, +): + """Compute an ns-package subpath for a filesystem or zipfile importer""" + + subpath = os.path.join(path_item, packageName.split('.')[-1]) + normalized = _normalize_cached(subpath) + for item in module.__path__: + if _normalize_cached(item) == normalized: + break + else: + # Only return the path if it's not already there + return subpath + + +if hasattr(pkgutil, 'ImpImporter'): + register_namespace_handler(pkgutil.ImpImporter, file_ns_handler) + +register_namespace_handler(zipimport.zipimporter, file_ns_handler) +register_namespace_handler(importlib.machinery.FileFinder, file_ns_handler) + + +def null_ns_handler( + importer: object, + path_item: str | None, + packageName: str | None, + module: _ModuleLike | None, +) -> None: + return None + + +register_namespace_handler(object, null_ns_handler) + + +@overload +def normalize_path(filename: StrPath) -> str: ... +@overload +def normalize_path(filename: BytesPath) -> bytes: ... +def normalize_path(filename: StrOrBytesPath) -> str | bytes: + """Normalize a file/dir name for comparison purposes""" + return os.path.normcase(os.path.realpath(os.path.normpath(_cygwin_patch(filename)))) + + +def _cygwin_patch(filename: StrOrBytesPath): # pragma: nocover + """ + Contrary to POSIX 2008, on Cygwin, getcwd (3) contains + symlink components. Using + os.path.abspath() works around this limitation. A fix in os.getcwd() + would probably better, in Cygwin even more so, except + that this seems to be by design... + """ + return os.path.abspath(filename) if sys.platform == 'cygwin' else filename + + +if TYPE_CHECKING: + # https://github.com/python/mypy/issues/16261 + # https://github.com/python/typeshed/issues/6347 + @overload + def _normalize_cached(filename: StrPath) -> str: ... + @overload + def _normalize_cached(filename: BytesPath) -> bytes: ... + def _normalize_cached(filename: StrOrBytesPath) -> str | bytes: ... + +else: + + @functools.cache + def _normalize_cached(filename): + return normalize_path(filename) + + +def _is_egg_path(path): + """ + Determine if given path appears to be an egg. + """ + return _is_zip_egg(path) or _is_unpacked_egg(path) + + +def _is_zip_egg(path): + return ( + path.lower().endswith('.egg') + and os.path.isfile(path) + and zipfile.is_zipfile(path) + ) + + +def _is_unpacked_egg(path): + """ + Determine if given path appears to be an unpacked egg. + """ + return path.lower().endswith('.egg') and os.path.isfile( + os.path.join(path, 'EGG-INFO', 'PKG-INFO') + ) + + +def _set_parent_ns(packageName) -> None: + parts = packageName.split('.') + name = parts.pop() + if parts: + parent = '.'.join(parts) + setattr(sys.modules[parent], name, sys.modules[packageName]) + + +MODULE = re.compile(r"\w+(\.\w+)*$").match +EGG_NAME = re.compile( + r""" + (?P<name>[^-]+) ( + -(?P<ver>[^-]+) ( + -py(?P<pyver>[^-]+) ( + -(?P<plat>.+) + )? + )? + )? + """, + re.VERBOSE | re.IGNORECASE, +).match + + +class EntryPoint: + """Object representing an advertised importable object""" + + def __init__( + self, + name: str, + module_name: str, + attrs: Iterable[str] = (), + extras: Iterable[str] = (), + dist: Distribution | None = None, + ) -> None: + if not MODULE(module_name): + raise ValueError("Invalid module name", module_name) + self.name = name + self.module_name = module_name + self.attrs = tuple(attrs) + self.extras = tuple(extras) + self.dist = dist + + def __str__(self) -> str: + s = f"{self.name} = {self.module_name}" + if self.attrs: + s += ':' + '.'.join(self.attrs) + if self.extras: + extras = ','.join(self.extras) + s += f' [{extras}]' + return s + + def __repr__(self) -> str: + return f"EntryPoint.parse({str(self)!r})" + + @overload + def load( + self, + require: Literal[True] = True, + env: Environment | None = None, + installer: _InstallerType | None = None, + ) -> _ResolvedEntryPoint: ... + @overload + def load( + self, + require: Literal[False], + *args: Any, + **kwargs: Any, + ) -> _ResolvedEntryPoint: ... + def load( + self, + require: bool = True, + *args: Environment | _InstallerType | None, + **kwargs: Environment | _InstallerType | None, + ) -> _ResolvedEntryPoint: + """ + Require packages for this EntryPoint, then resolve it. + """ + if not require or args or kwargs: + warnings.warn( + "Parameters to load are deprecated. Call .resolve and " + ".require separately.", + PkgResourcesDeprecationWarning, + stacklevel=2, + ) + if require: + # We could pass `env` and `installer` directly, + # but keeping `*args` and `**kwargs` for backwards compatibility + self.require(*args, **kwargs) # type: ignore[arg-type] + return self.resolve() + + def resolve(self) -> _ResolvedEntryPoint: + """ + Resolve the entry point from its module and attrs. + """ + module = __import__(self.module_name, fromlist=['__name__'], level=0) + try: + return functools.reduce(getattr, self.attrs, module) + except AttributeError as exc: + raise ImportError(str(exc)) from exc + + def require( + self, + env: Environment | None = None, + installer: _InstallerType | None = None, + ) -> None: + if not self.dist: + error_cls = UnknownExtra if self.extras else AttributeError + raise error_cls("Can't require() without a distribution", self) + + # Get the requirements for this entry point with all its extras and + # then resolve them. We have to pass `extras` along when resolving so + # that the working set knows what extras we want. Otherwise, for + # dist-info distributions, the working set will assume that the + # requirements for that extra are purely optional and skip over them. + reqs = self.dist.requires(self.extras) + items = working_set.resolve(reqs, env, installer, extras=self.extras) + list(map(working_set.add, items)) + + pattern = re.compile( + r'\s*' + r'(?P<name>.+?)\s*' + r'=\s*' + r'(?P<module>[\w.]+)\s*' + r'(:\s*(?P<attr>[\w.]+))?\s*' + r'(?P<extras>\[.*\])?\s*$' + ) + + @classmethod + def parse(cls, src: str, dist: Distribution | None = None) -> Self: + """Parse a single entry point from string `src` + + Entry point syntax follows the form:: + + name = some.module:some.attr [extra1, extra2] + + The entry name and module name are required, but the ``:attrs`` and + ``[extras]`` parts are optional + """ + m = cls.pattern.match(src) + if not m: + msg = "EntryPoint must be in 'name=module:attrs [extras]' format" + raise ValueError(msg, src) + res = m.groupdict() + extras = cls._parse_extras(res['extras']) + attrs = res['attr'].split('.') if res['attr'] else () + return cls(res['name'], res['module'], attrs, extras, dist) + + @classmethod + def _parse_extras(cls, extras_spec): + if not extras_spec: + return () + req = Requirement.parse('x' + extras_spec) + if req.specs: + raise ValueError + return req.extras + + @classmethod + def parse_group( + cls, + group: str, + lines: _NestedStr, + dist: Distribution | None = None, + ) -> dict[str, Self]: + """Parse an entry point group""" + if not MODULE(group): + raise ValueError("Invalid group name", group) + this: dict[str, Self] = {} + for line in yield_lines(lines): + ep = cls.parse(line, dist) + if ep.name in this: + raise ValueError("Duplicate entry point", group, ep.name) + this[ep.name] = ep + return this + + @classmethod + def parse_map( + cls, + data: str | Iterable[str] | dict[str, str | Iterable[str]], + dist: Distribution | None = None, + ) -> dict[str, dict[str, Self]]: + """Parse a map of entry point groups""" + _data: Iterable[tuple[str | None, str | Iterable[str]]] + if isinstance(data, dict): + _data = data.items() + else: + _data = split_sections(data) + maps: dict[str, dict[str, Self]] = {} + for group, lines in _data: + if group is None: + if not lines: + continue + raise ValueError("Entry points must be listed in groups") + group = group.strip() + if group in maps: + raise ValueError("Duplicate group name", group) + maps[group] = cls.parse_group(group, lines, dist) + return maps + + +def _version_from_file(lines): + """ + Given an iterable of lines from a Metadata file, return + the value of the Version field, if present, or None otherwise. + """ + + def is_version_line(line): + return line.lower().startswith('version:') + + version_lines = filter(is_version_line, lines) + line = next(iter(version_lines), '') + _, _, value = line.partition(':') + return safe_version(value.strip()) or None + + +class Distribution: + """Wrap an actual or potential sys.path entry w/metadata""" + + PKG_INFO = 'PKG-INFO' + + def __init__( + self, + location: str | None = None, + metadata: _MetadataType = None, + project_name: str | None = None, + version: str | None = None, + py_version: str | None = PY_MAJOR, + platform: str | None = None, + precedence: int = EGG_DIST, + ) -> None: + self.project_name = safe_name(project_name or 'Unknown') + if version is not None: + self._version = safe_version(version) + self.py_version = py_version + self.platform = platform + self.location = location + self.precedence = precedence + self._provider = metadata or empty_provider + + @classmethod + def from_location( + cls, + location: str, + basename: StrPath, + metadata: _MetadataType = None, + **kw: int, # We could set `precedence` explicitly, but keeping this as `**kw` for full backwards and subclassing compatibility + ) -> Distribution: + project_name, version, py_version, platform = [None] * 4 + basename, ext = os.path.splitext(basename) + if ext.lower() in _distributionImpl: + cls = _distributionImpl[ext.lower()] + + match = EGG_NAME(basename) + if match: + project_name, version, py_version, platform = match.group( + 'name', 'ver', 'pyver', 'plat' + ) + return cls( + location, + metadata, + project_name=project_name, + version=version, + py_version=py_version, + platform=platform, + **kw, + )._reload_version() + + def _reload_version(self): + return self + + @property + def hashcmp(self): + return ( + self._forgiving_parsed_version, + self.precedence, + self.key, + self.location, + self.py_version or '', + self.platform or '', + ) + + def __hash__(self) -> int: + return hash(self.hashcmp) + + def __lt__(self, other: Distribution) -> bool: + return self.hashcmp < other.hashcmp + + def __le__(self, other: Distribution) -> bool: + return self.hashcmp <= other.hashcmp + + def __gt__(self, other: Distribution) -> bool: + return self.hashcmp > other.hashcmp + + def __ge__(self, other: Distribution) -> bool: + return self.hashcmp >= other.hashcmp + + def __eq__(self, other: object) -> bool: + if not isinstance(other, self.__class__): + # It's not a Distribution, so they are not equal + return False + return self.hashcmp == other.hashcmp + + def __ne__(self, other: object) -> bool: + return not self == other + + # These properties have to be lazy so that we don't have to load any + # metadata until/unless it's actually needed. (i.e., some distributions + # may not know their name or version without loading PKG-INFO) + + @property + def key(self): + try: + return self._key + except AttributeError: + self._key = key = self.project_name.lower() + return key + + @property + def parsed_version(self): + if not hasattr(self, "_parsed_version"): + try: + self._parsed_version = parse_version(self.version) + except packaging.version.InvalidVersion as ex: + info = f"(package: {self.project_name})" + if hasattr(ex, "add_note"): + ex.add_note(info) # PEP 678 + raise + raise packaging.version.InvalidVersion(f"{str(ex)} {info}") from None + + return self._parsed_version + + @property + def _forgiving_parsed_version(self): + try: + return self.parsed_version + except packaging.version.InvalidVersion as ex: + self._parsed_version = parse_version(_forgiving_version(self.version)) + + notes = "\n".join(getattr(ex, "__notes__", [])) # PEP 678 + msg = f"""!!\n\n + ************************************************************************* + {str(ex)}\n{notes} + + This is a long overdue deprecation. + For the time being, `pkg_resources` will use `{self._parsed_version}` + as a replacement to avoid breaking existing environments, + but no future compatibility is guaranteed. + + If you maintain package {self.project_name} you should implement + the relevant changes to adequate the project to PEP 440 immediately. + ************************************************************************* + \n\n!! + """ + warnings.warn(msg, DeprecationWarning) + + return self._parsed_version + + @property + def version(self): + try: + return self._version + except AttributeError as e: + version = self._get_version() + if version is None: + path = self._get_metadata_path_for_display(self.PKG_INFO) + msg = f"Missing 'Version:' header and/or {self.PKG_INFO} file at path: {path}" + raise ValueError(msg, self) from e + + return version + + @property + def _dep_map(self): + """ + A map of extra to its list of (direct) requirements + for this distribution, including the null extra. + """ + try: + return self.__dep_map + except AttributeError: + self.__dep_map = self._filter_extras(self._build_dep_map()) + return self.__dep_map + + @staticmethod + def _filter_extras( + dm: dict[str | None, list[Requirement]], + ) -> dict[str | None, list[Requirement]]: + """ + Given a mapping of extras to dependencies, strip off + environment markers and filter out any dependencies + not matching the markers. + """ + for extra in list(filter(None, dm)): + new_extra: str | None = extra + reqs = dm.pop(extra) + new_extra, _, marker = extra.partition(':') + fails_marker = marker and ( + invalid_marker(marker) or not evaluate_marker(marker) + ) + if fails_marker: + reqs = [] + new_extra = safe_extra(new_extra) or None + + dm.setdefault(new_extra, []).extend(reqs) + return dm + + def _build_dep_map(self): + dm = {} + for name in 'requires.txt', 'depends.txt': + for extra, reqs in split_sections(self._get_metadata(name)): + dm.setdefault(extra, []).extend(parse_requirements(reqs)) + return dm + + def requires(self, extras: Iterable[str] = ()) -> list[Requirement]: + """List of Requirements needed for this distro if `extras` are used""" + dm = self._dep_map + deps: list[Requirement] = [] + deps.extend(dm.get(None, ())) + for ext in extras: + try: + deps.extend(dm[safe_extra(ext)]) + except KeyError as e: + raise UnknownExtra(f"{self} has no such extra feature {ext!r}") from e + return deps + + def _get_metadata_path_for_display(self, name): + """ + Return the path to the given metadata file, if available. + """ + try: + # We need to access _get_metadata_path() on the provider object + # directly rather than through this class's __getattr__() + # since _get_metadata_path() is marked private. + path = self._provider._get_metadata_path(name) + + # Handle exceptions e.g. in case the distribution's metadata + # provider doesn't support _get_metadata_path(). + except Exception: + return '[could not detect]' + + return path + + def _get_metadata(self, name): + if self.has_metadata(name): + yield from self.get_metadata_lines(name) + + def _get_version(self): + lines = self._get_metadata(self.PKG_INFO) + return _version_from_file(lines) + + def activate(self, path: list[str] | None = None, replace: bool = False) -> None: + """Ensure distribution is importable on `path` (default=sys.path)""" + if path is None: + path = sys.path + self.insert_on(path, replace=replace) + if path is sys.path and self.location is not None: + fixup_namespace_packages(self.location) + for pkg in self._get_metadata('namespace_packages.txt'): + if pkg in sys.modules: + declare_namespace(pkg) + + def egg_name(self): + """Return what this distribution's standard .egg filename should be""" + filename = f"{to_filename(self.project_name)}-{to_filename(self.version)}-py{self.py_version or PY_MAJOR}" + + if self.platform: + filename += '-' + self.platform + return filename + + def __repr__(self) -> str: + if self.location: + return f"{self} ({self.location})" + else: + return str(self) + + def __str__(self) -> str: + try: + version = getattr(self, 'version', None) + except ValueError: + version = None + version = version or "[unknown version]" + return f"{self.project_name} {version}" + + def __getattr__(self, attr: str): + """Delegate all unrecognized public attributes to .metadata provider""" + if attr.startswith('_'): + raise AttributeError(attr) + return getattr(self._provider, attr) + + def __dir__(self): + return list( + set(super().__dir__()) + | set(attr for attr in self._provider.__dir__() if not attr.startswith('_')) + ) + + @classmethod + def from_filename( + cls, + filename: StrPath, + metadata: _MetadataType = None, + **kw: int, # We could set `precedence` explicitly, but keeping this as `**kw` for full backwards and subclassing compatibility + ) -> Distribution: + return cls.from_location( + _normalize_cached(filename), os.path.basename(filename), metadata, **kw + ) + + def as_requirement(self): + """Return a ``Requirement`` that matches this distribution exactly""" + if isinstance(self.parsed_version, packaging.version.Version): + spec = f"{self.project_name}=={self.parsed_version}" + else: + spec = f"{self.project_name}==={self.parsed_version}" + + return Requirement.parse(spec) + + def load_entry_point(self, group: str, name: str) -> _ResolvedEntryPoint: + """Return the `name` entry point of `group` or raise ImportError""" + ep = self.get_entry_info(group, name) + if ep is None: + raise ImportError(f"Entry point {(group, name)!r} not found") + return ep.load() + + @overload + def get_entry_map(self, group: None = None) -> dict[str, dict[str, EntryPoint]]: ... + @overload + def get_entry_map(self, group: str) -> dict[str, EntryPoint]: ... + def get_entry_map(self, group: str | None = None): + """Return the entry point map for `group`, or the full entry map""" + if not hasattr(self, "_ep_map"): + self._ep_map = EntryPoint.parse_map( + self._get_metadata('entry_points.txt'), self + ) + if group is not None: + return self._ep_map.get(group, {}) + return self._ep_map + + def get_entry_info(self, group: str, name: str) -> EntryPoint | None: + """Return the EntryPoint object for `group`+`name`, or ``None``""" + return self.get_entry_map(group).get(name) + + # FIXME: 'Distribution.insert_on' is too complex (13) + def insert_on( # noqa: C901 + self, + path: list[str], + loc=None, + replace: bool = False, + ) -> None: + """Ensure self.location is on path + + If replace=False (default): + - If location is already in path anywhere, do nothing. + - Else: + - If it's an egg and its parent directory is on path, + insert just ahead of the parent. + - Else: add to the end of path. + If replace=True: + - If location is already on path anywhere (not eggs) + or higher priority than its parent (eggs) + do nothing. + - Else: + - If it's an egg and its parent directory is on path, + insert just ahead of the parent, + removing any lower-priority entries. + - Else: add it to the front of path. + """ + + loc = loc or self.location + if not loc: + return + + nloc = _normalize_cached(loc) + bdir = os.path.dirname(nloc) + npath = [(p and _normalize_cached(p) or p) for p in path] + + for p, item in enumerate(npath): + if item == nloc: + if replace: + break + else: + # don't modify path (even removing duplicates) if + # found and not replace + return + elif item == bdir and self.precedence == EGG_DIST: + # if it's an .egg, give it precedence over its directory + # UNLESS it's already been added to sys.path and replace=False + if (not replace) and nloc in npath[p:]: + return + if path is sys.path: + self.check_version_conflict() + path.insert(p, loc) + npath.insert(p, nloc) + break + else: + if path is sys.path: + self.check_version_conflict() + if replace: + path.insert(0, loc) + else: + path.append(loc) + return + + # p is the spot where we found or inserted loc; now remove duplicates + while True: + try: + np = npath.index(nloc, p + 1) + except ValueError: + break + else: + del npath[np], path[np] + # ha! + p = np + + return + + def check_version_conflict(self): + if self.key == 'setuptools': + # ignore the inevitable setuptools self-conflicts :( + return + + nsp = dict.fromkeys(self._get_metadata('namespace_packages.txt')) + loc = normalize_path(self.location) + for modname in self._get_metadata('top_level.txt'): + if ( + modname not in sys.modules + or modname in nsp + or modname in _namespace_packages + ): + continue + if modname in ('pkg_resources', 'setuptools', 'site'): + continue + fn = getattr(sys.modules[modname], '__file__', None) + if fn and ( + normalize_path(fn).startswith(loc) or fn.startswith(self.location) + ): + continue + issue_warning( + f"Module {modname} was already imported from {fn}, " + f"but {self.location} is being added to sys.path", + ) + + def has_version(self) -> bool: + try: + self.version + except ValueError: + issue_warning("Unbuilt egg for " + repr(self)) + return False + except SystemError: + # TODO: remove this except clause when python/cpython#103632 is fixed. + return False + return True + + def clone(self, **kw: str | int | IResourceProvider | None) -> Self: + """Copy this distribution, substituting in any changed keyword args""" + names = 'project_name version py_version platform location precedence' + for attr in names.split(): + kw.setdefault(attr, getattr(self, attr, None)) + kw.setdefault('metadata', self._provider) + # Unsafely unpacking. But keeping **kw for backwards and subclassing compatibility + return self.__class__(**kw) # type:ignore[arg-type] + + @property + def extras(self): + return [dep for dep in self._dep_map if dep] + + +class EggInfoDistribution(Distribution): + def _reload_version(self): + """ + Packages installed by distutils (e.g. numpy or scipy), + which uses an old safe_version, and so + their version numbers can get mangled when + converted to filenames (e.g., 1.11.0.dev0+2329eae to + 1.11.0.dev0_2329eae). These distributions will not be + parsed properly + downstream by Distribution and safe_version, so + take an extra step and try to get the version number from + the metadata file itself instead of the filename. + """ + md_version = self._get_version() + if md_version: + self._version = md_version + return self + + +class DistInfoDistribution(Distribution): + """ + Wrap an actual or potential sys.path entry + w/metadata, .dist-info style. + """ + + PKG_INFO = 'METADATA' + EQEQ = re.compile(r"([\(,])\s*(\d.*?)\s*([,\)])") + + @property + def _parsed_pkg_info(self): + """Parse and cache metadata""" + try: + return self._pkg_info + except AttributeError: + metadata = self.get_metadata(self.PKG_INFO) + self._pkg_info = email.parser.Parser().parsestr(metadata) + return self._pkg_info + + @property + def _dep_map(self): + try: + return self.__dep_map + except AttributeError: + self.__dep_map = self._compute_dependencies() + return self.__dep_map + + def _compute_dependencies(self) -> dict[str | None, list[Requirement]]: + """Recompute this distribution's dependencies.""" + self.__dep_map: dict[str | None, list[Requirement]] = {None: []} + + reqs: list[Requirement] = [] + # Including any condition expressions + for req in self._parsed_pkg_info.get_all('Requires-Dist') or []: + reqs.extend(parse_requirements(req)) + + def reqs_for_extra(extra): + for req in reqs: + if not req.marker or req.marker.evaluate({'extra': extra}): + yield req + + common = types.MappingProxyType(dict.fromkeys(reqs_for_extra(None))) + self.__dep_map[None].extend(common) + + for extra in self._parsed_pkg_info.get_all('Provides-Extra') or []: + s_extra = safe_extra(extra.strip()) + self.__dep_map[s_extra] = [ + r for r in reqs_for_extra(extra) if r not in common + ] + + return self.__dep_map + + +_distributionImpl = { + '.egg': Distribution, + '.egg-info': EggInfoDistribution, + '.dist-info': DistInfoDistribution, +} + + +def issue_warning(*args, **kw): + level = 1 + g = globals() + try: + # find the first stack frame that is *not* code in + # the pkg_resources module, to use for the warning + while sys._getframe(level).f_globals is g: + level += 1 + except ValueError: + pass + warnings.warn(stacklevel=level + 1, *args, **kw) + + +def parse_requirements(strs: _NestedStr) -> map[Requirement]: + """ + Yield ``Requirement`` objects for each specification in `strs`. + + `strs` must be a string, or a (possibly-nested) iterable thereof. + """ + return map(Requirement, join_continuation(map(drop_comment, yield_lines(strs)))) + + +class RequirementParseError(packaging.requirements.InvalidRequirement): + "Compatibility wrapper for InvalidRequirement" + + +class Requirement(packaging.requirements.Requirement): + # prefer variable length tuple to set (as found in + # packaging.requirements.Requirement) + extras: tuple[str, ...] # type: ignore[assignment] + + def __init__(self, requirement_string: str) -> None: + """DO NOT CALL THIS UNDOCUMENTED METHOD; use Requirement.parse()!""" + super().__init__(requirement_string) + self.unsafe_name = self.name + project_name = safe_name(self.name) + self.project_name, self.key = project_name, project_name.lower() + self.specs = [(spec.operator, spec.version) for spec in self.specifier] + self.extras = tuple(map(safe_extra, self.extras)) + self.hashCmp = ( + self.key, + self.url, + self.specifier, + frozenset(self.extras), + str(self.marker) if self.marker else None, + ) + self.__hash = hash(self.hashCmp) + + def __eq__(self, other: object) -> bool: + return isinstance(other, Requirement) and self.hashCmp == other.hashCmp + + def __ne__(self, other: object) -> bool: + return not self == other + + def __contains__( + self, item: Distribution | packaging.specifiers.UnparsedVersion + ) -> bool: + if isinstance(item, Distribution): + if item.key != self.key: + return False + + version = item.version + else: + version = item + + # Allow prereleases always in order to match the previous behavior of + # this method. In the future this should be smarter and follow PEP 440 + # more accurately. + return self.specifier.contains( + version, + prereleases=True, + ) + + def __hash__(self) -> int: + return self.__hash + + def __repr__(self) -> str: + return f"Requirement.parse({str(self)!r})" + + @staticmethod + def parse(s: str | Iterable[str]) -> Requirement: + (req,) = parse_requirements(s) + return req + + +def _always_object(classes): + """ + Ensure object appears in the mro even + for old-style classes. + """ + if object not in classes: + return classes + (object,) + return classes + + +def _find_adapter(registry: Mapping[type, _AdapterT], ob: object) -> _AdapterT: + """Return an adapter factory for `ob` from `registry`""" + types = _always_object(inspect.getmro(getattr(ob, '__class__', type(ob)))) + for t in types: + if t in registry: + return registry[t] + # _find_adapter would previously return None, and immediately be called. + # So we're raising a TypeError to keep backward compatibility if anyone depended on that behaviour. + raise TypeError(f"Could not find adapter for {registry} and {ob}") + + +def ensure_directory(path: StrOrBytesPath) -> None: + """Ensure that the parent directory of `path` exists""" + dirname = os.path.dirname(path) + os.makedirs(dirname, exist_ok=True) + + +def _bypass_ensure_directory(path) -> None: + """Sandbox-bypassing version of ensure_directory()""" + if not WRITE_SUPPORT: + raise OSError('"os.mkdir" not supported on this platform.') + dirname, filename = split(path) + if dirname and filename and not isdir(dirname): + _bypass_ensure_directory(dirname) + try: + mkdir(dirname, 0o755) + except FileExistsError: + pass + + +def split_sections(s: _NestedStr) -> Iterator[tuple[str | None, list[str]]]: + """Split a string or iterable thereof into (section, content) pairs + + Each ``section`` is a stripped version of the section header ("[section]") + and each ``content`` is a list of stripped lines excluding blank lines and + comment-only lines. If there are any such lines before the first section + header, they're returned in a first ``section`` of ``None``. + """ + section = None + content: list[str] = [] + for line in yield_lines(s): + if line.startswith("["): + if line.endswith("]"): + if section or content: + yield section, content + section = line[1:-1].strip() + content = [] + else: + raise ValueError("Invalid section heading", line) + else: + content.append(line) + + # wrap up last segment + yield section, content + + +def _mkstemp(*args, **kw): + old_open = os.open + try: + # temporarily bypass sandboxing + os.open = os_open + return tempfile.mkstemp(*args, **kw) + finally: + # and then put it back + os.open = old_open + + +# Silence the PEP440Warning by default, so that end users don't get hit by it +# randomly just because they use pkg_resources. We want to append the rule +# because we want earlier uses of filterwarnings to take precedence over this +# one. +warnings.filterwarnings("ignore", category=PEP440Warning, append=True) + + +class PkgResourcesDeprecationWarning(Warning): + """ + Base class for warning about deprecations in ``pkg_resources`` + + This class is not derived from ``DeprecationWarning``, and as such is + visible by default. + """ + + +# Ported from ``setuptools`` to avoid introducing an import inter-dependency: +_LOCALE_ENCODING = "locale" if sys.version_info >= (3, 10) else None + + +# This must go before calls to `_call_aside`. See https://github.com/pypa/setuptools/pull/4422 +def _read_utf8_with_fallback(file: str, fallback_encoding=_LOCALE_ENCODING) -> str: + """See setuptools.unicode_utils._read_utf8_with_fallback""" + try: + with open(file, "r", encoding="utf-8") as f: + return f.read() + except UnicodeDecodeError: # pragma: no cover + msg = f"""\ + ******************************************************************************** + `encoding="utf-8"` fails with {file!r}, trying `encoding={fallback_encoding!r}`. + + This fallback behaviour is considered **deprecated** and future versions of + `setuptools/pkg_resources` may not implement it. + + Please encode {file!r} with "utf-8" to ensure future builds will succeed. + + If this file was produced by `setuptools` itself, cleaning up the cached files + and re-building/re-installing the package with a newer version of `setuptools` + (e.g. by updating `build-system.requires` in its `pyproject.toml`) + might solve the problem. + ******************************************************************************** + """ + # TODO: Add a deadline? + # See comment in setuptools.unicode_utils._Utf8EncodingNeeded + warnings.warn(msg, PkgResourcesDeprecationWarning, stacklevel=2) + with open(file, "r", encoding=fallback_encoding) as f: + return f.read() + + +# from jaraco.functools 1.3 +def _call_aside(f, *args, **kwargs): + f(*args, **kwargs) + return f + + +@_call_aside +def _initialize(g=globals()) -> None: + "Set up global resource manager (deliberately not state-saved)" + manager = ResourceManager() + g['_manager'] = manager + g.update( + (name, getattr(manager, name)) + for name in dir(manager) + if not name.startswith('_') + ) + + +@_call_aside +def _initialize_master_working_set() -> None: + """ + Prepare the master working set and make the ``require()`` + API available. + + This function has explicit effects on the global state + of pkg_resources. It is intended to be invoked once at + the initialization of this module. + + Invocation by other packages is unsupported and done + at their own risk. + """ + working_set = _declare_state('object', 'working_set', WorkingSet._build_master()) + + require = working_set.require + iter_entry_points = working_set.iter_entry_points + add_activation_listener = working_set.subscribe + run_script = working_set.run_script + # backward compatibility + run_main = run_script + # Activate all distributions already on sys.path with replace=False and + # ensure that all distributions added to the working set in the future + # (e.g. by calling ``require()``) will get activated as well, + # with higher priority (replace=True). + tuple(dist.activate(replace=False) for dist in working_set) + add_activation_listener( + lambda dist: dist.activate(replace=True), + existing=False, + ) + working_set.entries = [] + # match order + list(map(working_set.add_entry, sys.path)) + globals().update(locals()) + + +if TYPE_CHECKING: + # All of these are set by the @_call_aside methods above + __resource_manager = ResourceManager() # Won't exist at runtime + resource_exists = __resource_manager.resource_exists + resource_isdir = __resource_manager.resource_isdir + resource_filename = __resource_manager.resource_filename + resource_stream = __resource_manager.resource_stream + resource_string = __resource_manager.resource_string + resource_listdir = __resource_manager.resource_listdir + set_extraction_path = __resource_manager.set_extraction_path + cleanup_resources = __resource_manager.cleanup_resources + + working_set = WorkingSet() + require = working_set.require + iter_entry_points = working_set.iter_entry_points + add_activation_listener = working_set.subscribe + run_script = working_set.run_script + run_main = run_script diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/api_tests.txt b/.venv/lib/python3.12/site-packages/pkg_resources/api_tests.txt new file mode 100644 index 00000000..d72b85aa --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/api_tests.txt @@ -0,0 +1,424 @@ +Pluggable Distributions of Python Software +========================================== + +Distributions +------------- + +A "Distribution" is a collection of files that represent a "Release" of a +"Project" as of a particular point in time, denoted by a +"Version":: + + >>> import sys, pkg_resources + >>> from pkg_resources import Distribution + >>> Distribution(project_name="Foo", version="1.2") + Foo 1.2 + +Distributions have a location, which can be a filename, URL, or really anything +else you care to use:: + + >>> dist = Distribution( + ... location="http://example.com/something", + ... project_name="Bar", version="0.9" + ... ) + + >>> dist + Bar 0.9 (http://example.com/something) + + +Distributions have various introspectable attributes:: + + >>> dist.location + 'http://example.com/something' + + >>> dist.project_name + 'Bar' + + >>> dist.version + '0.9' + + >>> dist.py_version == '{}.{}'.format(*sys.version_info) + True + + >>> print(dist.platform) + None + +Including various computed attributes:: + + >>> from pkg_resources import parse_version + >>> dist.parsed_version == parse_version(dist.version) + True + + >>> dist.key # case-insensitive form of the project name + 'bar' + +Distributions are compared (and hashed) by version first:: + + >>> Distribution(version='1.0') == Distribution(version='1.0') + True + >>> Distribution(version='1.0') == Distribution(version='1.1') + False + >>> Distribution(version='1.0') < Distribution(version='1.1') + True + +but also by project name (case-insensitive), platform, Python version, +location, etc.:: + + >>> Distribution(project_name="Foo",version="1.0") == \ + ... Distribution(project_name="Foo",version="1.0") + True + + >>> Distribution(project_name="Foo",version="1.0") == \ + ... Distribution(project_name="foo",version="1.0") + True + + >>> Distribution(project_name="Foo",version="1.0") == \ + ... Distribution(project_name="Foo",version="1.1") + False + + >>> Distribution(project_name="Foo",py_version="2.3",version="1.0") == \ + ... Distribution(project_name="Foo",py_version="2.4",version="1.0") + False + + >>> Distribution(location="spam",version="1.0") == \ + ... Distribution(location="spam",version="1.0") + True + + >>> Distribution(location="spam",version="1.0") == \ + ... Distribution(location="baz",version="1.0") + False + + + +Hash and compare distribution by prio/plat + +Get version from metadata +provider capabilities +egg_name() +as_requirement() +from_location, from_filename (w/path normalization) + +Releases may have zero or more "Requirements", which indicate +what releases of another project the release requires in order to +function. A Requirement names the other project, expresses some criteria +as to what releases of that project are acceptable, and lists any "Extras" +that the requiring release may need from that project. (An Extra is an +optional feature of a Release, that can only be used if its additional +Requirements are satisfied.) + + + +The Working Set +--------------- + +A collection of active distributions is called a Working Set. Note that a +Working Set can contain any importable distribution, not just pluggable ones. +For example, the Python standard library is an importable distribution that +will usually be part of the Working Set, even though it is not pluggable. +Similarly, when you are doing development work on a project, the files you are +editing are also a Distribution. (And, with a little attention to the +directory names used, and including some additional metadata, such a +"development distribution" can be made pluggable as well.) + + >>> from pkg_resources import WorkingSet + +A working set's entries are the sys.path entries that correspond to the active +distributions. By default, the working set's entries are the items on +``sys.path``:: + + >>> ws = WorkingSet() + >>> ws.entries == sys.path + True + +But you can also create an empty working set explicitly, and add distributions +to it:: + + >>> ws = WorkingSet([]) + >>> ws.add(dist) + >>> ws.entries + ['http://example.com/something'] + >>> dist in ws + True + >>> Distribution('foo',version="") in ws + False + +And you can iterate over its distributions:: + + >>> list(ws) + [Bar 0.9 (http://example.com/something)] + +Adding the same distribution more than once is a no-op:: + + >>> ws.add(dist) + >>> list(ws) + [Bar 0.9 (http://example.com/something)] + +For that matter, adding multiple distributions for the same project also does +nothing, because a working set can only hold one active distribution per +project -- the first one added to it:: + + >>> ws.add( + ... Distribution( + ... 'http://example.com/something', project_name="Bar", + ... version="7.2" + ... ) + ... ) + >>> list(ws) + [Bar 0.9 (http://example.com/something)] + +You can append a path entry to a working set using ``add_entry()``:: + + >>> ws.entries + ['http://example.com/something'] + >>> ws.add_entry(pkg_resources.__file__) + >>> ws.entries + ['http://example.com/something', '...pkg_resources...'] + +Multiple additions result in multiple entries, even if the entry is already in +the working set (because ``sys.path`` can contain the same entry more than +once):: + + >>> ws.add_entry(pkg_resources.__file__) + >>> ws.entries + ['...example.com...', '...pkg_resources...', '...pkg_resources...'] + +And you can specify the path entry a distribution was found under, using the +optional second parameter to ``add()``:: + + >>> ws = WorkingSet([]) + >>> ws.add(dist,"foo") + >>> ws.entries + ['foo'] + +But even if a distribution is found under multiple path entries, it still only +shows up once when iterating the working set: + + >>> ws.add_entry(ws.entries[0]) + >>> list(ws) + [Bar 0.9 (http://example.com/something)] + +You can ask a WorkingSet to ``find()`` a distribution matching a requirement:: + + >>> from pkg_resources import Requirement + >>> print(ws.find(Requirement.parse("Foo==1.0"))) # no match, return None + None + + >>> ws.find(Requirement.parse("Bar==0.9")) # match, return distribution + Bar 0.9 (http://example.com/something) + +Note that asking for a conflicting version of a distribution already in a +working set triggers a ``pkg_resources.VersionConflict`` error: + + >>> try: + ... ws.find(Requirement.parse("Bar==1.0")) + ... except pkg_resources.VersionConflict as exc: + ... print(str(exc)) + ... else: + ... raise AssertionError("VersionConflict was not raised") + (Bar 0.9 (http://example.com/something), Requirement.parse('Bar==1.0')) + +You can subscribe a callback function to receive notifications whenever a new +distribution is added to a working set. The callback is immediately invoked +once for each existing distribution in the working set, and then is called +again for new distributions added thereafter:: + + >>> def added(dist): print("Added %s" % dist) + >>> ws.subscribe(added) + Added Bar 0.9 + >>> foo12 = Distribution(project_name="Foo", version="1.2", location="f12") + >>> ws.add(foo12) + Added Foo 1.2 + +Note, however, that only the first distribution added for a given project name +will trigger a callback, even during the initial ``subscribe()`` callback:: + + >>> foo14 = Distribution(project_name="Foo", version="1.4", location="f14") + >>> ws.add(foo14) # no callback, because Foo 1.2 is already active + + >>> ws = WorkingSet([]) + >>> ws.add(foo12) + >>> ws.add(foo14) + >>> ws.subscribe(added) + Added Foo 1.2 + +And adding a callback more than once has no effect, either:: + + >>> ws.subscribe(added) # no callbacks + + # and no double-callbacks on subsequent additions, either + >>> just_a_test = Distribution(project_name="JustATest", version="0.99") + >>> ws.add(just_a_test) + Added JustATest 0.99 + + +Finding Plugins +--------------- + +``WorkingSet`` objects can be used to figure out what plugins in an +``Environment`` can be loaded without any resolution errors:: + + >>> from pkg_resources import Environment + + >>> plugins = Environment([]) # normally, a list of plugin directories + >>> plugins.add(foo12) + >>> plugins.add(foo14) + >>> plugins.add(just_a_test) + +In the simplest case, we just get the newest version of each distribution in +the plugin environment:: + + >>> ws = WorkingSet([]) + >>> ws.find_plugins(plugins) + ([JustATest 0.99, Foo 1.4 (f14)], {}) + +But if there's a problem with a version conflict or missing requirements, the +method falls back to older versions, and the error info dict will contain an +exception instance for each unloadable plugin:: + + >>> ws.add(foo12) # this will conflict with Foo 1.4 + >>> ws.find_plugins(plugins) + ([JustATest 0.99, Foo 1.2 (f12)], {Foo 1.4 (f14): VersionConflict(...)}) + +But if you disallow fallbacks, the failed plugin will be skipped instead of +trying older versions:: + + >>> ws.find_plugins(plugins, fallback=False) + ([JustATest 0.99], {Foo 1.4 (f14): VersionConflict(...)}) + + + +Platform Compatibility Rules +---------------------------- + +On the Mac, there are potential compatibility issues for modules compiled +on newer versions of macOS than what the user is running. Additionally, +macOS will soon have two platforms to contend with: Intel and PowerPC. + +Basic equality works as on other platforms:: + + >>> from pkg_resources import compatible_platforms as cp + >>> reqd = 'macosx-10.4-ppc' + >>> cp(reqd, reqd) + True + >>> cp("win32", reqd) + False + +Distributions made on other machine types are not compatible:: + + >>> cp("macosx-10.4-i386", reqd) + False + +Distributions made on earlier versions of the OS are compatible, as +long as they are from the same top-level version. The patchlevel version +number does not matter:: + + >>> cp("macosx-10.4-ppc", reqd) + True + >>> cp("macosx-10.3-ppc", reqd) + True + >>> cp("macosx-10.5-ppc", reqd) + False + >>> cp("macosx-9.5-ppc", reqd) + False + +Backwards compatibility for packages made via earlier versions of +setuptools is provided as well:: + + >>> cp("darwin-8.2.0-Power_Macintosh", reqd) + True + >>> cp("darwin-7.2.0-Power_Macintosh", reqd) + True + >>> cp("darwin-8.2.0-Power_Macintosh", "macosx-10.3-ppc") + False + + +Environment Markers +------------------- + + >>> from pkg_resources import invalid_marker as im, evaluate_marker as em + >>> import os + + >>> print(im("sys_platform")) + Expected marker operator, one of <=, <, !=, ==, >=, >, ~=, ===, in, not in + sys_platform + ^ + + >>> print(im("sys_platform==")) + Expected a marker variable or quoted string + sys_platform== + ^ + + >>> print(im("sys_platform=='win32'")) + False + + >>> print(im("sys=='x'")) + Expected a marker variable or quoted string + sys=='x' + ^ + + >>> print(im("(extra)")) + Expected marker operator, one of <=, <, !=, ==, >=, >, ~=, ===, in, not in + (extra) + ^ + + >>> print(im("(extra")) + Expected marker operator, one of <=, <, !=, ==, >=, >, ~=, ===, in, not in + (extra + ^ + + >>> print(im("os.open('foo')=='y'")) + Expected a marker variable or quoted string + os.open('foo')=='y' + ^ + + >>> print(im("'x'=='y' and os.open('foo')=='y'")) # no short-circuit! + Expected a marker variable or quoted string + 'x'=='y' and os.open('foo')=='y' + ^ + + >>> print(im("'x'=='x' or os.open('foo')=='y'")) # no short-circuit! + Expected a marker variable or quoted string + 'x'=='x' or os.open('foo')=='y' + ^ + + >>> print(im("r'x'=='x'")) + Expected a marker variable or quoted string + r'x'=='x' + ^ + + >>> print(im("'''x'''=='x'")) + Expected marker operator, one of <=, <, !=, ==, >=, >, ~=, ===, in, not in + '''x'''=='x' + ^ + + >>> print(im('"""x"""=="x"')) + Expected marker operator, one of <=, <, !=, ==, >=, >, ~=, ===, in, not in + """x"""=="x" + ^ + + >>> print(im(r"x\n=='x'")) + Expected a marker variable or quoted string + x\n=='x' + ^ + + >>> print(im("os.open=='y'")) + Expected a marker variable or quoted string + os.open=='y' + ^ + + >>> em("sys_platform=='win32'") == (sys.platform=='win32') + True + + >>> em("python_version >= '2.7'") + True + + >>> em("python_version > '2.6'") + True + + >>> im("implementation_name=='cpython'") + False + + >>> im("platform_python_implementation=='CPython'") + False + + >>> im("implementation_version=='3.5.1'") + False diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/py.typed b/.venv/lib/python3.12/site-packages/pkg_resources/py.typed new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/py.typed diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/__init__.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/__init__.py diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-source/setup.cfg b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-source/setup.cfg new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-source/setup.cfg diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-source/setup.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-source/setup.py new file mode 100644 index 00000000..ce908064 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-source/setup.py @@ -0,0 +1,7 @@ +import setuptools + +setuptools.setup( + name="my-test-package", + version="1.0", + zip_safe=True, +) diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-zip/my-test-package.zip b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-zip/my-test-package.zip Binary files differnew file mode 100644 index 00000000..81f9a017 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package-zip/my-test-package.zip diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/PKG-INFO b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/PKG-INFO new file mode 100644 index 00000000..7328e3f7 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/PKG-INFO @@ -0,0 +1,10 @@ +Metadata-Version: 1.0 +Name: my-test-package +Version: 1.0 +Summary: UNKNOWN +Home-page: UNKNOWN +Author: UNKNOWN +Author-email: UNKNOWN +License: UNKNOWN +Description: UNKNOWN +Platform: UNKNOWN diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/SOURCES.txt b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/SOURCES.txt new file mode 100644 index 00000000..3c4ee167 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/SOURCES.txt @@ -0,0 +1,7 @@ +setup.cfg +setup.py +my_test_package.egg-info/PKG-INFO +my_test_package.egg-info/SOURCES.txt +my_test_package.egg-info/dependency_links.txt +my_test_package.egg-info/top_level.txt +my_test_package.egg-info/zip-safe
\ No newline at end of file diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/dependency_links.txt b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/dependency_links.txt new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/top_level.txt b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/top_level.txt new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/top_level.txt @@ -0,0 +1 @@ + diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/zip-safe b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/zip-safe new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_unpacked-egg/my_test_package-1.0-py3.7.egg/EGG-INFO/zip-safe @@ -0,0 +1 @@ + diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_zipped-egg/my_test_package-1.0-py3.7.egg b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_zipped-egg/my_test_package-1.0-py3.7.egg Binary files differnew file mode 100644 index 00000000..5115b895 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/data/my-test-package_zipped-egg/my_test_package-1.0-py3.7.egg diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_find_distributions.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_find_distributions.py new file mode 100644 index 00000000..301b36d6 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_find_distributions.py @@ -0,0 +1,56 @@ +import shutil +from pathlib import Path + +import pytest + +import pkg_resources + +TESTS_DATA_DIR = Path(__file__).parent / 'data' + + +class TestFindDistributions: + @pytest.fixture + def target_dir(self, tmpdir): + target_dir = tmpdir.mkdir('target') + # place a .egg named directory in the target that is not an egg: + target_dir.mkdir('not.an.egg') + return target_dir + + def test_non_egg_dir_named_egg(self, target_dir): + dists = pkg_resources.find_distributions(str(target_dir)) + assert not list(dists) + + def test_standalone_egg_directory(self, target_dir): + shutil.copytree( + TESTS_DATA_DIR / 'my-test-package_unpacked-egg', + target_dir, + dirs_exist_ok=True, + ) + dists = pkg_resources.find_distributions(str(target_dir)) + assert [dist.project_name for dist in dists] == ['my-test-package'] + dists = pkg_resources.find_distributions(str(target_dir), only=True) + assert not list(dists) + + def test_zipped_egg(self, target_dir): + shutil.copytree( + TESTS_DATA_DIR / 'my-test-package_zipped-egg', + target_dir, + dirs_exist_ok=True, + ) + dists = pkg_resources.find_distributions(str(target_dir)) + assert [dist.project_name for dist in dists] == ['my-test-package'] + dists = pkg_resources.find_distributions(str(target_dir), only=True) + assert not list(dists) + + def test_zipped_sdist_one_level_removed(self, target_dir): + shutil.copytree( + TESTS_DATA_DIR / 'my-test-package-zip', target_dir, dirs_exist_ok=True + ) + dists = pkg_resources.find_distributions( + str(target_dir / "my-test-package.zip") + ) + assert [dist.project_name for dist in dists] == ['my-test-package'] + dists = pkg_resources.find_distributions( + str(target_dir / "my-test-package.zip"), only=True + ) + assert not list(dists) diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_integration_zope_interface.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_integration_zope_interface.py new file mode 100644 index 00000000..4e37c340 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_integration_zope_interface.py @@ -0,0 +1,54 @@ +import platform +from inspect import cleandoc + +import jaraco.path +import pytest + +pytestmark = pytest.mark.integration + + +# For the sake of simplicity this test uses fixtures defined in +# `setuptools.test.fixtures`, +# and it also exercise conditions considered deprecated... +# So if needed this test can be deleted. +@pytest.mark.skipif( + platform.system() != "Linux", + reason="only demonstrated to fail on Linux in #4399", +) +def test_interop_pkg_resources_iter_entry_points(tmp_path, venv): + """ + Importing pkg_resources.iter_entry_points on console_scripts + seems to cause trouble with zope-interface, when deprecates installation method + is used. See #4399. + """ + project = { + "pkg": { + "foo.py": cleandoc( + """ + from pkg_resources import iter_entry_points + + def bar(): + print("Print me if you can") + """ + ), + "setup.py": cleandoc( + """ + from setuptools import setup, find_packages + + setup( + install_requires=["zope-interface==6.4.post2"], + entry_points={ + "console_scripts": [ + "foo=foo:bar", + ], + }, + ) + """ + ), + } + } + jaraco.path.build(project, prefix=tmp_path) + cmd = ["pip", "install", "-e", ".", "--no-use-pep517"] + venv.run(cmd, cwd=tmp_path / "pkg") # Needs this version of pkg_resources installed + out = venv.run(["foo"]) + assert "Print me if you can" in out diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_markers.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_markers.py new file mode 100644 index 00000000..9306d5b3 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_markers.py @@ -0,0 +1,8 @@ +from unittest import mock + +from pkg_resources import evaluate_marker + + +@mock.patch('platform.python_version', return_value='2.7.10') +def test_ordering(python_version_mock): + assert evaluate_marker("python_full_version > '2.7.3'") is True diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_pkg_resources.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_pkg_resources.py new file mode 100644 index 00000000..cfc9b16c --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_pkg_resources.py @@ -0,0 +1,485 @@ +from __future__ import annotations + +import builtins +import datetime +import inspect +import os +import plistlib +import stat +import subprocess +import sys +import tempfile +import zipfile +from unittest import mock + +import pytest + +import pkg_resources +from pkg_resources import DistInfoDistribution, Distribution, EggInfoDistribution + +import distutils.command.install_egg_info +import distutils.dist + + +class EggRemover(str): + def __call__(self): + if self in sys.path: + sys.path.remove(self) + if os.path.exists(self): + os.remove(self) + + +class TestZipProvider: + finalizers: list[EggRemover] = [] + + ref_time = datetime.datetime(2013, 5, 12, 13, 25, 0) + "A reference time for a file modification" + + @classmethod + def setup_class(cls): + "create a zip egg and add it to sys.path" + egg = tempfile.NamedTemporaryFile(suffix='.egg', delete=False) + zip_egg = zipfile.ZipFile(egg, 'w') + zip_info = zipfile.ZipInfo() + zip_info.filename = 'mod.py' + zip_info.date_time = cls.ref_time.timetuple() + zip_egg.writestr(zip_info, 'x = 3\n') + zip_info = zipfile.ZipInfo() + zip_info.filename = 'data.dat' + zip_info.date_time = cls.ref_time.timetuple() + zip_egg.writestr(zip_info, 'hello, world!') + zip_info = zipfile.ZipInfo() + zip_info.filename = 'subdir/mod2.py' + zip_info.date_time = cls.ref_time.timetuple() + zip_egg.writestr(zip_info, 'x = 6\n') + zip_info = zipfile.ZipInfo() + zip_info.filename = 'subdir/data2.dat' + zip_info.date_time = cls.ref_time.timetuple() + zip_egg.writestr(zip_info, 'goodbye, world!') + zip_egg.close() + egg.close() + + sys.path.append(egg.name) + subdir = os.path.join(egg.name, 'subdir') + sys.path.append(subdir) + cls.finalizers.append(EggRemover(subdir)) + cls.finalizers.append(EggRemover(egg.name)) + + @classmethod + def teardown_class(cls): + for finalizer in cls.finalizers: + finalizer() + + def test_resource_listdir(self): + import mod # pyright: ignore[reportMissingImports] # Temporary package for test + + zp = pkg_resources.ZipProvider(mod) + + expected_root = ['data.dat', 'mod.py', 'subdir'] + assert sorted(zp.resource_listdir('')) == expected_root + + expected_subdir = ['data2.dat', 'mod2.py'] + assert sorted(zp.resource_listdir('subdir')) == expected_subdir + assert sorted(zp.resource_listdir('subdir/')) == expected_subdir + + assert zp.resource_listdir('nonexistent') == [] + assert zp.resource_listdir('nonexistent/') == [] + + import mod2 # pyright: ignore[reportMissingImports] # Temporary package for test + + zp2 = pkg_resources.ZipProvider(mod2) + + assert sorted(zp2.resource_listdir('')) == expected_subdir + + assert zp2.resource_listdir('subdir') == [] + assert zp2.resource_listdir('subdir/') == [] + + def test_resource_filename_rewrites_on_change(self): + """ + If a previous call to get_resource_filename has saved the file, but + the file has been subsequently mutated with different file of the + same size and modification time, it should not be overwritten on a + subsequent call to get_resource_filename. + """ + import mod # pyright: ignore[reportMissingImports] # Temporary package for test + + manager = pkg_resources.ResourceManager() + zp = pkg_resources.ZipProvider(mod) + filename = zp.get_resource_filename(manager, 'data.dat') + actual = datetime.datetime.fromtimestamp(os.stat(filename).st_mtime) + assert actual == self.ref_time + f = open(filename, 'w', encoding="utf-8") + f.write('hello, world?') + f.close() + ts = self.ref_time.timestamp() + os.utime(filename, (ts, ts)) + filename = zp.get_resource_filename(manager, 'data.dat') + with open(filename, encoding="utf-8") as f: + assert f.read() == 'hello, world!' + manager.cleanup_resources() + + +class TestResourceManager: + def test_get_cache_path(self): + mgr = pkg_resources.ResourceManager() + path = mgr.get_cache_path('foo') + type_ = str(type(path)) + message = "Unexpected type from get_cache_path: " + type_ + assert isinstance(path, str), message + + def test_get_cache_path_race(self, tmpdir): + # Patch to os.path.isdir to create a race condition + def patched_isdir(dirname, unpatched_isdir=pkg_resources.isdir): + patched_isdir.dirnames.append(dirname) + + was_dir = unpatched_isdir(dirname) + if not was_dir: + os.makedirs(dirname) + return was_dir + + patched_isdir.dirnames = [] + + # Get a cache path with a "race condition" + mgr = pkg_resources.ResourceManager() + mgr.set_extraction_path(str(tmpdir)) + + archive_name = os.sep.join(('foo', 'bar', 'baz')) + with mock.patch.object(pkg_resources, 'isdir', new=patched_isdir): + mgr.get_cache_path(archive_name) + + # Because this test relies on the implementation details of this + # function, these assertions are a sentinel to ensure that the + # test suite will not fail silently if the implementation changes. + called_dirnames = patched_isdir.dirnames + assert len(called_dirnames) == 2 + assert called_dirnames[0].split(os.sep)[-2:] == ['foo', 'bar'] + assert called_dirnames[1].split(os.sep)[-1:] == ['foo'] + + """ + Tests to ensure that pkg_resources runs independently from setuptools. + """ + + def test_setuptools_not_imported(self): + """ + In a separate Python environment, import pkg_resources and assert + that action doesn't cause setuptools to be imported. + """ + lines = ( + 'import pkg_resources', + 'import sys', + ('assert "setuptools" not in sys.modules, "setuptools was imported"'), + ) + cmd = [sys.executable, '-c', '; '.join(lines)] + subprocess.check_call(cmd) + + +def make_test_distribution(metadata_path, metadata): + """ + Make a test Distribution object, and return it. + + :param metadata_path: the path to the metadata file that should be + created. This should be inside a distribution directory that should + also be created. For example, an argument value might end with + "<project>.dist-info/METADATA". + :param metadata: the desired contents of the metadata file, as bytes. + """ + dist_dir = os.path.dirname(metadata_path) + os.mkdir(dist_dir) + with open(metadata_path, 'wb') as f: + f.write(metadata) + dists = list(pkg_resources.distributions_from_metadata(dist_dir)) + (dist,) = dists + + return dist + + +def test_get_metadata__bad_utf8(tmpdir): + """ + Test a metadata file with bytes that can't be decoded as utf-8. + """ + filename = 'METADATA' + # Convert the tmpdir LocalPath object to a string before joining. + metadata_path = os.path.join(str(tmpdir), 'foo.dist-info', filename) + # Encode a non-ascii string with the wrong encoding (not utf-8). + metadata = 'née'.encode('iso-8859-1') + dist = make_test_distribution(metadata_path, metadata=metadata) + + with pytest.raises(UnicodeDecodeError) as excinfo: + dist.get_metadata(filename) + + exc = excinfo.value + actual = str(exc) + expected = ( + # The error message starts with "'utf-8' codec ..." However, the + # spelling of "utf-8" can vary (e.g. "utf8") so we don't include it + "codec can't decode byte 0xe9 in position 1: " + 'invalid continuation byte in METADATA file at path: ' + ) + assert expected in actual, f'actual: {actual}' + assert actual.endswith(metadata_path), f'actual: {actual}' + + +def make_distribution_no_version(tmpdir, basename): + """ + Create a distribution directory with no file containing the version. + """ + dist_dir = tmpdir / basename + dist_dir.ensure_dir() + # Make the directory non-empty so distributions_from_metadata() + # will detect it and yield it. + dist_dir.join('temp.txt').ensure() + + dists = list(pkg_resources.distributions_from_metadata(dist_dir)) + assert len(dists) == 1 + (dist,) = dists + + return dist, dist_dir + + +@pytest.mark.parametrize( + ("suffix", "expected_filename", "expected_dist_type"), + [ + ('egg-info', 'PKG-INFO', EggInfoDistribution), + ('dist-info', 'METADATA', DistInfoDistribution), + ], +) +@pytest.mark.xfail( + sys.version_info[:2] == (3, 12) and sys.version_info.releaselevel != 'final', + reason="https://github.com/python/cpython/issues/103632", +) +def test_distribution_version_missing( + tmpdir, suffix, expected_filename, expected_dist_type +): + """ + Test Distribution.version when the "Version" header is missing. + """ + basename = f'foo.{suffix}' + dist, dist_dir = make_distribution_no_version(tmpdir, basename) + + expected_text = ( + f"Missing 'Version:' header and/or {expected_filename} file at path: " + ) + metadata_path = os.path.join(dist_dir, expected_filename) + + # Now check the exception raised when the "version" attribute is accessed. + with pytest.raises(ValueError) as excinfo: + dist.version + + err = str(excinfo.value) + # Include a string expression after the assert so the full strings + # will be visible for inspection on failure. + assert expected_text in err, str((expected_text, err)) + + # Also check the args passed to the ValueError. + msg, dist = excinfo.value.args + assert expected_text in msg + # Check that the message portion contains the path. + assert metadata_path in msg, str((metadata_path, msg)) + assert type(dist) is expected_dist_type + + +@pytest.mark.xfail( + sys.version_info[:2] == (3, 12) and sys.version_info.releaselevel != 'final', + reason="https://github.com/python/cpython/issues/103632", +) +def test_distribution_version_missing_undetected_path(): + """ + Test Distribution.version when the "Version" header is missing and + the path can't be detected. + """ + # Create a Distribution object with no metadata argument, which results + # in an empty metadata provider. + dist = Distribution('/foo') + with pytest.raises(ValueError) as excinfo: + dist.version + + msg, dist = excinfo.value.args + expected = ( + "Missing 'Version:' header and/or PKG-INFO file at path: [could not detect]" + ) + assert msg == expected + + +@pytest.mark.parametrize('only', [False, True]) +def test_dist_info_is_not_dir(tmp_path, only): + """Test path containing a file with dist-info extension.""" + dist_info = tmp_path / 'foobar.dist-info' + dist_info.touch() + assert not pkg_resources.dist_factory(str(tmp_path), str(dist_info), only) + + +def test_macos_vers_fallback(monkeypatch, tmp_path): + """Regression test for pkg_resources._macos_vers""" + orig_open = builtins.open + + # Pretend we need to use the plist file + monkeypatch.setattr('platform.mac_ver', mock.Mock(return_value=('', (), ''))) + + # Create fake content for the fake plist file + with open(tmp_path / 'fake.plist', 'wb') as fake_file: + plistlib.dump({"ProductVersion": "11.4"}, fake_file) + + # Pretend the fake file exists + monkeypatch.setattr('os.path.exists', mock.Mock(return_value=True)) + + def fake_open(file, *args, **kwargs): + return orig_open(tmp_path / 'fake.plist', *args, **kwargs) + + # Ensure that the _macos_vers works correctly + with mock.patch('builtins.open', mock.Mock(side_effect=fake_open)) as m: + pkg_resources._macos_vers.cache_clear() + assert pkg_resources._macos_vers() == ["11", "4"] + pkg_resources._macos_vers.cache_clear() + + m.assert_called() + + +class TestDeepVersionLookupDistutils: + @pytest.fixture + def env(self, tmpdir): + """ + Create a package environment, similar to a virtualenv, + in which packages are installed. + """ + + class Environment(str): + pass + + env = Environment(tmpdir) + tmpdir.chmod(stat.S_IRWXU) + subs = 'home', 'lib', 'scripts', 'data', 'egg-base' + env.paths = dict((dirname, str(tmpdir / dirname)) for dirname in subs) + list(map(os.mkdir, env.paths.values())) + return env + + def create_foo_pkg(self, env, version): + """ + Create a foo package installed (distutils-style) to env.paths['lib'] + as version. + """ + ld = "This package has unicode metadata! ❄" + attrs = dict(name='foo', version=version, long_description=ld) + dist = distutils.dist.Distribution(attrs) + iei_cmd = distutils.command.install_egg_info.install_egg_info(dist) + iei_cmd.initialize_options() + iei_cmd.install_dir = env.paths['lib'] + iei_cmd.finalize_options() + iei_cmd.run() + + def test_version_resolved_from_egg_info(self, env): + version = '1.11.0.dev0+2329eae' + self.create_foo_pkg(env, version) + + # this requirement parsing will raise a VersionConflict unless the + # .egg-info file is parsed (see #419 on BitBucket) + req = pkg_resources.Requirement.parse('foo>=1.9') + dist = pkg_resources.WorkingSet([env.paths['lib']]).find(req) + assert dist.version == version + + @pytest.mark.parametrize( + ("unnormalized", "normalized"), + [ + ('foo', 'foo'), + ('foo/', 'foo'), + ('foo/bar', 'foo/bar'), + ('foo/bar/', 'foo/bar'), + ], + ) + def test_normalize_path_trailing_sep(self, unnormalized, normalized): + """Ensure the trailing slash is cleaned for path comparison. + + See pypa/setuptools#1519. + """ + result_from_unnormalized = pkg_resources.normalize_path(unnormalized) + result_from_normalized = pkg_resources.normalize_path(normalized) + assert result_from_unnormalized == result_from_normalized + + @pytest.mark.skipif( + os.path.normcase('A') != os.path.normcase('a'), + reason='Testing case-insensitive filesystems.', + ) + @pytest.mark.parametrize( + ("unnormalized", "normalized"), + [ + ('MiXeD/CasE', 'mixed/case'), + ], + ) + def test_normalize_path_normcase(self, unnormalized, normalized): + """Ensure mixed case is normalized on case-insensitive filesystems.""" + result_from_unnormalized = pkg_resources.normalize_path(unnormalized) + result_from_normalized = pkg_resources.normalize_path(normalized) + assert result_from_unnormalized == result_from_normalized + + @pytest.mark.skipif( + os.path.sep != '\\', + reason='Testing systems using backslashes as path separators.', + ) + @pytest.mark.parametrize( + ("unnormalized", "expected"), + [ + ('forward/slash', 'forward\\slash'), + ('forward/slash/', 'forward\\slash'), + ('backward\\slash\\', 'backward\\slash'), + ], + ) + def test_normalize_path_backslash_sep(self, unnormalized, expected): + """Ensure path seps are cleaned on backslash path sep systems.""" + result = pkg_resources.normalize_path(unnormalized) + assert result.endswith(expected) + + +class TestWorkdirRequire: + def fake_site_packages(self, tmp_path, monkeypatch, dist_files): + site_packages = tmp_path / "site-packages" + site_packages.mkdir() + for file, content in self.FILES.items(): + path = site_packages / file + path.parent.mkdir(exist_ok=True, parents=True) + path.write_text(inspect.cleandoc(content), encoding="utf-8") + + monkeypatch.setattr(sys, "path", [site_packages]) + return os.fspath(site_packages) + + FILES = { + "pkg1_mod-1.2.3.dist-info/METADATA": """ + Metadata-Version: 2.4 + Name: pkg1.mod + Version: 1.2.3 + """, + "pkg2.mod-0.42.dist-info/METADATA": """ + Metadata-Version: 2.1 + Name: pkg2.mod + Version: 0.42 + """, + "pkg3_mod.egg-info/PKG-INFO": """ + Name: pkg3.mod + Version: 1.2.3.4 + """, + "pkg4.mod.egg-info/PKG-INFO": """ + Name: pkg4.mod + Version: 0.42.1 + """, + } + + @pytest.mark.parametrize( + ("version", "requirement"), + [ + ("1.2.3", "pkg1.mod>=1"), + ("0.42", "pkg2.mod>=0.4"), + ("1.2.3.4", "pkg3.mod<=2"), + ("0.42.1", "pkg4.mod>0.2,<1"), + ], + ) + def test_require_non_normalised_name( + self, tmp_path, monkeypatch, version, requirement + ): + # https://github.com/pypa/setuptools/issues/4853 + site_packages = self.fake_site_packages(tmp_path, monkeypatch, self.FILES) + ws = pkg_resources.WorkingSet([site_packages]) + + for req in [requirement, requirement.replace(".", "-")]: + [dist] = ws.require(req) + assert dist.version == version + assert os.path.samefile( + os.path.commonpath([dist.location, site_packages]), site_packages + ) diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_resources.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_resources.py new file mode 100644 index 00000000..70436c08 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_resources.py @@ -0,0 +1,869 @@ +import itertools +import os +import platform +import string +import sys + +import pytest +from packaging.specifiers import SpecifierSet + +import pkg_resources +from pkg_resources import ( + Distribution, + EntryPoint, + Requirement, + VersionConflict, + WorkingSet, + parse_requirements, + parse_version, + safe_name, + safe_version, +) + + +# from Python 3.6 docs. Available from itertools on Python 3.10 +def pairwise(iterable): + "s -> (s0,s1), (s1,s2), (s2, s3), ..." + a, b = itertools.tee(iterable) + next(b, None) + return zip(a, b) + + +class Metadata(pkg_resources.EmptyProvider): + """Mock object to return metadata as if from an on-disk distribution""" + + def __init__(self, *pairs) -> None: + self.metadata = dict(pairs) + + def has_metadata(self, name) -> bool: + return name in self.metadata + + def get_metadata(self, name): + return self.metadata[name] + + def get_metadata_lines(self, name): + return pkg_resources.yield_lines(self.get_metadata(name)) + + +dist_from_fn = pkg_resources.Distribution.from_filename + + +class TestDistro: + def testCollection(self): + # empty path should produce no distributions + ad = pkg_resources.Environment([], platform=None, python=None) + assert list(ad) == [] + assert ad['FooPkg'] == [] + ad.add(dist_from_fn("FooPkg-1.3_1.egg")) + ad.add(dist_from_fn("FooPkg-1.4-py2.4-win32.egg")) + ad.add(dist_from_fn("FooPkg-1.2-py2.4.egg")) + + # Name is in there now + assert ad['FooPkg'] + # But only 1 package + assert list(ad) == ['foopkg'] + + # Distributions sort by version + expected = ['1.4', '1.3-1', '1.2'] + assert [dist.version for dist in ad['FooPkg']] == expected + + # Removing a distribution leaves sequence alone + ad.remove(ad['FooPkg'][1]) + assert [dist.version for dist in ad['FooPkg']] == ['1.4', '1.2'] + + # And inserting adds them in order + ad.add(dist_from_fn("FooPkg-1.9.egg")) + assert [dist.version for dist in ad['FooPkg']] == ['1.9', '1.4', '1.2'] + + ws = WorkingSet([]) + foo12 = dist_from_fn("FooPkg-1.2-py2.4.egg") + foo14 = dist_from_fn("FooPkg-1.4-py2.4-win32.egg") + (req,) = parse_requirements("FooPkg>=1.3") + + # Nominal case: no distros on path, should yield all applicable + assert ad.best_match(req, ws).version == '1.9' + # If a matching distro is already installed, should return only that + ws.add(foo14) + assert ad.best_match(req, ws).version == '1.4' + + # If the first matching distro is unsuitable, it's a version conflict + ws = WorkingSet([]) + ws.add(foo12) + ws.add(foo14) + with pytest.raises(VersionConflict): + ad.best_match(req, ws) + + # If more than one match on the path, the first one takes precedence + ws = WorkingSet([]) + ws.add(foo14) + ws.add(foo12) + ws.add(foo14) + assert ad.best_match(req, ws).version == '1.4' + + def checkFooPkg(self, d): + assert d.project_name == "FooPkg" + assert d.key == "foopkg" + assert d.version == "1.3.post1" + assert d.py_version == "2.4" + assert d.platform == "win32" + assert d.parsed_version == parse_version("1.3-1") + + def testDistroBasics(self): + d = Distribution( + "/some/path", + project_name="FooPkg", + version="1.3-1", + py_version="2.4", + platform="win32", + ) + self.checkFooPkg(d) + + d = Distribution("/some/path") + assert d.py_version == f'{sys.version_info.major}.{sys.version_info.minor}' + assert d.platform is None + + def testDistroParse(self): + d = dist_from_fn("FooPkg-1.3.post1-py2.4-win32.egg") + self.checkFooPkg(d) + d = dist_from_fn("FooPkg-1.3.post1-py2.4-win32.egg-info") + self.checkFooPkg(d) + + def testDistroMetadata(self): + d = Distribution( + "/some/path", + project_name="FooPkg", + py_version="2.4", + platform="win32", + metadata=Metadata(('PKG-INFO', "Metadata-Version: 1.0\nVersion: 1.3-1\n")), + ) + self.checkFooPkg(d) + + def distRequires(self, txt): + return Distribution("/foo", metadata=Metadata(('depends.txt', txt))) + + def checkRequires(self, dist, txt, extras=()): + assert list(dist.requires(extras)) == list(parse_requirements(txt)) + + def testDistroDependsSimple(self): + for v in "Twisted>=1.5", "Twisted>=1.5\nZConfig>=2.0": + self.checkRequires(self.distRequires(v), v) + + needs_object_dir = pytest.mark.skipif( + not hasattr(object, '__dir__'), + reason='object.__dir__ necessary for self.__dir__ implementation', + ) + + def test_distribution_dir(self): + d = pkg_resources.Distribution() + dir(d) + + @needs_object_dir + def test_distribution_dir_includes_provider_dir(self): + d = pkg_resources.Distribution() + before = d.__dir__() + assert 'test_attr' not in before + d._provider.test_attr = None + after = d.__dir__() + assert len(after) == len(before) + 1 + assert 'test_attr' in after + + @needs_object_dir + def test_distribution_dir_ignores_provider_dir_leading_underscore(self): + d = pkg_resources.Distribution() + before = d.__dir__() + assert '_test_attr' not in before + d._provider._test_attr = None + after = d.__dir__() + assert len(after) == len(before) + assert '_test_attr' not in after + + def testResolve(self): + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + # Resolving no requirements -> nothing to install + assert list(ws.resolve([], ad)) == [] + # Request something not in the collection -> DistributionNotFound + with pytest.raises(pkg_resources.DistributionNotFound): + ws.resolve(parse_requirements("Foo"), ad) + + Foo = Distribution.from_filename( + "/foo_dir/Foo-1.2.egg", + metadata=Metadata(('depends.txt', "[bar]\nBaz>=2.0")), + ) + ad.add(Foo) + ad.add(Distribution.from_filename("Foo-0.9.egg")) + + # Request thing(s) that are available -> list to activate + for i in range(3): + targets = list(ws.resolve(parse_requirements("Foo"), ad)) + assert targets == [Foo] + list(map(ws.add, targets)) + with pytest.raises(VersionConflict): + ws.resolve(parse_requirements("Foo==0.9"), ad) + ws = WorkingSet([]) # reset + + # Request an extra that causes an unresolved dependency for "Baz" + with pytest.raises(pkg_resources.DistributionNotFound): + ws.resolve(parse_requirements("Foo[bar]"), ad) + Baz = Distribution.from_filename( + "/foo_dir/Baz-2.1.egg", metadata=Metadata(('depends.txt', "Foo")) + ) + ad.add(Baz) + + # Activation list now includes resolved dependency + assert list(ws.resolve(parse_requirements("Foo[bar]"), ad)) == [Foo, Baz] + # Requests for conflicting versions produce VersionConflict + with pytest.raises(VersionConflict) as vc: + ws.resolve(parse_requirements("Foo==1.2\nFoo!=1.2"), ad) + + msg = 'Foo 0.9 is installed but Foo==1.2 is required' + assert vc.value.report() == msg + + def test_environment_marker_evaluation_negative(self): + """Environment markers are evaluated at resolution time.""" + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + res = ws.resolve(parse_requirements("Foo;python_version<'2'"), ad) + assert list(res) == [] + + def test_environment_marker_evaluation_positive(self): + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + Foo = Distribution.from_filename("/foo_dir/Foo-1.2.dist-info") + ad.add(Foo) + res = ws.resolve(parse_requirements("Foo;python_version>='2'"), ad) + assert list(res) == [Foo] + + def test_environment_marker_evaluation_called(self): + """ + If one package foo requires bar without any extras, + markers should pass for bar without extras. + """ + (parent_req,) = parse_requirements("foo") + (req,) = parse_requirements("bar;python_version>='2'") + req_extras = pkg_resources._ReqExtras({req: parent_req.extras}) + assert req_extras.markers_pass(req) + + (parent_req,) = parse_requirements("foo[]") + (req,) = parse_requirements("bar;python_version>='2'") + req_extras = pkg_resources._ReqExtras({req: parent_req.extras}) + assert req_extras.markers_pass(req) + + def test_marker_evaluation_with_extras(self): + """Extras are also evaluated as markers at resolution time.""" + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + Foo = Distribution.from_filename( + "/foo_dir/Foo-1.2.dist-info", + metadata=Metadata(( + "METADATA", + "Provides-Extra: baz\nRequires-Dist: quux; extra=='baz'", + )), + ) + ad.add(Foo) + assert list(ws.resolve(parse_requirements("Foo"), ad)) == [Foo] + quux = Distribution.from_filename("/foo_dir/quux-1.0.dist-info") + ad.add(quux) + res = list(ws.resolve(parse_requirements("Foo[baz]"), ad)) + assert res == [Foo, quux] + + def test_marker_evaluation_with_extras_normlized(self): + """Extras are also evaluated as markers at resolution time.""" + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + Foo = Distribution.from_filename( + "/foo_dir/Foo-1.2.dist-info", + metadata=Metadata(( + "METADATA", + "Provides-Extra: baz-lightyear\n" + "Requires-Dist: quux; extra=='baz-lightyear'", + )), + ) + ad.add(Foo) + assert list(ws.resolve(parse_requirements("Foo"), ad)) == [Foo] + quux = Distribution.from_filename("/foo_dir/quux-1.0.dist-info") + ad.add(quux) + res = list(ws.resolve(parse_requirements("Foo[baz-lightyear]"), ad)) + assert res == [Foo, quux] + + def test_marker_evaluation_with_multiple_extras(self): + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + Foo = Distribution.from_filename( + "/foo_dir/Foo-1.2.dist-info", + metadata=Metadata(( + "METADATA", + "Provides-Extra: baz\n" + "Requires-Dist: quux; extra=='baz'\n" + "Provides-Extra: bar\n" + "Requires-Dist: fred; extra=='bar'\n", + )), + ) + ad.add(Foo) + quux = Distribution.from_filename("/foo_dir/quux-1.0.dist-info") + ad.add(quux) + fred = Distribution.from_filename("/foo_dir/fred-0.1.dist-info") + ad.add(fred) + res = list(ws.resolve(parse_requirements("Foo[baz,bar]"), ad)) + assert sorted(res) == [fred, quux, Foo] + + def test_marker_evaluation_with_extras_loop(self): + ad = pkg_resources.Environment([]) + ws = WorkingSet([]) + a = Distribution.from_filename( + "/foo_dir/a-0.2.dist-info", + metadata=Metadata(("METADATA", "Requires-Dist: c[a]")), + ) + b = Distribution.from_filename( + "/foo_dir/b-0.3.dist-info", + metadata=Metadata(("METADATA", "Requires-Dist: c[b]")), + ) + c = Distribution.from_filename( + "/foo_dir/c-1.0.dist-info", + metadata=Metadata(( + "METADATA", + "Provides-Extra: a\n" + "Requires-Dist: b;extra=='a'\n" + "Provides-Extra: b\n" + "Requires-Dist: foo;extra=='b'", + )), + ) + foo = Distribution.from_filename("/foo_dir/foo-0.1.dist-info") + for dist in (a, b, c, foo): + ad.add(dist) + res = list(ws.resolve(parse_requirements("a"), ad)) + assert res == [a, c, b, foo] + + @pytest.mark.xfail( + sys.version_info[:2] == (3, 12) and sys.version_info.releaselevel != 'final', + reason="https://github.com/python/cpython/issues/103632", + ) + def testDistroDependsOptions(self): + d = self.distRequires( + """ + Twisted>=1.5 + [docgen] + ZConfig>=2.0 + docutils>=0.3 + [fastcgi] + fcgiapp>=0.1""" + ) + self.checkRequires(d, "Twisted>=1.5") + self.checkRequires( + d, "Twisted>=1.5 ZConfig>=2.0 docutils>=0.3".split(), ["docgen"] + ) + self.checkRequires(d, "Twisted>=1.5 fcgiapp>=0.1".split(), ["fastcgi"]) + self.checkRequires( + d, + "Twisted>=1.5 ZConfig>=2.0 docutils>=0.3 fcgiapp>=0.1".split(), + ["docgen", "fastcgi"], + ) + self.checkRequires( + d, + "Twisted>=1.5 fcgiapp>=0.1 ZConfig>=2.0 docutils>=0.3".split(), + ["fastcgi", "docgen"], + ) + with pytest.raises(pkg_resources.UnknownExtra): + d.requires(["foo"]) + + +class TestWorkingSet: + def test_find_conflicting(self): + ws = WorkingSet([]) + Foo = Distribution.from_filename("/foo_dir/Foo-1.2.egg") + ws.add(Foo) + + # create a requirement that conflicts with Foo 1.2 + req = next(parse_requirements("Foo<1.2")) + + with pytest.raises(VersionConflict) as vc: + ws.find(req) + + msg = 'Foo 1.2 is installed but Foo<1.2 is required' + assert vc.value.report() == msg + + def test_resolve_conflicts_with_prior(self): + """ + A ContextualVersionConflict should be raised when a requirement + conflicts with a prior requirement for a different package. + """ + # Create installation where Foo depends on Baz 1.0 and Bar depends on + # Baz 2.0. + ws = WorkingSet([]) + md = Metadata(('depends.txt', "Baz==1.0")) + Foo = Distribution.from_filename("/foo_dir/Foo-1.0.egg", metadata=md) + ws.add(Foo) + md = Metadata(('depends.txt', "Baz==2.0")) + Bar = Distribution.from_filename("/foo_dir/Bar-1.0.egg", metadata=md) + ws.add(Bar) + Baz = Distribution.from_filename("/foo_dir/Baz-1.0.egg") + ws.add(Baz) + Baz = Distribution.from_filename("/foo_dir/Baz-2.0.egg") + ws.add(Baz) + + with pytest.raises(VersionConflict) as vc: + ws.resolve(parse_requirements("Foo\nBar\n")) + + msg = "Baz 1.0 is installed but Baz==2.0 is required by " + msg += repr(set(['Bar'])) + assert vc.value.report() == msg + + +class TestEntryPoints: + def assertfields(self, ep): + assert ep.name == "foo" + assert ep.module_name == "pkg_resources.tests.test_resources" + assert ep.attrs == ("TestEntryPoints",) + assert ep.extras == ("x",) + assert ep.load() is TestEntryPoints + expect = "foo = pkg_resources.tests.test_resources:TestEntryPoints [x]" + assert str(ep) == expect + + def setup_method(self, method): + self.dist = Distribution.from_filename( + "FooPkg-1.2-py2.4.egg", metadata=Metadata(('requires.txt', '[x]')) + ) + + def testBasics(self): + ep = EntryPoint( + "foo", + "pkg_resources.tests.test_resources", + ["TestEntryPoints"], + ["x"], + self.dist, + ) + self.assertfields(ep) + + def testParse(self): + s = "foo = pkg_resources.tests.test_resources:TestEntryPoints [x]" + ep = EntryPoint.parse(s, self.dist) + self.assertfields(ep) + + ep = EntryPoint.parse("bar baz= spammity[PING]") + assert ep.name == "bar baz" + assert ep.module_name == "spammity" + assert ep.attrs == () + assert ep.extras == ("ping",) + + ep = EntryPoint.parse(" fizzly = wocka:foo") + assert ep.name == "fizzly" + assert ep.module_name == "wocka" + assert ep.attrs == ("foo",) + assert ep.extras == () + + # plus in the name + spec = "html+mako = mako.ext.pygmentplugin:MakoHtmlLexer" + ep = EntryPoint.parse(spec) + assert ep.name == 'html+mako' + + reject_specs = "foo", "x=a:b:c", "q=x/na", "fez=pish:tush-z", "x=f[a]>2" + + @pytest.mark.parametrize("reject_spec", reject_specs) + def test_reject_spec(self, reject_spec): + with pytest.raises(ValueError): + EntryPoint.parse(reject_spec) + + def test_printable_name(self): + """ + Allow any printable character in the name. + """ + # Create a name with all printable characters; strip the whitespace. + name = string.printable.strip() + spec = "{name} = module:attr".format(**locals()) + ep = EntryPoint.parse(spec) + assert ep.name == name + + def checkSubMap(self, m): + assert len(m) == len(self.submap_expect) + for key, ep in self.submap_expect.items(): + assert m.get(key).name == ep.name + assert m.get(key).module_name == ep.module_name + assert sorted(m.get(key).attrs) == sorted(ep.attrs) + assert sorted(m.get(key).extras) == sorted(ep.extras) + + submap_expect = dict( + feature1=EntryPoint('feature1', 'somemodule', ['somefunction']), + feature2=EntryPoint( + 'feature2', 'another.module', ['SomeClass'], ['extra1', 'extra2'] + ), + feature3=EntryPoint('feature3', 'this.module', extras=['something']), + ) + submap_str = """ + # define features for blah blah + feature1 = somemodule:somefunction + feature2 = another.module:SomeClass [extra1,extra2] + feature3 = this.module [something] + """ + + def testParseList(self): + self.checkSubMap(EntryPoint.parse_group("xyz", self.submap_str)) + with pytest.raises(ValueError): + EntryPoint.parse_group("x a", "foo=bar") + with pytest.raises(ValueError): + EntryPoint.parse_group("x", ["foo=baz", "foo=bar"]) + + def testParseMap(self): + m = EntryPoint.parse_map({'xyz': self.submap_str}) + self.checkSubMap(m['xyz']) + assert list(m.keys()) == ['xyz'] + m = EntryPoint.parse_map("[xyz]\n" + self.submap_str) + self.checkSubMap(m['xyz']) + assert list(m.keys()) == ['xyz'] + with pytest.raises(ValueError): + EntryPoint.parse_map(["[xyz]", "[xyz]"]) + with pytest.raises(ValueError): + EntryPoint.parse_map(self.submap_str) + + def testDeprecationWarnings(self): + ep = EntryPoint( + "foo", "pkg_resources.tests.test_resources", ["TestEntryPoints"], ["x"] + ) + with pytest.warns(pkg_resources.PkgResourcesDeprecationWarning): + ep.load(require=False) + + +class TestRequirements: + def testBasics(self): + r = Requirement.parse("Twisted>=1.2") + assert str(r) == "Twisted>=1.2" + assert repr(r) == "Requirement.parse('Twisted>=1.2')" + assert r == Requirement("Twisted>=1.2") + assert r == Requirement("twisTed>=1.2") + assert r != Requirement("Twisted>=2.0") + assert r != Requirement("Zope>=1.2") + assert r != Requirement("Zope>=3.0") + assert r != Requirement("Twisted[extras]>=1.2") + + def testOrdering(self): + r1 = Requirement("Twisted==1.2c1,>=1.2") + r2 = Requirement("Twisted>=1.2,==1.2c1") + assert r1 == r2 + assert str(r1) == str(r2) + assert str(r2) == "Twisted==1.2c1,>=1.2" + assert Requirement("Twisted") != Requirement( + "Twisted @ https://localhost/twisted.zip" + ) + + def testBasicContains(self): + r = Requirement("Twisted>=1.2") + foo_dist = Distribution.from_filename("FooPkg-1.3_1.egg") + twist11 = Distribution.from_filename("Twisted-1.1.egg") + twist12 = Distribution.from_filename("Twisted-1.2.egg") + assert parse_version('1.2') in r + assert parse_version('1.1') not in r + assert '1.2' in r + assert '1.1' not in r + assert foo_dist not in r + assert twist11 not in r + assert twist12 in r + + def testOptionsAndHashing(self): + r1 = Requirement.parse("Twisted[foo,bar]>=1.2") + r2 = Requirement.parse("Twisted[bar,FOO]>=1.2") + assert r1 == r2 + assert set(r1.extras) == set(("foo", "bar")) + assert set(r2.extras) == set(("foo", "bar")) + assert hash(r1) == hash(r2) + assert hash(r1) == hash(( + "twisted", + None, + SpecifierSet(">=1.2"), + frozenset(["foo", "bar"]), + None, + )) + assert hash( + Requirement.parse("Twisted @ https://localhost/twisted.zip") + ) == hash(( + "twisted", + "https://localhost/twisted.zip", + SpecifierSet(), + frozenset(), + None, + )) + + def testVersionEquality(self): + r1 = Requirement.parse("foo==0.3a2") + r2 = Requirement.parse("foo!=0.3a4") + d = Distribution.from_filename + + assert d("foo-0.3a4.egg") not in r1 + assert d("foo-0.3a1.egg") not in r1 + assert d("foo-0.3a4.egg") not in r2 + + assert d("foo-0.3a2.egg") in r1 + assert d("foo-0.3a2.egg") in r2 + assert d("foo-0.3a3.egg") in r2 + assert d("foo-0.3a5.egg") in r2 + + def testSetuptoolsProjectName(self): + """ + The setuptools project should implement the setuptools package. + """ + + assert Requirement.parse('setuptools').project_name == 'setuptools' + # setuptools 0.7 and higher means setuptools. + assert Requirement.parse('setuptools == 0.7').project_name == 'setuptools' + assert Requirement.parse('setuptools == 0.7a1').project_name == 'setuptools' + assert Requirement.parse('setuptools >= 0.7').project_name == 'setuptools' + + +class TestParsing: + def testEmptyParse(self): + assert list(parse_requirements('')) == [] + + def testYielding(self): + for inp, out in [ + ([], []), + ('x', ['x']), + ([[]], []), + (' x\n y', ['x', 'y']), + (['x\n\n', 'y'], ['x', 'y']), + ]: + assert list(pkg_resources.yield_lines(inp)) == out + + def testSplitting(self): + sample = """ + x + [Y] + z + + a + [b ] + # foo + c + [ d] + [q] + v + """ + assert list(pkg_resources.split_sections(sample)) == [ + (None, ["x"]), + ("Y", ["z", "a"]), + ("b", ["c"]), + ("d", []), + ("q", ["v"]), + ] + with pytest.raises(ValueError): + list(pkg_resources.split_sections("[foo")) + + def testSafeName(self): + assert safe_name("adns-python") == "adns-python" + assert safe_name("WSGI Utils") == "WSGI-Utils" + assert safe_name("WSGI Utils") == "WSGI-Utils" + assert safe_name("Money$$$Maker") == "Money-Maker" + assert safe_name("peak.web") != "peak-web" + + def testSafeVersion(self): + assert safe_version("1.2-1") == "1.2.post1" + assert safe_version("1.2 alpha") == "1.2.alpha" + assert safe_version("2.3.4 20050521") == "2.3.4.20050521" + assert safe_version("Money$$$Maker") == "Money-Maker" + assert safe_version("peak.web") == "peak.web" + + def testSimpleRequirements(self): + assert list(parse_requirements('Twis-Ted>=1.2-1')) == [ + Requirement('Twis-Ted>=1.2-1') + ] + assert list(parse_requirements('Twisted >=1.2, \\ # more\n<2.0')) == [ + Requirement('Twisted>=1.2,<2.0') + ] + assert Requirement.parse("FooBar==1.99a3") == Requirement("FooBar==1.99a3") + with pytest.raises(ValueError): + Requirement.parse(">=2.3") + with pytest.raises(ValueError): + Requirement.parse("x\\") + with pytest.raises(ValueError): + Requirement.parse("x==2 q") + with pytest.raises(ValueError): + Requirement.parse("X==1\nY==2") + with pytest.raises(ValueError): + Requirement.parse("#") + + def test_requirements_with_markers(self): + assert Requirement.parse("foobar;os_name=='a'") == Requirement.parse( + "foobar;os_name=='a'" + ) + assert Requirement.parse( + "name==1.1;python_version=='2.7'" + ) != Requirement.parse("name==1.1;python_version=='3.6'") + assert Requirement.parse( + "name==1.0;python_version=='2.7'" + ) != Requirement.parse("name==1.2;python_version=='2.7'") + assert Requirement.parse( + "name[foo]==1.0;python_version=='3.6'" + ) != Requirement.parse("name[foo,bar]==1.0;python_version=='3.6'") + + def test_local_version(self): + parse_requirements('foo==1.0+org1') + + def test_spaces_between_multiple_versions(self): + parse_requirements('foo>=1.0, <3') + parse_requirements('foo >= 1.0, < 3') + + @pytest.mark.parametrize( + ("lower", "upper"), + [ + ('1.2-rc1', '1.2rc1'), + ('0.4', '0.4.0'), + ('0.4.0.0', '0.4.0'), + ('0.4.0-0', '0.4-0'), + ('0post1', '0.0post1'), + ('0pre1', '0.0c1'), + ('0.0.0preview1', '0c1'), + ('0.0c1', '0-rc1'), + ('1.2a1', '1.2.a.1'), + ('1.2.a', '1.2a'), + ], + ) + def testVersionEquality(self, lower, upper): + assert parse_version(lower) == parse_version(upper) + + torture = """ + 0.80.1-3 0.80.1-2 0.80.1-1 0.79.9999+0.80.0pre4-1 + 0.79.9999+0.80.0pre2-3 0.79.9999+0.80.0pre2-2 + 0.77.2-1 0.77.1-1 0.77.0-1 + """ + + @pytest.mark.parametrize( + ("lower", "upper"), + [ + ('2.1', '2.1.1'), + ('2a1', '2b0'), + ('2a1', '2.1'), + ('2.3a1', '2.3'), + ('2.1-1', '2.1-2'), + ('2.1-1', '2.1.1'), + ('2.1', '2.1post4'), + ('2.1a0-20040501', '2.1'), + ('1.1', '02.1'), + ('3.2', '3.2.post0'), + ('3.2post1', '3.2post2'), + ('0.4', '4.0'), + ('0.0.4', '0.4.0'), + ('0post1', '0.4post1'), + ('2.1.0-rc1', '2.1.0'), + ('2.1dev', '2.1a0'), + ] + + list(pairwise(reversed(torture.split()))), + ) + def testVersionOrdering(self, lower, upper): + assert parse_version(lower) < parse_version(upper) + + def testVersionHashable(self): + """ + Ensure that our versions stay hashable even though we've subclassed + them and added some shim code to them. + """ + assert hash(parse_version("1.0")) == hash(parse_version("1.0")) + + +class TestNamespaces: + ns_str = "__import__('pkg_resources').declare_namespace(__name__)\n" + + @pytest.fixture + def symlinked_tmpdir(self, tmpdir): + """ + Where available, return the tempdir as a symlink, + which as revealed in #231 is more fragile than + a natural tempdir. + """ + if not hasattr(os, 'symlink'): + yield str(tmpdir) + return + + link_name = str(tmpdir) + '-linked' + os.symlink(str(tmpdir), link_name) + try: + yield type(tmpdir)(link_name) + finally: + os.unlink(link_name) + + @pytest.fixture(autouse=True) + def patched_path(self, tmpdir): + """ + Patch sys.path to include the 'site-pkgs' dir. Also + restore pkg_resources._namespace_packages to its + former state. + """ + saved_ns_pkgs = pkg_resources._namespace_packages.copy() + saved_sys_path = sys.path[:] + site_pkgs = tmpdir.mkdir('site-pkgs') + sys.path.append(str(site_pkgs)) + try: + yield + finally: + pkg_resources._namespace_packages = saved_ns_pkgs + sys.path = saved_sys_path + + issue591 = pytest.mark.xfail(platform.system() == 'Windows', reason="#591") + + @issue591 + def test_two_levels_deep(self, symlinked_tmpdir): + """ + Test nested namespace packages + Create namespace packages in the following tree : + site-packages-1/pkg1/pkg2 + site-packages-2/pkg1/pkg2 + Check both are in the _namespace_packages dict and that their __path__ + is correct + """ + real_tmpdir = symlinked_tmpdir.realpath() + tmpdir = symlinked_tmpdir + sys.path.append(str(tmpdir / 'site-pkgs2')) + site_dirs = tmpdir / 'site-pkgs', tmpdir / 'site-pkgs2' + for site in site_dirs: + pkg1 = site / 'pkg1' + pkg2 = pkg1 / 'pkg2' + pkg2.ensure_dir() + (pkg1 / '__init__.py').write_text(self.ns_str, encoding='utf-8') + (pkg2 / '__init__.py').write_text(self.ns_str, encoding='utf-8') + with pytest.warns(DeprecationWarning, match="pkg_resources.declare_namespace"): + import pkg1 # pyright: ignore[reportMissingImports] # Temporary package for test + assert "pkg1" in pkg_resources._namespace_packages + # attempt to import pkg2 from site-pkgs2 + with pytest.warns(DeprecationWarning, match="pkg_resources.declare_namespace"): + import pkg1.pkg2 # pyright: ignore[reportMissingImports] # Temporary package for test + # check the _namespace_packages dict + assert "pkg1.pkg2" in pkg_resources._namespace_packages + assert pkg_resources._namespace_packages["pkg1"] == ["pkg1.pkg2"] + # check the __path__ attribute contains both paths + expected = [ + str(real_tmpdir / "site-pkgs" / "pkg1" / "pkg2"), + str(real_tmpdir / "site-pkgs2" / "pkg1" / "pkg2"), + ] + assert pkg1.pkg2.__path__ == expected + + @issue591 + def test_path_order(self, symlinked_tmpdir): + """ + Test that if multiple versions of the same namespace package subpackage + are on different sys.path entries, that only the one earliest on + sys.path is imported, and that the namespace package's __path__ is in + the correct order. + + Regression test for https://github.com/pypa/setuptools/issues/207 + """ + + tmpdir = symlinked_tmpdir + site_dirs = ( + tmpdir / "site-pkgs", + tmpdir / "site-pkgs2", + tmpdir / "site-pkgs3", + ) + + vers_str = "__version__ = %r" + + for number, site in enumerate(site_dirs, 1): + if number > 1: + sys.path.append(str(site)) + nspkg = site / 'nspkg' + subpkg = nspkg / 'subpkg' + subpkg.ensure_dir() + (nspkg / '__init__.py').write_text(self.ns_str, encoding='utf-8') + (subpkg / '__init__.py').write_text(vers_str % number, encoding='utf-8') + + with pytest.warns(DeprecationWarning, match="pkg_resources.declare_namespace"): + import nspkg # pyright: ignore[reportMissingImports] # Temporary package for test + import nspkg.subpkg # pyright: ignore[reportMissingImports] # Temporary package for test + expected = [str(site.realpath() / 'nspkg') for site in site_dirs] + assert nspkg.__path__ == expected + assert nspkg.subpkg.__version__ == 1 diff --git a/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_working_set.py b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_working_set.py new file mode 100644 index 00000000..ed20c59d --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pkg_resources/tests/test_working_set.py @@ -0,0 +1,505 @@ +import functools +import inspect +import re +import textwrap + +import pytest + +import pkg_resources + +from .test_resources import Metadata + + +def strip_comments(s): + return '\n'.join( + line + for line in s.split('\n') + if line.strip() and not line.strip().startswith('#') + ) + + +def parse_distributions(s): + """ + Parse a series of distribution specs of the form: + {project_name}-{version} + [optional, indented requirements specification] + + Example: + + foo-0.2 + bar-1.0 + foo>=3.0 + [feature] + baz + + yield 2 distributions: + - project_name=foo, version=0.2 + - project_name=bar, version=1.0, + requires=['foo>=3.0', 'baz; extra=="feature"'] + """ + s = s.strip() + for spec in re.split(r'\n(?=[^\s])', s): + if not spec: + continue + fields = spec.split('\n', 1) + assert 1 <= len(fields) <= 2 + name, version = fields.pop(0).rsplit('-', 1) + if fields: + requires = textwrap.dedent(fields.pop(0)) + metadata = Metadata(('requires.txt', requires)) + else: + metadata = None + dist = pkg_resources.Distribution( + project_name=name, version=version, metadata=metadata + ) + yield dist + + +class FakeInstaller: + def __init__(self, installable_dists) -> None: + self._installable_dists = installable_dists + + def __call__(self, req): + return next( + iter(filter(lambda dist: dist in req, self._installable_dists)), None + ) + + +def parametrize_test_working_set_resolve(*test_list): + idlist = [] + argvalues = [] + for test in test_list: + ( + name, + installed_dists, + installable_dists, + requirements, + expected1, + expected2, + ) = ( + strip_comments(s.lstrip()) + for s in textwrap.dedent(test).lstrip().split('\n\n', 5) + ) + installed_dists = list(parse_distributions(installed_dists)) + installable_dists = list(parse_distributions(installable_dists)) + requirements = list(pkg_resources.parse_requirements(requirements)) + for id_, replace_conflicting, expected in ( + (name, False, expected1), + (name + '_replace_conflicting', True, expected2), + ): + idlist.append(id_) + expected = strip_comments(expected.strip()) + if re.match(r'\w+$', expected): + expected = getattr(pkg_resources, expected) + assert issubclass(expected, Exception) + else: + expected = list(parse_distributions(expected)) + argvalues.append( + pytest.param( + installed_dists, + installable_dists, + requirements, + replace_conflicting, + expected, + ) + ) + return pytest.mark.parametrize( + ( + "installed_dists", + "installable_dists", + "requirements", + "replace_conflicting", + "resolved_dists_or_exception", + ), + argvalues, + ids=idlist, + ) + + +@parametrize_test_working_set_resolve( + """ + # id + noop + + # installed + + # installable + + # wanted + + # resolved + + # resolved [replace conflicting] + """, + """ + # id + already_installed + + # installed + foo-3.0 + + # installable + + # wanted + foo>=2.1,!=3.1,<4 + + # resolved + foo-3.0 + + # resolved [replace conflicting] + foo-3.0 + """, + """ + # id + installable_not_installed + + # installed + + # installable + foo-3.0 + foo-4.0 + + # wanted + foo>=2.1,!=3.1,<4 + + # resolved + foo-3.0 + + # resolved [replace conflicting] + foo-3.0 + """, + """ + # id + not_installable + + # installed + + # installable + + # wanted + foo>=2.1,!=3.1,<4 + + # resolved + DistributionNotFound + + # resolved [replace conflicting] + DistributionNotFound + """, + """ + # id + no_matching_version + + # installed + + # installable + foo-3.1 + + # wanted + foo>=2.1,!=3.1,<4 + + # resolved + DistributionNotFound + + # resolved [replace conflicting] + DistributionNotFound + """, + """ + # id + installable_with_installed_conflict + + # installed + foo-3.1 + + # installable + foo-3.5 + + # wanted + foo>=2.1,!=3.1,<4 + + # resolved + VersionConflict + + # resolved [replace conflicting] + foo-3.5 + """, + """ + # id + not_installable_with_installed_conflict + + # installed + foo-3.1 + + # installable + + # wanted + foo>=2.1,!=3.1,<4 + + # resolved + VersionConflict + + # resolved [replace conflicting] + DistributionNotFound + """, + """ + # id + installed_with_installed_require + + # installed + foo-3.9 + baz-0.1 + foo>=2.1,!=3.1,<4 + + # installable + + # wanted + baz + + # resolved + foo-3.9 + baz-0.1 + + # resolved [replace conflicting] + foo-3.9 + baz-0.1 + """, + """ + # id + installed_with_conflicting_installed_require + + # installed + foo-5 + baz-0.1 + foo>=2.1,!=3.1,<4 + + # installable + + # wanted + baz + + # resolved + VersionConflict + + # resolved [replace conflicting] + DistributionNotFound + """, + """ + # id + installed_with_installable_conflicting_require + + # installed + foo-5 + baz-0.1 + foo>=2.1,!=3.1,<4 + + # installable + foo-2.9 + + # wanted + baz + + # resolved + VersionConflict + + # resolved [replace conflicting] + baz-0.1 + foo-2.9 + """, + """ + # id + installed_with_installable_require + + # installed + baz-0.1 + foo>=2.1,!=3.1,<4 + + # installable + foo-3.9 + + # wanted + baz + + # resolved + foo-3.9 + baz-0.1 + + # resolved [replace conflicting] + foo-3.9 + baz-0.1 + """, + """ + # id + installable_with_installed_require + + # installed + foo-3.9 + + # installable + baz-0.1 + foo>=2.1,!=3.1,<4 + + # wanted + baz + + # resolved + foo-3.9 + baz-0.1 + + # resolved [replace conflicting] + foo-3.9 + baz-0.1 + """, + """ + # id + installable_with_installable_require + + # installed + + # installable + foo-3.9 + baz-0.1 + foo>=2.1,!=3.1,<4 + + # wanted + baz + + # resolved + foo-3.9 + baz-0.1 + + # resolved [replace conflicting] + foo-3.9 + baz-0.1 + """, + """ + # id + installable_with_conflicting_installable_require + + # installed + foo-5 + + # installable + foo-2.9 + baz-0.1 + foo>=2.1,!=3.1,<4 + + # wanted + baz + + # resolved + VersionConflict + + # resolved [replace conflicting] + baz-0.1 + foo-2.9 + """, + """ + # id + conflicting_installables + + # installed + + # installable + foo-2.9 + foo-5.0 + + # wanted + foo>=2.1,!=3.1,<4 + foo>=4 + + # resolved + VersionConflict + + # resolved [replace conflicting] + VersionConflict + """, + """ + # id + installables_with_conflicting_requires + + # installed + + # installable + foo-2.9 + dep==1.0 + baz-5.0 + dep==2.0 + dep-1.0 + dep-2.0 + + # wanted + foo + baz + + # resolved + VersionConflict + + # resolved [replace conflicting] + VersionConflict + """, + """ + # id + installables_with_conflicting_nested_requires + + # installed + + # installable + foo-2.9 + dep1 + dep1-1.0 + subdep<1.0 + baz-5.0 + dep2 + dep2-1.0 + subdep>1.0 + subdep-0.9 + subdep-1.1 + + # wanted + foo + baz + + # resolved + VersionConflict + + # resolved [replace conflicting] + VersionConflict + """, + """ + # id + wanted_normalized_name_installed_canonical + + # installed + foo.bar-3.6 + + # installable + + # wanted + foo-bar==3.6 + + # resolved + foo.bar-3.6 + + # resolved [replace conflicting] + foo.bar-3.6 + """, +) +def test_working_set_resolve( + installed_dists, + installable_dists, + requirements, + replace_conflicting, + resolved_dists_or_exception, +): + ws = pkg_resources.WorkingSet([]) + list(map(ws.add, installed_dists)) + resolve_call = functools.partial( + ws.resolve, + requirements, + installer=FakeInstaller(installable_dists), + replace_conflicting=replace_conflicting, + ) + if inspect.isclass(resolved_dists_or_exception): + with pytest.raises(resolved_dists_or_exception): + resolve_call() + else: + assert sorted(resolve_call()) == sorted(resolved_dists_or_exception) |