Testen mit Logfire¶

Möglicherweise müssen Sie überprüfen , ob Ihre APIs die von Ihnen erwarteten Daten aufzeichnen , die Arbeit ihrer Pakete richtig verfolgen usw . Dies ist oft schwierig , einschließlich der Verwendung der integrierten Protokollierung von Python und des SDKs von OpenTelemetry .

Logfire erleichtert das Testen von Logs und Span mit dem Dienstprogramm im [logfire.testing ][logfire.testinglogfire.testing]-Modul , das intern von Logfire verwendet wird , um sich selbst zu testen .

`capfire` Leuchten .¶

Es hat zwei Eigenschaften. exporter und metrics_reader。

`exporter`¶

Dies ist eine Instanz von TestExporter und ist ein SPAN-Exporter , der mit dem OpenTelemetry SDK kompatibel ist und den exportierten span im Speicher behält .

exporter.exported_spans_as_dict() Methode ermöglicht es Ihnen, eine einfache Wörterbuchdarstellung der exportierten Spanne zu erhalten, und Sie können diesen Bereich leicht behaupten und gute Unterschiede daraus ziehen Dieser Ansatz führt einige Datenverarbeitung durch, um die Ausgabe lesbarer und deterministischer zu machen, zum Beispiel mit. 123 Ersetzen Sie den Pfad durch den Dateinamen.

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 ist eine von DevTools bereitgestellte nützliche Funktion .

Im Folgenden finden Sie weitere Informationen zu .

Sie können 按exporter.exported_spans auf den exportierten span zugreifen .

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

Sie können [exporter.clear() ][] [logfire.testing.TestExporter.clear] aufrufen , um die im Test erfasste Spanne zurückzusetzen .

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

Dies ist eine Instanz von InMemoryMetricReader , die den Indikator in den Speicher einliest .

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

Werfen wir einen Blick auf die von uns verwendeten Verfahren .

¶

IncrementalId生成器

Eines der komplexesten Dinge beim Vergleich der Log-Ausgabe mit den erwarteten Ergebnissen ist die Quelle der Unbestimmtheit für die OpenTelemetry-Spanne , die beiden größten Spannweiten sind Spanne und Tracking-ID und Zeitstempel .

IncrementalIdGenerator Erhöhen Sie die Spanne und die Tracking-ID nacheinander , sodass die Testausgabe immer gleich ist .

¶

时间生成器

Diese Art erzeugt einen Nanosekunden-Zeitstempel , der sich jedes Mal um eine Sekunde erhöht , wenn ein Zeitstempel erzeugt wird .

¶

logfire.配置

Es ist die gleiche Konfigurationsfunktion , die Sie in der Produktion verwenden , und alles ist gebündelt .

Bitte beachten Sie , dass wir speziell für Sie reserviert sind :

send_to_logfire=False Weil wir nicht auf die tatsächliche Produktion eingehen wollen .
id_generator=IncrementalIdGenerator() Machen Sie die Pass-ID sicher
ns_timestamp_generator=TimeGenerator() Zeitstempel mit Sicherheit .
processors=[SimpleSpanProcessor(exporter)] verwendet unser TestExporter zum Erfassen von Spannweiten und wir verwenden SimpleSpanProcessor zum Exportieren von Spannbreiten ohne Verzögerung .

`insert_assert`¶

Dies ist DevTools einrichten Eine nützliche Funktion, die bereitgestellt wird, wenn sie mit pytest ausgeführt wird, fügt automatisch die Ausgabe des aufrufenden Codes in die Testdatei ein, das heißt, wenn Sie die Zeile auskommentieren, sehen Sie sie.断言 capfire.exported_spans_as_dict() == [...] werden können capfire.exported_spans_as_dict() Die aktuelle Ausgabe Ersatz, da unsere Tests sind deterministisch, sollte dies genau das gleiche sein!

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

本文总阅读量次

Testen mit Logfire¶

capfire Leuchten .¶

exporter¶

metrics_reader¶

¶

¶

¶

insert_assert¶

`capfire` Leuchten .¶

`exporter`¶

`metrics_reader`¶

`insert_assert`¶