X-iZhang
c57e6132a5
- Introduced a new test file `test_rich_escape.py` to validate the safety of Rich markup escaping in the ToolResultFormatter. - Added tests to ensure that tool names and error messages containing brackets do not cause crashes during formatting.
370 lines
14 KiB
Python
370 lines
14 KiB
Python
"""Tests for EvoScientist/backends.py — validate_command, path conversion, resolve_path."""
|
|
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from EvoScientist.backends import (
|
|
validate_command,
|
|
convert_virtual_paths_in_command,
|
|
CustomSandboxBackend,
|
|
)
|
|
|
|
|
|
# === validate_command ===
|
|
|
|
class TestValidateCommand:
|
|
def test_safe_ls(self):
|
|
assert validate_command("ls -la") is None
|
|
|
|
def test_safe_python(self):
|
|
assert validate_command("python script.py") is None
|
|
|
|
def test_safe_pip(self):
|
|
assert validate_command("pip install pandas") is None
|
|
|
|
def test_blocked_traversal(self):
|
|
result = validate_command("cat ../../../etc/passwd")
|
|
assert result is not None
|
|
assert "blocked" in result.lower()
|
|
|
|
def test_blocked_sudo(self):
|
|
result = validate_command("sudo rm -rf /")
|
|
assert result is not None
|
|
assert "blocked" in result.lower()
|
|
|
|
def test_blocked_chmod(self):
|
|
result = validate_command("chmod 777 file.py")
|
|
assert result is not None
|
|
|
|
def test_blocked_dd(self):
|
|
result = validate_command("dd if=/dev/zero of=file bs=1M count=100")
|
|
assert result is not None
|
|
|
|
def test_blocked_home_tilde(self):
|
|
result = validate_command("cat ~/secrets.txt")
|
|
assert result is not None
|
|
|
|
def test_blocked_rm_rf_absolute(self):
|
|
result = validate_command("rm -rf /important")
|
|
assert result is not None
|
|
|
|
def test_blocked_cd_absolute(self):
|
|
result = validate_command("cd /etc && cat passwd")
|
|
assert result is not None
|
|
|
|
def test_safe_echo(self):
|
|
assert validate_command("echo hello world") is None
|
|
|
|
def test_safe_grep(self):
|
|
assert validate_command("grep -r 'pattern' .") is None
|
|
|
|
|
|
# === convert_virtual_paths_in_command ===
|
|
|
|
class TestConvertVirtualPaths:
|
|
def test_absolute_to_relative(self):
|
|
result = convert_virtual_paths_in_command("python /main.py")
|
|
assert result == "python ./main.py"
|
|
|
|
def test_nested_path(self):
|
|
result = convert_virtual_paths_in_command("cat /data/file.txt")
|
|
assert result == "cat ./data/file.txt"
|
|
|
|
def test_root_only(self):
|
|
result = convert_virtual_paths_in_command("ls /")
|
|
assert result == "ls ."
|
|
|
|
def test_no_change_relative(self):
|
|
result = convert_virtual_paths_in_command("python main.py")
|
|
assert result == "python main.py"
|
|
|
|
def test_url_preserved(self):
|
|
result = convert_virtual_paths_in_command("curl https://example.com/path")
|
|
# URLs should not be converted
|
|
assert "https://example.com/path" in result
|
|
|
|
def test_no_op_no_paths(self):
|
|
result = convert_virtual_paths_in_command("echo hello")
|
|
assert result == "echo hello"
|
|
|
|
def test_system_path_with_workspace_converted(self):
|
|
"""Hallucinated system path containing workspace dir should be fixed."""
|
|
result = convert_virtual_paths_in_command(
|
|
"mkdir -p /Users/user/project/workspace/swarm-discussion",
|
|
workspace_name="workspace",
|
|
)
|
|
assert result == "mkdir -p ./swarm-discussion"
|
|
|
|
def test_system_path_with_workspace_nested(self):
|
|
result = convert_virtual_paths_in_command(
|
|
"python /home/user/workspace/src/main.py",
|
|
workspace_name="workspace",
|
|
)
|
|
assert result == "python ./src/main.py"
|
|
|
|
def test_system_path_workspace_only(self):
|
|
result = convert_virtual_paths_in_command(
|
|
"ls /Users/user/Downloads/project/workspace",
|
|
workspace_name="workspace",
|
|
)
|
|
assert result == "ls ."
|
|
|
|
def test_system_path_with_shell_expansion(self):
|
|
"""Paths with $(whoami) or similar should still be caught."""
|
|
result = convert_virtual_paths_in_command(
|
|
"mkdir -p /Users/$(whoami)/workspace/notes",
|
|
workspace_name="workspace",
|
|
)
|
|
assert result == "mkdir -p ./notes"
|
|
|
|
def test_system_path_custom_workspace_name(self):
|
|
"""Should work with any workspace directory name, not just 'workspace'."""
|
|
result = convert_virtual_paths_in_command(
|
|
"mkdir -p /Users/user/my-project/data",
|
|
workspace_name="my-project",
|
|
)
|
|
assert result == "mkdir -p ./data"
|
|
|
|
def test_system_path_custom_workspace_name_only(self):
|
|
result = convert_virtual_paths_in_command(
|
|
"ls /home/user/experiment-1",
|
|
workspace_name="experiment-1",
|
|
)
|
|
assert result == "ls ."
|
|
|
|
def test_system_path_no_workspace_name_fallthrough(self):
|
|
"""Without workspace_name, system paths get normal ./ treatment."""
|
|
result = convert_virtual_paths_in_command(
|
|
"cat /Users/user/workspace/file.txt",
|
|
workspace_name=None,
|
|
)
|
|
assert result == "cat ./Users/user/workspace/file.txt"
|
|
|
|
def test_system_path_without_workspace_unchanged(self):
|
|
"""System paths not referencing workspace fall through to normal ./"""
|
|
result = convert_virtual_paths_in_command(
|
|
"cat /tmp/somefile", workspace_name="workspace",
|
|
)
|
|
assert result == "cat ./tmp/somefile"
|
|
|
|
|
|
# === CustomSandboxBackend._resolve_path ===
|
|
|
|
class TestResolvePath:
|
|
def test_strip_workspace_prefix(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
# /workspace/main.py should resolve to root/main.py
|
|
resolved = backend._resolve_path("/workspace/main.py")
|
|
assert str(resolved).endswith("main.py")
|
|
assert "workspace/workspace" not in str(resolved)
|
|
|
|
def test_workspace_root(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
resolved = backend._resolve_path("/workspace")
|
|
# Should resolve to root dir
|
|
assert resolved == backend._resolve_path("/")
|
|
|
|
def test_system_path_with_workspace_marker(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
resolved = backend._resolve_path("/Users/someone/project/workspace/main.py")
|
|
assert str(resolved).endswith("main.py")
|
|
|
|
def test_system_path_without_workspace(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
resolved = backend._resolve_path("/Users/someone/file.py")
|
|
# Falls back to basename
|
|
assert str(resolved).endswith("file.py")
|
|
|
|
def test_custom_workspace_name_prefix_stripped(self, tmp_path):
|
|
"""_resolve_path uses the actual dir name, not hardcoded 'workspace'."""
|
|
ws = tmp_path / "my-project"
|
|
ws.mkdir()
|
|
backend = CustomSandboxBackend(root_dir=str(ws), virtual_mode=True)
|
|
resolved = backend._resolve_path("/my-project/main.py")
|
|
assert str(resolved).endswith("main.py")
|
|
assert "my-project/my-project" not in str(resolved)
|
|
|
|
def test_custom_workspace_name_system_path(self, tmp_path):
|
|
ws = tmp_path / "experiment-1"
|
|
ws.mkdir()
|
|
backend = CustomSandboxBackend(root_dir=str(ws), virtual_mode=True)
|
|
resolved = backend._resolve_path("/Users/someone/experiment-1/data/out.csv")
|
|
assert str(resolved).endswith("data/out.csv")
|
|
|
|
def test_normal_virtual_path(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
resolved = backend._resolve_path("/src/main.py")
|
|
assert str(resolved).endswith("src/main.py")
|
|
|
|
|
|
# === CustomSandboxBackend.id ===
|
|
|
|
class TestSandboxId:
|
|
def test_sandbox_has_id(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
assert isinstance(backend.id, str)
|
|
assert backend.id.startswith("evosci-")
|
|
assert len(backend.id) == len("evosci-") + 8
|
|
|
|
def test_sandbox_id_is_stable(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
assert backend.id == backend.id # same instance → same id
|
|
|
|
def test_sandbox_id_unique(self, tmp_workspace):
|
|
b1 = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
b2 = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
assert b1.id != b2.id
|
|
|
|
def test_sandbox_id_hex_suffix(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
suffix = backend.id[len("evosci-"):]
|
|
assert re.fullmatch(r"[0-9a-f]{8}", suffix)
|
|
|
|
|
|
# === execute() literal cwd sanitization ===
|
|
|
|
class TestExecuteCwdSanitization:
|
|
def test_literal_workspace_path_replaced(self, tmp_workspace):
|
|
"""execute() should replace literal workspace root path with ./"""
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
# Create a subdir via the sanitized path
|
|
resp = backend.execute(f"mkdir -p {tmp_workspace}/test-sanitized && echo ok")
|
|
assert resp.exit_code == 0
|
|
# The dir should be created at workspace/test-sanitized, not nested
|
|
assert (Path(tmp_workspace) / "test-sanitized").is_dir()
|
|
assert not (Path(tmp_workspace) / tmp_workspace.lstrip("/")).exists()
|
|
|
|
|
|
# === execute() output truncation ===
|
|
|
|
class TestExecuteTruncation:
|
|
def test_execute_truncates_large_output(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(
|
|
root_dir=tmp_workspace,
|
|
virtual_mode=True, max_output_bytes=100,
|
|
)
|
|
# Generate output larger than 100 bytes
|
|
resp = backend.execute("python3 -c \"print('A' * 200)\"")
|
|
assert resp.truncated is True
|
|
assert "... Output truncated at 100 bytes" in resp.output
|
|
# Output body (before truncation message) should be ≤ 100 bytes
|
|
before_marker = resp.output.split("\n\n... Output truncated")[0]
|
|
assert len(before_marker) <= 100
|
|
|
|
def test_execute_no_truncation_small_output(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(
|
|
root_dir=tmp_workspace,
|
|
virtual_mode=True, max_output_bytes=100_000,
|
|
)
|
|
resp = backend.execute("echo hello")
|
|
assert resp.truncated is False
|
|
assert "truncated" not in resp.output.lower()
|
|
|
|
|
|
# === execute() stderr attribution ===
|
|
|
|
class TestExecuteStderr:
|
|
def test_execute_stderr_attribution(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(
|
|
root_dir=tmp_workspace, virtual_mode=True,
|
|
)
|
|
resp = backend.execute("python3 -c \"import sys; sys.stderr.write('warning\\n')\"")
|
|
assert "[stderr] warning" in resp.output
|
|
|
|
def test_execute_nonzero_exit_code_in_output(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(
|
|
root_dir=tmp_workspace, virtual_mode=True,
|
|
)
|
|
resp = backend.execute("python3 -c \"raise SystemExit(42)\"")
|
|
assert resp.exit_code == 42
|
|
assert "Exit code: 42" in resp.output
|
|
|
|
def test_execute_mixed_stdout_stderr(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(
|
|
root_dir=tmp_workspace, virtual_mode=True,
|
|
)
|
|
resp = backend.execute(
|
|
"python3 -c \"import sys; print('out'); sys.stderr.write('err\\n')\""
|
|
)
|
|
assert "out" in resp.output
|
|
assert "[stderr] err" in resp.output
|
|
|
|
def test_execute_success_no_exit_code(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(
|
|
root_dir=tmp_workspace, virtual_mode=True,
|
|
)
|
|
resp = backend.execute("echo ok")
|
|
assert resp.exit_code == 0
|
|
assert "Exit code:" not in resp.output
|
|
|
|
|
|
# === execute() timeout kwarg ===
|
|
|
|
class TestExecuteTimeout:
|
|
def test_execute_accepts_timeout_kwarg(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
resp = backend.execute("echo hello", timeout=60)
|
|
assert resp.exit_code == 0
|
|
assert "hello" in resp.output
|
|
|
|
def test_execute_timeout_none_uses_default(self, tmp_workspace):
|
|
backend = CustomSandboxBackend(root_dir=tmp_workspace, virtual_mode=True)
|
|
resp = backend.execute("echo ok", timeout=None)
|
|
assert resp.exit_code == 0
|
|
|
|
def test_execute_accepts_timeout_introspection(self):
|
|
from deepagents.backends.protocol import execute_accepts_timeout
|
|
execute_accepts_timeout.cache_clear()
|
|
assert execute_accepts_timeout(CustomSandboxBackend) is True
|
|
|
|
|
|
# === '..' traversal false-positive fix ===
|
|
|
|
class TestTraversalFalsePositiveFix:
|
|
def test_dotdot_in_filename_allowed(self):
|
|
assert validate_command("echo foo..bar.txt") is None
|
|
|
|
def test_dotdot_path_component_still_blocked(self):
|
|
result = validate_command("cat ../secret")
|
|
assert result is not None
|
|
assert "blocked" in result.lower()
|
|
|
|
def test_dotdot_nested_still_blocked(self):
|
|
result = validate_command("cat foo/../../etc/passwd")
|
|
assert result is not None
|
|
|
|
|
|
# === Pipeline command validation ===
|
|
|
|
class TestPipelineCommandValidation:
|
|
def test_pipe_blocked_command(self):
|
|
"""sudo after pipe should be caught."""
|
|
result = validate_command("echo hi | sudo tee /etc/passwd")
|
|
assert result is not None
|
|
assert "sudo" in result
|
|
|
|
def test_chained_blocked_command(self):
|
|
"""chmod after && should be caught."""
|
|
result = validate_command("echo ok && chmod 777 file")
|
|
assert result is not None
|
|
assert "chmod" in result
|
|
|
|
def test_semicolon_blocked_command(self):
|
|
"""dd after ; should be caught."""
|
|
result = validate_command("echo start ; dd if=/dev/zero of=disk")
|
|
assert result is not None
|
|
assert "dd" in result
|
|
|
|
def test_safe_pipe_allowed(self):
|
|
"""Normal pipes should be fine."""
|
|
assert validate_command("cat file.txt | grep pattern") is None
|
|
|
|
def test_safe_chain_allowed(self):
|
|
"""Normal && chains should be fine."""
|
|
assert validate_command("mkdir build && cd build") is None
|
|
|
|
def test_quoted_pipe_not_split(self):
|
|
"""Pipe inside quotes is not a shell operator."""
|
|
assert validate_command("echo 'hello | world'") is None
|