Add POE_PWD for original working directory and allow use of env vars …

…for cwd

poethepoet/context.py

@@ -1,4 +1,3 @@
-import os
 import re
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Tuple
@@ -106,15 +105,17 @@ def get_executor(
     ) -> "PoeExecutor":
         from .executor import PoeExecutor
 
-        cwd_option = task_options.get("cwd", ".")
-        if cwd_option == '$exec_cwd':
-            cwd_option = os.getcwd()
+        cwd_option = self.env.fill_template(task_options.get("cwd", "."))
+        working_dir = Path(cwd_option)
+
+        if not working_dir.is_absolute():
+            working_dir = self.project_dir / working_dir
 
         return PoeExecutor.get(
             invocation=invocation,
             context=self,
             env=env,
-            working_dir=self.project_dir / cwd_option,
+            working_dir=working_dir,
             dry=self.dry,
             executor_config=task_options.get("executor"),
             capture_stdout=task_options.get("capture_stdout", False),

poethepoet/env/manager.py

@@ -1,3 +1,4 @@
+import os
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union
 
@@ -51,6 +52,9 @@ def __init__(
 
         self._vars["POE_ROOT"] = str(self._config.project_dir)
 
+        if "POE_PWD" not in self._vars:
+            self._vars["POE_PWD"] = os.getcwd()
+
     def get(self, key: str, default: Optional[str] = None) -> Optional[str]:
         return self._vars.get(key, default)
 

tests/conftest.py

@@ -141,6 +141,7 @@ def run_poe_subproc(
 
         subproc_env = dict(os.environ)
         subproc_env.pop("VIRTUAL_ENV", None)
+        subproc_env.pop("POE_PWD", None)  # do not inherit this from the test
         if env:
             subproc_env.update(env)
 

tests/fixtures/cwd_project/pyproject.toml

@@ -1,3 +1,11 @@
 [tool.poe.tasks.cwd]
 cmd = "poe_test_pwd"
 cwd = "./subdir/foo"
+
+[tool.poe.tasks.cwd_env]
+cmd = "poe_test_pwd"
+cwd = "./subdir/${BAR_ENV}"
+
+[tool.poe.tasks.cwd_poe_pwd]
+cmd = "poe_test_pwd"
+cwd = "${POE_PWD}"

tests/fixtures/monorepo_project/pyproject.toml

@@ -4,6 +4,8 @@ path = "subproject_1/pyproject.toml"
 [[tool.poe.include]]
 path = "subproject_2/pyproject.toml"
 cwd  = "subproject_2"
+[[tool.poe.include]]
+path = "subproject_3/pyproject.toml"
 
 
 [tool.poe.tasks.get_cwd_0]

tests/fixtures/monorepo_project/subproject_3/pyproject.toml

@@ -0,0 +1,7 @@
+
+
+
+[tool.poe.tasks.get_cwd_3]
+interpreter = "python"
+shell       = "import os; print(os.getcwd())"
+cwd         = "${POE_PWD}"

tests/test_cmd_tasks.py

@@ -1,3 +1,6 @@
+import os
+
+
 def test_call_echo_task(run_poe_subproc, projects, esc_prefix):
     result = run_poe_subproc("echo", "foo", "!", project="cmds")
     assert (
@@ -63,3 +66,59 @@ def test_cmd_task_with_cwd_option(run_poe_subproc, poe_project_path):
         == f'{poe_project_path.joinpath("tests", "fixtures", "cwd_project", "subdir", "foo")}\n'
     )
     assert result.stderr == ""
+
+
+def test_cmd_task_with_cwd_option_env(run_poe_subproc, poe_project_path):
+    result = run_poe_subproc("cwd_env", project="cwd", env={"BAR_ENV": "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 == ""
+
+
+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)
+
+
+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": 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)

tests/test_includes.py

@@ -1,3 +1,6 @@
+import os
+
+
 def test_docs_for_include_toml_file(run_poe_subproc):
     result = run_poe_subproc(project="includes")
     assert (
@@ -88,7 +91,8 @@ def test_monorepo_contains_only_expected_tasks(run_poe_subproc, projects):
         "CONFIGURED TASKS\n"
         "  get_cwd_0      \n"
         "  get_cwd_1      \n"
-        "  get_cwd_2      \n\n\n"
+        "  get_cwd_2      \n"
+        "  get_cwd_3      \n\n\n"
     )
     assert result.stdout == ""
     assert result.stderr == ""
@@ -148,3 +152,18 @@ def test_monorepo_runs_each_task_with_expected_cwd(
             "poethepoet/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)