लॉगफायर के साथ परीक्षण

आपको यह जांचने की आवश्यकता हो सकती है कि आपके एपीआई आपके द्वारा अपेक्षित डेटा लॉग कर रहे हैं, उस काम को ट्रैक कर रहे हैं जो वे सही ढंग से लपेट रहे हैं, आदि। यह अक्सर मुश्किल होता है, जिसमें पायथन के अंतर्निहित लॉगिंग और ओपनटेलीमेट्री के एसडीके का उपयोग करना शामिल है।

लॉगफायर का उपयोग करता हैlogfire.testing मॉड्यूल में उपयोगिता जारी किए गए लॉग और स्पैन का परीक्षण करना बहुत आसान बनाती है। यह भी है कि लॉगफायर आंतरिक रूप से कैसे परीक्षण करता है।

capfire लैंप

इसके दो गुण हैं। exporter और metrics_reader...

exporter

यहाँ यह है। TestExporter और एक स्पैन निर्यातक है जो ओपनटेलीमेट्री एसडीके के साथ संगत है जो निर्यात किए गए स्पैन को मेमोरी में रखता है।

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 हाँ उपकरण उपयोगिता कार्य प्रदान किए गए।

   नीचे इसके बारे में अधिक जानकारी देखें...

आप कर सकते हैं按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生成器

लॉग आउटपुट की तुलना अपेक्षित परिणामों से करते समय सबसे जटिल चीजों में से एक गैर-निर्धारक का स्रोत है। के लिए, दो सबसे बड़े स्पैन स्पैन और ट्रेस आईडी और टाइमस्टैम्प हैं।

IncrementalIdGenerator स्पैन बढ़ाएं और आईडी को क्रमिक रूप से ट्रेस करें ताकि परीक्षण आउटपुट हमेशा समान हो।

时间生成器

यह वर्ग एक नैनोसेकंड टाइमस्टैम्प उत्पन्न करता है जो हर बार टाइमस्टैम्प उत्पन्न होने पर 1 सेकंड तक बढ़ जाता है।

logfire.配置

यह वही कॉन्फ़िगरेशन फ़ंक्शन है जिसका उपयोग आप उत्पादन में करते हैं, और यह सब एक साथ आता है।

कृपया ध्यान दें कि हम विशेष रूप से कॉन्फ़िगर करते हैं:

• send_to_logfire=False, क्योंकि हम वास्तविक उत्पादन सेवा को छूना नहीं चाहते हैं

• id_generator=IncrementalIdGenerator() स्पैन आईडी निर्धारक बनाएं

• ns_timestamp_generator=TimeGenerator() टाइमस्टैम्प निर्धारक बनाएं

• processors=[SimpleSpanProcessor(exporter)] हमारा उपयोग करें TestExporter स्पैन को पकड़ने के लिए। हम उपयोग करते हैं SimpleSpanProcessor बिना देरी के स्पैन प्राप्त करना।

insert_assert

यहाँ यह है। देवउपकरण एक उपयोगिता फ़ंक्शन प्रदान करता है, जो पाइटेस्ट के माध्यम से चलाने पर, स्वचालित रूप से कोड के आउटपुट को सम्मिलित करेगा जो इसे परीक्षण फ़ाइल में कॉल करता है। यही है, यदि आप उस पंक्ति पर टिप्पणी करते हैं, तो आप देखेंगे断言 capfire.exported_spans_as_dict() == [...] रजाई capfire.exported_spans_as_dict() प्रतिस्थापन का वर्तमान आउटपुट, यह देखते हुए कि हमारा परीक्षण निर्धारक है, बिल्कुल वैसा ही होना चाहिए!

in-memory-metric-reader https://opentelemetry-python.readthedocs.io/en/latest/sdk/metrics.export.html#opentelemetry.sdk.metrics.export.InMemoryMetricReader

---

本文总阅读量次