Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 57 additions & 78 deletions docs/features/executing_commands.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -121,23 +97,27 @@ 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"])

# Set permissions
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"])
Expand All @@ -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
34 changes: 32 additions & 2 deletions src/testcontainers/core/container.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
"""

Expand Down Expand Up @@ -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])

Expand Down
Loading
Loading