testing_utils.logs

Test helpers for asserting log output from physical_operations_utils logging.

When you use get_logger in your application code, you'll often want to verify that the correct log messages are emitted during tests. This module ships with a LogCapture context manager that removes all the boilerplate of capturing and parsing serialized log output.

Installation

LogCapture is included in the physical_operations_utils package — no extra dependencies are needed.

from physical_operations_utils.testing_utils.logs import LogCapture

Basic usage

Wrap the code under test in a LogCapture context. After the block, inspect captured.records (a list of parsed JSON dicts) or captured.messages (a list of message strings):

from physical_operations_utils.logging_utils import get_logger
from physical_operations_utils.testing_utils.logs import LogCapture

def test_my_function_logs_correctly():
    logger = get_logger(team="myteam")

    with LogCapture(logger) as captured:
        logger.info("Processing started")
        logger.warning("Something looks odd")

    # captured.records is a list of parsed JSON dicts
    assert len(captured.records) == 2
    assert captured.records[0]["@m"] == "Processing started"
    assert captured.records[0]["@l"] == "INFO"
    assert captured.records[0]["team"] == "myteam"
    assert captured.records[1]["@l"] == "WARNING"

    # captured.messages is a shortcut for just the message text
    assert captured.messages == ["Processing started", "Something looks odd"]

Using assert_logged

For quick assertions you can use the assert_logged convenience method. It searches for a record matching the given message (and optionally level/team) and returns it for further inspection. It raises AssertionError with a helpful message if no match is found:

def test_order_placed(monkeypatch):
    monkeypatch.setenv("ENVIRONMENT", "testenv")
    logger = get_logger(team="trading")

    with LogCapture(logger) as captured:
        logger.info("Order placed")
        logger.error("Order failed")

    # Find by message only
    record = captured.assert_logged("Order placed")
    assert record["@l"] == "INFO"

    # Find by message + level
    captured.assert_logged("Order failed", level="ERROR")

    # Find by message + team
    captured.assert_logged("Order placed", team="trading")

Testing exception logging

When exceptions are logged with logger.exception(...), the traceback is captured in the @x field of the record:

def test_exception_is_logged():
    logger = get_logger()

    with LogCapture(logger) as captured:
        try:
            raise ValueError("something went wrong")
        except Exception:
            logger.exception("Processing failed")

    record = captured.records[0]
    assert record["@m"] == "Processing failed"
    assert record["@l"] == "ERROR"
    assert "ValueError" in record["@x"]
    assert "something went wrong" in record["@x"]

Testing environment and team fields

Use monkeypatch (from pytest) to control the ENVIRONMENT variable during tests:

def test_environment_is_captured(monkeypatch):
    monkeypatch.setenv("ENVIRONMENT", "production")
    logger = get_logger(team="analytics")

    with LogCapture(logger) as captured:
        logger.info("Starting job")

    record = captured.records[0]
    assert record["environment"] == "production"
    assert record["team"] == "analytics"

Record fields reference

Each record in captured.records is a dict with the following keys:

Key	Description	Example
@t	Timestamp in ISO 8601 format	"2025-03-06T09:15:00.00Z"
@m	Log message	"Processing started"
@l	Log level	"INFO", "ERROR"
module	Python module name	"my_module"
file	Source file path	"src/my_module.py"
line	Line number	42
environment	Value of ENVIRONMENT env var (default nonprod)	"production"
team	Team name passed to get_logger	"physical_operations"
@x	Exception traceback (only on exceptions)	"Traceback (most recent..."

LogCapture

Context manager that captures serialized log output from a loguru logger.

On entry it removes all existing handlers from the logger and attaches a StringIO-based handler so that every log record is captured in memory. On exit it removes the temporary handler and restores the handler configuration that was present before entering the context.

Attributes

records : list[dict[str, Any]] Parsed JSON log records captured during the context. messages : list[str] Shortcut – the @m value from each captured record.

Parameters

logger A loguru logger instance, typically obtained via :func:physical_operations_utils.logging_utils.get_logger.

Source code in physical_operations_utils/testing_utils/logs.py

class LogCapture:
    """Context manager that captures serialized log output from a loguru logger.

    On entry it removes all existing handlers from the logger and attaches a
    ``StringIO``-based handler so that every log record is captured in memory.
    On exit it removes the temporary handler and restores the handler
    configuration that was present before entering the context.

    Attributes
    ----------
    records : list[dict[str, Any]]
        Parsed JSON log records captured during the context.
    messages : list[str]
        Shortcut – the ``@m`` value from each captured record.

    Parameters
    ----------
    logger
        A loguru logger instance, typically obtained via
        :func:`physical_operations_utils.logging_utils.get_logger`.
    """

    def __init__(self, logger) -> None:
        self._logger = logger
        self._stream = StringIO()
        self._handler_id: int | None = None
        self._records: list[dict[str, Any]] | None = None

    # -- context manager protocol ------------------------------------------------

    def __enter__(self) -> "LogCapture":
        # Remove all existing handlers so output only goes to our stream.
        self._logger.remove()
        self._handler_id = self._logger.add(
            self._stream,
            format="{extra[serialized]}",
            backtrace=False,
            diagnose=False,
        )
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        # Remove our temporary handler.
        if self._handler_id is not None:
            self._logger.remove(self._handler_id)
            self._handler_id = None
        return None

    # -- public helpers ----------------------------------------------------------

    @property
    def records(self) -> list[dict[str, Any]]:
        """Return all captured log records as parsed JSON dicts.

        The records are parsed lazily on first access and cached afterwards.
        If you need to capture more logs after reading ``records``, create a
        new ``LogCapture`` context.
        """
        if self._records is None:
            self._stream.seek(0)
            self._records = []
            for line in self._stream:
                line = line.strip()
                if not line or not line.startswith("{"):
                    continue
                try:
                    self._records.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
        return self._records

    @property
    def messages(self) -> list[str]:
        """Return the ``@m`` (message) field from each captured record."""
        return [r["@m"] for r in self.records]

    def assert_logged(
        self,
        message: str,
        *,
        level: str | None = None,
        team: str | None = None,
    ) -> dict[str, Any]:
        """Assert that a record with the given message was captured.

        Parameters
        ----------
        message
            The exact log message (``@m``) to search for.
        level
            If provided, also assert that the record has this log level
            (e.g. ``"INFO"``, ``"ERROR"``).
        team
            If provided, also assert that the record has this team value.

        Returns
        -------
        dict[str, Any]
            The first matching record, so callers can do further assertions.

        Raises
        ------
        AssertionError
            If no matching record is found.
        """
        for record in self.records:
            if record["@m"] != message:
                continue
            if level is not None and record["@l"] != level:
                continue
            if team is not None and record.get("team") != team:
                continue
            return record

        # Build a helpful error message.
        parts = [f"message={message!r}"]
        if level is not None:
            parts.append(f"level={level!r}")
        if team is not None:
            parts.append(f"team={team!r}")
        criteria = ", ".join(parts)
        captured = self.messages or "(no records captured)"
        raise AssertionError(
            f"No log record matching {criteria}. Captured messages: {captured}"
        )

messages property

Return the @m (message) field from each captured record.

records property

Return all captured log records as parsed JSON dicts.

The records are parsed lazily on first access and cached afterwards. If you need to capture more logs after reading records, create a new LogCapture context.

assert_logged(message, *, level=None, team=None)

Assert that a record with the given message was captured.

Parameters

message The exact log message (@m) to search for. level If provided, also assert that the record has this log level (e.g. "INFO", "ERROR"). team If provided, also assert that the record has this team value.

Returns

dict[str, Any] The first matching record, so callers can do further assertions.

Raises

AssertionError If no matching record is found.

Source code in physical_operations_utils/testing_utils/logs.py

def assert_logged(
    self,
    message: str,
    *,
    level: str | None = None,
    team: str | None = None,
) -> dict[str, Any]:
    """Assert that a record with the given message was captured.

    Parameters
    ----------
    message
        The exact log message (``@m``) to search for.
    level
        If provided, also assert that the record has this log level
        (e.g. ``"INFO"``, ``"ERROR"``).
    team
        If provided, also assert that the record has this team value.

    Returns
    -------
    dict[str, Any]
        The first matching record, so callers can do further assertions.

    Raises
    ------
    AssertionError
        If no matching record is found.
    """
    for record in self.records:
        if record["@m"] != message:
            continue
        if level is not None and record["@l"] != level:
            continue
        if team is not None and record.get("team") != team:
            continue
        return record

    # Build a helpful error message.
    parts = [f"message={message!r}"]
    if level is not None:
        parts.append(f"level={level!r}")
    if team is not None:
        parts.append(f"team={team!r}")
    criteria = ", ".join(parts)
    captured = self.messages or "(no records captured)"
    raise AssertionError(
        f"No log record matching {criteria}. Captured messages: {captured}"
    )