Assign POE_PWD from passed-down cwd parameter

poethepoet/app.py

@@ -113,9 +113,9 @@ def __call__(self, cli_args: Sequence[str], internal: bool = False) -> int:
             return 1
 
         if task.has_deps():
-            return self.run_task_graph(task) or 0
+            return self.run_task_graph(task, cwd=self.cwd) or 0
         else:
-            return self.run_task(task) or 0
+            return self.run_task(task, cwd=self.cwd) or 0
 
     def resolve_task(self, allow_hidden: bool = False) -> Optional["PoeTask"]:
         from .task import PoeTask
@@ -143,10 +143,13 @@ def resolve_task(self, allow_hidden: bool = False) -> Optional["PoeTask"]:
         )
 
     def run_task(
-        self, task: "PoeTask", context: Optional["RunContext"] = None
+        self,
+        task: "PoeTask",
+        context: Optional["RunContext"] = None,
+        cwd: Optional[Union[Path, str]] = None,
     ) -> Optional[int]:
         if context is None:
-            context = self.get_run_context()
+            context = self.get_run_context(cwd=cwd)
         try:
             return task.run(context=context, extra_args=task.invocation[1:])
         except PoeException as error:
@@ -156,18 +159,22 @@ def run_task(
             self.ui.print_error(error=error)
             return 1
 
-    def run_task_graph(self, task: "PoeTask") -> Optional[int]:
+    def run_task_graph(
+        self,
+        task: "PoeTask",
+        cwd: Optional[Union[Path, str]] = None,
+    ) -> Optional[int]:
         from .task.graph import TaskExecutionGraph
 
-        context = self.get_run_context(multistage=True)
+        context = self.get_run_context(multistage=True, cwd=cwd)
         graph = TaskExecutionGraph(task, context)
         plan = graph.get_execution_plan()
 
         for stage in plan:
             for stage_task in stage:
                 if stage_task == task:
                     # The final sink task gets special treatment
-                    return self.run_task(stage_task, context)
+                    return self.run_task(stage_task, context, cwd=cwd)
 
                 try:
                     task_result = stage_task.run(
@@ -185,7 +192,11 @@ def run_task_graph(self, task: "PoeTask") -> Optional[int]:
                     return 1
         return 0
 
-    def get_run_context(self, multistage: bool = False) -> "RunContext":
+    def get_run_context(
+        self,
+        multistage: bool = False,
+        cwd: Optional[Union[Path, str]] = None,
+    ) -> "RunContext":
         from .context import RunContext
 
         result = RunContext(
@@ -195,6 +206,7 @@ def get_run_context(self, multistage: bool = False) -> "RunContext":
             dry=self.ui["dry_run"],
             poe_active=os.environ.get("POE_ACTIVE"),
             multistage=multistage,
+            cwd=cwd,
         )
         if self._poetry_env_path:
             # This allows the PoetryExecutor to use the venv from poetry directly

poethepoet/context.py

@@ -1,6 +1,6 @@
 import re
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Tuple, Union
 
 if TYPE_CHECKING:
     from .config import PoeConfig
@@ -28,6 +28,7 @@ def __init__(
         dry: bool,
         poe_active: Optional[str],
         multistage: bool = False,
+        cwd: Optional[Union[Path, str]] = None,
     ):
         from .env.manager import EnvVarsManager
 
@@ -39,7 +40,7 @@ def __init__(
         self.multistage = multistage
         self.exec_cache = {}
         self.captured_stdout = {}
-        self.env = EnvVarsManager(self.config, self.ui, base_env=env)
+        self.env = EnvVarsManager(self.config, self.ui, base_env=env, cwd=cwd)
 
     @property
     def executor_type(self) -> Optional[str]:

poethepoet/env/manager.py

@@ -1,4 +1,3 @@
-import os
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union
 
@@ -22,6 +21,7 @@ def __init__(
         ui: Optional["PoeUi"],
         parent_env: Optional["EnvVarsManager"] = None,
         base_env: Optional[Mapping[str, str]] = None,
+        cwd: Optional[Union[Path, str]] = None,
     ):
         from .cache import EnvFileCache
 
@@ -53,7 +53,7 @@ def __init__(
         self._vars["POE_ROOT"] = str(self._config.project_dir)
 
         if "POE_PWD" not in self._vars:
-            self._vars["POE_PWD"] = os.getcwd()
+            self._vars["POE_PWD"] = str(cwd)
 
     def get(self, key: str, default: Optional[str] = None) -> Optional[str]:
         return self._vars.get(key, default)
@@ -80,12 +80,15 @@ def _apply_env_config(
             )
 
     def for_task(
-        self, task_envfile: Optional[str], task_env: Optional[Mapping[str, str]]
+        self,
+        task_envfile: Optional[str],
+        task_env: Optional[Mapping[str, str]],
+        cwd: Optional[Union[Path, str]] = None,
     ) -> "EnvVarsManager":
         """
         Create a copy of self and extend it to include vars for the task.
         """
-        result = EnvVarsManager(self._config, self._ui, parent_env=self)
+        result = EnvVarsManager(self._config, self._ui, parent_env=self, cwd=cwd)
 
         # Include env vars from envfile(s) referenced in task options
         if isinstance(task_envfile, str):

tests/conftest.py

@@ -115,13 +115,14 @@ def run_poe_subproc(projects, temp_file, tmp_path, is_windows):
 
     def run_poe_subproc(
         *run_args: str,
-        cwd: str = projects["example"],
+        cwd: Optional[str] = None,
         config: Optional[Mapping[str, Any]] = None,
         coverage: bool = not is_windows,
         env: Dict[str, str] = None,
         project: Optional[str] = None,
     ) -> PoeRunResult:
-        cwd = projects.get(project, cwd)
+        if cwd is None:
+            cwd = projects.get(project, projects["example"])
         if config is not None:
             config_path = tmp_path.joinpath("tmp_test_config_file")
             with config_path.open("w+") as config_file:

tests/test_cmd_tasks.py

@@ -79,51 +79,42 @@ def test_cmd_task_with_cwd_option_env(run_poe_subproc, poe_project_path):
 
 
 def test_cmd_task_with_cwd_option_pwd(run_poe_subproc, poe_project_path):
-    prev_cwd = os.getcwd()
-    try:
-        os.chdir(
-            poe_project_path.joinpath(
-                "tests", "fixtures", "cwd_project", "subdir", "foo"
-            )
-        )
-        result = run_poe_subproc("cwd_poe_pwd", project="cwd")
-        assert result.capture == "Poe => poe_test_pwd\n"
-        assert (
-            result.stdout
-            == f'{poe_project_path.joinpath("tests", "fixtures", "cwd_project", "subdir", "foo")}\n'
-        )
-        assert result.stderr == ""
-    finally:
-        os.chdir(prev_cwd)
+    result = run_poe_subproc(
+        "cwd_poe_pwd",
+        project="cwd",
+        cwd=poe_project_path.joinpath(
+            "tests", "fixtures", "cwd_project", "subdir", "foo"
+        ),
+    )
+    assert result.capture == "Poe => poe_test_pwd\n"
+    assert (
+        result.stdout
+        == f'{poe_project_path.joinpath("tests", "fixtures", "cwd_project", "subdir", "foo")}\n'
+    )
+    assert result.stderr == ""
 
 
 def test_cmd_task_with_cwd_option_pwd_override(run_poe_subproc, poe_project_path):
-    prev_cwd = os.getcwd()
-    try:
-        os.chdir(
-            poe_project_path.joinpath(
-                "tests", "fixtures", "cwd_project", "subdir", "foo"
-            )
-        )
-        result = run_poe_subproc(
-            "cwd_poe_pwd",
-            project="cwd",
-            env={
-                "POE_PWD": str(
-                    poe_project_path.joinpath(
-                        "tests", "fixtures", "cwd_project", "subdir", "bar"
-                    )
+    result = run_poe_subproc(
+        "cwd_poe_pwd",
+        project="cwd",
+        env={
+            "POE_PWD": str(
+                poe_project_path.joinpath(
+                    "tests", "fixtures", "cwd_project", "subdir", "bar"
                 )
-            },
-        )
-        assert result.capture == "Poe => poe_test_pwd\n"
-        assert (
-            result.stdout
-            == f'{poe_project_path.joinpath("tests", "fixtures", "cwd_project", "subdir", "bar")}\n'
-        )
-        assert result.stderr == ""
-    finally:
-        os.chdir(prev_cwd)
+            )
+        },
+        cwd=poe_project_path.joinpath(
+            "tests", "fixtures", "cwd_project", "subdir", "foo"
+        ),
+    )
+    assert result.capture == "Poe => poe_test_pwd\n"
+    assert (
+        result.stdout
+        == f'{poe_project_path.joinpath("tests", "fixtures", "cwd_project", "subdir", "bar")}\n'
+    )
+    assert result.stderr == ""
 
 
 def test_cmd_task_with_cwd_option_arg(run_poe_subproc, poe_project_path):

tests/test_includes.py

@@ -158,17 +158,10 @@ def test_monorepo_runs_each_task_with_expected_cwd(
         assert result.stdout.endswith("/tests/fixtures/monorepo_project/subproject_2\n")
     assert result.stderr == ""
 
-    prev_cwd = os.getcwd()
-    try:
-        os.chdir(projects["example"])
-        result = run_poe_subproc("get_cwd_3", project="monorepo")
-        assert result.capture == "Poe => import os; print(os.getcwd())\n"
-        if is_windows:
-            assert result.stdout.endswith(
-                "poethepoet\\tests\\fixtures\\example_project\n"
-            )
-        else:
-            assert result.stdout.endswith("poethepoet/tests/fixtures/example_project\n")
-        assert result.stderr == ""
-    finally:
-        os.chdir(prev_cwd)
+    result = run_poe_subproc("get_cwd_3", project="monorepo", cwd=projects["example"])
+    assert result.capture == "Poe => import os; print(os.getcwd())\n"
+    if is_windows:
+        assert result.stdout.endswith("poethepoet\\tests\\fixtures\\example_project\n")
+    else:
+        assert result.stdout.endswith("poethepoet/tests/fixtures/example_project\n")
+    assert result.stderr == ""