Python library for throwaway instances of anything that can run in a Docker container
npx @tessl/cli install tessl/pypi-testcontainers@4.12.0A comprehensive Python library for managing throwaway Docker instances in tests. Testcontainers enables developers to create isolated, reproducible test environments by programmatically spinning up Docker containers for databases, message queues, web services, and other infrastructure components during test execution.
pip install testcontainersfrom testcontainers.core.container import DockerContainer
from testcontainers.core.network import Network
from testcontainers.core.waiting_utils import wait_for_logs, wait_forFor specialized containers:
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from testcontainers.mysql import MySqlContainer
from testcontainers.mongodb import MongoDbContainer
from testcontainers.kafka import KafkaContainer
from testcontainers.elasticsearch import ElasticSearchContainerFor compose orchestration:
from testcontainers.compose import DockerComposeFor configuration and exceptions:
from testcontainers.core.config import testcontainers_config
from testcontainers.core.exceptions import (
ContainerStartException,
ContainerConnectException,
NoSuchPortExposed
)from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs
# Basic container usage with context manager
with DockerContainer("hello-world") as container:
delay = wait_for_logs(container, "Hello from Docker!")
# Database container example
from testcontainers.postgres import PostgresContainer
with PostgresContainer("postgres:13") as postgres:
# Container automatically configured with database, user, password
connection_url = postgres.get_connection_url()
# Use with your database client
import psycopg2
conn = psycopg2.connect(connection_url)
cursor = conn.cursor()
cursor.execute("SELECT version();")
result = cursor.fetchone()
print(result)
# Compose orchestration example
from testcontainers.compose import DockerCompose
with DockerCompose(".", compose_file_name="docker-compose.test.yml") as compose:
# Get specific service container
web_container = compose.get_container("web")
db_container = compose.get_container("db")
# Get service endpoints
web_url = compose.get_service_host("web", 80)
db_host = compose.get_service_host("db", 5432)Testcontainers follows a layered architecture enabling flexible container management:
This design provides both simplicity for common use cases and extensibility for complex testing scenarios, integrating seamlessly with pytest, unittest, and other Python testing frameworks.
Fundamental Docker container lifecycle management with comprehensive configuration options, networking, volume mounting, and environment setup. Provides the foundation for all specialized containers.
class DockerContainer:
def __init__(
self,
image: str,
docker_client_kw: Optional[dict] = None,
command: Optional[str] = None,
env: Optional[dict] = None,
name: Optional[str] = None,
ports: Optional[list] = None,
volumes: Optional[list] = None,
network: Optional[Network] = None,
network_aliases: Optional[list] = None,
**kwargs
): ...
def start(self) -> "DockerContainer": ...
def stop(self, force: bool = True, delete_volume: bool = True) -> None: ...
def get_exposed_port(self, port: int) -> str: ...
def get_container_host_ip(self) -> str: ...
def with_env(self, key: str, value: str) -> "DockerContainer": ...
def with_exposed_ports(self, *ports: int) -> "DockerContainer": ...
def with_bind_ports(self, container: int, host: Optional[int] = None) -> "DockerContainer": ...
def with_volume_mapping(self, host: str, container: str, mode: str = "ro") -> "DockerContainer": ...Pre-configured containers for popular databases including PostgreSQL, MySQL, MongoDB, Redis, and many others. Each provides service-specific configuration options and connection utilities.
class PostgresContainer:
def __init__(
self,
image: str = "postgres:latest",
port: int = 5432,
username: Optional[str] = None,
password: Optional[str] = None,
dbname: Optional[str] = None,
**kwargs
): ...
def get_connection_url(self, host: Optional[str] = None) -> str: ...
class MySqlContainer:
def __init__(
self,
image: str = "mysql:latest",
username: Optional[str] = None,
password: Optional[str] = None,
dbname: Optional[str] = None,
**kwargs
): ...
def get_connection_url(self) -> str: ...
class MongoDbContainer:
def __init__(
self,
image: str = "mongo:latest",
port: int = 27017,
username: Optional[str] = None,
password: Optional[str] = None,
**kwargs
): ...
def get_connection_url(self) -> str: ...
def get_connection_client(self): ...Containers for caching systems, message queues, and pub/sub services including Redis, Kafka, RabbitMQ, NATS, and messaging brokers with client integration.
class RedisContainer:
def __init__(
self,
image: str = "redis:latest",
port: int = 6379,
password: Optional[str] = None,
**kwargs
): ...
def get_client(self, **kwargs): ...
class KafkaContainer:
def __init__(
self,
image: str = "confluentinc/cp-kafka:7.6.0",
port: int = 9093,
**kwargs
): ...
def get_bootstrap_server(self) -> str: ...
def with_kraft(self) -> "KafkaContainer": ...Complete Docker Compose integration for managing multi-container environments, service discovery, and complex application stacks during testing.
class DockerCompose:
def __init__(
self,
context: str,
compose_file_name: Optional[str] = None,
pull: bool = False,
build: bool = False,
wait: bool = True,
**kwargs
): ...
def start(self) -> "DockerCompose": ...
def stop(self, down: bool = True) -> None: ...
def get_container(self, service_name: str) -> ComposeContainer: ...
def get_service_host(self, service_name: str, port: int) -> str: ...
def get_service_port(self, service_name: str, port: int) -> int: ...
def exec_in_container(self, command: str, service_name: str): ...Robust container readiness detection, log monitoring, and condition waiting utilities for reliable test execution across different container types and startup behaviors.
def wait_container_is_ready(*transient_exceptions) -> Callable: ...
def wait_for_logs(
container: DockerContainer,
predicate: Union[str, Callable],
timeout: float = 120,
interval: float = 1,
**kwargs
) -> float: ...
def wait_for(condition: Callable, timeout: float = 120, interval: float = 1) -> bool: ...Specialized containers for search engines, analytics platforms, and data processing including Elasticsearch, OpenSearch, ClickHouse, and vector databases.
class ElasticSearchContainer:
def __init__(self, image: str = "elasticsearch", port: int = 9200, **kwargs): ...
def get_url(self) -> str: ...
class ClickHouseContainer:
def __init__(self, image: str = "clickhouse/clickhouse-server", **kwargs): ...
def get_connection_url(self) -> str: ...Containers for cloud service emulation and integration including LocalStack for AWS services, Azure emulators, and Google Cloud Platform services for local development and testing.
class LocalStackContainer:
def __init__(
self,
image: str = "localstack/localstack:2.0.1",
edge_port: int = 4566,
**kwargs
): ...
def with_services(self, *services: str) -> "LocalStackContainer": ...
def get_url(self) -> str: ...
def get_client(self, name: str, **kwargs): ...Containers for web services, browser automation, and testing infrastructure including Nginx, Selenium WebDriver, and specialized testing utilities.
class BrowserWebDriverContainer:
def __init__(
self,
capabilities: dict,
image: Optional[str] = None,
port: int = 4444,
**kwargs
): ...
def get_driver(self): ...
def get_connection_url(self) -> str: ...
def with_options(self, options) -> "BrowserWebDriverContainer": ...
class NginxContainer:
def __init__(self, image: str = "nginx:alpine", port: int = 80, **kwargs): ...
def get_url(self) -> str: ...Additional specialized containers for various development and testing needs.
class VaultContainer:
def __init__(self, image: str = "vault:latest", port: int = 8200, **kwargs): ...
def get_url(self) -> str: ...
class MailpitContainer:
def __init__(self, image: str = "axllent/mailpit:latest", **kwargs): ...
def get_smtp_host(self) -> str: ...
def get_web_url(self) -> str: ...
class OllamaContainer:
def __init__(self, image: str = "ollama/ollama:latest", **kwargs): ...
def get_endpoint_url(self) -> str: ...
class SftpContainer:
def __init__(self, image: str = "atmoz/sftp:latest", **kwargs): ...
def get_connection_url(self) -> str: ...from testcontainers.core.config import testcontainers_config
# Configuration properties
testcontainers_config.max_tries: int
testcontainers_config.sleep_time: int
testcontainers_config.timeout: int
testcontainers_config.ryuk_disabled: boolclass ContainerStartException(RuntimeError): ...
class ContainerConnectException(RuntimeError): ...
class ContainerIsNotRunning(RuntimeError): ...
class NoSuchPortExposed(RuntimeError): ...The library includes 45+ specialized container modules providing pre-configured containers for popular services:
Databases: postgres, mysql, mongodb, redis, cassandra, clickhouse, cockroachdb, cosmosdb, db2, influxdb, mssql, neo4j, oracle-free, scylla, trino
Vector Databases: chroma, milvus, qdrant, weaviate
Message Brokers: kafka, mqtt, nats, rabbitmq
Search & Analytics: elasticsearch, opensearch
Storage & Cache: azurite, memcached, minio, registry, vault
Development Tools: generic, keycloak, k3s, mailpit, nginx, ollama, selenium, sftp
Cloud Services: aws, google, localstack, openfga
Each specialized container follows similar patterns with service-specific configuration methods and client getters appropriate for the service type.
from testcontainers.core.exceptions import (
ContainerStartException,
ContainerConnectException,
ContainerIsNotRunning,
NoSuchPortExposed
)
class ContainerStartException(RuntimeError):
"""Raised when container fails to start properly."""
class ContainerConnectException(RuntimeError):
"""Raised when connection to container fails."""
class ContainerIsNotRunning(RuntimeError):
"""Raised when operation requires running container but container is not running."""
class NoSuchPortExposed(RuntimeError):
"""Raised when trying to access a port that was not exposed."""