docker-agent-sandbox/tests/integration/test_sandbox.py

"""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_timeout(sandbox: DockerSandbox):
    code, out = sandbox.exec("sleep 60", timeout=2)
    assert code == 124
    assert "timed out" 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",
    )
    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",
    )
    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()