5d44c2e7e2
Implement DockerSandbox extending deepagents' BaseSandbox to manage Docker containers via docker-py. This completes Task 1 of Phase 2. Key implementation details: - Extends BaseSandbox which auto-implements file I/O (read/write/ls/grep) by delegating to execute() - Synchronous execute() method called via loop.run_in_executor() by server.py - Container lifecycle management (create/connect/close) - Upload/download file support via tar archives - Configurable resource limits (memory, CPU, PIDs) - Timeout support with proper exit code handling - Environment variable configuration via create_sandbox() factory Tests: - 6 new tests covering container creation, command execution, and cleanup - All 46 tests passing (40 existing + 6 new)
82 lines
3.6 KiB
Python
82 lines
3.6 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_docker_client():
|
|
client = MagicMock()
|
|
container = MagicMock()
|
|
container.id = "abc123def456"
|
|
container.status = "running"
|
|
client.containers.run.return_value = container
|
|
client.containers.get.return_value = container
|
|
return client, container
|
|
|
|
|
|
def test_sandbox_create_new_container(mock_docker_client):
|
|
client, container = mock_docker_client
|
|
with patch("agent.integrations.docker_sandbox.docker.DockerClient", return_value=client):
|
|
from agent.integrations.docker_sandbox import DockerSandbox
|
|
sandbox = DockerSandbox()
|
|
assert sandbox.id == "abc123def456"
|
|
client.containers.run.assert_called_once()
|
|
call_kwargs = client.containers.run.call_args.kwargs
|
|
assert call_kwargs["detach"] is True
|
|
assert call_kwargs["labels"] == {"galaxis-agent-sandbox": "true"}
|
|
|
|
|
|
def test_sandbox_connect_existing_container(mock_docker_client):
|
|
client, container = mock_docker_client
|
|
with patch("agent.integrations.docker_sandbox.docker.DockerClient", return_value=client):
|
|
from agent.integrations.docker_sandbox import DockerSandbox
|
|
sandbox = DockerSandbox(container_id="abc123def456")
|
|
assert sandbox.id == "abc123def456"
|
|
client.containers.get.assert_called_once_with("abc123def456")
|
|
client.containers.run.assert_not_called()
|
|
|
|
|
|
def test_sandbox_execute_success(mock_docker_client):
|
|
client, container = mock_docker_client
|
|
container.exec_run.return_value = MagicMock(exit_code=0, output=(b"hello world\n", None))
|
|
with patch("agent.integrations.docker_sandbox.docker.DockerClient", return_value=client):
|
|
from agent.integrations.docker_sandbox import DockerSandbox
|
|
sandbox = DockerSandbox(container_id="abc123def456")
|
|
result = sandbox.execute("echo hello world")
|
|
assert "hello world" in result.output
|
|
assert result.exit_code == 0
|
|
assert result.truncated is False
|
|
|
|
|
|
def test_sandbox_execute_with_stderr(mock_docker_client):
|
|
client, container = mock_docker_client
|
|
container.exec_run.return_value = MagicMock(exit_code=1, output=(b"", b"error: not found\n"))
|
|
with patch("agent.integrations.docker_sandbox.docker.DockerClient", return_value=client):
|
|
from agent.integrations.docker_sandbox import DockerSandbox
|
|
sandbox = DockerSandbox(container_id="abc123def456")
|
|
result = sandbox.execute("cat missing.txt")
|
|
assert "error: not found" in result.output
|
|
assert result.exit_code == 1
|
|
|
|
|
|
def test_sandbox_execute_with_timeout(mock_docker_client):
|
|
client, container = mock_docker_client
|
|
container.exec_run.return_value = MagicMock(exit_code=0, output=(b"ok\n", None))
|
|
with patch("agent.integrations.docker_sandbox.docker.DockerClient", return_value=client):
|
|
from agent.integrations.docker_sandbox import DockerSandbox
|
|
sandbox = DockerSandbox(container_id="abc123def456")
|
|
sandbox.execute("sleep 1", timeout=30)
|
|
call_args = container.exec_run.call_args
|
|
cmd = call_args.kwargs.get("cmd", call_args.args[0] if call_args.args else [])
|
|
assert cmd[0] == "timeout"
|
|
assert cmd[1] == "30"
|
|
|
|
|
|
def test_sandbox_close(mock_docker_client):
|
|
client, container = mock_docker_client
|
|
with patch("agent.integrations.docker_sandbox.docker.DockerClient", return_value=client):
|
|
from agent.integrations.docker_sandbox import DockerSandbox
|
|
sandbox = DockerSandbox(container_id="abc123def456")
|
|
sandbox.close()
|
|
container.stop.assert_called_once()
|
|
container.remove.assert_called_once()
|