Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 34 additions & 6 deletions tests/tools/test_base_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand All @@ -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
Expand Down
11 changes: 8 additions & 3 deletions tools/environments/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
# ------------------------------------------------------------------
Expand Down Expand Up @@ -335,13 +340,15 @@ 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"
f"alias -p >> {self._snapshot_path}\n"
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"
)
Expand Down Expand Up @@ -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
Expand Down