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()
|