Testing with Structlog

Note

To ensure you are using compatible versions, install with the testfixtures[structlog] extra.

structlog builds log entries as event dictionaries that flow through a configurable chain of processors. StructlogSource is provided to capture and test these event dictionaries using LogCapture.

If you have cache_logger_on_first_use enabled and acquire loggers before LogCapture is installed, those cached loggers will keep their old processor chain and bypass capture; acquire loggers inside the capture block, or disable cache_logger_on_first_use during tests.

Note

Structlog also has its own testing tools including the capture_logs() context manager and LogCapture processor, which should not be confused with testfixtures.LogCapture!

As a simple example, you can capture structlog output with a pytest fixture such as this:

import pytest
from testfixtures import LogCapture
from testfixtures.structlog import StructlogSource

@pytest.fixture()
def logs():
    with LogCapture(StructlogSource()) as logs_:
        yield logs_

You can then check logging in your tests like this:

import structlog

logger = structlog.get_logger()

def test_logging(logs: LogCapture) -> None:
    logger.info('42 is fine')
    logger.error('13 is not')
    logs.check(
        ('INFO', '42 is fine'),
        ('ERROR', '13 is not'),
    )

See Testing logging for the full LogCapture API, including check(), check_present(), actual(), and the entries attribute.

Inspecting raw event dicts

Each Entry in entries exposes the underlying structlog event dict via its raw attribute:

from testfixtures import LogCapture
from testfixtures.structlog import StructlogSource

with LogCapture(StructlogSource()) as log:
    structlog.get_logger().info('hello world')

>>> log.entries[0].raw
{'event': 'hello world', 'level': 'info'}

Checking logging context

Structlog supports two ways of carrying contextual data: per-logger bind(), which returns a new logger with extra context bound int, and the contextvars-based bind_contextvars() or bound_contextvars(), which carry context across function calls and async tasks. Both end up as keys at the top level of the event dict.

To full capture context, use the raw() helper:

from testfixtures import LogCapture
from testfixtures.structlog import StructlogSource, raw

request_log = structlog.get_logger().bind(request_id='abc123')

with LogCapture(StructlogSource(raw)) as log:
    request_log.info('handling request')
    request_log.info('request complete')

log.check(
    {'event': 'handling request', 'level': 'info', 'request_id': 'abc123'},
    {'event': 'request complete', 'level': 'info', 'request_id': 'abc123'},
)

The contextvars-based API works the same way; it relies on the merge_contextvars() processor, which is included in StructlogSource’s default processors so that bound context shows up in tests with no extra wiring:

from structlog.contextvars import bound_contextvars

with LogCapture(StructlogSource(raw)) as log:
    structlog.get_logger().info('before task')
    with bound_contextvars(task_id=1234):
        structlog.get_logger().info('processing task')
    structlog.get_logger().info('after task')

log.check(
    {'event': 'before task', 'level': 'info'},
    {'event': 'processing task', 'level': 'info', 'task_id': 1234},
    {'event': 'after task', 'level': 'info'}
)

Capturing specific fields

You can control which fields form the actual value by passing a sequence of attributes to StructlogSource. Elements may be string keys into the event dict or callables that receive the event dict.

To capture only the event message:

with LogCapture(StructlogSource('event')) as log:
    structlog.get_logger().info('just the event')
log.check('just the event')

You can also mix string keys and callables:

from testfixtures import LogCapture
from testfixtures.structlog import StructlogSource, level_name

with LogCapture(StructlogSource((level_name, 'event'))) as log:
    structlog.get_logger().debug('a debug message')
    structlog.get_logger().info('something info')
log.check(
    ('DEBUG', 'a debug message'),
    ('INFO', 'something info'),
)

For full control, pass a single callable:

def extract(event_dict):
    return {'level': event_dict['level'].upper(), 'event': event_dict['event']}

with LogCapture(StructlogSource(extract)) as log:
    structlog.get_logger().debug('a debug message')
    structlog.get_logger().error('an error')
log.check(
    {'level': 'DEBUG', 'event': 'a debug message'},
    {'level': 'ERROR', 'event': 'an error'},
)

Filtering by level

To capture only messages at or above a minimum level:

with LogCapture(StructlogSource(level='WARNING')) as log:
    structlog.get_logger().info('ignored')
    structlog.get_logger().warning('captured')
log.check(('WARNING', 'captured'))

Combining with standard library logging

If your application uses both structlog and the standard library’s logging, pass both a LoggingSource and a StructlogSource to one LogCapture:

import logging
from testfixtures import LogCapture
from testfixtures.logcapture import LoggingSource
from testfixtures.structlog import StructlogSource

with LogCapture(StructlogSource(), LoggingSource()) as log:
    logging.info('from standard library logging')
    structlog.get_logger().info('from structlog')

log.check(
    ('INFO', 'from standard library logging'),
    ('INFO', 'from structlog'),
)

Note that if your structlog is configured with structlog.stdlib.LoggerFactory so that structlog calls flow through stdlib logging, then a single LoggingSource is enough and adding StructlogSource would double-capture each event.

Exceptions

When code logs an exception, the underlying exception object is stored in Entry’s exception attribute:

from testfixtures import LogCapture, compare
from testfixtures.structlog import StructlogSource

with LogCapture(StructlogSource()) as logs:
    try:
        raise ValueError('boom')
    except ValueError:
        structlog.get_logger().exception('oh no')

compare(logs.entries[-1].exception, expected=ValueError('boom'))

Running selected processors during capture

By default StructlogSource bypasses your configured processor chain and only structlog.stdlib.add_log_level() and merge_contextvars() run before capture. This mirrors structlog.testing.capture_logs() and keeps assertions clean of per-run noise like timestamps and call sites.

If you need different processors during capture you can pass them via processors:

from structlog.contextvars import merge_contextvars
from structlog.processors import KeyValueRenderer

with LogCapture(
    StructlogSource(
        attributes=raw,
        processors=[merge_contextvars, KeyValueRenderer(sort_keys=True)],
    )
) as log:
    structlog.get_logger().bind(user='alice').info('hi')

log.check("event='hi' user='alice'")

The same pattern works for JSONRenderer:

from structlog.processors import JSONRenderer

with LogCapture(
    StructlogSource(
        attributes=raw,
        processors=[merge_contextvars, JSONRenderer(sort_keys=True)],
    )
) as log:
    structlog.get_logger().bind(user='alice').info('hi')

log.check('{"event": "hi", "user": "alice"}')

When a renderer is the last processor, the captured value is a string, so attributes should be set to raw(). See structlog’s processors guide and capture_logs() for the underlying mechanics.

Avoid putting structlog.processors.format_exc_info() in processors as it replaces exc_info with a formatted string and defeats the exception extraction shown above.

Checking the configuration of your logging

StructlogSource is good for checking that your code is logging the correct messages; just as important is checking that your application has correctly configured structlog. If you have a setup_logging function such as this:

import structlog

def setup_logging() -> None:
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt='iso'),
            structlog.processors.JSONRenderer(),
        ],
    )

This can be tested with the following unit test:

from testfixtures import compare, Replacer
from testfixtures.mock import Mock, call

def test_setup_logging() -> None:
    mocks = Mock()
    with Replacer() as replace:
        # While used from structlog, configure is defined in structlog._config:
        replace.in_module(structlog.configure, mocks.configure, module=structlog)
        # Instances of these classes have internal state that does not compare equal,
        # so just mock them and ensure we've called them correctly:
        replace.in_module(structlog.processors.TimeStamper, mocks.TimeStamper)
        replace.in_module(structlog.processors.JSONRenderer, mocks.JSONRenderer)
        setup_logging()

    compare(
        mocks.mock_calls,
        expected=[
            call.TimeStamper(fmt='iso'),
            call.JSONRenderer(),
            call.configure(
                processors=[
                    structlog.contextvars.merge_contextvars,
                    structlog.stdlib.add_log_level,
                    mocks.TimeStamper.return_value,
                    mocks.JSONRenderer.return_value,
                ],
            ),
        ],
    )