使用 Logfire 进行测试¶

您可能需要检查您的 API 是否记录了您期望的数据，是否正确跟踪了它们包装的工作等。这通常很困难，包括使用 Python 的内置日志记录和 OpenTelemetry 的 SDK。

Logfire使得使用logfire.testing模块中的实用程序测试发出的日志和跨度变得非常容易。这也是 Logfire 内部用来测试自身的方法。

`capfire` 灯具¶

它有两个属性 exporter 和 metrics_reader。

`exporter`¶

这是 TestExporter 的一个实例，并且是与 OpenTelemetry SDK 兼容的 span 导出器，可将导出的 span 保留在内存中。

exporter.exported_spans_as_dict() 方法让您获得导出的跨度的简单字典表示，您可以轻松地断言该范围并从中获得很好的差异。这种方法会进行一些数据处理，以使输出更具可读性和确定性，例如，用 123 替换行号，用文件名替换文件路径。

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',
                    },
                }
            ],
        },
    ]

insert_assert 是 DevTools 提供的实用函数。

在下面查看有关它的更多信息。

您可以按exporter.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

您可以调用 exporter.clear() 来重置测试中捕获的跨度。

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`¶

这是 InMemoryMetricReader 的一个实例，它将指标读入内存。

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

让我们来看看我们使用的实用程序。

¶

IncrementalId生成器

将日志输出与预期结果进行比较时，最复杂的事情之一是非确定性的来源。对于 OpenTelemetry 跨度，两个最大的跨度是跨度和跟踪 ID 和时间戳。

IncrementalIdGenerator 按顺序增加跨度和跟踪 ID，以便测试输出始终相同。

¶

时间生成器

此类生成纳秒时间戳，每次生成时间戳时，该时间戳都会增加 1 秒。

¶

logfire.配置

这与您在生产中使用的配置函数相同，并且所有内容都汇集在一起。

请注意，我们专门配置：

send_to_logfire=False，因为我们不想触及实际的生产服务
id_generator=IncrementalIdGenerator() 使跨度 ID 具有确定性
ns_timestamp_generator=TimeGenerator() 使时间戳具有确定性
processors=[SimpleSpanProcessor(exporter)] 使用我们的 TestExporter 来捕获跨度。我们使用 SimpleSpanProcessor 来无延迟地导出跨度。

`insert_assert`¶

这是 devtools 提供的一个实用函数，当通过 pytest 运行时，它会自动将调用它的代码输出插入到测试文件中。也就是说，如果你注释掉那行，你会看到断言 capfire.exported_spans_as_dict() == [...] 行被 capfire.exported_spans_as_dict() 的当前输出替换，鉴于我们的测试是确定性的，这应该完全相同！

本文总阅读量次

使用 Logfire 进行测试¶

capfire 灯具¶

exporter¶

metrics_reader¶

¶

¶

¶

insert_assert¶

`capfire` 灯具¶

`exporter`¶

`metrics_reader`¶

`insert_assert`¶