282 lines
8.4 KiB
Python
282 lines
8.4 KiB
Python
"""Integration tests for DockerSandbox core (exec, file I/O, path resolution)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from docker_agent_sandbox import DockerSandbox
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# exec()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_exec_simple_command(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo hello")
|
|
assert code == 0
|
|
assert "hello" in out
|
|
|
|
|
|
def test_exec_nonzero_exit_code(sandbox: DockerSandbox):
|
|
code, _ = sandbox.exec("exit 42")
|
|
assert code == 42
|
|
|
|
|
|
def test_exec_stderr_captured(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo msg_on_stderr >&2")
|
|
assert code == 0
|
|
assert "msg_on_stderr" in out
|
|
|
|
|
|
def test_exec_combined_stdout_and_stderr(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo stdout_line; echo stderr_line >&2")
|
|
assert code == 0
|
|
assert "stdout_line" in out
|
|
assert "stderr_line" in out
|
|
|
|
|
|
def test_exec_returns_error_when_container_not_running():
|
|
# Construct a sandbox without starting it to exercise the guard.
|
|
sb = DockerSandbox.__new__(DockerSandbox)
|
|
sb._container = None
|
|
sb._working_dir = None
|
|
code, out = sb.exec("echo hi")
|
|
assert code == 1
|
|
assert "not running" in out.lower()
|
|
|
|
|
|
def test_exec_instant_command(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("true")
|
|
assert code == 0
|
|
assert out == ""
|
|
|
|
|
|
def test_exec_instant_nonzero(sandbox: DockerSandbox):
|
|
code, _ = sandbox.exec("false")
|
|
assert code == 1
|
|
|
|
|
|
def test_exec_delayed_command_within_timeout(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("sleep 1 && echo done", timeout=10)
|
|
assert code == 0
|
|
assert "done" in out
|
|
|
|
|
|
def test_exec_timeout(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("sleep 60", timeout=2)
|
|
assert code == 124
|
|
assert "timed out" in out
|
|
|
|
|
|
def test_exec_timeout_longer_than_sleep(sandbox: DockerSandbox):
|
|
# Command finishes before timeout — must not raise or return 124.
|
|
code, out = sandbox.exec("sleep 1 && echo ok", timeout=10)
|
|
assert code == 0
|
|
assert "ok" in out
|
|
|
|
|
|
def test_exec_and_chain_both_succeed(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo first && echo second")
|
|
assert code == 0
|
|
assert "first" in out
|
|
assert "second" in out
|
|
|
|
|
|
def test_exec_and_chain_short_circuits_on_failure(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("false && echo should_not_print")
|
|
assert code != 0
|
|
assert "should_not_print" not in out
|
|
|
|
|
|
def test_exec_pipe(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo hello world | tr ' ' '_'")
|
|
assert code == 0
|
|
assert "hello_world" in out
|
|
|
|
|
|
def test_exec_pipe_exit_code_is_last_command(sandbox: DockerSandbox):
|
|
# grep finds no match → exit 1, even though echo succeeded
|
|
code, _ = sandbox.exec("echo hello | grep nomatch")
|
|
assert code == 1
|
|
|
|
|
|
def test_exec_stdout_redirect_to_file(sandbox: DockerSandbox, workdir: str):
|
|
code, out = sandbox.exec(f"echo redirected > {workdir}/out.txt && cat {workdir}/out.txt")
|
|
assert code == 0
|
|
assert "redirected" in out
|
|
|
|
|
|
def test_exec_stderr_redirect_to_stdout(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo err_msg >&2 2>&1")
|
|
assert code == 0
|
|
assert "err_msg" in out
|
|
|
|
|
|
def test_exec_subshell(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("result=$(echo inner) && echo $result")
|
|
assert code == 0
|
|
assert "inner" in out
|
|
|
|
|
|
def test_exec_multiline_via_semicolons(sandbox: DockerSandbox):
|
|
code, out = sandbox.exec("echo a; echo b; echo c")
|
|
assert code == 0
|
|
assert "a" in out
|
|
assert "b" in out
|
|
assert "c" in out
|
|
|
|
|
|
def test_exec_working_dir_respected():
|
|
"""When working_dir is set, exec uses it as cwd."""
|
|
sb = DockerSandbox(
|
|
container_name="test-workdir-check",
|
|
image="python:3.11-slim",
|
|
command="sleep infinity",
|
|
working_dir="/tmp",
|
|
cpu_limit=1,
|
|
memory_limit="256m",
|
|
)
|
|
sb.start()
|
|
try:
|
|
code, out = sb.exec("pwd")
|
|
assert code == 0
|
|
assert "/tmp" in out
|
|
finally:
|
|
sb.stop()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# write_file() / read_file()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_write_read_roundtrip(sandbox: DockerSandbox, workdir: str):
|
|
path = f"{workdir}/hello.txt"
|
|
sandbox.write_file(path, "hello world\n")
|
|
assert sandbox.read_file(path) == b"hello world\n"
|
|
|
|
|
|
def test_write_read_unicode(sandbox: DockerSandbox, workdir: str):
|
|
path = f"{workdir}/unicode.txt"
|
|
content = "héllo wörld 你好\n"
|
|
sandbox.write_file(path, content)
|
|
assert sandbox.read_file(path).decode("utf-8") == content
|
|
|
|
|
|
def test_write_creates_parent_directories(sandbox: DockerSandbox, workdir: str):
|
|
path = f"{workdir}/deep/nested/dir/file.txt"
|
|
sandbox.write_file(path, "content")
|
|
assert sandbox.read_file(path) == b"content"
|
|
|
|
|
|
def test_write_overwrites_existing_file(sandbox: DockerSandbox, workdir: str):
|
|
path = f"{workdir}/overwrite.txt"
|
|
sandbox.write_file(path, "first")
|
|
sandbox.write_file(path, "second")
|
|
assert sandbox.read_file(path) == b"second"
|
|
|
|
|
|
def test_read_file_not_found_raises(sandbox: DockerSandbox, workdir: str):
|
|
with pytest.raises(FileNotFoundError):
|
|
sandbox.read_file(f"{workdir}/no_such_file.txt")
|
|
|
|
|
|
def test_read_directory_raises(sandbox: DockerSandbox, workdir: str):
|
|
sandbox.exec(f"mkdir -p {workdir}/subdir")
|
|
with pytest.raises(IsADirectoryError):
|
|
sandbox.read_file(f"{workdir}/subdir")
|
|
|
|
|
|
def test_read_file_when_container_not_running_raises():
|
|
sb = DockerSandbox.__new__(DockerSandbox)
|
|
sb._container = None
|
|
sb._working_dir = None
|
|
with pytest.raises(RuntimeError, match="not running"):
|
|
sb.read_file("/tmp/anything.txt")
|
|
|
|
|
|
def test_write_file_when_container_not_running_raises():
|
|
sb = DockerSandbox.__new__(DockerSandbox)
|
|
sb._container = None
|
|
sb._working_dir = None
|
|
with pytest.raises(RuntimeError, match="not running"):
|
|
sb.write_file("/tmp/anything.txt", "data")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_host_port()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_host_port_raises_when_not_running():
|
|
sb = DockerSandbox.__new__(DockerSandbox)
|
|
sb._container = None
|
|
with pytest.raises(RuntimeError, match="not running"):
|
|
sb.get_host_port("8080/tcp")
|
|
|
|
|
|
def test_get_host_port_raises_for_unmapped_port(sandbox: DockerSandbox):
|
|
with pytest.raises(RuntimeError, match="not mapped"):
|
|
sandbox.get_host_port("9999/tcp")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_path()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path, working_dir, expected",
|
|
[
|
|
("/absolute/path", "/work", "/absolute/path"),
|
|
("relative/file", "/work", "/work/relative/file"),
|
|
("relative/file", None, "relative/file"),
|
|
("/absolute/path", None, "/absolute/path"),
|
|
],
|
|
)
|
|
def test_resolve_path(path, working_dir, expected):
|
|
sb = DockerSandbox.__new__(DockerSandbox)
|
|
sb._working_dir = working_dir
|
|
assert str(sb._resolve_path(path)) == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# context manager
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_context_manager_stops_container():
|
|
sb = DockerSandbox(
|
|
container_name="test-ctx-manager",
|
|
image="python:3.11-slim",
|
|
command="sleep infinity",
|
|
cpu_limit=1,
|
|
memory_limit="256m",
|
|
)
|
|
sb.start()
|
|
with sb:
|
|
code, _ = sb.exec("echo alive")
|
|
assert code == 0
|
|
assert sb._container is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_image_if_missing()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_image_if_missing_skips_when_present(sandbox: DockerSandbox):
|
|
# python:3.11-slim was already pulled by the session fixture; this must not
|
|
# raise even though dockerfile_dir is None.
|
|
sandbox.build_image_if_missing()
|
|
|
|
|
|
def test_build_image_if_missing_raises_without_dockerfile_dir():
|
|
sb = DockerSandbox(
|
|
container_name="irrelevant",
|
|
image="image-that-does-not-exist-xyzzy123",
|
|
)
|
|
with pytest.raises((RuntimeError, ValueError)):
|
|
sb.build_image_if_missing()
|