Tester avec Logfire¶

Vous voudrez peut-être vérifier que vos API enregistrent les données que vous attendez, suivent correctement le travail qu 'elles enveloppent, etc. Cela est souvent difficile, y compris l'utilisation de la journalisation intégrée de Python et du SDK d'OpenTelemetry.

Logfire facilite grandement le test des logs et des portées émis à l'aide de l'utilitaire du module logfire.testing , qui est également utilisé en interne par Logfire pour se tester.

`capfire` Lampe à¶

Il possède deux propriétés exporter et metrics_reader ].

`exporter`¶

Il s'agit d'une instance de TestExporter et d'un exportateur de span compatible avec le SDK OpenTelemetry, qui conserve les span exportés en mémoire.

exporter.exported_spans_as_dict() La méthode vous permet d'obtenir une représentation simple du dictionnaire de l'étendue exportée, que vous pouvez facilement affirmer et obtenir de bonnes différences à partir de laquelle cette méthode effectue un certain traitement des données pour rendre la sortie plus lisible et déterministe, par exemple, avec 123 Remplacez les numéros de ligne, remplacez les chemins de fichiers par des noms de fichiers.

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 est une fonction utilitaire fournie par DevTools.

Voir plus d'informations à ce sujet ci-dessous.

Vous pouvez accéder à la portée exportée par 按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

Vous pouvez appeler exporter.clear() pour réinitialiser les étendues capturées dans un test.

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

Il s'agit d'une instance de InMemoryMetricReader , qui lit un indicateur en mémoire.

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

Jetons un coup d'œil à l'utilitaire que nous utilisons

¶

IncrementalId生成器

L'une des choses les plus compliquées lors de la comparaison de la sortie du journal avec le résultat attendu est une source de non-déterminisme-pour la portée OpenTelemetry, les deux plus grandes portées sont la portée et l'ID de suivi et l'horodatage.

IncrementalIdGenerator Augmentez la portée et l'ID de suivi dans l'ordre afin que le résultat du test soit toujours le même.

¶

时间生成器

Cette classe génère un horodatage nanoseconde qui est ajouté d'une seconde chaque fois qu 'il est généré.

¶

logfire.配置

C'est la même fonction de configuration que vous utilisez en production, et tout est rassemblé.

Veuillez noter que nous avons spécifiquement configuré:

send_to_logfire=False, parce que nous ne voulons pas toucher aux services de production réels
id_generator=IncrementalIdGenerator() Rendre le Span ID déterministe
ns_timestamp_generator=TimeGenerator() Rendre l'horodatage déterministe
processors=[SimpleSpanProcessor(exporter)] utilise notre TestExporter pour capturer les span. nous utilisons SimpleSpanProcessor pour exporter les span sans latence.

`insert_assert`¶

Il s'agit d'une fonction utilitaire fournie par devtools qui, lorsqu 'elle est exécutée via pytest, insère automatiquement la sortie du code qui l'appelle dans le fichier de test. c'est-à-dire que si vous commentez cette ligne, vous verrez que la ligne 断言 capfire.exported_spans_as_dict() == [...] est remplacée par la sortie actuelle de capfire.exported_spans_as_dict(), ce qui devrait être exactement la même chose étant donné que notre test est déterministe!

本文总阅读量次

Tester avec Logfire¶

capfire Lampe à¶

exporter¶

metrics_reader¶

¶

¶

¶

insert_assert¶

`capfire` Lampe à¶

`exporter`¶

`metrics_reader`¶

`insert_assert`¶