Перейти к содержанию

Тестирование с помощью Logfire

Возможно, вам придется проверить, записывает ли ваш API желаемые данные, правильно ли отслеживает работу с ними и т. д. Это часто бывает трудно, включая встроенные журналы на Python и SDK OpenTelemetry.

Logfire делает очень простым тестирование журналов и диапазонов с помощью утилиты logfire.testing модуль. Это также метод, используемый внутри Logfire для тестирования себя.

capfire Светильники

У него два свойства. exporter и metrics_reader。

exporter

Это экземпляр TestExporter и совместимый с OpenTelemetry SDK экспортируемый 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',
                    },
                }
            ],
        },
    ]

  1. 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 два самых больших диапазона-это диапазон и идентификатор отслеживания и временные метки.

IncrementalIdGenerator Последовательно увеличивать диапазон и идентификатор отслеживания, чтобы результаты тестирования были одинаковыми.

时间生成器

Эти типы генерируют наносекундную метку времени, которая увеличивается на 1 секунду при каждом создании метки времени.

logfire.配置

Это то же самое, что и функция конфигурации, которую вы используете в производстве, и все вместе.

Пожалуйста, обратите внимание, что мы специально настроены:

  • send_to_logfire=False Потому что мы не хотим касаться реальных производственных услуг.

  • id_generator=IncrementalIdGenerator() Обеспечивает определенность идентификатора пролета

  • ns_timestamp_generator=TimeGenerator() Обеспечивает определенность метки времени

  • processors=[SimpleSpanProcessor(exporter)] использует наш TestExporter для захвата пролетов. Мы используем SimpleSpanProcessor для экспорта пролетов без задержек.

insert_assert

Это функциональная функция devtools, которая автоматически вставляет вызывающий ее код в тестовый файл при запуске через pytest. То есть, если вы заметите эту строку, вы увидите, что строка 断言 capfire.exported_spans_as_dict() == [...] заменяется текущим выводом capfire.exported_spans_as_dict(). Это должно быть точно то же самое, учитывая, что наш тест является окончательным!

本文总阅读量次