common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix: update tests for unified spawn-per-call execution model

Browse Source 
- Docker env tests: verify _build_init_env_args() instead of per-execute
  Popen flags (env forwarding is now init-time only)
- Docker: preserve explicit forward_env bypass of blocklist from main
- Daytona tests: adapt to SDK-native timeout, _ThreadedProcessHandle,
  base.py interrupt handling, HERMES_STDIN_ heredoc prefix
- Modal tests: fix _load_module to include _ThreadedProcessHandle stub,
  check ensurepip in _resolve_modal_image instead of __init__
- SSH tests: mock time.sleep on base module instead of removed ssh import
- Add missing BaseEnvironment attributes to __new__()-based test fixtures
This commit is contained in:
Teknium
2026-04-08 17:17:41 -07:00
committed by Teknium
parent d684d7ee7e
commit e19252afc4
5 changed files with 120 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
tests/tools/test_daytona_environment.py
View File

@@ -59,8 +59,8 @@ def daytona_sdk(monkeypatch):
@pytest.fixture()
def make_env(daytona_sdk, monkeypatch):
"""Factory that creates a DaytonaEnvironment with a mocked SDK."""
# Prevent is_interrupted from interfering
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
# Prevent is_interrupted from interfering — patch where it's used (base.py)
monkeypatch.setattr("tools.environments.base.is_interrupted", lambda: False)
# Prevent skills/credential sync from consuming mock exec calls
monkeypatch.setattr("tools.credential_files.get_credential_file_mounts", lambda: [])
monkeypatch.setattr("tools.credential_files.get_skills_directory_mount", lambda **kw: None)
@@ -221,41 +221,45 @@ class TestCleanup:
class TestExecute:
def test_basic_command(self, make_env):
sb = _make_sandbox()
# First call: $HOME detection; subsequent calls: actual commands
# Calls: (1) $HOME detection, (2) init_session bootstrap, (3) actual command
sb.process.exec.side_effect = [
_make_exec_response(result="/root"), # $HOME
_make_exec_response(result="", exit_code=0), # init_session
_make_exec_response(result="hello", exit_code=0), # actual cmd
]
sb.state = "started"
env = make_env(sandbox=sb)
result = env.execute("echo hello")
assert result["output"] == "hello"
assert "hello" in result["output"]
assert result["returncode"] == 0
def test_command_wrapped_with_shell_timeout(self, make_env):
def test_sdk_timeout_passed_to_exec(self, make_env):
"""SDK native timeout is passed to sandbox.process.exec()."""
sb = _make_sandbox()
sb.process.exec.side_effect = [
_make_exec_response(result="/root"),
_make_exec_response(result="", exit_code=0), # init_session
_make_exec_response(result="ok", exit_code=0),
]
sb.state = "started"
env = make_env(sandbox=sb, timeout=42)
env.execute("echo hello")
# The command sent to exec should be wrapped with `timeout N sh -c '...'`
# The exec call should receive timeout= kwarg (SDK native timeout)
call_args = sb.process.exec.call_args_list[-1]
assert call_args[1]["timeout"] == 42
# The command should NOT have a shell `timeout` prefix
cmd = call_args[0][0]
assert cmd.startswith("timeout 42 sh -c ")
# SDK timeout param should NOT be passed
assert "timeout" not in call_args[1]
assert not cmd.startswith("timeout ")
def test_timeout_returns_exit_code_124(self, make_env):
"""Shell timeout utility returns exit code 124."""
"""SDK-level timeout surfaces as exit code 124 via _wait_for_process."""
sb = _make_sandbox()
sb.process.exec.side_effect = [
_make_exec_response(result="/root"),
_make_exec_response(result="", exit_code=124),
_make_exec_response(result="", exit_code=0), # init_session
_make_exec_response(result="", exit_code=124), # actual cmd
]
sb.state = "started"
env = make_env(sandbox=sb)
@@ -267,6 +271,7 @@ class TestExecute:
sb = _make_sandbox()
sb.process.exec.side_effect = [
_make_exec_response(result="/root"),
_make_exec_response(result="", exit_code=0), # init_session
_make_exec_response(result="not found", exit_code=127),
]
sb.state = "started"
@@ -279,6 +284,7 @@ class TestExecute:
sb = _make_sandbox()
sb.process.exec.side_effect = [
_make_exec_response(result="/root"),
_make_exec_response(result="", exit_code=0), # init_session
_make_exec_response(result="ok", exit_code=0),
]
sb.state = "started"
@@ -286,39 +292,47 @@ class TestExecute:
env.execute("python3", stdin_data="print('hi')")
# Check that the command passed to exec contains heredoc markers
# (single quotes get shell-escaped by shlex.quote, so check components)
# Base class uses HERMES_STDIN_ prefix for heredoc delimiters
call_args = sb.process.exec.call_args_list[-1]
cmd = call_args[0][0]
assert "HERMES_EOF_" in cmd
assert "HERMES_STDIN_" in cmd
assert "print" in cmd
assert "hi" in cmd
def test_custom_cwd_passed_through(self, make_env):
def test_custom_cwd_in_command_wrapper(self, make_env):
"""CWD is handled by _wrap_command() in the command string, not as a kwarg."""
sb = _make_sandbox()
sb.process.exec.side_effect = [
_make_exec_response(result="/root"),
_make_exec_response(result="", exit_code=0), # init_session
_make_exec_response(result="/tmp", exit_code=0),
]
sb.state = "started"
env = make_env(sandbox=sb)
env.execute("pwd", cwd="/tmp")
call_kwargs = sb.process.exec.call_args_list[-1][1]
assert call_kwargs["cwd"] == "/tmp"
# CWD should be embedded in the command string via _wrap_command
call_args = sb.process.exec.call_args_list[-1]
cmd = call_args[0][0]
assert "cd /tmp" in cmd
# CWD should NOT be passed as a kwarg to exec
assert "cwd" not in call_args[1]
def test_daytona_error_triggers_retry(self, make_env, daytona_sdk):
sb = _make_sandbox()
sb.state = "started"
sb.process.exec.side_effect = [
_make_exec_response(result="/root"), # $HOME
_make_exec_response(result="", exit_code=0), # init_session
daytona_sdk.DaytonaError("transient"), # first attempt fails
_make_exec_response(result="ok", exit_code=0), # retry succeeds
]
env = make_env(sandbox=sb)
result = env.execute("echo retry")
assert result["output"] == "ok"
assert result["returncode"] == 0
# DaytonaError now surfaces directly through _ThreadedProcessHandle
# (no retry logic) — the error becomes returncode=1
assert result["returncode"] == 1
# ---------------------------------------------------------------------------
@@ -359,14 +373,18 @@ class TestInterrupt:
calls["n"] += 1
if calls["n"] == 1:
return _make_exec_response(result="/root") # $HOME detection
if calls["n"] == 2:
return _make_exec_response(result="", exit_code=0) # init_session
event.wait(timeout=5) # simulate long-running command
return _make_exec_response(result="done", exit_code=0)
sb.process.exec.side_effect = exec_side_effect
env = make_env(sandbox=sb)
# is_interrupted is checked by base.py's _wait_for_process,
# patch where it's actually referenced (base.py's local binding)
monkeypatch.setattr(
"tools.environments.daytona.is_interrupted", lambda: True
"tools.environments.base.is_interrupted", lambda: True
)
try:
result = env.execute("sleep 10")
@@ -377,23 +395,24 @@ class TestInterrupt:
# ---------------------------------------------------------------------------
# Retry exhaustion
# DaytonaError surfaces directly (no retry)
# ---------------------------------------------------------------------------
class TestRetryExhausted:
def test_both_attempts_fail(self, make_env, daytona_sdk):
"""DaytonaError surfaces directly as rc=1 (retry logic was removed)."""
sb = _make_sandbox()
sb.state = "started"
sb.process.exec.side_effect = [
_make_exec_response(result="/root"), # $HOME
daytona_sdk.DaytonaError("fail1"), # first attempt
daytona_sdk.DaytonaError("fail2"), # retry
_make_exec_response(result="", exit_code=0), # init_session
daytona_sdk.DaytonaError("fail1"), # actual command fails
]
env = make_env(sandbox=sb)
result = env.execute("echo x")
# Error surfaces directly through _ThreadedProcessHandle (rc=1)
assert result["returncode"] == 1
assert "Daytona execution error" in result["output"]
# ---------------------------------------------------------------------------