diff --git a/tests/tools/test_base_environment.py b/tests/tools/test_base_environment.py index 913ad0387c54..0368fb6cdaa2 100644 --- a/tests/tools/test_base_environment.py +++ b/tests/tools/test_base_environment.py @@ -130,6 +130,32 @@ def test_unique_delimiter_each_call(self): class TestInitSessionFailure: + def test_init_session_bootstrap_changes_to_configured_cwd(self): + env = _TestableEnv(cwd="/tmp/configured") + + calls = [] + + def mock_run_bash(cmd, *, login=False, timeout=120, stdin_data=None): + calls.append({"cmd": cmd, "login": login}) + return MagicMock() + + env._run_bash = mock_run_bash + env._wait_for_process = MagicMock( + return_value={ + "output": f"{env._cwd_marker}/tmp/configured{env._cwd_marker}", + "returncode": 0, + } + ) + env.init_session() + + assert len(calls) == 1 + assert calls[0]["login"] is True + assert ( + "cd /tmp/configured" in calls[0]["cmd"] + or "cd '/tmp/configured'" in calls[0]["cmd"] + ) + assert env._snapshot_ready is True + def test_snapshot_ready_false_on_failure(self): env = _TestableEnv() @@ -147,16 +173,18 @@ def test_login_flag_when_snapshot_not_ready(self): env._snapshot_ready = False calls = [] + def mock_run_bash(cmd, *, login=False, timeout=120, stdin_data=None): calls.append({"login": login}) - # Return a mock process handle - mock = MagicMock() - mock.poll.return_value = 0 - mock.returncode = 0 - mock.stdout = iter([]) - return mock + return MagicMock() env._run_bash = mock_run_bash + env._wait_for_process = MagicMock( + return_value={ + "output": f"{env._cwd_marker}/tmp{env._cwd_marker}", + "returncode": 0, + } + ) env.execute("echo test") assert len(calls) == 1 diff --git a/tools/environments/base.py b/tools/environments/base.py index d89b66f19de9..92ecc1e92eb9 100644 --- a/tools/environments/base.py +++ b/tools/environments/base.py @@ -299,6 +299,11 @@ def __init__(self, cwd: str, timeout: int, env: dict = None): self._cwd_marker = _cwd_marker(self._session_id) self._snapshot_ready = False + @staticmethod + def _quote_cwd_for_shell(cwd: str) -> str: + """Quote a cwd for shell ``cd`` while preserving ``~`` expansion.""" + return shlex.quote(cwd) if cwd != "~" and not cwd.startswith("~/") else cwd + # ------------------------------------------------------------------ # Abstract methods # ------------------------------------------------------------------ @@ -335,6 +340,7 @@ def init_session(self): instead of running with ``bash -l``. """ # Full capture: env vars, functions (filtered), aliases, shell options. + quoted_cwd = self._quote_cwd_for_shell(self.cwd) bootstrap = ( f"export -p > {self._snapshot_path}\n" f"declare -f | grep -vE '^_[^_]' >> {self._snapshot_path}\n" @@ -342,6 +348,7 @@ def init_session(self): f"echo 'shopt -s expand_aliases' >> {self._snapshot_path}\n" f"echo 'set +e' >> {self._snapshot_path}\n" f"echo 'set +u' >> {self._snapshot_path}\n" + f"cd {quoted_cwd} || exit 126\n" f"pwd -P > {self._cwd_file} 2>/dev/null || true\n" f"printf '\\n{self._cwd_marker}%s{self._cwd_marker}\\n' \"$(pwd -P)\"\n" ) @@ -380,9 +387,7 @@ def _wrap_command(self, command: str, cwd: str) -> str: parts.append(f"source {self._snapshot_path} 2>/dev/null || true") # cd to working directory — let bash expand ~ natively - quoted_cwd = ( - shlex.quote(cwd) if cwd != "~" and not cwd.startswith("~/") else cwd - ) + quoted_cwd = self._quote_cwd_for_shell(cwd) parts.append(f"builtin cd {quoted_cwd} || exit 126") # Run the actual command