Logfire로 테스트¶

API가 예상되는 데이터를 기록하고 있는지, 올바르게 포장하는 작업을 추적하고 있는지 등을 확인해야 할 수도 있습니다. 이것은 파이썬의 내장 로깅과 OpenTelemetry의 SDK를 사용하는 것을 포함하여 종종 어렵습니다.

Logfire를 사용하면logfire.testing 모듈의 유틸리티를 사용하여 로그 및 스팬을 쉽게 테스트할 수 있습니다. 이것이 Logfire가 내부적으로 테스트하는 방법입니다.

`capfire` 램프¶

exporter 와metrics_reader metrics_reader 의 두 가지 속성이 있습니다.

`exporter`¶

이것은TestExporter 의 인스턴스이며 내보낸 스팬을 메모리에 유지하는 OpenTelemetry SDK 호환 스팬 내보내기입니다.

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 는DevTores에서 제공하는 유틸리티 기능입니다.

아래에서 자세히 확인하십시오.

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

이것은dev에서 제공하는 유틸리티 기능으로, pyest를 실행할 때 테스트 파일에 호출하는 출력을 자동으로 삽입합니다. 즉, 해당 라인에 댓글을 달면 断言 capfire.exported_spans_as_dict() == [...] 라인이 capfire.exported_spans_as_dict() 의 현재 출력으로 대체되는 것을 볼 수 있습니다. 테스트가 결정적이라는 점을 감안할 때 정확히 동일해야 합니다!

本文总阅读量次

Logfire로 테스트¶

capfire 램프¶

exporter¶

metrics_reader¶

¶

¶

¶

insert_assert¶

`capfire` 램프¶

`exporter`¶

`metrics_reader`¶

`insert_assert`¶