210 lines
9.1 KiB
Python
210 lines
9.1 KiB
Python
"""Commands must return after timeout even when descendants inherit output."""
|
|
|
|
import ctypes
|
|
import io
|
|
import os
|
|
import shlex
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from deerflow.community.aio_sandbox import aio_sandbox as aio_module
|
|
from deerflow.config.sandbox_config import SandboxConfig
|
|
from deerflow.sandbox.local.local_sandbox import LocalSandbox
|
|
|
|
|
|
@pytest.fixture
|
|
def python_sandbox(monkeypatch):
|
|
sandbox = LocalSandbox("command-liveness-test")
|
|
# A Python parent stands in for a shell, keeping these process-tree tests
|
|
# independent of bash installation and PowerShell quoting differences.
|
|
monkeypatch.setattr(sandbox, "_get_shell", lambda: sys.executable)
|
|
monkeypatch.setattr(sandbox, "_command_timeout_seconds", lambda: 10)
|
|
return sandbox
|
|
|
|
|
|
def _is_running(pid: int) -> bool:
|
|
if os.name == "nt":
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
kernel32.OpenProcess.restype = ctypes.c_void_p
|
|
kernel32.OpenProcess.argtypes = [ctypes.c_uint, ctypes.c_bool, ctypes.c_uint]
|
|
kernel32.WaitForSingleObject.argtypes = [ctypes.c_void_p, ctypes.c_uint]
|
|
kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
|
|
handle = kernel32.OpenProcess(0x00100000, False, pid) # SYNCHRONIZE
|
|
if not handle:
|
|
return False
|
|
try:
|
|
return kernel32.WaitForSingleObject(handle, 0) == 0x102 # WAIT_TIMEOUT
|
|
finally:
|
|
kernel32.CloseHandle(handle)
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
# A killed grandchild can briefly remain as a zombie until its new parent
|
|
# reaps it. It cannot execute work or retain pipe handles in this state.
|
|
stat = Path(f"/proc/{pid}/stat")
|
|
if stat.exists() and stat.read_text().rsplit(")", 1)[1].split()[0] == "Z":
|
|
return False
|
|
return True
|
|
|
|
|
|
def test_command_preserves_stdout_stderr_and_exit_code(python_sandbox):
|
|
output = python_sandbox.execute_command("import sys; print('stdout marker'); print('stderr marker', file=sys.stderr); sys.exit(7)")
|
|
assert "stdout marker" in output
|
|
assert "Std Error:\nstderr marker" in output
|
|
assert "Exit Code: 7" in output
|
|
|
|
|
|
def test_command_stdin_is_closed(python_sandbox):
|
|
output = python_sandbox.execute_command("import sys; print(repr(sys.stdin.read()))")
|
|
assert output.strip() == "''"
|
|
|
|
|
|
def test_command_timeout_terminates_child_that_inherits_output(python_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(python_sandbox, "_command_timeout_seconds", lambda: 2)
|
|
pid_file = tmp_path / "child.pid"
|
|
child = "import time; time.sleep(60)"
|
|
command = f"import pathlib,subprocess,sys,time; child=subprocess.Popen([sys.executable,'-c',{child!r}]); pathlib.Path({str(pid_file)!r}).write_text(str(child.pid)); print('started', flush=True); time.sleep(60)"
|
|
started = time.monotonic()
|
|
child_pid = None
|
|
try:
|
|
output = python_sandbox.execute_command(command)
|
|
assert time.monotonic() - started < 8
|
|
assert output.startswith("Error: Command timed out after 2 seconds")
|
|
assert "started" in output
|
|
child_pid = int(pid_file.read_text())
|
|
deadline = time.monotonic() + 2
|
|
while _is_running(child_pid) and time.monotonic() < deadline:
|
|
time.sleep(0.02)
|
|
assert not _is_running(child_pid), "Timed-out skill child was left running"
|
|
finally:
|
|
if child_pid is None and pid_file.exists():
|
|
child_pid = int(pid_file.read_text())
|
|
if child_pid is not None and _is_running(child_pid):
|
|
os.kill(child_pid, signal.SIGTERM)
|
|
|
|
|
|
def test_shell_exit_does_not_wait_for_descendants_output_handles(python_sandbox, tmp_path):
|
|
pid_file = tmp_path / "background-child.pid"
|
|
command = f"import pathlib,subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)']); pathlib.Path({str(pid_file)!r}).write_text(str(child.pid)); print('parent done')"
|
|
try:
|
|
started = time.monotonic()
|
|
output = python_sandbox.execute_command(command)
|
|
assert time.monotonic() - started < 8
|
|
assert output.strip() == "parent done"
|
|
finally:
|
|
if pid_file.exists():
|
|
child_pid = int(pid_file.read_text())
|
|
if _is_running(child_pid):
|
|
os.kill(child_pid, signal.SIGTERM)
|
|
|
|
|
|
def test_large_command_output_is_bounded_before_tool_truncation(python_sandbox):
|
|
output = python_sandbox.execute_command("import sys; sys.stdout.write('HEAD' + 'x' * (3 * 1024 * 1024) + 'TAIL')")
|
|
assert output.startswith("HEAD") and output.endswith("TAIL")
|
|
assert "command output truncated" in output
|
|
assert len(output) < LocalSandbox._COMMAND_OUTPUT_MAX_BYTES + 100
|
|
|
|
|
|
def test_runaway_output_is_stopped_before_command_timeout(python_sandbox, monkeypatch):
|
|
monkeypatch.setattr(python_sandbox, "_COMMAND_CAPTURE_MAX_BYTES", 8192)
|
|
output = python_sandbox.execute_command("import sys,time\nwhile True:\n sys.stdout.write('x' * 4096); sys.stdout.flush(); time.sleep(0.01)")
|
|
assert output.startswith("Error: Command output exceeded 8192 bytes")
|
|
|
|
|
|
def test_command_timeout_setting_and_default(python_sandbox, monkeypatch):
|
|
from deerflow import config
|
|
|
|
settings = SandboxConfig(use="test", command_timeout_seconds=12.5)
|
|
monkeypatch.setattr(config, "get_app_config", lambda: SimpleNamespace(sandbox=settings))
|
|
assert LocalSandbox._command_timeout_seconds(python_sandbox) == 12.5
|
|
assert SandboxConfig(use="test").command_timeout_seconds == 600
|
|
for invalid in (0, -1, float("inf"), float("nan")):
|
|
with pytest.raises(ValidationError):
|
|
SandboxConfig(use="test", command_timeout_seconds=invalid)
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["read_file", "read_file_range"])
|
|
def test_local_file_read_rejects_pipe_before_reading(monkeypatch, operation):
|
|
read_fd, write_fd = os.pipe()
|
|
try:
|
|
monkeypatch.setattr(os, "open", lambda *_args: os.dup(read_fd))
|
|
with pytest.raises(OSError, match="Only regular files"):
|
|
getattr(LocalSandbox("test"), operation)("/virtual/fifo")
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(write_fd)
|
|
|
|
|
|
def test_local_range_reader_stops_at_budget_without_draining_long_line(monkeypatch):
|
|
class LongLine(io.StringIO):
|
|
reads = 0
|
|
|
|
def readline(self, size=-1):
|
|
self.reads += 1
|
|
if self.reads > 1:
|
|
raise AssertionError("Read the rest of the long line after filling the result")
|
|
return "x" * 4097
|
|
|
|
sandbox = LocalSandbox("test")
|
|
monkeypatch.setattr(sandbox, "_open_regular_text", lambda *_args, **_kwargs: LongLine())
|
|
assert sandbox.read_file_range("/virtual/long-line", max_chars=100) == ("x" * 100, 2, True)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("operation", "args"),
|
|
[
|
|
("execute_command", ("echo test",)),
|
|
("read_file_range", ("/tmp/skill.md",)),
|
|
("list_dir", ("/tmp",)),
|
|
("write_file", ("/tmp/output", "test")),
|
|
("update_file", ("/tmp/output", b"test")),
|
|
],
|
|
)
|
|
def test_aio_busy_session_fails_without_executing_or_releasing_other_request(monkeypatch, operation, args):
|
|
client = Mock()
|
|
monkeypatch.setattr(aio_module, "AioSandboxClient", Mock(return_value=client))
|
|
sandbox = aio_module.AioSandbox("test", "http://127.0.0.1:1")
|
|
sandbox._LOCK_WAIT_TIMEOUT = 0.01
|
|
sandbox._lock.acquire()
|
|
try:
|
|
started = time.monotonic()
|
|
with pytest.raises(TimeoutError, match="waiting for the previous operation"):
|
|
getattr(sandbox, operation)(*args)
|
|
assert time.monotonic() - started < 1
|
|
assert sandbox._lock.locked()
|
|
assert client.mock_calls == []
|
|
finally:
|
|
sandbox._lock.release()
|
|
client.shell.exec_command.return_value = SimpleNamespace(data=SimpleNamespace(output="next command"))
|
|
assert sandbox.execute_command("echo next") == "next command"
|
|
assert not sandbox._lock.locked()
|
|
|
|
|
|
def test_aio_range_reader_preserves_continuation_and_rejects_non_files(monkeypatch, tmp_path):
|
|
def execute_script(*, command, no_change_timeout):
|
|
# Run only the generated bounded-reader program against test fixtures;
|
|
# no container or external sandbox service is contacted.
|
|
args = shlex.split(command)
|
|
completed = subprocess.run([sys.executable, *args[1:]], capture_output=True, encoding="utf-8", timeout=5)
|
|
return SimpleNamespace(data=SimpleNamespace(output=completed.stdout + completed.stderr))
|
|
|
|
client = Mock()
|
|
client.shell.exec_command.side_effect = execute_script
|
|
monkeypatch.setattr(aio_module, "AioSandboxClient", Mock(return_value=client))
|
|
sandbox = aio_module.AioSandbox("test", "http://127.0.0.1:1")
|
|
source = tmp_path / "long.txt"
|
|
source.write_text("x" * 10_000 + "\nsecond\nthird", encoding="utf-8")
|
|
assert sandbox.read_file_range(str(source), max_chars=100) == ("x" * 100, 2, True)
|
|
assert sandbox.read_file_range(str(source), start_line=2, end_line=2, max_chars=100) == ("second", None, False)
|
|
with pytest.raises(OSError, match="bounded sandbox read failed"):
|
|
sandbox.read_file_range(str(tmp_path))
|