tessl/pypi-testcontainers
Python library for throwaway instances of anything that can run in a Docker container
compose.mddocs/
Docker Compose Orchestration
Complete Docker Compose integration for managing multi-container environments, service discovery, and complex application stacks during testing. Enables full orchestration of interconnected services with configuration management and lifecycle control.
Capabilities
Compose Environment Management
Manage entire Docker Compose environments with automatic service startup, configuration loading, and coordinated shutdown.
@dataclass
class DockerCompose:
context: Union[str, PathLike[str]]
compose_file_name: Optional[Union[str, list[str]]] = None
pull: bool = False
build: bool = False
wait: bool = True
keep_volumes: bool = False
env_file: Optional[str] = None
services: Optional[list[str]] = None
docker_command_path: Optional[str] = None
profiles: Optional[list[str]] = None
"""
Initialize Docker Compose environment.
Args:
context: Path to compose context directory
compose_file_name: Compose file name (default: docker-compose.yml)
pull: Pull images before starting
build: Build images before starting
wait: Wait for services to be ready
keep_volumes: Preserve volumes on shutdown
env_file: Environment file path
services: Specific services to run
docker_command_path: Custom docker-compose command path
profiles: Compose profiles to activate
**kwargs: Additional compose options
"""
def start(self) -> "DockerCompose":
"""
Start the compose environment.
Returns:
Self for method chaining
"""
def stop(self, down: bool = True) -> None:
"""
Stop the compose environment.
Args:
down: Use 'docker-compose down' instead of 'stop'
"""
def __enter__(self) -> "DockerCompose":
"""Context manager entry - starts compose environment."""
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Context manager exit - stops compose environment."""Service Discovery and Access
Access individual services within the compose environment, retrieve connection information, and interact with running containers.
def get_container(self, service_name: Optional[str] = None, include_all: bool = False) -> ComposeContainer:
"""
Get container for specific service.
Args:
service_name: Service name (first service if None)
include_all: Include stopped containers
Returns:
ComposeContainer instance
"""
def get_containers(self, include_all: bool = False) -> list[ComposeContainer]:
"""
Get all containers in the compose environment.
Args:
include_all: Include stopped containers
Returns:
List of ComposeContainer instances
"""
def get_service_host(self, service_name: Optional[str] = None, port: Optional[int] = None) -> str:
"""
Get host address for service.
Args:
service_name: Service name
port: Service port
Returns:
Host address string
"""
def get_service_port(self, service_name: Optional[str] = None, port: Optional[int] = None) -> int:
"""
Get mapped port for service.
Args:
service_name: Service name
port: Internal service port
Returns:
Mapped host port number
"""
def get_service_host_and_port(self, service_name: Optional[str] = None, port: Optional[int] = None) -> tuple[str, int]:
"""
Get host and port for service.
Args:
service_name: Service name
port: Internal service port
Returns:
Tuple of (host, port)
"""Container Operations
Execute commands in running services, retrieve logs, and interact with the compose environment.
def exec_in_container(self, command: str, service_name: Optional[str] = None) -> str:
"""
Execute command in service container.
Args:
command: Command to execute
service_name: Target service name
Returns:
Command output string
"""
def get_logs(self, *services: str) -> str:
"""
Get logs from services.
Args:
*services: Service names (all services if none specified)
Returns:
Combined log output string
"""
def get_config(
self,
path_resolution: bool = True,
normalize: bool = True,
interpolate: bool = True
) -> dict:
"""
Get compose configuration.
Args:
path_resolution: Resolve file paths
normalize: Normalize configuration format
interpolate: Interpolate environment variables
Returns:
Compose configuration dictionary
"""Service Health Checking
Wait for services to become available and ready for connections.
def wait_for(self, url: str) -> None:
"""
Wait for URL to become available.
Args:
url: URL to check for availability
"""Container Information
Access detailed information about individual containers within the compose environment.
class ComposeContainer:
ID: str # Container ID
Name: str # Container name
Command: str # Container command
Project: str # Compose project name
Service: str # Service name
State: str # Container state
Health: str # Health status
ExitCode: int # Exit code
Publishers: list[PublishedPortModel] # Published ports
def get_publisher(
self,
by_port: Optional[int] = None,
by_host: Optional[str] = None,
prefer_ip_version: str = "IPv4"
) -> PublishedPortModel:
"""
Get port publisher information.
Args:
by_port: Filter by port number
by_host: Filter by host address
prefer_ip_version: Preferred IP version ("IPv4" or "IPv6")
Returns:
PublishedPortModel instance
"""
class PublishedPortModel:
URL: str # Published URL
TargetPort: int # Target container port
PublishedPort: int # Published host port
Protocol: str # Protocol (tcp/udp)
def normalize(self) -> "PublishedPortModel":
"""
Normalize for Windows compatibility.
Returns:
Normalized PublishedPortModel
"""Usage Examples
Basic Compose Usage
from testcontainers.compose import DockerCompose
import requests
# docker-compose.yml in current directory with web and db services
with DockerCompose(".") as compose:
# Get service endpoints
web_host = compose.get_service_host("web", 80)
web_port = compose.get_service_port("web", 80)
# Make request to web service
response = requests.get(f"http://{web_host}:{web_port}/health")
assert response.status_code == 200
# Get database connection info
db_host = compose.get_service_host("db", 5432)
db_port = compose.get_service_port("db", 5432)
print(f"Database available at {db_host}:{db_port}")Custom Compose File
from testcontainers.compose import DockerCompose
# Use specific compose file and environment
compose = DockerCompose(
context="./docker",
compose_file_name="docker-compose.test.yml",
pull=True, # Pull latest images
build=True, # Build custom images
env_file="test.env"
)
with compose:
# Execute command in service
result = compose.exec_in_container("ls -la", service_name="app")
print(f"Container contents: {result}")
# Get logs from specific services
logs = compose.get_logs("app", "worker")
print(f"Service logs: {logs}")Service-Specific Operations
from testcontainers.compose import DockerCompose
with DockerCompose(".", compose_file_name="microservices.yml") as compose:
# Get all running containers
containers = compose.get_containers()
for container in containers:
print(f"Service: {container.Service}")
print(f"State: {container.State}")
print(f"Health: {container.Health}")
# Get port information
for publisher in container.Publishers:
print(f"Port {publisher.TargetPort} -> {publisher.PublishedPort}")
# Access specific service container
api_container = compose.get_container("api")
print(f"API container ID: {api_container.ID}")Profile-Based Deployment
from testcontainers.compose import DockerCompose
# Use compose profiles for different environments
test_compose = DockerCompose(
context=".",
profiles=["test", "monitoring"],
services=["app", "db", "redis"] # Only start specific services
)
with test_compose:
# Only services in 'test' and 'monitoring' profiles are started
app_url = f"http://{test_compose.get_service_host('app', 8080)}:{test_compose.get_service_port('app', 8080)}"
print(f"Test app available at: {app_url}")Integration Testing Setup
from testcontainers.compose import DockerCompose
import pytest
import requests
@pytest.fixture(scope="session")
def app_stack():
"""Pytest fixture for full application stack."""
with DockerCompose(".", compose_file_name="test-stack.yml") as compose:
# Wait for services to be ready
compose.wait_for(f"http://{compose.get_service_host('app', 8080)}:{compose.get_service_port('app', 8080)}/health")
yield compose
def test_api_endpoints(app_stack):
"""Test API endpoints with full stack."""
compose = app_stack
# Get API endpoint
api_host = compose.get_service_host("api", 3000)
api_port = compose.get_service_port("api", 3000)
base_url = f"http://{api_host}:{api_port}"
# Test endpoints
response = requests.get(f"{base_url}/users")
assert response.status_code == 200
response = requests.post(f"{base_url}/users", json={"name": "Test User"})
assert response.status_code == 201
def test_database_integration(app_stack):
"""Test database operations."""
compose = app_stack
# Execute database command
result = compose.exec_in_container("psql -U postgres -c 'SELECT version();'", "db")
assert "PostgreSQL" in resultComplex Multi-Service Architecture
from testcontainers.compose import DockerCompose
import time
# docker-compose.yml with web, api, worker, db, redis, elasticsearch
with DockerCompose(".", compose_file_name="full-stack.yml") as compose:
# Get all service endpoints
services = {
"web": compose.get_service_host_and_port("web", 80),
"api": compose.get_service_host_and_port("api", 3000),
"db": compose.get_service_host_and_port("db", 5432),
"redis": compose.get_service_host_and_port("redis", 6379),
"elasticsearch": compose.get_service_host_and_port("elasticsearch", 9200)
}
print("Service endpoints:")
for service, (host, port) in services.items():
print(f" {service}: {host}:{port}")
# Wait for all services to be healthy
for container in compose.get_containers():
while container.Health not in ["healthy", ""]:
print(f"Waiting for {container.Service} to be healthy...")
time.sleep(2)
# Refresh container info
container = compose.get_container(container.Service)
print("All services are ready!")
# Run integration tests
web_host, web_port = services["web"]
response = requests.get(f"http://{web_host}:{web_port}")
print(f"Web response status: {response.status_code}")Environment Configuration
from testcontainers.compose import DockerCompose
import os
# Set environment variables for compose
os.environ["DATABASE_URL"] = "postgres://test:test@db:5432/testdb"
os.environ["REDIS_URL"] = "redis://redis:6379"
os.environ["DEBUG"] = "true"
# Compose with environment file and variable interpolation
compose = DockerCompose(
context="./infrastructure",
env_file="test.env",
keep_volumes=False # Clean up volumes after testing
)
with compose:
# Environment variables are available in compose services
config = compose.get_config()
print("Compose configuration:", config)
# Services use interpolated environment variables
app_logs = compose.get_logs("app")
print("Application logs:", app_logs)Install with Tessl CLI
npx tessl i tessl/pypi-testcontainers