Waiting Strategies and Utilities

def wait_container_is_ready(*transient_exceptions: type[BaseException]) -> Callable:
    """
    Decorator for container readiness checks with retry logic.
    
    Automatically retries decorated function until success or timeout.
    Handles common transient exceptions plus any additional specified exceptions.
    
    Args:
        *transient_exceptions: Additional exception types to treat as transient
        
    Returns:
        Decorator function that wraps the target method
        
    Usage:
        @wait_container_is_ready(CustomException)
        def _connect(self):
            # Connection logic that may fail transiently
            pass
    """

def wait_for_logs(
    container: DockerContainer,
    predicate: Union[Callable[..., bool], str],
    timeout: Union[float, None] = None,
    interval: float = 1,
    predicate_streams_and: bool = False,
    raise_on_exit: bool = False
) -> float:
    """
    Wait for specific log output from container.
    
    Args:
        container: Container to monitor
        predicate: String to search for or callable returning bool
        timeout: Maximum wait time in seconds
        interval: Polling interval in seconds
        predicate_streams_and: Apply predicate to both stdout and stderr
        raise_on_exit: Raise exception if container exits
        
    Returns:
        Time elapsed until condition was met
        
    Raises:
        TimeoutError: If timeout reached without condition being met
        ContainerStartException: If container exits unexpectedly
    """

def wait_for(
    condition: Callable[[], bool],
    timeout: float = 120,
    interval: float = 1
) -> bool:
    """
    Wait for generic condition to be met.
    
    Args:
        condition: Function returning True when condition is met
        timeout: Maximum wait time in seconds
        interval: Polling interval in seconds
        
    Returns:
        True if condition was met, False if timeout reached
    """

from testcontainers.core.config import testcontainers_config

# Configure waiting behavior
testcontainers_config.max_tries: int = 120    # Maximum retry attempts
testcontainers_config.sleep_time: int = 1     # Sleep between retries (seconds)
testcontainers_config.timeout: int = 120      # Total timeout (seconds)

TRANSIENT_EXCEPTIONS = (TimeoutError, ConnectionError)

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

# Wait for application startup message
with DockerContainer("my-app:latest") as container:
    # Wait for specific log message indicating readiness
    delay = wait_for_logs(container, "Server started successfully")
    print(f"Application ready after {delay:.2f} seconds")
    
    # Now safe to connect to the application
    app_port = container.get_exposed_port(8080)
    # Make requests to the application...

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

with DockerContainer("postgres:13") as postgres:
    postgres.with_env("POSTGRES_PASSWORD", "test")
    
    # Wait for PostgreSQL to be ready using regex pattern
    def postgres_ready(log_line):
        return re.search(r"database system is ready to accept connections", log_line) is not None
    
    delay = wait_for_logs(postgres, postgres_ready, timeout=30)
    print(f"PostgreSQL ready after {delay:.2f} seconds")

from testcontainers.redis import RedisContainer
from testcontainers.core.waiting_utils import wait_for
import redis
import time

with RedisContainer() as redis_container:
    redis_client = redis_container.get_client()
    
    # Wait for Redis to accept connections
    def redis_ready():
        try:
            return redis_client.ping()
        except:
            return False
    
    success = wait_for(redis_ready, timeout=30, interval=0.5)
    if success:
        print("Redis is ready for connections")
    else:
        print("Redis failed to become ready within timeout")

from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for
import requests

with DockerContainer("nginx:alpine") as web_server:
    web_server.with_exposed_ports(80)
    
    host = web_server.get_container_host_ip()
    port = web_server.get_exposed_port(80)
    
    # Wait for HTTP endpoint to respond
    def http_ready():
        try:
            response = requests.get(f"http://{host}:{port}/", timeout=1)
            return response.status_code == 200
        except:
            return False
    
    if wait_for(http_ready, timeout=60, interval=2):
        print("Web server is responding to HTTP requests")
        # Proceed with tests...

from testcontainers.postgres import PostgresContainer
from testcontainers.core.waiting_utils import wait_container_is_ready
import psycopg2

class CustomPostgresContainer(PostgresContainer):
    @wait_container_is_ready(psycopg2.OperationalError)
    def _connect(self):
        """Custom connection method with automatic retry."""
        conn = psycopg2.connect(self.get_connection_url())
        cursor = conn.cursor()
        cursor.execute("SELECT 1")
        cursor.fetchone()
        conn.close()

# Use custom container with automatic connection retry
with CustomPostgresContainer("postgres:13") as postgres:
    # Container automatically waits for successful connection
    connection_url = postgres.get_connection_url()
    print(f"PostgreSQL ready at: {connection_url}")

from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs, wait_for
import requests
import time

class WebAppContainer(DockerContainer):
    def __init__(self, image):
        super().__init__(image)
        self.with_exposed_ports(8080)
    
    def wait_for_readiness(self):
        """Wait for multiple readiness conditions."""
        # First, wait for application startup logs
        wait_for_logs(self, "Application started", timeout=60)
        
        # Then wait for health endpoint to respond
        host = self.get_container_host_ip()
        port = self.get_exposed_port(8080)
        
        def health_check():
            try:
                response = requests.get(f"http://{host}:{port}/health", timeout=2)
                return response.status_code == 200 and response.json().get("status") == "healthy"
            except:
                return False
        
        if not wait_for(health_check, timeout=30):
            raise Exception("Application failed health check")
        
        print("Application is fully ready")

# Use comprehensive readiness checking
with WebAppContainer("my-web-app:latest") as app:
    app.wait_for_readiness()
    # Application is now fully ready for testing

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

# Different containers may need different timeout strategies
containers = [
    ("redis:6", "Ready to accept connections", 15),
    ("postgres:13", "database system is ready", 45),
    ("elasticsearch:7.15.0", "started", 120)
]

for image, log_pattern, timeout in containers:
    with DockerContainer(image) as container:
        try:
            delay = wait_for_logs(container, log_pattern, timeout=timeout)
            print(f"{image} ready after {delay:.2f}s")
        except TimeoutError:
            print(f"{image} failed to start within {timeout}s")
            # Handle timeout appropriately

from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer  
from testcontainers.core.waiting_utils import wait_for
import threading
import time

def start_and_wait(container, name):
    """Start container and wait for readiness."""
    container.start()
    
    # Different waiting strategies per container type
    if isinstance(container, PostgresContainer):
        def pg_ready():
            try:
                import psycopg2
                conn = psycopg2.connect(container.get_connection_url())
                conn.close()
                return True
            except:
                return False
        wait_for(pg_ready, timeout=45)
    
    elif isinstance(container, RedisContainer):
        client = container.get_client()
        wait_for(lambda: client.ping(), timeout=15)
    
    print(f"{name} is ready")

# Start containers in parallel
postgres = PostgresContainer("postgres:13")
redis = RedisContainer("redis:6")

threads = [
    threading.Thread(target=start_and_wait, args=(postgres, "PostgreSQL")),
    threading.Thread(target=start_and_wait, args=(redis, "Redis"))
]

start_time = time.time()
for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print(f"All containers ready in {time.time() - start_time:.2f} seconds")

# Clean up
postgres.stop()
redis.stop()

from testcontainers.core.exceptions import (
    ContainerStartException,
    ContainerConnectException,
    TimeoutError
)

try:
    with DockerContainer("problematic-image") as container:
        wait_for_logs(container, "ready", timeout=30)
except ContainerStartException:
    print("Container failed to start")
except TimeoutError:
    print("Container did not become ready within timeout")
except ContainerConnectException:
    print("Failed to connect to container")

from testcontainers.core.waiting_utils import wait_for
import logging

def wait_with_fallback(condition, primary_timeout=60, fallback_timeout=30):
    """Wait with fallback strategy."""
    try:
        if wait_for(condition, timeout=primary_timeout):
            return True
        else:
            logging.warning(f"Primary wait timed out after {primary_timeout}s, trying fallback")
            return wait_for(condition, timeout=fallback_timeout, interval=0.1)
    except Exception as e:
        logging.error(f"Wait failed: {e}")
        return False