Testcontainers Core¶

testcontainers-core is the core functionality for spinning up Docker containers in test environments.

class BytesExecResult(exit_code, output)¶

Basic container object to spin up Docker instances.

Parameters:

image – The name of the image to start.
docker_client_kw – Dictionary with arguments that will be passed to the docker.DockerClient init.
command – Optional execution command for the container.
name – Optional name for the container.
ports – Ports to be exposed by the container. The port number will be automatically assigned on the host, use get_exposed_port(PORT) method to get the port number on the host.
volumes – Volumes to mount into the container. Each entry should be a tuple with three values: host path, container path and mode (default ‘ro’).
network – Optional network to connect the container to.
network_aliases – Optional list of aliases for the container in the network.

>>> from testcontainers.core.container import DockerContainer
>>> from testcontainers.core.waiting_utils import wait_for_logs

>>> with DockerContainer("hello-world") as container:
...    delay = wait_for_logs(container, "Hello from Docker!")

copy_from_container(source_in_container: str, destination_on_host: Path) → None¶

copy_into_container(transferable: bytes | Path, destination_in_container: str, mode: int = 420) → None¶

exec(command: str | list[str] | ExecConfig) → BytesExecResult¶

get_container_host_ip() → str¶

get_container_id() → str¶

get_container_info() → ContainerInspectInfo | None¶

Get container information via docker inspect (lazy loaded).

Returns:: Container inspect information or None if container is not started.

get_exposed_port(port: int) → int¶

get_logs() → tuple[bytes, bytes]¶

get_wrapped_container() → Container¶

maybe_emulate_amd64() → Self¶

reload() → None¶: Reload container information for compatibility with wait strategies.

start() → Self¶

property status: str¶: Get container status for compatibility with wait strategies.

stop(force: bool = True, delete_volume: bool = True) → None¶

wait() → int¶: Wait for the container to stop and return its exit code.

waiting_for(strategy: WaitStrategy) → Self¶: Set a wait strategy to be used after container start.

with_bind_ports(container: str | int, host: str | int | None = None) → Self¶

Bind container port to host port

Parameters:

container – container port
host – host port

Doctest:

>>> from testcontainers.core.container import DockerContainer
>>> container = DockerContainer("nginx")
>>> container = container.with_bind_ports("8080/tcp", 8080)
>>> container = container.with_bind_ports("8081/tcp", 8081)

with_command(command: str | list[str]) → Self¶

with_copy_into_container(transferable: bytes | Path, destination_in_container: str, mode: int = 420) → Self¶

with_env(key: str, value: str) → Self¶

with_env_file(env_file: str | PathLike[str]) → Self¶

with_envs(**variables: str) → Self¶

with_exposed_ports(*ports: str | int) → Self¶

Expose ports from the container without binding them to the host.

Parameters:: ports – ports to expose
Doctest:

>>> from testcontainers.core.container import DockerContainer
>>> container = DockerContainer("nginx")
>>> container = container.with_exposed_ports("8080/tcp", "8081/tcp")

with_kwargs(**kwargs: Any) → Self¶

with_name(name: str) → Self¶

with_network(network: Network) → Self¶

with_network_aliases(*aliases: str) → Self¶

with_tmpfs_mount(container_path: str, size: str | None = None) → Self¶

Mount a tmpfs volume on the container.

Parameters:

container_path – Container path to mount tmpfs on (e.g., ‘/data’)
size – Optional size limit (e.g., ‘256m’, ‘1g’). If None, unbounded.

Returns:

Self for chaining

with_volume_mapping(host: str | PathLike[str], container: str, mode: str = 'ro') → Self¶

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: str | list[str]¶

environment: dict[str, str] | None = None¶

privileged: bool = False¶

to_exec_run_kwargs() → dict[str, Any]¶: Yield the kwargs-dict equivalent for a docker-py exec_run() call.

user: str | None = None¶

workdir: str | PathLike[str] | None = None¶

class Mount¶

bind: str¶

mode: str¶

class Network(docker_client_kw: dict[str, Any] | None = None, docker_network_kw: dict[str, Any] | None = None)¶: Network context manager for programmatically connecting containers.

class DockerImage(path: str | PathLike[str], docker_client_kw: dict[str, Any] | None = None, tag: str | None = None, clean_up: bool = True, dockerfile_path: str = 'Dockerfile', no_cache: bool = False, **kwargs: Any)¶

Basic image object to build Docker images.

>>> from testcontainers.core.image import DockerImage

>>> with DockerImage(path=f"{TEST_DIR}/core/image_fixtures/sample/", tag="test-image") as image:
...    logs = image.get_logs()

Parameters:

path – Path to the build context
docker_client_kw – Keyword arguments to pass to the DockerClient
tag – Tag for the image to be built (default: None)
clean_up – Remove the image after exiting the context (default: True)
dockerfile_path – Path to the Dockerfile within the build context path (default: Dockerfile)
no_cache – Bypass build cache; CLI’s –no-cache
kwargs – Additional keyword arguments to pass to the underlying docker-py

DEPRECATED (for removal) Please use database-specific container classes or SqlContainer instead. # from testcontainers.community.generic.sql import SqlContainer

Generic database container.

class WaitStrategy¶: Base class for all wait strategies.

Transferable¶: alias of bytes | Path

Compose¶

It is also possible to use Docker Compose functionality:

A container class that represents a container managed by compose. It is not a true testcontainers.core.container.DockerContainer, but you can use the id with DockerClient to get that one too.

get_container_host_ip() → str¶: Get the host IP for the container.

get_container_info() → ContainerInspectInfo | None¶

Get container information via docker inspect (lazy loaded).

Returns:: Container inspect information or None if container is not started.

get_exposed_port(port: int) → int¶: Get the exposed port mapping for the given internal port.

get_logs() → tuple[bytes, bytes]¶: Get container logs.

get_wrapped_container() → Self¶: Get the underlying container object for compatibility.

reload() → None¶: Reload container information for compatibility with wait strategies.

property status: str¶: Get container status for compatibility with wait strategies.

class DockerCompose(context: str | PathLike[str], compose_file_name: str | list[str] | None = None, pull: bool = False, build: bool = False, wait: bool = True, keep_volumes: bool = False, env_file: str | list[str] | None = None, services: list[str] | None = None, docker_command_path: str | None = None, profiles: list[str] | None = None, quiet_pull: bool = False, quiet_build: bool = False)¶

Manage docker compose environments.

Parameters:

context – The docker context. It corresponds to the directory containing the docker compose configuration file.
compose_file_name – Optional. File name of the docker compose configuration file. If specified, you need to also specify the overrides if any.
pull – Pull images before launching environment.
build – Run docker compose build before running the environment.
wait – Wait for the services to be healthy (as per healthcheck definitions in the docker compose configuration)
env_file – Path(s) to an ‘.env’ file containing environment variables to pass to docker compose.
services – The list of services to use from this DockerCompose.
docker_command_path – The docker compose command to run.
profiles – The list of profiles to enable.
quiet_pull – Whether to suppress output when pulling images.
quiet_build – Whether to suppress output when building images.

Example

This example spins up chrome and firefox containers using docker compose.

>>> from testcontainers.compose import DockerCompose

>>> compose = DockerCompose(f"{TEST_DIR}/core/compose_fixtures/basic", compose_file_name="hello.yaml",
...                         pull=True)
>>> with compose:
...     stdout, stderr = compose.get_logs()
>>> "Hello from Docker!" in stdout
True

services:
  hello-world:
    image: "hello-world"

docker_compose_command() → list[str]¶

Returns command parts used for the docker compose commands

Returns:: Docker compose command parts.
Return type:: list[str]

exec_in_container(command: list[str], service_name: str | None = None) → tuple[str, str, int]¶

Executes a command in the container of one of the services.

Parameters:

service_name – Name of the docker compose service to run the command in.
command – Command to execute.
service_name – specify the service name
command – the command to run in the container

Returns:

stdout: Standard output stream. str: stderr: Standard error stream. int: exit_code: The command’s exit code.

Return type:

str

get_config(*, path_resolution: bool = True, normalize: bool = True, interpolate: bool = True) → dict[str, Any]¶

Parse, resolve and returns compose file via docker config –format json. In case of multiple compose files, the returned value will be a merge of all files.

See: https://docs.docker.com/reference/cli/docker/compose/config/ for more details

Parameters:

path_resolution – whether to resolve file paths
normalize – whether to normalize compose model
interpolate – whether to interpolate environment variables

Returns:

Compose file

get_containers(include_all: bool = False) → list[ComposeContainer]¶

Fetch information about running containers via docker compose ps –format json. Available only in V2 of compose.

Returns:: The list of running containers.

get_logs(*services: str) → tuple[str, str]¶

Returns all log output from stdout and stderr of a specific container.

Parameters:: services – which services to get the logs for (or omit, for all)
Returns:: stdout: Standard output stream. str: stderr: Standard error stream.
Return type:: str

get_service_host(service_name: str | None = None, port: int | None = None) → str | None¶

Returns the host for one of the services.

Parameters:

service_name (str) – Name of the docker compose service
port (int) – The internal port to get the host for

Returns:

The hostname for the service

Return type:

str

get_service_port(service_name: str | None = None, port: int | None = None) → int | None¶

Returns the mapped port for one of the services.

Parameters:

service_name (str) – Name of the docker compose service
port (int) – The internal port to get the mapping for

Returns:

The mapped port on the host

Return type:

int

start() → None¶: Starts the docker compose environment.

stop(down: bool = True) → None¶: Stops the docker compose environment.

wait_for(url: str) → Self¶

Waits for a response from a given URL. This is typically used to block until a service in the environment has started and is responding. Note that it does not assert any sort of return code, only check that the connection was successful.

This is a convenience method that internally uses HttpWaitStrategy. For more complex wait scenarios, consider using the structured wait strategies with waiting_for().

Parameters:: url – URL from one of the services in the environment to use to wait on.

Example

# Simple URL wait (legacy style) compose.wait_for(”http://localhost:8080”) # For more complex scenarios, use structured wait strategies: from testcontainers.core.waiting_utils import HttpWaitStrategy, LogMessageWaitStrategy compose.waiting_for({ “web”: HttpWaitStrategy(8080).for_status_code(200), “db”: LogMessageWaitStrategy(“database system is ready to accept connections”) })

waiting_for(strategies: dict[str, WaitStrategy]) → Self¶

Set wait strategies for specific services.

Parameters:: strategies – Dictionary mapping service names to wait strategies

class PublishedPortModel(URL: str | None = None, TargetPort: int | None = None, PublishedPort: int | None = None, Protocol: str | None = None)¶: Class that represents the response we get from compose when inquiring status via DockerCompose.get_running_containers().

Examples¶

Using DockerContainer and DockerImage to create a container:

>>> from testcontainers.core.container import DockerContainer
>>> from testcontainers.core.waiting_utils import wait_for_logs
>>> from testcontainers.core.image import DockerImage

>>> with DockerImage(path=f"{TEST_DIR}/core/image_fixtures/sample/", tag="test-sample:latest") as image:
...     with DockerContainer(str(image)) as container:
...         delay = wait_for_logs(container, "Test Sample Image")

The DockerImage class is used to build the image from the specified path and tag. The DockerContainer class is then used to create a container from the image.

Copying a file from disk into a container:

>>> import tempfile
>>> from pathlib import Path
>>> from testcontainers.core.container import DockerContainer

>>> with tempfile.TemporaryDirectory() as tmp:
...     my_file = Path(tmp) / "my_file.txt"
...     _ = my_file.write_text("file content")
...     with DockerContainer("bash", command="sleep infinity") as container:
...         container.copy_into_container(my_file, "/tmp/my_file.txt")
...         result = container.exec("cat /tmp/my_file.txt")
...         result.output
b'file content'

Testcontainers Core¶

Compose¶

Examples¶

testcontainers

Navigation

Related Topics