helpers.py (poetry-1.1.15) | : | helpers.py (poetry-1.2.0) | ||
---|---|---|---|---|
from __future__ import annotations | ||||
import contextlib | ||||
import os | import os | |||
import re | ||||
import shutil | import shutil | |||
import sys | ||||
import urllib.parse | ||||
from pathlib import Path | ||||
from typing import TYPE_CHECKING | ||||
from typing import Any | ||||
from poetry.console import Application | ||||
from poetry.core.masonry.utils.helpers import escape_name | from poetry.core.masonry.utils.helpers import escape_name | |||
from poetry.core.masonry.utils.helpers import escape_version | from poetry.core.masonry.utils.helpers import escape_version | |||
from poetry.core.packages import Dependency | from poetry.core.packages.package import Package | |||
from poetry.core.packages import Link | from poetry.core.packages.utils.link import Link | |||
from poetry.core.packages import Package | ||||
from poetry.core.toml.file import TOMLFile | from poetry.core.toml.file import TOMLFile | |||
from poetry.core.vcs.git import ParsedUrl | from poetry.core.vcs.git import ParsedUrl | |||
from poetry.config.config import Config | ||||
from poetry.console.application import Application | ||||
from poetry.factory import Factory | from poetry.factory import Factory | |||
from poetry.installation.executor import Executor | from poetry.installation.executor import Executor | |||
from poetry.packages import Locker | from poetry.packages import Locker | |||
from poetry.repositories import Repository | from poetry.repositories import Repository | |||
from poetry.repositories.exceptions import PackageNotFound | from poetry.repositories.exceptions import PackageNotFound | |||
from poetry.utils._compat import PY2 | from poetry.utils._compat import metadata | |||
from poetry.utils._compat import WINDOWS | ||||
from poetry.utils._compat import Path | if TYPE_CHECKING: | |||
from poetry.utils._compat import urlparse | from collections.abc import Iterator | |||
from poetry.core.packages.dependency import Dependency | ||||
from poetry.core.semver.version import Version | ||||
from pytest_mock import MockerFixture | ||||
from tomlkit.toml_document import TOMLDocument | ||||
from poetry.installation.operations.operation import Operation | ||||
from poetry.poetry import Poetry | ||||
FIXTURE_PATH = Path(__file__).parent / "fixtures" | FIXTURE_PATH = Path(__file__).parent / "fixtures" | |||
def get_package(name, version): | # Used as a mock for latest git revision. | |||
return Package(name, version) | MOCK_DEFAULT_GIT_REVISION = "9cf87a285a2d3fbb0b9fa621997b3acc3631ed24" | |||
def get_package( | ||||
name: str, version: str | Version, yanked: str | bool = False | ||||
) -> Package: | ||||
return Package(name, version, yanked=yanked) | ||||
def get_dependency( | def get_dependency( | |||
name, constraint=None, category="main", optional=False, allows_prereleases=F | name: str, | |||
alse | constraint: str | dict[str, Any] | None = None, | |||
): | groups: list[str] | None = None, | |||
return Dependency( | optional: bool = False, | |||
name, | allows_prereleases: bool = False, | |||
constraint or "*", | ) -> Dependency: | |||
category=category, | if constraint is None: | |||
optional=optional, | constraint = "*" | |||
allows_prereleases=allows_prereleases, | ||||
) | if isinstance(constraint, str): | |||
constraint = {"version": constraint} | ||||
constraint["optional"] = optional | ||||
constraint["allow-prereleases"] = allows_prereleases | ||||
return Factory.create_dependency(name, constraint or "*", groups=groups) | ||||
def fixture(path=None): | def fixture(path: str | None = None) -> Path: | |||
if path: | if path: | |||
return FIXTURE_PATH / path | return FIXTURE_PATH / path | |||
else: | else: | |||
return FIXTURE_PATH | return FIXTURE_PATH | |||
def copy_or_symlink(source, dest): | def copy_or_symlink(source: Path, dest: Path) -> None: | |||
if dest.exists(): | if dest.is_symlink() or dest.is_file(): | |||
if dest.is_symlink(): | dest.unlink() # missing_ok is only available in Python >= 3.8 | |||
os.unlink(str(dest)) | elif dest.is_dir(): | |||
elif dest.is_dir(): | shutil.rmtree(dest) | |||
shutil.rmtree(str(dest)) | ||||
else: | ||||
os.unlink(str(dest)) | ||||
# Python2 does not support os.symlink on Windows whereas Python3 does. | ||||
# os.symlink requires either administrative privileges or developer mode on Win10, | # os.symlink requires either administrative privileges or developer mode on Win10, | |||
# throwing an OSError if neither is active. | # throwing an OSError if neither is active. | |||
if WINDOWS: | if sys.platform == "win32": | |||
if PY2: | try: | |||
os.symlink(str(source), str(dest), target_is_directory=source.is_dir | ||||
()) | ||||
except OSError: | ||||
if source.is_dir(): | if source.is_dir(): | |||
shutil.copytree(str(source), str(dest)) | shutil.copytree(str(source), str(dest)) | |||
else: | else: | |||
shutil.copyfile(str(source), str(dest)) | shutil.copyfile(str(source), str(dest)) | |||
else: | ||||
try: | ||||
os.symlink(str(source), str(dest), target_is_directory=source.is | ||||
_dir()) | ||||
except OSError: | ||||
if source.is_dir(): | ||||
shutil.copytree(str(source), str(dest)) | ||||
else: | ||||
shutil.copyfile(str(source), str(dest)) | ||||
else: | else: | |||
os.symlink(str(source), str(dest)) | os.symlink(str(source), str(dest)) | |||
def mock_clone(_, source, dest): | class MockDulwichRepo: | |||
def __init__(self, root: Path | str, **__: Any) -> None: | ||||
self.path = str(root) | ||||
def head(self) -> bytes: | ||||
return MOCK_DEFAULT_GIT_REVISION.encode() | ||||
def mock_clone( | ||||
url: str, | ||||
*_: Any, | ||||
source_root: Path | None = None, | ||||
**__: Any, | ||||
) -> MockDulwichRepo: | ||||
# Checking source to determine which folder we need to copy | # Checking source to determine which folder we need to copy | |||
parsed = ParsedUrl.parse(source) | parsed = ParsedUrl.parse(url) | |||
path = re.sub(r"(.git)?$", "", parsed.pathname.lstrip("/")) | ||||
folder = ( | folder = Path(__file__).parent / "fixtures" / "git" / parsed.resource / path | |||
Path(__file__).parent | ||||
/ "fixtures" | if not source_root: | |||
/ "git" | source_root = Path(Config.create().get("cache-dir")) / "src" | |||
/ parsed.resource | ||||
/ parsed.pathname.lstrip("/").rstrip(".git") | dest = source_root / path | |||
) | dest.parent.mkdir(parents=True, exist_ok=True) | |||
copy_or_symlink(folder, dest) | copy_or_symlink(folder, dest) | |||
return MockDulwichRepo(dest) | ||||
def mock_download(url, dest, **__): | def mock_download(url: str, dest: Path) -> None: | |||
parts = urlparse.urlparse(url) | parts = urllib.parse.urlparse(url) | |||
fixtures = Path(__file__).parent / "fixtures" | fixtures = Path(__file__).parent / "fixtures" | |||
fixture = fixtures / parts.path.lstrip("/") | fixture = fixtures / parts.path.lstrip("/") | |||
copy_or_symlink(fixture, Path(dest)) | copy_or_symlink(fixture, dest) | |||
class TestExecutor(Executor): | class TestExecutor(Executor): | |||
def __init__(self, *args, **kwargs): | def __init__(self, *args: Any, **kwargs: Any) -> None: | |||
super(TestExecutor, self).__init__(*args, **kwargs) | super().__init__(*args, **kwargs) | |||
self._installs = [] | self._installs = [] | |||
self._updates = [] | self._updates = [] | |||
self._uninstalls = [] | self._uninstalls = [] | |||
@property | @property | |||
def installations(self): | def installations(self) -> list[Package]: | |||
return self._installs | return self._installs | |||
@property | @property | |||
def updates(self): | def updates(self) -> list[Package]: | |||
return self._updates | return self._updates | |||
@property | @property | |||
def removals(self): | def removals(self) -> list[Package]: | |||
return self._uninstalls | return self._uninstalls | |||
def _do_execute_operation(self, operation): | def _do_execute_operation(self, operation: Operation) -> None: | |||
super(TestExecutor, self)._do_execute_operation(operation) | super()._do_execute_operation(operation) | |||
if not operation.skipped: | if not operation.skipped: | |||
getattr(self, "_{}s".format(operation.job_type)).append(operation.pa ckage) | getattr(self, f"_{operation.job_type}s").append(operation.package) | |||
def _execute_install(self, operation): | def _execute_install(self, operation: Operation) -> int: | |||
return 0 | return 0 | |||
def _execute_update(self, operation): | def _execute_update(self, operation: Operation) -> int: | |||
return 0 | return 0 | |||
def _execute_remove(self, operation): | def _execute_remove(self, operation: Operation) -> int: | |||
return 0 | return 0 | |||
class TestApplication(Application): | class PoetryTestApplication(Application): | |||
def __init__(self, poetry): | def __init__(self, poetry: Poetry) -> None: | |||
super(TestApplication, self).__init__() | super().__init__() | |||
self._poetry = poetry | self._poetry = poetry | |||
def reset_poetry(self): | def reset_poetry(self) -> None: | |||
poetry = self._poetry | poetry = self._poetry | |||
self._poetry = Factory().create_poetry(self._poetry.file.path.parent) | self._poetry = Factory().create_poetry(self._poetry.file.path.parent) | |||
self._poetry.set_pool(poetry.pool) | self._poetry.set_pool(poetry.pool) | |||
self._poetry.set_config(poetry.config) | self._poetry.set_config(poetry.config) | |||
self._poetry.set_locker( | self._poetry.set_locker( | |||
TestLocker(poetry.locker.lock.path, self._poetry.local_config) | TestLocker(poetry.locker.lock.path, self._poetry.local_config) | |||
) | ) | |||
class TestLocker(Locker): | class TestLocker(Locker): | |||
def __init__(self, lock, local_config): # noqa | def __init__(self, lock: str | Path, local_config: dict) -> None: | |||
self._lock = TOMLFile(lock) | self._lock = TOMLFile(lock) | |||
self._local_config = local_config | self._local_config = local_config | |||
self._lock_data = None | self._lock_data = None | |||
self._content_hash = self._get_content_hash() | self._content_hash = self._get_content_hash() | |||
self._locked = False | self._locked = False | |||
self._lock_data = None | self._lock_data = None | |||
self._write = False | self._write = False | |||
def write(self, write=True): | def write(self, write: bool = True) -> None: | |||
self._write = write | self._write = write | |||
def is_locked(self): | def is_locked(self) -> bool: | |||
return self._locked | return self._locked | |||
def locked(self, is_locked=True): | def locked(self, is_locked: bool = True) -> TestLocker: | |||
self._locked = is_locked | self._locked = is_locked | |||
return self | return self | |||
def mock_lock_data(self, data): | def mock_lock_data(self, data: dict) -> None: | |||
self.locked() | self.locked() | |||
self._lock_data = data | self._lock_data = data | |||
def is_fresh(self): | def is_fresh(self) -> bool: | |||
return True | return True | |||
def _write_lock_data(self, data): | def _write_lock_data(self, data: TOMLDocument) -> None: | |||
if self._write: | if self._write: | |||
super(TestLocker, self)._write_lock_data(data) | super()._write_lock_data(data) | |||
self._locked = True | self._locked = True | |||
return | return | |||
self._lock_data = data | self._lock_data = data | |||
class TestRepository(Repository): | class TestRepository(Repository): | |||
def find_packages(self, dependency): | def find_packages(self, dependency: Dependency) -> list[Package]: | |||
packages = super(TestRepository, self).find_packages(dependency) | packages = super().find_packages(dependency) | |||
if len(packages) == 0: | if len(packages) == 0: | |||
raise PackageNotFound("Package [{}] not found.".format(dependency.na me)) | raise PackageNotFound(f"Package [{dependency.name}] not found.") | |||
return packages | return packages | |||
def find_links_for_package(self, package): | def find_links_for_package(self, package: Package) -> list[Link]: | |||
return [ | return [ | |||
Link( | Link( | |||
"https://foo.bar/files/{}-{}-py2.py3-none-any.whl".format( | f"https://foo.bar/files/{escape_name(package.name)}" | |||
escape_name(package.name), escape_version(package.version.te | f"-{escape_version(package.version.text)}-py2.py3-none-any.whl" | |||
xt) | ||||
) | ||||
) | ) | |||
] | ] | |||
@contextlib.contextmanager | ||||
def isolated_environment( | ||||
environ: dict[str, Any] | None = None, clear: bool = False | ||||
) -> Iterator[None]: | ||||
original_environ = dict(os.environ) | ||||
if clear: | ||||
os.environ.clear() | ||||
if environ: | ||||
os.environ.update(environ) | ||||
yield | ||||
os.environ.clear() | ||||
os.environ.update(original_environ) | ||||
def make_entry_point_from_plugin( | ||||
name: str, cls: type[Any], dist: metadata.Distribution | None = None | ||||
) -> metadata.EntryPoint: | ||||
ep = metadata.EntryPoint( | ||||
name=name, | ||||
group=getattr(cls, "group", None), | ||||
value=f"{cls.__module__}:{cls.__name__}", | ||||
) | ||||
if dist: | ||||
return ep._for(dist) | ||||
return ep | ||||
def mock_metadata_entry_points( | ||||
mocker: MockerFixture, | ||||
cls: type[Any], | ||||
name: str = "my-plugin", | ||||
dist: metadata.Distribution | None = None, | ||||
) -> None: | ||||
mocker.patch.object( | ||||
metadata, | ||||
"entry_points", | ||||
return_value=[make_entry_point_from_plugin(name, cls, dist)], | ||||
) | ||||
End of changes. 44 change blocks. | ||||
84 lines changed or deleted | 116 lines changed or added |