Improved tests

pyproject.toml

@@ -22,6 +22,7 @@ mypy = {version="^1.13.0", extras=["faster-cache"]}
 pre-commit = "^4.0.0"
 pre-commit-hooks = "^5.0.0"
 pytest = "^8.3.3"
+pytest-mock = "^3.14.0"
 
 [tool.poetry.group.ci]
 optional = true

tests/conftest.py

@@ -1,20 +1,25 @@
+from __future__ import annotations
+
 from pathlib import Path
+from typing import TYPE_CHECKING
 from unittest.mock import Mock
 
 import pytest
+from cleo.events.console_command_event import ConsoleCommandEvent
 from poetry.core.packages.dependency import Dependency
 from poetry.core.packages.dependency_group import MAIN_GROUP, DependencyGroup
 from poetry.core.packages.directory_dependency import DirectoryDependency
 from poetry.poetry import Poetry
 
+from tests.helpers import MockRepoManager, _poetry_run
 
-@pytest.fixture
-def mock_event_gen():
+if TYPE_CHECKING:
     from poetry.console.commands.command import Command
 
-    def _factory(command_cls: type[Command], disable_cache: bool):
-        from cleo.events.console_command_event import ConsoleCommandEvent
 
+@pytest.fixture
+def mock_event_gen():
+    def _factory(command_cls: type[Command], disable_cache: bool):
         main_grp = DependencyGroup(MAIN_GROUP)
         main_grp.add_dependency(Dependency("numpy", "==1.5.0"))
         main_grp.add_dependency(
@@ -52,21 +57,12 @@ def _factory(command_cls: type[Command], disable_cache: bool):
 
 
 @pytest.fixture
-def mock_terminate_event_gen(mock_event_gen):
-    from poetry.console.commands.command import Command
-
-    def _factory(command_cls: type[Command], disable_cache: bool):
-        from cleo.events.console_terminate_event import ConsoleTerminateEvent
+def poetry_run():
+    return _poetry_run
 
-        mock_event = mock_event_gen(command_cls, disable_cache)
-        mock_io = mock_event.io
-        mock_command = mock_event.command
-        del mock_event
 
-        mock_terminate_event = Mock(spec=ConsoleTerminateEvent)
-        mock_terminate_event.command = mock_command
-        mock_terminate_event.io = mock_io
-
-        return mock_terminate_event
-
-    return _factory
+@pytest.fixture(scope="session")
+def repo_manager():
+    obj = MockRepoManager()
+    yield obj
+    del obj

tests/fixtures/v1/pkg_one/poetry.toml

@@ -0,0 +1,2 @@
+[virtualenvs]
+create = false

tests/fixtures/v1/pkg_one/pyproject.toml

@@ -0,0 +1,25 @@
+[tool.poetry]
+name = "pkg-one"
+version = "0.1.0"
+description = ""
+authors = ["Example <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.9"
+
+# A list of all of the optional dependencies, some of which are included in the
+# below `extras`. They can be opted into by apps.
+pandas = { version = "^2.0.0", optional = true }
+
+[tool.poetry.extras]
+withpandas = ["pandas"]
+
+[tool.poetry-monoranger-plugin]
+enabled = true
+monorepo-root = "../"
+version-rewrite-rule = '=='
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

tests/fixtures/v1/pkg_three/poetry.toml

@@ -0,0 +1,2 @@
+[virtualenvs]
+create = false

tests/fixtures/v1/pkg_three/pyproject.toml

@@ -0,0 +1,19 @@
+[tool.poetry]
+name = "pkg-three"
+version = "0.1.0"
+description = ""
+authors = ["Example <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.9"
+pkg-two = { path = '../pkg_two', develop = true, extras = ["withpandas"] }
+
+[tool.poetry-monoranger-plugin]
+enabled = true
+monorepo-root = "../"
+version-rewrite-rule = '=='
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

tests/fixtures/v1/pkg_two/poetry.toml

@@ -0,0 +1,2 @@
+[virtualenvs]
+create = false

tests/fixtures/v1/pkg_two/pyproject.toml

@@ -0,0 +1,19 @@
+[tool.poetry]
+name = "pkg-two"
+version = "0.1.0"
+description = ""
+authors = ["Example <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.9"
+tqdm = "^4.67.1"
+
+[tool.poetry-monoranger-plugin]
+enabled = true
+monorepo-root = "../"
+version-rewrite-rule = '=='
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

tests/fixtures/v1/poetry.toml

@@ -0,0 +1,2 @@
+[virtualenvs]
+in-project = true

tests/fixtures/v1/pyproject.toml

@@ -0,0 +1,21 @@
+[tool.poetry]
+name = "v1"
+version = "0.1.0"
+description = ""
+authors = ["Example <[email protected]>"]
+readme = "README.md"
+package-mode = false
+
+[tool.poetry.dependencies]
+python = "^3.9"
+pkg-one = { path = 'pkg_one', develop = true }
+pkg-two = { path = 'pkg_two', develop = true }
+pkg-three = { path = 'pkg_three', develop = true }
+
+[tool.poetry.group.dev.dependencies]
+ruff = "^0.6.6"
+poetry = ">=1.8.0,<2"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

tests/helpers.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import os
+import shutil
+import tempfile
+import weakref
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from cleo.events.console_command_event import ConsoleCommandEvent
+from cleo.events.console_events import COMMAND
+from cleo.io.inputs.argv_input import ArgvInput
+from cleo.io.outputs.stream_output import StreamOutput
+from poetry.console.application import Application
+from poetry.factory import Factory
+from poetry.utils.env import EnvManager
+from poetry.utils.env.system_env import SystemEnv
+
+if TYPE_CHECKING:
+    from cleo.events.event import Event
+    from cleo.events.event_dispatcher import EventDispatcher
+    from poetry.console.commands.command import Command
+    from poetry.utils.env.base_env import Env
+
+
+class MockApplication(Application):
+    """A mock application that stored the last command class that was executed
+
+    Not currently used in tests but could be useful in the future
+    """
+
+    def __init__(self):
+        super().__init__()
+        dispatcher = self.event_dispatcher
+        if dispatcher is not None:
+            dispatcher.add_listener(COMMAND, self.configure_command_spy)
+
+        self._last_cmd: Command = None  # type: ignore[assignment]
+
+    def configure_command_spy(self, event: Event, event_name: str, _: EventDispatcher) -> None:
+        """Store the last command class that was executed"""
+        if isinstance(event, ConsoleCommandEvent):
+            self._last_cmd = event.command  # type: ignore[assignment]
+
+
+class MockRepoManager:
+    """A helper class to generate test repositories"""
+
+    def __init__(self):
+        self._src_repos: dict[str, Path] = {"v1": None}  # type: ignore[assignment, dict-item]
+        self._preinstalled_repos: dict[str, Path] = {"v1": None}  # type: ignore[assignment, dict-item]
+
+        self._tmp_path_obj = tempfile.TemporaryDirectory()
+        self._tmp_path = Path(self._tmp_path_obj.name)
+
+        for repo in self._src_repos:
+            src = Path(__file__).parent / "fixtures" / repo
+            self._src_repos[repo] = src
+
+        for repo, src in self._src_repos.items():
+            dst = self._tmp_path / "preinstalled" / repo
+            dst.mkdir(parents=True, exist_ok=True)
+
+            shutil.copytree(src, dst, dirs_exist_ok=True)
+            _poetry_run(dst, None, "install")
+            self._preinstalled_repos[repo] = dst
+
+        weakref.finalize(self, self._tmp_path_obj.cleanup)
+
+    def get_repo(self, repo: str, preinstalled: bool = False):
+        """Create a test monorepo and return the path to it
+
+        If preinstalled is True, the monorepo will have its dependencies
+        preinstalled (i.e. poetry install will be executed)
+        """
+        src = self._preinstalled_repos[repo] if preinstalled else self._src_repos[repo]
+
+        dst = Path(tempfile.mkdtemp(prefix=f"{repo}_", dir=self._tmp_path))
+        shutil.copytree(src, dst, dirs_exist_ok=True)
+
+        if preinstalled:
+            _poetry_run(dst, None, "install")
+
+        return dst
+
+    @staticmethod
+    def get_envs(path: Path) -> EnvCollection:
+        """Get the root and package environments for a monorepo"""
+        copied_env = os.environ.copy()
+        copied_env.pop("VIRTUAL_ENV", None)
+        copied_env.pop("CONDA_PREFIX", None)
+        copied_env.pop("CONDA_DEFAULT_ENV", None)
+        with patch.dict(os.environ, copied_env, clear=True):
+            root_env: Env = EnvManager(Factory().create_poetry(cwd=path)).get()
+            envs: list[Env] = []
+            for pkg in path.iterdir():
+                if pkg.is_dir() and (pkg / "pyproject.toml").exists():
+                    envs.append(EnvManager(Factory().create_poetry(cwd=pkg)).get())
+
+        return EnvCollection(root_env, envs)
+
+
+@dataclass(frozen=True)
+class PoetryRunResult:
+    """The result of running a poetry command"""
+
+    cmd: str
+    cmd_obj: Command
+    run_dir: Path
+    stdout: str
+    stderr: str
+    exit_code: int
+
+
+@dataclass(frozen=True)
+class EnvCollection:
+    """A collection of environments for a monorepo"""
+
+    root_env: Env
+    pkg_envs: list[Env]
+
+
+@contextlib.contextmanager
+def new_cd(direc: str | Path):
+    """Context manager to temporarily change the current working directory"""
+    curr_dir = os.getcwd()
+    os.chdir(str(direc))
+    try:
+        yield
+    finally:
+        os.chdir(str(curr_dir))
+
+
+def _poetry_run(monorepo_root: Path, sub_project: str | None = None, cmd: str = "help"):
+    """Run a poetry command in a monorepo
+
+    Args:
+        monorepo_root (Path): The path to the monorepo
+        sub_project (str, optional): The subproject to run the command in. Defaults to None.
+        cmd (str, optional): The command to run. Defaults to "help".
+    """
+    pkg_dir = monorepo_root / sub_project if sub_project else monorepo_root
+    cmd = f"poetry {cmd}"
+
+    input_stream = io.StringIO()
+    input_obj = ArgvInput(cmd.split(" "))
+    input_obj.set_stream(input_stream)
+
+    output_stream = io.StringIO()
+    output_obj = StreamOutput(output_stream)
+
+    error_stream = io.StringIO()
+    error_obj = StreamOutput(error_stream)
+
+    copied_env = os.environ.copy()
+    copied_env.pop("VIRTUAL_ENV", None)
+    copied_env.pop("CONDA_PREFIX", None)
+    copied_env.pop("CONDA_DEFAULT_ENV", None)
+    with new_cd(pkg_dir), patch.dict(os.environ, copied_env, clear=True):
+        app = MockApplication()
+        app.auto_exits(False)
+        exit_code = app.run(input_obj, output_obj, error_obj)
+
+    return PoetryRunResult(
+        cmd=cmd,
+        cmd_obj=app._last_cmd,
+        run_dir=pkg_dir,
+        stdout=output_stream.getvalue(),
+        stderr=error_stream.getvalue(),
+        exit_code=exit_code,
+    )
+
+
+def is_system_env(env: Env) -> bool:
+    return isinstance(env, SystemEnv) or (hasattr(env, "_child_env") and isinstance(env._child_env, SystemEnv))