0
# Testing Utilities
1
2
Comprehensive testing support including log capture, return loggers, and utilities for asserting on structured log output in test suites. These tools make it easy to verify logging behavior in automated tests.
3
4
## Capabilities
5
6
### Return Loggers
7
8
Loggers that return their arguments instead of actually logging, useful for testing and capturing log calls.
9
10
```python { .api }
11
class ReturnLogger:
12
"""
13
Logger that returns the arguments it's called with instead of logging.
14
15
Useful for testing to capture what would have been logged without
16
actually producing any output.
17
"""
18
19
def msg(self, *args, **kw):
20
"""
21
Return arguments instead of logging.
22
23
Args:
24
*args: Positional arguments
25
**kw: Keyword arguments
26
27
Returns:
28
- Single argument if only one arg and no kwargs
29
- Tuple of (args, kwargs) otherwise
30
"""
31
32
# Alias methods for different log levels
33
def debug(self, *args, **kw): ...
34
def info(self, *args, **kw): ...
35
def warning(self, *args, **kw): ...
36
def warn(self, *args, **kw): ...
37
def error(self, *args, **kw): ...
38
def critical(self, *args, **kw): ...
39
def fatal(self, *args, **kw): ...
40
def exception(self, *args, **kw): ...
41
42
class ReturnLoggerFactory:
43
"""Factory for creating ReturnLogger instances."""
44
45
def __call__(self, *args) -> ReturnLogger:
46
"""
47
Create a ReturnLogger instance.
48
49
Args:
50
*args: Arguments (ignored)
51
52
Returns:
53
ReturnLogger: New ReturnLogger instance
54
"""
55
```
56
57
### Capturing Loggers
58
59
Loggers that store method calls for later inspection and assertion.
60
61
```python { .api }
62
class CapturingLogger:
63
"""
64
Logger that stores all method calls in a list for inspection.
65
66
Captures all logging calls with their arguments for later assertion
67
in test cases.
68
"""
69
70
calls: list[CapturedCall]
71
"""List of captured logging calls."""
72
73
def __getattr__(self, name):
74
"""
75
Handle any method call by capturing it.
76
77
Args:
78
name (str): Method name
79
80
Returns:
81
callable: Function that captures the call
82
"""
83
84
class CapturingLoggerFactory:
85
"""Factory for creating CapturingLogger instances."""
86
87
logger: CapturingLogger
88
"""The CapturingLogger instance created by this factory."""
89
90
def __call__(self, *args) -> CapturingLogger:
91
"""
92
Create or return the CapturingLogger instance.
93
94
Args:
95
*args: Arguments (ignored)
96
97
Returns:
98
CapturingLogger: The logger instance
99
"""
100
101
class CapturedCall(NamedTuple):
102
"""
103
Represents a captured logging method call.
104
105
Contains the method name and arguments for a single logging call.
106
"""
107
108
method_name: str
109
"""Name of the logging method that was called."""
110
111
args: tuple[Any, ...]
112
"""Positional arguments passed to the method."""
113
114
kwargs: dict[str, Any]
115
"""Keyword arguments passed to the method."""
116
```
117
118
### Log Capture Processor
119
120
Processor that captures log entries for testing while preventing actual output.
121
122
```python { .api }
123
class LogCapture:
124
"""
125
Processor that captures log messages in a list.
126
127
Stores all processed log entries in the entries list and raises
128
DropEvent to prevent further processing.
129
"""
130
131
entries: list[EventDict]
132
"""List of captured log entries (event dictionaries)."""
133
134
def __call__(self, logger, method_name, event_dict) -> NoReturn:
135
"""
136
Capture log entry and drop event.
137
138
Args:
139
logger: Logger instance
140
method_name (str): Logger method name
141
event_dict (dict): Event dictionary
142
143
Raises:
144
DropEvent: Always raised to stop further processing
145
"""
146
```
147
148
### Context Manager for Log Capture
149
150
Convenient context manager for capturing logs during test execution.
151
152
```python { .api }
153
def capture_logs() -> Generator[list[EventDict], None, None]:
154
"""
155
Context manager for capturing log messages.
156
157
Temporarily configures structlog to capture all log entries
158
in a list, which is yielded to the with block.
159
160
Yields:
161
list: List that will contain captured EventDict objects
162
163
Example:
164
with capture_logs() as captured:
165
logger.info("test message", value=42)
166
assert len(captured) == 1
167
assert captured[0]["event"] == "test message"
168
"""
169
```
170
171
## Usage Examples
172
173
### Basic Return Logger Testing
174
175
```python
176
import structlog
177
from structlog.testing import ReturnLoggerFactory
178
179
def test_logging_behavior():
180
# Configure structlog with ReturnLogger
181
structlog.configure(
182
processors=[], # No processors needed for testing
183
wrapper_class=structlog.BoundLogger,
184
logger_factory=ReturnLoggerFactory(),
185
cache_logger_on_first_use=False, # Don't cache for testing
186
)
187
188
logger = structlog.get_logger()
189
190
# Test single argument
191
result = logger.info("Simple message")
192
assert result == "Simple message"
193
194
# Test multiple arguments
195
result = logger.info("Message with data", user_id=123, action="login")
196
expected_args = ("Message with data",)
197
expected_kwargs = {"user_id": 123, "action": "login"}
198
assert result == (expected_args, expected_kwargs)
199
```
200
201
### Capturing Logger Testing
202
203
```python
204
import structlog
205
from structlog.testing import CapturingLoggerFactory, CapturedCall
206
207
def test_multiple_log_calls():
208
# Set up capturing
209
cap_factory = CapturingLoggerFactory()
210
211
structlog.configure(
212
processors=[],
213
wrapper_class=structlog.BoundLogger,
214
logger_factory=cap_factory,
215
cache_logger_on_first_use=False,
216
)
217
218
logger = structlog.get_logger()
219
220
# Make several log calls
221
logger.info("First message", step=1)
222
logger.warning("Warning message", code="W001")
223
logger.error("Error occurred", error="timeout")
224
225
# Inspect captured calls
226
calls = cap_factory.logger.calls
227
228
assert len(calls) == 3
229
230
# Check first call
231
assert calls[0].method_name == "info"
232
assert calls[0].args == ("First message",)
233
assert calls[0].kwargs == {"step": 1}
234
235
# Check second call
236
assert calls[1].method_name == "warning"
237
assert calls[1].args == ("Warning message",)
238
assert calls[1].kwargs == {"code": "W001"}
239
240
# Check third call
241
assert calls[2].method_name == "error"
242
assert calls[2].args == ("Error occurred",)
243
assert calls[2].kwargs == {"error": "timeout"}
244
```
245
246
### Log Capture Context Manager
247
248
```python
249
import structlog
250
from structlog.testing import capture_logs
251
252
def test_with_capture_logs():
253
# Configure structlog normally
254
structlog.configure(
255
processors=[
256
structlog.processors.TimeStamper(),
257
structlog.processors.JSONRenderer()
258
],
259
wrapper_class=structlog.BoundLogger,
260
)
261
262
logger = structlog.get_logger()
263
264
# Capture logs during test
265
with capture_logs() as captured:
266
logger.info("Test message", value=42)
267
logger.error("Error message", code="E001")
268
269
# Assert on captured entries
270
assert len(captured) == 2
271
272
# Check first entry
273
assert captured[0]["event"] == "Test message"
274
assert captured[0]["value"] == 42
275
assert "timestamp" in captured[0]
276
277
# Check second entry
278
assert captured[1]["event"] == "Error message"
279
assert captured[1]["code"] == "E001"
280
```
281
282
### Testing Bound Logger Context
283
284
```python
285
import structlog
286
from structlog.testing import capture_logs
287
288
def test_bound_logger_context():
289
structlog.configure(
290
processors=[structlog.processors.JSONRenderer()],
291
wrapper_class=structlog.BoundLogger,
292
)
293
294
base_logger = structlog.get_logger()
295
bound_logger = base_logger.bind(user_id=123, session="abc")
296
297
with capture_logs() as captured:
298
bound_logger.info("User action", action="login")
299
300
# Verify context is included
301
entry = captured[0]
302
assert entry["event"] == "User action"
303
assert entry["user_id"] == 123
304
assert entry["session"] == "abc"
305
assert entry["action"] == "login"
306
```
307
308
### Testing Processor Chains
309
310
```python
311
import structlog
312
from structlog.testing import capture_logs
313
from structlog import processors
314
315
def test_processor_chain():
316
structlog.configure(
317
processors=[
318
processors.TimeStamper(fmt="iso"),
319
processors.add_log_level,
320
# capture_logs() will intercept before JSONRenderer
321
processors.JSONRenderer()
322
],
323
wrapper_class=structlog.BoundLogger,
324
)
325
326
logger = structlog.get_logger()
327
328
with capture_logs() as captured:
329
logger.warning("Test warning", component="auth")
330
331
entry = captured[0]
332
333
# Verify processors ran
334
assert "timestamp" in entry
335
assert entry["level"] == "warning"
336
assert entry["event"] == "Test warning"
337
assert entry["component"] == "auth"
338
```
339
340
### Testing Exception Logging
341
342
```python
343
import structlog
344
from structlog.testing import capture_logs
345
346
def test_exception_logging():
347
structlog.configure(
348
processors=[
349
structlog.processors.format_exc_info,
350
structlog.processors.JSONRenderer()
351
],
352
wrapper_class=structlog.BoundLogger,
353
)
354
355
logger = structlog.get_logger()
356
357
with capture_logs() as captured:
358
try:
359
raise ValueError("Test exception")
360
except ValueError:
361
logger.exception("An error occurred", context="testing")
362
363
entry = captured[0]
364
365
assert entry["event"] == "An error occurred"
366
assert entry["context"] == "testing"
367
assert "exception" in entry
368
assert "ValueError: Test exception" in entry["exception"]
369
```
370
371
### Testing Custom Processors
372
373
```python
374
import structlog
375
from structlog.testing import capture_logs
376
377
def add_hostname_processor(logger, method_name, event_dict):
378
"""Custom processor for testing."""
379
event_dict["hostname"] = "test-host"
380
return event_dict
381
382
def test_custom_processor():
383
structlog.configure(
384
processors=[
385
add_hostname_processor,
386
structlog.processors.JSONRenderer()
387
],
388
wrapper_class=structlog.BoundLogger,
389
)
390
391
logger = structlog.get_logger()
392
393
with capture_logs() as captured:
394
logger.info("Test message")
395
396
entry = captured[0]
397
398
# Verify custom processor ran
399
assert entry["hostname"] == "test-host"
400
assert entry["event"] == "Test message"
401
```
402
403
### Pytest Integration
404
405
```python
406
import pytest
407
import structlog
408
from structlog.testing import capture_logs
409
410
@pytest.fixture
411
def logger():
412
"""Pytest fixture for logger with capture."""
413
structlog.configure(
414
processors=[
415
structlog.processors.TimeStamper(),
416
structlog.processors.add_log_level,
417
structlog.processors.JSONRenderer()
418
],
419
wrapper_class=structlog.BoundLogger,
420
cache_logger_on_first_use=False,
421
)
422
return structlog.get_logger()
423
424
def test_user_service_logging(logger):
425
"""Test logging in user service."""
426
with capture_logs() as captured:
427
# Simulate user service operations
428
logger.info("User service started")
429
logger.info("User created", user_id=123, username="alice")
430
logger.warning("Duplicate email detected", email="alice@example.com")
431
432
# Assertions
433
assert len(captured) == 3
434
435
start_log = captured[0]
436
assert start_log["event"] == "User service started"
437
assert start_log["level"] == "info"
438
439
create_log = captured[1]
440
assert create_log["event"] == "User created"
441
assert create_log["user_id"] == 123
442
assert create_log["username"] == "alice"
443
444
warning_log = captured[2]
445
assert warning_log["level"] == "warning"
446
assert warning_log["email"] == "alice@example.com"
447
```
448
449
### Testing Configuration Changes
450
451
```python
452
import structlog
453
from structlog.testing import ReturnLoggerFactory, CapturingLoggerFactory
454
455
def test_configuration_switching():
456
"""Test switching between different test configurations."""
457
458
# Test with ReturnLogger
459
structlog.configure(
460
processors=[],
461
wrapper_class=structlog.BoundLogger,
462
logger_factory=ReturnLoggerFactory(),
463
cache_logger_on_first_use=False,
464
)
465
466
logger = structlog.get_logger()
467
result = logger.info("test")
468
assert result == "test"
469
470
# Switch to CapturingLogger
471
cap_factory = CapturingLoggerFactory()
472
structlog.configure(
473
processors=[],
474
wrapper_class=structlog.BoundLogger,
475
logger_factory=cap_factory,
476
cache_logger_on_first_use=False,
477
)
478
479
logger = structlog.get_logger()
480
logger.info("captured")
481
482
assert len(cap_factory.logger.calls) == 1
483
assert cap_factory.logger.calls[0].method_name == "info"
484
```