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