diff --git a/docs/features/executing_commands.md b/docs/features/executing_commands.md index de44ec3aa..410169c0e 100644 --- a/docs/features/executing_commands.md +++ b/docs/features/executing_commands.md @@ -4,104 +4,80 @@ Testcontainers-Python provides several ways to execute commands inside container ## Basic Command Execution -The simplest way to execute a command is using the `exec` method: +The simplest way to execute a command is using the `exec` method, passing either an argv list or a string: ```python -from testcontainers.community.generic import GenericContainer +from testcontainers.core.container import DockerContainer -with GenericContainer("alpine:latest") as container: - # Execute a simple command - exit_code, output = container.exec(["ls", "-la"]) - print(output) # Command output as string -``` +with DockerContainer("alpine:latest") as container: + # Execute a simple command (argv form) + result = container.exec(["ls", "-la"]) + print(result.exit_code) # 0 + print(result.output) # command output as bytes + print(result.output.decode()) # ...decoded to str -## Command Execution with Options + # A string is also accepted + result = container.exec("ls -la") +``` -You can customize command execution with various options: +`exec` returns a named `(exit_code, output)` tuple, so you can also unpack it directly: ```python -with GenericContainer("alpine:latest") as container: - # Execute command with user - exit_code, output = container.exec( - ["whoami"], - user="nobody" - ) - - # Execute command with environment variables - exit_code, output = container.exec( - ["echo", "$TEST_VAR"], - environment={"TEST_VAR": "test_value"} - ) - - # Execute command with working directory - exit_code, output = container.exec( - ["pwd"], - workdir="/tmp" - ) +exit_code, output = container.exec(["ls", "-la"]) ``` -## Interactive Commands +> **A string command is *not* run through a shell.** `docker-py` tokenizes it with `shlex.split`, so shell features such as pipes, redirections, and variable expansion are passed through literally — `container.exec("echo $HOME")` prints the text `$HOME`, not your home directory. When you need shell behavior, invoke a shell explicitly: `container.exec(["sh", "-c", "echo $HOME"])`. -For interactive commands, you can use the `exec_interactive` method: +## Command Execution with Options + +To customize how a command runs — the user, environment, or working directory — pass an `ExecConfig`: ```python -with GenericContainer("alpine:latest") as container: - # Start an interactive shell - container.exec_interactive(["sh"]) -``` +from testcontainers.core.container import DockerContainer, ExecConfig -## Command Execution with Timeout +with DockerContainer("alpine:latest") as container: + # Execute command as a specific user + result = container.exec(ExecConfig(command=["whoami"], user="nobody")) -You can set a timeout for command execution: + # Execute command with environment variables + # (use a command that reads the environment, e.g. printenv -- a bare + # argv command is not shell-expanded, see the note above) + result = container.exec(ExecConfig(command=["printenv", "TEST_VAR"], environment={"TEST_VAR": "test_value"})) -```python -with GenericContainer("alpine:latest") as container: - # Execute command with timeout - try: - exit_code, output = container.exec( - ["sleep", "10"], - timeout=5 # Timeout in seconds - ) - except TimeoutError: - print("Command timed out") + # Execute command in a working directory (str or pathlib.Path) + result = container.exec(ExecConfig(command=["pwd"], workdir="/tmp")) ``` -## Command Execution with Privileges - -For commands that require elevated privileges: +`ExecConfig` is a frozen dataclass: only `command` is required, and `user`, `environment`, `workdir`, and `privileged` are optional. Because it is immutable, the idiomatic way to derive a variant is `dataclasses.replace`: ```python -with GenericContainer("alpine:latest") as container: - # Execute command with privileges - exit_code, output = container.exec( - ["mount"], - privileged=True - ) +from dataclasses import replace + +base = ExecConfig(command=["pwd"], workdir="/tmp") +in_var = replace(base, workdir="/var") ``` -## Command Execution with TTY +## Command Execution with Privileges -For commands that require a TTY: +For commands that require elevated privileges, set `privileged=True`: ```python -with GenericContainer("alpine:latest") as container: - # Execute command with TTY - exit_code, output = container.exec( - ["top"], - tty=True - ) +from testcontainers.core.container import DockerContainer, ExecConfig + +with DockerContainer("alpine:latest") as container: + # Execute command with privileges + result = container.exec(ExecConfig(command=["mount"], privileged=True)) ``` ## Best Practices -1. Use appropriate timeouts for long-running commands -2. Handle command failures gracefully -3. Use environment variables for configuration -4. Consider security implications of privileged commands -5. Clean up after command execution -6. Use appropriate user permissions -7. Handle command output appropriately -8. Consider using shell scripts for complex commands +1. Handle command failures gracefully — check `exit_code` rather than assuming success +2. Use environment variables for configuration +3. Consider security implications of privileged commands +4. Clean up after command execution +5. Use appropriate user permissions +6. Decode `output` from bytes when you need text +7. Use an explicit `["sh", "-c", ...]` invocation for commands that rely on shell features ## Common Use Cases @@ -121,7 +97,9 @@ with PostgresContainer() as postgres: ### File Operations ```python -with GenericContainer("alpine:latest") as container: +from testcontainers.core.container import DockerContainer + +with DockerContainer("alpine:latest") as container: # Create a directory container.exec(["mkdir", "-p", "/data"]) @@ -129,15 +107,17 @@ with GenericContainer("alpine:latest") as container: container.exec(["chmod", "755", "/data"]) # List files - exit_code, output = container.exec(["ls", "-la", "/data"]) + result = container.exec(["ls", "-la", "/data"]) ``` ### Service Management ```python -with GenericContainer("nginx:alpine") as container: +from testcontainers.core.container import DockerContainer + +with DockerContainer("nginx:alpine") as container: # Check service status - exit_code, output = container.exec(["nginx", "-t"]) + result = container.exec(["nginx", "-t"]) # Reload configuration container.exec(["nginx", "-s", "reload"]) @@ -151,7 +131,6 @@ If you encounter issues with command execution: 2. Verify user permissions 3. Check container state 4. Verify command availability -5. Check for timeout issues -6. Verify environment variables -7. Check working directory -8. Verify TTY requirements +5. Verify environment variables +6. Check the working directory +7. Remember that string commands are tokenized, not shell-interpreted diff --git a/src/testcontainers/core/container.py b/src/testcontainers/core/container.py index 9b2148ae0..76fce9c14 100644 --- a/src/testcontainers/core/container.py +++ b/src/testcontainers/core/container.py @@ -4,6 +4,7 @@ import pathlib import sys import tarfile +from dataclasses import dataclass from os import PathLike from socket import socket from types import TracebackType @@ -45,6 +46,34 @@ class BytesExecResult(ExecResult): output: bytes +@dataclass(frozen=True) +class ExecConfig: + """Configuration for a command executed inside a running container. + + Backend-agnostic data carrier. Only `command` is required, and every other field + defaults to `None`/`False`. + + `command` accepts either an argv `list[str]` or a `str`. Both are forwarded untouched. + """ + + command: Union[str, list[str]] + user: Optional[str] = None + environment: Optional[dict[str, str]] = None + workdir: Optional[Union[str, PathLike[str]]] = None + privileged: bool = False + + def to_exec_run_kwargs(self) -> dict[str, Any]: + """Yield the kwargs-dict equivalent for a `docker-py` `exec_run()` call.""" + + return { + "cmd": self.command, + "user": self.user or "", + "environment": self.environment, + "workdir": str(self.workdir) if self.workdir is not None else None, + "privileged": self.privileged, + } + + class DockerContainer: """ @@ -358,10 +387,11 @@ def status(self) -> str: return "not_started" return self._container.status - def exec(self, command: Union[str, list[str]]) -> BytesExecResult: + def exec(self, command: Union[str, list[str], ExecConfig]) -> BytesExecResult: if not self._container: raise ContainerStartException("Container should be started before executing a command") - result = self._container.exec_run(command) + config = command if isinstance(command, ExecConfig) else ExecConfig(command=command) + result = self._container.exec_run(**config.to_exec_run_kwargs()) assert isinstance(result.output, bytes) return BytesExecResult(result[0], result[1]) diff --git a/tests/core/test_exec.py b/tests/core/test_exec.py new file mode 100644 index 000000000..949fdfac4 --- /dev/null +++ b/tests/core/test_exec.py @@ -0,0 +1,216 @@ +"""Behavioral tests for ``DockerContainer.exec``. + +The tests in this module are split in two: + +* The functions below pin the *current* behavior of ``.exec()`` as it forwards + ``str`` and ``list[str]`` commands to ``docker-py``'s ``exec_run``. They exist + to make any future change in interpretation a visible, intentional diff. +* In particular, they nail down a sharp edge: ``docker-py`` tokenizes a ``str`` + command with ``shlex.split`` and never wraps it in a shell. Pipes, redirects, + and variable expansion are therefore *not* interpreted -- a fact that is easy + to assume otherwise and was previously untested. +""" + +from collections.abc import Iterator +from dataclasses import FrozenInstanceError, replace +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pytest +from docker.models.containers import ExecResult + +from testcontainers.core.container import DockerContainer, ExecConfig +from testcontainers.core.exceptions import ContainerStartException + +if TYPE_CHECKING: + from pytest_mock import MockerFixture + + +@pytest.fixture(scope="module") +def running_container() -> Iterator[DockerContainer]: + """A long-lived alpine container to exec against (exec is read-only here).""" + container = DockerContainer("alpine").with_command("tail -f /dev/null") + with container: + yield container + + +def test_exec_str_command(running_container: DockerContainer) -> None: + result = running_container.exec("echo hello") + assert result.exit_code == 0 + assert isinstance(result.output, bytes) + assert result.output.strip() == b"hello" + + +def test_exec_list_command(running_container: DockerContainer) -> None: + result = running_container.exec(["echo", "hello"]) + assert result.exit_code == 0 + assert result.output.strip() == b"hello" + + +def test_exec_nonzero_exit_code_is_propagated(running_container: DockerContainer) -> None: + result = running_container.exec(["sh", "-c", "exit 3"]) + assert result.exit_code == 3 + + +def test_str_command_is_tokenized_not_shell_interpreted(running_container: DockerContainer) -> None: + """A ``str`` command is split with ``shlex``, not run through a shell. + + ``echo a | wc -l`` becomes the argv ``["echo", "a", "|", "wc", "-l"]``, so + ``echo`` prints the pipe and ``wc`` literally instead of counting lines. + """ + result = running_container.exec("echo a | wc -l") + assert result.exit_code == 0 + assert result.output.strip() == b"a | wc -l" + + +def test_str_command_does_not_redirect(running_container: DockerContainer) -> None: + result = running_container.exec("echo hi > /tmp/frob") + assert result.exit_code == 0 + assert result.output.strip() == b"hi > /tmp/frob" + + +def test_str_command_does_not_expand_variables(running_container: DockerContainer) -> None: + """No shell means no parameter expansion: ``$HOME`` reaches ``echo`` verbatim.""" + result = running_container.exec("echo $HOME") + assert result.exit_code == 0 + assert result.output.strip() == b"$HOME" + + +# --- Pure unit tests for the config -> exec_run kwargs seam (no Docker needed) --- + + +def test_exec_config_requires_only_command() -> None: + config = ExecConfig(command=["echo", "hi"]) + assert config.command == ["echo", "hi"] + assert config.user is None + assert config.environment is None + assert config.workdir is None + assert config.privileged is False + + +def test_exec_config_is_frozen() -> None: + config = ExecConfig(command="true") + with pytest.raises(FrozenInstanceError): + config.user = "frob" # type: ignore[misc] + + +def test_exec_config_supports_dataclasses_replace() -> None: + base = ExecConfig(command="true") + derived = replace(base, workdir="/tmp", user="frob") + assert base.workdir is None # original untouched + assert base.user is None + assert derived.command == "true" + assert derived.workdir == "/tmp" + assert derived.user == "frob" + + +def test_kwargs_defaults_collapse_user_to_docker_sentinel() -> None: + kwargs = ExecConfig(command=["whoami"]).to_exec_run_kwargs() + assert kwargs == { + "cmd": ["whoami"], + "user": "", # docker-py's empty-string sentinel, not our None + "environment": None, + "workdir": None, + "privileged": False, + } + + +def test_kwargs_forward_str_command_verbatim() -> None: + # docker-py does its own shlex tokenization; we must not pre-split. + assert ExecConfig(command="echo a | wc -l").to_exec_run_kwargs()["cmd"] == "echo a | wc -l" + + +def test_kwargs_pass_through_user_environment_and_privileged() -> None: + kwargs = ExecConfig(command=["env"], user="frob", environment={"FROB": "243"}, privileged=True).to_exec_run_kwargs() + assert kwargs["user"] == "frob" + assert kwargs["environment"] == {"FROB": "243"} + assert kwargs["privileged"] is True + + +def test_kwargs_stringify_pathlike_workdir() -> None: + kwargs = ExecConfig(command=["pwd"], workdir=Path("/tmp/xyzzy")).to_exec_run_kwargs() + assert kwargs["workdir"] == "/tmp/xyzzy" + assert isinstance(kwargs["workdir"], str) + + +def test_kwargs_leave_unset_workdir_as_none() -> None: + assert ExecConfig(command=["pwd"]).to_exec_run_kwargs()["workdir"] is None + + +@pytest.fixture +def offline_container(mocker: "MockerFixture") -> DockerContainer: + """A DockerContainer whose client is mocked away, so exec() can be exercised + without a running daemon. ``_container`` is wired with a stub ``exec_run``.""" + mocker.patch("testcontainers.core.container.DockerClient") + return DockerContainer("alpine") + + +def test_exec_wraps_str_into_config_and_forwards_kwargs(offline_container: DockerContainer) -> None: + offline_container._container = MagicMock() + offline_container._container.exec_run.return_value = ExecResult(exit_code=0, output=b"hi") + result = offline_container.exec("echo hi") + offline_container._container.exec_run.assert_called_once_with( + cmd="echo hi", user="", environment=None, workdir=None, privileged=False + ) + assert result.exit_code == 0 + assert result.output == b"hi" + + +def test_exec_accepts_exec_config_directly(offline_container: DockerContainer) -> None: + offline_container._container = MagicMock() + offline_container._container.exec_run.return_value = ExecResult(exit_code=0, output=b"hi") + offline_container.exec(ExecConfig(command=["pwd"], workdir=Path("/tmp"), user="frob")) + offline_container._container.exec_run.assert_called_once_with( + cmd=["pwd"], user="frob", environment=None, workdir="/tmp", privileged=False + ) + + +def test_exec_before_start_raises(mocker: "MockerFixture") -> None: + mocker.patch("testcontainers.core.container.DockerClient") + with pytest.raises(ContainerStartException): + DockerContainer("alpine").exec("true") + + +# --- Integration tests for the new ExecConfig fields (require a Docker daemon) --- + + +def test_exec_config_workdir_str(running_container: DockerContainer) -> None: + result = running_container.exec(ExecConfig(command=["pwd"], workdir="/tmp")) + assert result.exit_code == 0 + assert result.output.strip() == b"/tmp" + + +def test_exec_config_workdir_accepts_pathlib_path(running_container: DockerContainer) -> None: + result = running_container.exec(ExecConfig(command=["pwd"], workdir=Path("/tmp"))) + assert result.exit_code == 0 + assert result.output.strip() == b"/tmp" + + +def test_exec_config_environment(running_container: DockerContainer) -> None: + result = running_container.exec(ExecConfig(command=["env"], environment={"FROB": "243"})) + assert result.exit_code == 0 + assert b"FROB=243" in result.output + + +def test_exec_config_user(running_container: DockerContainer) -> None: + as_default = running_container.exec(ExecConfig(command=["whoami"])) + as_nobody = running_container.exec(ExecConfig(command=["whoami"], user="nobody")) + assert as_default.output.strip() == b"root" + assert as_nobody.output.strip() == b"nobody" + + +def test_exec_config_privileged_grants_more_capabilities(running_container: DockerContainer) -> None: + """A privileged exec receives the full capability set, even though the + container itself is unprivileged -- so its CapEff strictly exceeds the + default exec's. We read the effective set straight from procfs.""" + + def cap_eff(config: ExecConfig) -> int: + output = running_container.exec(config).output.decode() + # /proc/self/status line looks like: "CapEff:\t00000000a80425fb" + line = next(ln for ln in output.splitlines() if ln.startswith("CapEff:")) + return int(line.split()[1], 16) + + default_caps = cap_eff(ExecConfig(command=["grep", "CapEff", "/proc/self/status"])) + privileged_caps = cap_eff(ExecConfig(command=["grep", "CapEff", "/proc/self/status"], privileged=True)) + assert privileged_caps > default_caps