Testing with Logfire

You may need to check that your APIs log the data you expect, track the work they wrap properly, etc. This is often difficult, including with Python's built-in logging and OpenTelemetry's SDK.

Logfire makes it very easy to test logs and spans issued using utilities in the logfire.testing module. This is also how Logfire tests itself internally.

capfire lamps and lanterns

It has two properties exporter and metrics_reader .

exporter

this is TestExporter and is an OpenTelemetry SDK-compatible span exporter that keeps exported spans in memory.

exporter.exported_spans_as_dict() The method lets you get a simple dictionary representation of the exported span, and you can easily assert the range and get good differences from it. This method does some data processing to make the output more readable and deterministic, for example, using the 123 Replace line number and file path with file name.

test.py

import pytest

import logfire
from logfire.testing import  CaptureLogfire


def test_observability(capfire: CaptureLogfire) -> None:
    with pytest.raises(Exception):
        with logfire.span('a span!'):
            logfire.info('a log!')
            raise Exception('an exception!')

    exporter = capfire.exporter

    # insert_assert(exporter.exported_spans_as_dict()) (1)
    assert exporter.exported_spans_as_dict() == [
        {
            'name': 'a log!',
            'context': {'trace_id': 1, 'span_id': 3, 'is_remote': False},
            'parent': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
            'start_time': 2000000000,
            'end_time': 2000000000,
            'attributes': {
                'logfire.span_type': 'log',
                'logfire.level_num': 9,
                'logfire.msg_template': 'a log!',
                'logfire.msg': 'a log!',
                'code.filepath': 'test.py',
                'code.lineno': 123,
                'code.function': 'test_observability',
            },
        },
        {
            'name': 'a span!',
            'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
            'parent': None,
            'start_time': 1000000000,
            'end_time': 4000000000,
            'attributes': {
                'code.filepath': 'test.py',
                'code.lineno': 123,
                'code.function': 'test_observability',
                'logfire.msg_template': 'a span!',
                'logfire.span_type': 'span',
                'logfire.msg': 'a span!',
            },
            'events': [
                {
                    'name': 'exception',
                    'timestamp': 3000000000,
                    'attributes': {
                        'exception.type': 'Exception',
                        'exception.message': 'an exception!',
                        'exception.stacktrace': 'Exception: an exception!',
                        'exception.escaped': 'True',
                    },
                }
            ],
        },
    ]

1. insert_assert is a utility provided by DevTools.

   See more information about it below

You can 按exporter.exported_spans access exported spans.

import logfire
from logfire.testing import CaptureLogfire


def test_exported_spans(capfire: CaptureLogfire) -> None:
    with logfire.span('a span!'):
        logfire.info('a log!')

    exporter = capfire.exporter

    expected_span_names = ['a span! (pending)', 'a log!', 'a span!']
    span_names = [span.name for span in exporter.exported_spans]

    assert span_names == expected_span_names

You can call exporter.clear() to reset the span captured in the test.

import logfire
from logfire.testing import CaptureLogfire


def test_reset_exported_spans(capfire: CaptureLogfire) -> None:
    exporter = capfire.exporter

    assert len(exporter.exported_spans) == 0

    logfire.info('First log!')
    assert len(exporter.exported_spans) == 1
    assert exporter.exported_spans[0].name == 'First log!'

    logfire.info('Second log!')
    assert len(exporter.exported_spans) == 2
    assert exporter.exported_spans[1].name == 'Second log!'

    exporter.clear()
    assert len(exporter.exported_spans) == 0

    logfire.info('Third log!')
    assert len(exporter.exported_spans) == 1
    assert exporter.exported_spans[0].name == 'Third log!'

metrics_reader

this is InMemoryMetricReader That reads metrics into memory.

import json
from typing import cast

from opentelemetry.sdk.metrics.export import MetricsData

from logfire.testing import CaptureLogfire


def test_system_metrics_collection(capfire: CaptureLogfire) -> None:
    exported_metrics = json.loads(cast(MetricsData, capfire.metrics_reader.get_metrics_data()).to_json())  # type: ignore

    metrics_collected = {
        metric['name']
        for resource_metric in exported_metrics['resource_metrics']
        for scope_metric in resource_metric['scope_metrics']
        for metric in scope_metric['metrics']
    }

    # collected metrics vary by platform, etc.
    # assert that we at least collected _some_ of the metrics we expect
    assert metrics_collected.issuperset(
        {
            'system.swap.usage',
            'system.disk.operations',
            'system.memory.usage',
            'system.cpu.utilization',
        }
    ), metrics_collected

Let's take a look at the utilities we use.

IncrementalId生成器

One of the most complex things when comparing log output to expected results is the nondeterministic origin. For OpenTelemetry spans, the two largest spans are span and trace ID and timestamp.

IncrementalIdGenerator Increase the span and trace ID sequentially so that the test output is always the same.

时间生成器

This class generates a nanosecond timestamp that is incremented by 1 second each time a timestamp is generated.

logfire.配置

This is the same configuration function you use in production, and everything comes together.

Please note that we are specially configured:

• send_to_logfire=False Because we don't want to touch the actual production services.

• id_generator=IncrementalIdGenerator() Make span ID deterministic

• ns_timestamp_generator=TimeGenerator() Make Timestamps Deterministic

• processors=[SimpleSpanProcessor(exporter)] uses our TestExporter to capture spans. We use SimpleSpanProcessor to derive spans without delay.

insert_assert

this is devtools provides a utility function that, when run through pytest, automatically inserts the code output that calls it into the test file. That is, if you comment out that line, you will see 断言 capfire.exported_spans_as_dict() == [...] row is capfire.exported_spans_as_dict() Given that our test is deterministic, this should be exactly the same!

---

本文总阅读量次