or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

tessl/pypi-pytest-docker

Simple pytest fixtures for Docker and Docker Compose based tests

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pytest-docker@3.2.x

To install, run

npx @tessl/cli install tessl/pypi-pytest-docker@3.2.0

index.mddocs/

pytest-docker

Simple pytest fixtures for Docker and Docker Compose based tests. This library simplifies integration testing with containerized services by providing automatic lifecycle management, port discovery, and service readiness checking.

Package Information

  • Package Name: pytest-docker
  • Package Type: pypi
  • Language: Python
  • Installation: pip install pytest-docker
  • Plugin Entry Point: pytest11 = docker = pytest_docker

Core Imports

import pytest_docker

For accessing fixtures directly (optional, fixtures are automatically available):

from pytest_docker import (
    docker_ip,
    docker_services,
    docker_compose_file,
    docker_compose_project_name,
    docker_compose_command,
    docker_setup,
    docker_cleanup,
    Services
)

Basic Usage

Create a docker-compose.yml file in your tests directory:

version: '2'
services:
  httpbin:
    image: "kennethreitz/httpbin"
    ports:
      - "8000:80"

Write integration tests using the fixtures:

import pytest
import requests
from requests.exceptions import ConnectionError


def is_responsive(url):
    """Check if service is responsive."""
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return True
    except ConnectionError:
        return False


@pytest.fixture(scope="session")
def http_service(docker_ip, docker_services):
    """Ensure that HTTP service is up and responsive."""
    # Get the host port for container port 80
    port = docker_services.port_for("httpbin", 80)
    url = "http://{}:{}".format(docker_ip, port)
    
    # Wait for service to be ready
    docker_services.wait_until_responsive(
        timeout=30.0, 
        pause=0.1, 
        check=lambda: is_responsive(url)
    )
    return url


def test_status_code(http_service):
    """Test HTTP service returns expected status codes."""
    status = 418
    response = requests.get(http_service + "/status/{}".format(status))
    assert response.status_code == status

Capabilities

Docker IP Discovery

Automatically determines the correct IP address for connecting to Docker containers based on your Docker configuration.

@pytest.fixture(scope="session")
def docker_ip() -> str:
    """
    Determine the IP address for TCP connections to Docker containers.
    
    Returns:
        str: IP address for Docker connections (usually "127.0.0.1")
    """

Docker Compose Configuration

Configures Docker Compose command, file locations, and project naming for container management.

@pytest.fixture(scope="session")
def docker_compose_command() -> str:
    """
    Docker Compose command to use for container operations.
    
    Returns:
        str: Command name, default "docker compose" (Docker Compose V2)
    """

@pytest.fixture(scope="session")
def docker_compose_file(pytestconfig) -> Union[List[str], str]:
    """
    Get absolute path to docker-compose.yml file(s).
    
    Args:
        pytestconfig: pytest configuration object
    
    Returns:
        Union[List[str], str]: Path or list of paths to compose files
    """

@pytest.fixture(scope="session")
def docker_compose_project_name() -> str:
    """
    Generate unique project name for Docker Compose.
    
    Returns:
        str: Project name (default: "pytest{PID}")
    """

Container Lifecycle Management

Controls Docker container startup and cleanup behavior with configurable commands.

@pytest.fixture(scope="session")
def docker_setup() -> Union[List[str], str]:
    """
    Get docker-compose commands for container setup.
    
    Returns:
        Union[List[str], str]: Setup commands (default: ["up --build -d"])
    """

@pytest.fixture(scope="session")
def docker_cleanup() -> Union[List[str], str]:
    """
    Get docker-compose commands for container cleanup.
    
    Returns:
        Union[List[str], str]: Cleanup commands (default: ["down -v"])
    """

def get_setup_command() -> Union[List[str], str]:
    """
    Get default setup command for Docker containers.
    
    Returns:
        Union[List[str], str]: Default setup command list
    """

def get_cleanup_command() -> Union[List[str], str]:
    """
    Get default cleanup command for Docker containers.
    
    Returns:
        Union[List[str], str]: Default cleanup command list
    """

Service Management

Main interface for interacting with running Docker services, including port discovery and readiness checking.

@pytest.fixture(scope="session")
def docker_services(
    docker_compose_command: str,
    docker_compose_file: Union[List[str], str],
    docker_compose_project_name: str,
    docker_setup: Union[List[str], str],
    docker_cleanup: Union[List[str], str]
) -> Iterator[Services]:
    """
    Start Docker services and provide Services interface.
    
    Automatically starts containers before tests and cleans up afterward.
    
    Args:
        docker_compose_command: Docker Compose command to use
        docker_compose_file: Path(s) to docker-compose.yml file(s)
        docker_compose_project_name: Project name for container isolation
        docker_setup: Command(s) to run for container setup
        docker_cleanup: Command(s) to run for container cleanup
    
    Yields:
        Services: Interface for service interaction
    """

Service Interaction

The Services class provides methods for port discovery and service readiness verification.

class Services:
    """Interface for interacting with Docker services."""
    
    def port_for(self, service: str, container_port: int) -> int:
        """
        Get host port mapped to container port for a service.
        
        Args:
            service: Name of the Docker service from docker-compose.yml
            container_port: Port number inside the container
            
        Returns:
            int: Host port number that maps to the container port
            
        Raises:
            ValueError: If port mapping cannot be determined
        """
    
    def wait_until_responsive(
        self,
        check: Any,
        timeout: float,
        pause: float,
        clock: Any = timeit.default_timer
    ) -> None:
        """
        Wait until a service responds or timeout is reached.
        
        Args:
            check: Function that returns True when service is ready
            timeout: Maximum time to wait in seconds
            pause: Time to wait between checks in seconds
            clock: Timer function (default: timeit.default_timer)
            
        Raises:
            Exception: If timeout is reached before service responds
        """

Command Execution

Low-level utilities for executing Docker Compose commands and shell operations.

class DockerComposeExecutor:
    """Executes Docker Compose commands with proper file and project configuration."""
    
    def __init__(
        self,
        compose_command: str,
        compose_files: Union[List[str], str],
        compose_project_name: str
    ):
        """
        Initialize executor with Docker Compose configuration.
        
        Args:
            compose_command: Docker Compose command ("docker compose" or "docker-compose")
            compose_files: Path or list of paths to compose files
            compose_project_name: Project name for container isolation
        """
    
    def execute(self, subcommand: str, **kwargs) -> bytes:
        """
        Execute Docker Compose subcommand.
        
        Args:
            subcommand: Docker Compose subcommand (e.g., "up -d", "down")
            **kwargs: Additional arguments passed to shell execution
            
        Returns:
            bytes: Command output
            
        Raises:
            Exception: If command fails with non-zero exit code
        """

def execute(
    command: str, 
    success_codes: Iterable[int] = (0,), 
    ignore_stderr: bool = False
) -> bytes:
    """
    Execute shell command with error handling.
    
    Args:
        command: Shell command to execute
        success_codes: Acceptable exit codes (default: (0,))
        ignore_stderr: Whether to ignore stderr output
        
    Returns:
        bytes: Command output
        
    Raises:
        Exception: If command returns unacceptable exit code
    """

def get_docker_ip() -> str:
    """
    Determine Docker host IP from DOCKER_HOST environment variable.
    
    Returns:
        str: IP address for Docker connections
    """

def container_scope_fixture(request: FixtureRequest) -> Any:
    """
    Get container scope from pytest request configuration.
    
    Args:
        request: pytest fixture request object
        
    Returns:
        Any: Container scope configuration
    """

def containers_scope(fixture_name: str, config: Config) -> Any:
    """
    Determine container scope for fixtures based on pytest configuration.
    
    Args:
        fixture_name: Name of the fixture
        config: pytest configuration object
        
    Returns:
        Any: Container scope for the fixture
    """

Plugin Configuration

Configures pytest plugin integration and command-line options.

def pytest_addoption(parser: pytest.Parser) -> None:
    """
    Add pytest command-line options for Docker container configuration.
    
    Args:
        parser: pytest argument parser
    """

Configuration Options

Container Scope

Control fixture scope using the --container-scope command line option:

pytest --container-scope session  # Default: containers shared across test session
pytest --container-scope module   # New containers for each test module
pytest --container-scope class    # New containers for each test class
pytest --container-scope function # New containers for each test function

Docker Compose V1 Support

For legacy Docker Compose V1 (docker-compose command):

@pytest.fixture(scope="session")
def docker_compose_command() -> str:
    return "docker-compose"

Or install with V1 support:

pip install pytest-docker[docker-compose-v1]

Custom Docker Compose File Location

Override the default location (tests/docker-compose.yml):

import os
import pytest

@pytest.fixture(scope="session")
def docker_compose_file(pytestconfig):
    return os.path.join(str(pytestconfig.rootdir), "custom", "docker-compose.yml")

Multiple Compose Files

Use multiple compose files for complex configurations:

@pytest.fixture(scope="session")
def docker_compose_file(pytestconfig):
    return [
        os.path.join(str(pytestconfig.rootdir), "tests", "compose.yml"),
        os.path.join(str(pytestconfig.rootdir), "tests", "compose.override.yml"),
    ]

Custom Project Name

Pin project name to avoid conflicts during debugging:

@pytest.fixture(scope="session")
def docker_compose_project_name() -> str:
    return "my-test-project"

Custom Setup and Cleanup

Modify container lifecycle commands:

@pytest.fixture(scope="session")
def docker_setup():
    return ["down -v", "up --build -d"]  # Cleanup first, then start

@pytest.fixture(scope="session") 
def docker_cleanup():
    return ["down -v", "system prune -f"]  # Extended cleanup

Types

from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
from _pytest.config import Config
from _pytest.fixtures import FixtureRequest
import timeit
import pytest

# Type aliases for clarity
PortMapping = Dict[str, Dict[int, int]]  # Service name -> {container_port: host_port}

Error Handling

Common exceptions and error patterns:

  • ValueError: Raised by Services.port_for() when port mapping cannot be determined
  • Exception: Raised by Services.wait_until_responsive() when timeout is reached
  • Exception: Raised by execute() functions when shell commands fail
  • subprocess.CalledProcessError: Underlying exception for command execution failures

Usage Patterns

Database Testing

@pytest.fixture(scope="session")
def postgres_service(docker_ip, docker_services):
    """Ensure PostgreSQL is ready for connections."""
    port = docker_services.port_for("postgres", 5432)
    
    def is_ready():
        try:
            conn = psycopg2.connect(
                host=docker_ip, 
                port=port, 
                user="test", 
                password="test",
                database="testdb"
            )
            conn.close()
            return True
        except psycopg2.OperationalError:
            return False
    
    docker_services.wait_until_responsive(
        timeout=60.0,
        pause=1.0,
        check=is_ready
    )
    
    return {
        "host": docker_ip,
        "port": port,
        "user": "test",
        "password": "test",
        "database": "testdb"
    }

API Testing

@pytest.fixture(scope="session")
def api_service(docker_ip, docker_services):
    """Ensure API service is ready."""
    port = docker_services.port_for("api", 3000)
    url = f"http://{docker_ip}:{port}"
    
    docker_services.wait_until_responsive(
        timeout=30.0,
        pause=0.5,
        check=lambda: requests.get(f"{url}/health").status_code == 200
    )
    
    return url

Multi-Service Testing

@pytest.fixture(scope="session")
def full_stack(docker_ip, docker_services):
    """Ensure all services are ready."""
    # Wait for database
    db_port = docker_services.port_for("database", 5432)
    # Wait for cache
    cache_port = docker_services.port_for("redis", 6379)
    # Wait for API (depends on db and cache)
    api_port = docker_services.port_for("api", 8000)
    
    # Check services in dependency order  
    docker_services.wait_until_responsive(
        timeout=60.0, pause=1.0,
        check=lambda: check_postgres(docker_ip, db_port)
    )
    
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.5,
        check=lambda: check_redis(docker_ip, cache_port)
    )
    
    docker_services.wait_until_responsive(
        timeout=45.0, pause=1.0,
        check=lambda: check_api_health(docker_ip, api_port)
    )
    
    return {
        "database": {"host": docker_ip, "port": db_port},
        "cache": {"host": docker_ip, "port": cache_port},
        "api": {"host": docker_ip, "port": api_port}
    }