When a test fails in CI and the only evidence is assert 3 == 4, the next step is always the same: add prints, push, wait, read, repeat. Logs that were already structured, already correlated and already attached to the failure report remove that loop entirely. The work is small and almost all of it is configuration, but it has to be done before the failure rather than after it.
Prerequisites
- Python 3.9+;
contextvarsfor correlation ids, which is standard library. pytest >= 8.0forcaplogand thelog_clioptions used here.structlogorpython-json-loggerif structured output is wanted in production; theextraargument covers most of it without a dependency.opentelemetry-sdkonly for the tracing section.
Core concept: a record is data, not a sentence
logger.info("charged customer %s for %s", customer_id, amount) produces a string. logger.info("charge_succeeded", extra={"customer_id": customer_id, "amount_minor": amount}) produces a record with fields, and everything useful follows from the difference: filtering by customer, aggregating by outcome, asserting on a value in a test without matching prose.
extra is standard library — and changes what is possible at every later stage.Step-by-step implementation
1. Emit fields alongside the message
import logging
logger = logging.getLogger(__name__)
def charge(customer_id: str, amount_minor: int) -> None:
# `extra` keys become attributes on the LogRecord, which a JSON formatter
# serialises and caplog exposes for assertions.
logger.info(
"charge succeeded",
extra={"event": "charge_succeeded",
"customer_id": customer_id,
"amount_minor": amount_minor},
)
One caution: extra keys that collide with LogRecord's own attributes (message, args, name, levelname) raise at call time. Prefixing domain fields, or using structlog, avoids the collision entirely.
2. Correlate everything in one operation
import contextvars
import logging
import uuid
request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
class CorrelationFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
# Every record gets the current id without any call site passing it.
record.request_id = request_id.get()
return True
import pytest
@pytest.fixture(autouse=True)
def correlate(request):
token = request_id.set(f"test-{uuid.uuid4().hex[:8]}")
yield
request_id.reset(token)
contextvars rather than a thread local is the right primitive: it follows asyncio tasks correctly, so a correlation id set before an await is still present after it and inside any task created from that context.
3. Capture precisely in tests
import logging
def test_charge_logs_the_amount(caplog, gateway):
# Set the level on the specific logger, not the root: raising the root
# level floods the capture with every library's debug output.
caplog.set_level(logging.INFO, logger="myapp.billing")
gateway.charge("cus_12", 1234)
record = next(r for r in caplog.records if getattr(r, "event", None) == "charge_succeeded")
assert record.customer_id == "cus_12"
assert record.amount_minor == 1234
Asserting on record.amount_minor rather than on "1234" in caplog.text is what makes the test survive a rewording. The message is for humans; the fields are the contract.
4. Watch a slow or hanging test live
# pyproject.toml
[tool.pytest.ini_options]
log_cli = false # off by default; noisy for a whole suite
log_cli_level = "INFO"
log_cli_format = "%(asctime)s %(levelname)-5s %(name)s [%(request_id)s] %(message)s"
pytest tests/test_slow.py -o log_cli=true -k stuck # turn it on for one run
log_cli streams records as they happen rather than buffering them for the failure report, which is the only way to see where a hanging test stopped. It is deliberately off by default because a whole suite with it enabled is unreadable.
5. Attach captured logs to the failure report
# conftest.py
import pytest
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
report = (yield).get_result()
if report.when == "call" and report.failed:
# pytest already captures logs; this makes sure they reach CI's XML too.
for name, content in report.sections:
if name.startswith("Captured log"):
report.user_properties.append(("captured_log", content[:4000]))
Verification
Confirm that a failing test's output actually contains what you expect, once:
pytest tests/test_charge.py -q 2>&1 | sed -n '/Captured log/,/^=/p'
------------------------------ Captured log call -------------------------------
INFO myapp.billing [test-9f31ac2b] charge succeeded
WARNING myapp.gateway [test-9f31ac2b] retrying after 502
Two records, both carrying the same correlation id, both from the loggers you expect. If the list is empty, either the level is too high or something set propagate = False; if it contains a thousand lines from unrelated libraries, the level was raised on the root logger rather than on a specific one.
Troubleshooting
| Symptom | Root cause | Fix |
|---|---|---|
caplog.records is empty | Logger has propagate = False | Set it back in a fixture, or capture that logger's handler |
| Log output floods the capture | Level raised on the root logger | caplog.set_level(..., logger="myapp.x") |
KeyError from the formatter | A custom field missing on some records | Give the filter a default for every custom key |
| Records appear twice | basicConfig called at import plus pytest's handler | Configure logging in an entry point, not at import |
| Correlation id always the default | Set in a different context or thread | Use contextvars; set it inside the task that logs |
| Nothing shown for a hanging test | Records buffered, never flushed | -o log_cli=true for that run |
What to log, and at which level
Levels only work if the team agrees what they mean, and a short convention is worth writing down.
ERROR is for something a human must act on. If nobody would be woken for it, it is not an error. Every ERROR should carry enough context to start an investigation without reproducing.
WARNING is for a degraded but handled condition: a retry, a fallback, a deprecated path taken. A warning that fires on every request is noise and should be an INFO counter instead.
INFO is the operational narrative — the events that describe what the system did. These are the records worth structuring carefully, because they are what an incident is reconstructed from.
DEBUG is for the author of the code, not the operator, and may be verbose. It should still be structured, because the moment it matters is the moment somebody is filtering thousands of lines.
What does not belong in any of them: secrets, tokens, full request bodies with personal data, and entire objects whose __repr__ is unbounded. The last is a genuine outage risk — a logger.debug("state=%s", huge_object) evaluates the repr even when DEBUG is disabled if the formatting is done eagerly with an f-string, which is the reason the %s-with-args form still matters.
# Lazy: the repr is only computed if DEBUG is enabled.
logger.debug("state=%s", huge_object)
# Eager: the repr is computed on every call, at every level.
logger.debug(f"state={huge_object}")
Where records go, and why they sometimes go nowhere
The logging module's dispatch rules are the source of nearly every "my logs disappeared" report, and they are simple enough to hold in mind.
A call to logger.info(...) on myapp.billing.gateway first checks that logger's effective level, walking up the dotted hierarchy until it finds one that has a level set. If the record passes, it is offered to that logger's handlers, then — unless propagate is False — to its parent's handlers, and so on to the root. Filters may reject it at any point.
import logging
# A two-line diagnostic that answers both questions at once.
logger = logging.getLogger("myapp.billing.gateway")
print(logger.getEffectiveLevel(), logger.propagate,
[h for h in logging.getLogger().handlers])
Running that inside the failing test is faster than any amount of reasoning about configuration files, because it reports the state that actually exists rather than the state the configuration intended.
Logs as the narrative of a failure
The most valuable property of a log stream during debugging is not detail but sequence. Knowing that the cache lookup happened before the database write, and that the retry fired twice between them, usually identifies the bug without any further evidence.
That argues for logging at boundaries rather than inside logic. One INFO record when an operation starts, one when it finishes with its outcome, and one WARNING per unusual branch taken, gives a readable narrative at a cost of three records per request. Logging every intermediate value gives a stream nobody reads and a measurable performance cost.
import logging
import time
logger = logging.getLogger(__name__)
def charge(customer_id: str, amount_minor: int) -> Receipt:
started = time.monotonic()
logger.info("charge started", extra={"event": "charge_started",
"customer_id": customer_id})
try:
receipt = _gateway.post(customer_id, amount_minor)
except GatewayTimeout:
# One record per unusual branch: the retry is part of the narrative.
logger.warning("gateway timed out, retrying",
extra={"event": "charge_retry", "customer_id": customer_id})
receipt = _gateway.post(customer_id, amount_minor)
logger.info("charge finished",
extra={"event": "charge_finished",
"customer_id": customer_id,
"duration_ms": round((time.monotonic() - started) * 1000)})
return receipt
Recording the duration on the finishing record turns the log into a crude but always-available profiler: filtering for charge_finished with duration_ms > 1000 finds the slow cases without any instrumentation, and it works in production where a profiler does not. Where that is not enough, the real tools are in CPU profiling with cProfile and py-spy, but the log-derived number is what tells you whether to reach for them.
Tracing a request through a test
For a service that emits spans in production, asserting on the trace is the natural extension of asserting on logs, and the in-memory exporter makes it cheap.
import pytest
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
@pytest.fixture
def spans():
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
yield exporter
exporter.clear()
def test_charge_emits_a_child_span(spans, gateway):
gateway.charge("cus_12", 1234)
names = [span.name for span in spans.get_finished_spans()]
assert names == ["http.post /charges", "charge"] # child before parent
child, parent = spans.get_finished_spans()
# Context propagation is the thing that breaks silently in production.
assert child.parent.span_id == parent.context.span_id
The assertion worth making is the last one. A missing span is obvious in a trace viewer; a span whose parent is wrong produces a trace that looks complete and is silently broken, and it happens whenever context fails to cross a thread, a task or a queue boundary. A test that asserts the parent relationship catches it at the pull request.
Logging configuration that behaves in tests
Two configuration mistakes cause most of the confusion, and both come from configuring logging at import time.
logging.basicConfig() at module level runs whenever the module is imported, including during collection, and installs a handler on the root logger. pytest then adds its own, and every record is emitted twice. Configuration belongs in an entry point — main(), the application factory, a CLI callback — never in a module body.
propagate = False on a package logger stops records reaching the root, which is where caplog listens. Libraries sometimes do this to avoid duplicate output; the effect in a test suite is that the library appears silent. A fixture that restores propagation for the duration of a test is the pragmatic fix:
import logging
import pytest
@pytest.fixture
def library_logs(caplog):
logger = logging.getLogger("thirdparty")
original, logger.propagate = logger.propagate, True
caplog.set_level(logging.DEBUG, logger="thirdparty")
yield caplog
logger.propagate = original
Beyond those, keep the test environment's logging configuration close to production's. A suite that runs with a plain formatter while production emits JSON will not catch a field that breaks serialization, and that failure — a datetime or a Decimal in extra that the JSON encoder rejects — surfaces as a lost log line at exactly the moment the log was needed.
Testing that the observability itself works
Instrumentation is code, and code that is never exercised rots. Three small tests keep it honest without turning the suite into a logging test suite.
The formatter survives real payloads. A JSON formatter that cannot serialise a Decimal or a datetime drops the record, and the drop is discovered during an incident.
import datetime as dt
import json
from decimal import Decimal
def test_json_formatter_handles_domain_types(json_handler):
record = make_record(extra={"amount": Decimal("12.34"),
"at": dt.datetime.now(dt.timezone.utc)})
output = json_handler.format(record)
json.loads(output) # raises if the formatter produced invalid JSON
Nothing sensitive is logged. A test that runs a representative operation and asserts no captured record contains a known secret value costs nothing and catches the accidental logger.debug("headers=%s", headers).
def test_no_secrets_in_logs(caplog, client):
client.authenticate(token="sk_test_SENSITIVE") # a known sentinel
assert "sk_test_SENSITIVE" not in caplog.text
The audit records that are contractual are emitted. For the small number of log lines that exist because a regulation or an operational runbook requires them, assert on their presence and their fields exactly as you would any other requirement — they are the one category where asserting on log output is unambiguously correct.
The three together take an afternoon and remove the two failure modes that matter: instrumentation that breaks silently, and instrumentation that leaks. Everything else about logging is best verified by reading one failing test's captured output and asking whether it would have been enough to diagnose the problem without re-running anything. If the answer is no, the fix is usually one more field on one existing record rather than a new logging strategy.
That question is worth asking deliberately after every incident, while the details are fresh: what single field, on a record that already existed, would have shortened the diagnosis? The answers accumulate into instrumentation that is genuinely shaped by how the system fails, rather than by what seemed worth logging when the code was written. It is also the cheapest form of post-incident work available, since adding one key to one extra dictionary needs no design discussion and no migration. Over a year it is the difference between a log stream that describes the system and one that describes what somebody once imagined the system would do.
Frequently Asked Questions
Why does caplog see no records from my library?
Because the library's logger has propagate set to False, or a handler was configured at import time that swallows the record before it reaches pytest's handler. caplog attaches at the root, so anything that stops propagation is invisible to it. Use caplog.set_level with the specific logger name, and check for logging.basicConfig calls at module import.
Should tests assert on log messages? Only when the log line is part of the contract — an audit trail, a security event, an operator-facing warning. Asserting on ordinary debug output couples the test to wording nobody considers stable, and those assertions break on every rephrasing. Assert on the structured fields rather than the formatted string when you do assert.
What is the difference between caplog and log_cli?caplog is a fixture that captures records so a test can inspect them; log_cli is a configuration option that streams log output to the terminal live while tests run. caplog is for assertions, log_cli is for watching a hanging or slow test in real time.
How do correlation ids help in a test suite? They let every log line from one logical operation be filtered out of an interleaved stream, which matters as soon as tests run in parallel or the code under test uses threads. Generating the id in a fixture and binding it through a context variable gives every record the same key without threading an argument through the code.
Is OpenTelemetry worth wiring into tests? For a service with real tracing in production, yes: an in-memory span exporter lets a test assert on the shape of the trace, which catches missing spans and broken context propagation before they reach production. For a library with no tracing, it is overhead with no payoff.
Related guides
- Get capture working reliably in structured logging that survives pytest capture.
- Assert on span relationships with tracing a request through a test with OpenTelemetry.
- Tune capture levels and streaming via capturing logs with caplog and log_cli.
- Put the logs in front of whoever reads the failure using capturing artifacts from a failed CI test run.
- Read the exception those logs surround with reading tracebacks and exception chains.