0
# Waiting Strategies and Utilities
1

2
Robust container readiness detection, log monitoring, and condition waiting utilities for reliable test execution across different container types and startup behaviors. Essential for ensuring containers are fully ready before test execution begins.
3

4
## Capabilities
5

6
### Container Readiness Decorator
7

8
Decorator for automatic retry logic when connecting to containers, handling transient errors and ensuring reliable container readiness detection.
9

10
```python { .api }
11
def wait_container_is_ready(*transient_exceptions: type[BaseException]) -> Callable:
12
    """
13
    Decorator for container readiness checks with retry logic.
14
    
15
    Automatically retries decorated function until success or timeout.
16
    Handles common transient exceptions plus any additional specified exceptions.
17
    
18
    Args:
19
        *transient_exceptions: Additional exception types to treat as transient
20
        
21
    Returns:
22
        Decorator function that wraps the target method
23
        
24
    Usage:
25
        @wait_container_is_ready(CustomException)
26
        def _connect(self):
27
            # Connection logic that may fail transiently
28
            pass
29
    """
30
```
31

32
### Log-Based Waiting
33

34
Wait for specific log output to appear in container logs, supporting both string patterns and custom predicates.
35

36
```python { .api }
37
def wait_for_logs(
38
    container: DockerContainer,
39
    predicate: Union[Callable[..., bool], str],
40
    timeout: Union[float, None] = None,
41
    interval: float = 1,
42
    predicate_streams_and: bool = False,
43
    raise_on_exit: bool = False
44
) -> float:
45
    """
46
    Wait for specific log output from container.
47
    
48
    Args:
49
        container: Container to monitor
50
        predicate: String to search for or callable returning bool
51
        timeout: Maximum wait time in seconds
52
        interval: Polling interval in seconds
53
        predicate_streams_and: Apply predicate to both stdout and stderr
54
        raise_on_exit: Raise exception if container exits
55
        
56
    Returns:
57
        Time elapsed until condition was met
58
        
59
    Raises:
60
        TimeoutError: If timeout reached without condition being met
61
        ContainerStartException: If container exits unexpectedly
62
    """
63
```
64

65
### Generic Condition Waiting
66

67
Wait for arbitrary conditions to be met with configurable timeout and polling intervals.
68

69
```python { .api }
70
def wait_for(
71
    condition: Callable[[], bool],
72
    timeout: float = 120,
73
    interval: float = 1
74
) -> bool:
75
    """
76
    Wait for generic condition to be met.
77
    
78
    Args:
79
        condition: Function returning True when condition is met
80
        timeout: Maximum wait time in seconds
81
        interval: Polling interval in seconds
82
        
83
    Returns:
84
        True if condition was met, False if timeout reached
85
    """
86
```
87

88
## Configuration
89

90
### Global Timeout Settings
91

92
Container readiness waiting behavior is controlled by global configuration:
93

94
```python { .api }
95
from testcontainers.core.config import testcontainers_config
96

97
# Configure waiting behavior
98
testcontainers_config.max_tries: int = 120    # Maximum retry attempts
99
testcontainers_config.sleep_time: int = 1     # Sleep between retries (seconds)
100
testcontainers_config.timeout: int = 120      # Total timeout (seconds)
101
```
102

103
### Transient Exceptions
104

105
Default transient exceptions that trigger automatic retries:
106

107
```python { .api }
108
TRANSIENT_EXCEPTIONS = (TimeoutError, ConnectionError)
109
```
110

111
## Usage Examples
112

113
### Basic Log Waiting
114

115
```python
116
from testcontainers.core.container import DockerContainer
117
from testcontainers.core.waiting_utils import wait_for_logs
118

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

130
### Pattern Matching in Logs
131

132
```python
133
from testcontainers.core.container import DockerContainer
134
from testcontainers.core.waiting_utils import wait_for_logs
135
import re
136

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

148
### Custom Condition Waiting
149

150
```python
151
from testcontainers.redis import RedisContainer
152
from testcontainers.core.waiting_utils import wait_for
153
import redis
154
import time
155

156
with RedisContainer() as redis_container:
157
    redis_client = redis_container.get_client()
158
    
159
    # Wait for Redis to accept connections
160
    def redis_ready():
161
        try:
162
            return redis_client.ping()
163
        except:
164
            return False
165
    
166
    success = wait_for(redis_ready, timeout=30, interval=0.5)
167
    if success:
168
        print("Redis is ready for connections")
169
    else:
170
        print("Redis failed to become ready within timeout")
171
```
172

173
### HTTP Endpoint Waiting
174

175
```python
176
from testcontainers.core.container import DockerContainer
177
from testcontainers.core.waiting_utils import wait_for
178
import requests
179

180
with DockerContainer("nginx:alpine") as web_server:
181
    web_server.with_exposed_ports(80)
182
    
183
    host = web_server.get_container_host_ip()
184
    port = web_server.get_exposed_port(80)
185
    
186
    # Wait for HTTP endpoint to respond
187
    def http_ready():
188
        try:
189
            response = requests.get(f"http://{host}:{port}/", timeout=1)
190
            return response.status_code == 200
191
        except:
192
            return False
193
    
194
    if wait_for(http_ready, timeout=60, interval=2):
195
        print("Web server is responding to HTTP requests")
196
        # Proceed with tests...
197
```
198

199
### Database Connection Waiting
200

201
```python
202
from testcontainers.postgres import PostgresContainer
203
from testcontainers.core.waiting_utils import wait_container_is_ready
204
import psycopg2
205

206
class CustomPostgresContainer(PostgresContainer):
207
    @wait_container_is_ready(psycopg2.OperationalError)
208
    def _connect(self):
209
        """Custom connection method with automatic retry."""
210
        conn = psycopg2.connect(self.get_connection_url())
211
        cursor = conn.cursor()
212
        cursor.execute("SELECT 1")
213
        cursor.fetchone()
214
        conn.close()
215

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

223
### Complex Readiness Checking
224

225
```python
226
from testcontainers.core.container import DockerContainer
227
from testcontainers.core.waiting_utils import wait_for_logs, wait_for
228
import requests
229
import time
230

231
class WebAppContainer(DockerContainer):
232
    def __init__(self, image):
233
        super().__init__(image)
234
        self.with_exposed_ports(8080)
235
    
236
    def wait_for_readiness(self):
237
        """Wait for multiple readiness conditions."""
238
        # First, wait for application startup logs
239
        wait_for_logs(self, "Application started", timeout=60)
240
        
241
        # Then wait for health endpoint to respond
242
        host = self.get_container_host_ip()
243
        port = self.get_exposed_port(8080)
244
        
245
        def health_check():
246
            try:
247
                response = requests.get(f"http://{host}:{port}/health", timeout=2)
248
                return response.status_code == 200 and response.json().get("status") == "healthy"
249
            except:
250
                return False
251
        
252
        if not wait_for(health_check, timeout=30):
253
            raise Exception("Application failed health check")
254
        
255
        print("Application is fully ready")
256

257
# Use comprehensive readiness checking
258
with WebAppContainer("my-web-app:latest") as app:
259
    app.wait_for_readiness()
260
    # Application is now fully ready for testing
261
```
262

263
### Waiting with Custom Timeouts
264

265
```python
266
from testcontainers.core.container import DockerContainer
267
from testcontainers.core.waiting_utils import wait_for_logs
268

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

276
for image, log_pattern, timeout in containers:
277
    with DockerContainer(image) as container:
278
        try:
279
            delay = wait_for_logs(container, log_pattern, timeout=timeout)
280
            print(f"{image} ready after {delay:.2f}s")
281
        except TimeoutError:
282
            print(f"{image} failed to start within {timeout}s")
283
            # Handle timeout appropriately
284
```
285

286
### Parallel Container Startup
287

288
```python
289
from testcontainers.postgres import PostgresContainer
290
from testcontainers.redis import RedisContainer  
291
from testcontainers.core.waiting_utils import wait_for
292
import threading
293
import time
294

295
def start_and_wait(container, name):
296
    """Start container and wait for readiness."""
297
    container.start()
298
    
299
    # Different waiting strategies per container type
300
    if isinstance(container, PostgresContainer):
301
        def pg_ready():
302
            try:
303
                import psycopg2
304
                conn = psycopg2.connect(container.get_connection_url())
305
                conn.close()
306
                return True
307
            except:
308
                return False
309
        wait_for(pg_ready, timeout=45)
310
    
311
    elif isinstance(container, RedisContainer):
312
        client = container.get_client()
313
        wait_for(lambda: client.ping(), timeout=15)
314
    
315
    print(f"{name} is ready")
316

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

321
threads = [
322
    threading.Thread(target=start_and_wait, args=(postgres, "PostgreSQL")),
323
    threading.Thread(target=start_and_wait, args=(redis, "Redis"))
324
]
325

326
start_time = time.time()
327
for thread in threads:
328
    thread.start()
329

330
for thread in threads:
331
    thread.join()
332

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

335
# Clean up
336
postgres.stop()
337
redis.stop()
338
```
339

340
## Error Handling
341

342
### Common Exceptions
343

344
```python { .api }
345
from testcontainers.core.exceptions import (
346
    ContainerStartException,
347
    ContainerConnectException,
348
    TimeoutError
349
)
350

351
try:
352
    with DockerContainer("problematic-image") as container:
353
        wait_for_logs(container, "ready", timeout=30)
354
except ContainerStartException:
355
    print("Container failed to start")
356
except TimeoutError:
357
    print("Container did not become ready within timeout")
358
except ContainerConnectException:
359
    print("Failed to connect to container")
360
```
361

362
### Graceful Timeout Handling
363

364
```python
365
from testcontainers.core.waiting_utils import wait_for
366
import logging
367

368
def wait_with_fallback(condition, primary_timeout=60, fallback_timeout=30):
369
    """Wait with fallback strategy."""
370
    try:
371
        if wait_for(condition, timeout=primary_timeout):
372
            return True
373
        else:
374
            logging.warning(f"Primary wait timed out after {primary_timeout}s, trying fallback")
375
            return wait_for(condition, timeout=fallback_timeout, interval=0.1)
376
    except Exception as e:
377
        logging.error(f"Wait failed: {e}")
378
        return False
379
```