Pengujian dengan Logfire¶
Anda mungkin perlu memeriksa apakah API Anda mencatat data yang Anda harapkan, melacak pekerjaan yang mereka bungkus dengan benar, dll. Ini seringkali sulit, termasuk menggunakan built-in logging Python dan OpenTelemetry 's SDK.
Logfire memanfaatkan logfire.testing Utilitas dalam modul membuatnya sangat mudah untuk menguji log dan bentang yang dikeluarkan. Ini juga cara Logfire menguji dirinya sendiri secara internal.
capfire Lampu¶
Ini memiliki dua sifat,exporter dan metrics_reader .
exporter¶
Ini adalah contoh TestExporter , dan merupakan eksportir bentang yang kompatibel dengan OpenTelemetry SDK yang membuat bentang yang diekspor dalam memori.
Metode exporter.exported_spans_as_dict() memungkinkan Anda memperoleh representasi kamus sederhana dari bentang yang diekspor, yang dapat dengan mudah Anda tegaskan dan peroleh perbedaan yang baik.
Pendekatan ini melakukan beberapa pemrosesan data untuk membuat output lebih mudah dibaca dan deterministik, misalnya, mengganti nomor baris dengan 123 dan jalur file dengan nama file.
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_assertadalah fungsi utilitas yang disediakan oleh DevTools.
Anda dapat mengakses bentang yang diekspor dengan 按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
Anda dapat memanggil exporter.clear() untuk mengatur ulang rentang yang ditangkap dalam pengujian.
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¶
Ini adalah contoh InMemoryMetricReader yang membaca metrik ke dalam memori.
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
Mari kita lihat utilitas yang kita gunakan.
¶
Salah satu hal yang paling kompleks ketika membandingkan output log dengan hasil yang diharapkan adalah sumber non-determinisme. Untuk rentang OpenTelemetry, dua bentang terbesar adalah bentang dan jejak ID dan stempel waktu.
IncrementalIdGenerator Tingkatkan bentang dan lacak ID secara berurutan sehingga output pengujian selalu sama.
¶
Kelas ini menghasilkan timestamp nanodetik yang bertambah 1 detik setiap kali timestamp dihasilkan.
¶
Ini adalah fungsi konfigurasi yang sama yang Anda gunakan dalam produksi, dan semuanya menyatu.
Harap dicatat bahwa kami secara khusus mengkonfigurasi:
-
send_to_logfire=False, karena kami tidak ingin menyentuh layanan produksi yang sebenarnya -
id_generator=IncrementalIdGenerator()Buat penentuan ID rentang -
ns_timestamp_generator=TimeGenerator()Buat stempel waktu menjadi deterministik -
processors=[SimpleSpanProcessor(exporter)]menggunakanTestExporterkami untuk menangkap bentang. Kami menggunakanSimpleSpanProcessoruntuk mengekspor bentang tanpa penundaan.
insert_assert¶
Ini adalah fungsi utilitas yang disediakan oleh devtools yang secara otomatis memasukkan output dari memanggilnya ke dalam file pengujian ketika di jalankan melalui pytest.
Artinya, jika Anda mengomentari baris itu, Anda akan melihat bahwa garis 断言 capfire.exported_spans_as_dict() == [...] digantikan oleh keluaran capfire.exported_spans_as_dict() saat ini, yang seharusnya persis sama mengingat tes kami deterministik!
本文总阅读量次