Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Testing

Testing utilities for Logfire. Seethe guide for examples.

TestExporter

TestExporter()

Bases:SpanExporter

A SpanExporter that stores exported spans in a list for asserting in tests.

Source code inlogfire/_internal/exporters/test.py
3031
def__init__(self)->None:self.exported_spans:list[ReadableSpan]=[]

export

Exports a batch of telemetry data.

Source code inlogfire/_internal/exporters/test.py
33343536
defexport(self,spans:Sequence[ReadableSpan])->SpanExportResult:"""Exports a batch of telemetry data."""self.exported_spans.extend(spans)returnSpanExportResult.SUCCESS

clear

clear()->None

Clears the collected spans.

Source code inlogfire/_internal/exporters/test.py
383940
defclear(self)->None:"""Clears the collected spans."""self.exported_spans=[]

exported_spans_as_dict

exported_spans_as_dict(fixed_line_number:int|None=123,strip_filepaths:bool=True,include_resources:bool=False,include_instrumentation_scope:bool=False,_include_pending_spans:bool=False,_strip_function_qualname:bool=True,parse_json_attributes:bool=False,)->list[dict[str,Any]]

The exported spans as a list of dicts.

Parameters:

NameTypeDescriptionDefault

fixed_line_number

int | None

The line number to use for all spans.

123

strip_filepaths

bool

Whether to strip the filepaths from the exported spans.

True

include_resources

bool

Whether to include the resource attributes in the exported spans.

False

include_instrumentation_scope

bool

Whether to include the instrumentation scope in the exported spans.

False

parse_json_attributes

bool

Whether to parse strings containing JSON arrays/objects.

False

Returns:

TypeDescription
list[dict[str,Any]]

A list of dicts representing the exported spans.

Source code inlogfire/_internal/exporters/test.py
 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99100101102103104105106107108109110111112113114115116117118119120121122123124125126
defexported_spans_as_dict(self,fixed_line_number:int|None=123,strip_filepaths:bool=True,include_resources:bool=False,include_instrumentation_scope:bool=False,_include_pending_spans:bool=False,_strip_function_qualname:bool=True,parse_json_attributes:bool=False,)->list[dict[str,Any]]:"""The exported spans as a list of dicts.    Args:        fixed_line_number: The line number to use for all spans.        strip_filepaths: Whether to strip the filepaths from the exported spans.        include_resources: Whether to include the resource attributes in the exported spans.        include_instrumentation_scope: Whether to include the instrumentation scope in the exported spans.        parse_json_attributes: Whether to parse strings containing JSON arrays/objects.    Returns:        A list of dicts representing the exported spans.    """_build_attributes=partial(build_attributes,fixed_line_number=fixed_line_number,strip_filepaths=strip_filepaths,strip_function_qualname=_strip_function_qualname,parse_json_attributes=parse_json_attributes,)defbuild_context(context:trace.SpanContext)->dict[str,Any]:return{'trace_id':context.trace_id,'span_id':context.span_id,'is_remote':context.is_remote}defbuild_link(link:trace.Link)->dict[str,Any]:context=link.contextortrace.INVALID_SPAN_CONTEXTreturn{'context':build_context(context),'attributes':_build_attributes(link.attributes)}defbuild_event(event:Event)->dict[str,Any]:res:dict[str,Any]={'name':event.name,'timestamp':event.timestamp}ifevent.attributes:# pragma: no branchres['attributes']=attributes=dict(event.attributes)if'exception.stacktrace'inattributes:last_line=next(# pragma: no branchline.strip()forlineinreversed(cast(str,event.attributes['exception.stacktrace']).split('\n'))ifline.strip())attributes['exception.stacktrace']=last_linereturnresdefbuild_instrumentation_scope(span:ReadableSpan)->dict[str,Any]:ifinclude_instrumentation_scope:return{'instrumentation_scope':span.instrumentation_scopeandspan.instrumentation_scope.name}else:return{}defbuild_span(span:ReadableSpan)->dict[str,Any]:context=span.contextortrace.INVALID_SPAN_CONTEXTres:dict[str,Any]={'name':span.name,'context':build_context(context),'parent':build_context(span.parent)ifspan.parentelseNone,'start_time':span.start_time,'end_time':span.end_time,**build_instrumentation_scope(span),'attributes':_build_attributes(span.attributes),}ifspan.links:res['links']=[build_link(link)forlinkinspan.links]ifspan.events:res['events']=[build_event(event)foreventinspan.events]ifinclude_resources:resource_attributes=_build_attributes(span.resource.attributes)res['resource']={'attributes':resource_attributes,}returnresspans=[build_span(span)forspaninself.exported_spans]return[spanforspaninspansif_include_pending_spansisTrueor(span.get('attributes',{}).get(ATTRIBUTES_SPAN_TYPE_KEY,'span')!='pending_span')]

TestLogExporter

TestLogExporter(ns_timestamp_generator:Callable[[],int])

Bases:InMemoryLogExporter

A LogExporter that stores exported logs in a list for asserting in tests.

Source code inlogfire/_internal/exporters/test.py
187188189
def__init__(self,ns_timestamp_generator:typing.Callable[[],int])->None:super().__init__()self.ns_timestamp_generator=ns_timestamp_generator

SeededRandomIdGeneratordataclass

SeededRandomIdGenerator(seed:int|None=0,_ms_timestamp_generator:Callable[[],int]=_default_ms_timestamp_generator,)

Bases:IdGenerator

Generate random span/trace IDs from a seed for deterministic tests.

Similar to RandomIdGenerator from OpenTelemetry, but with a seed.Set the seed to None for non-deterministic randomness.In that case the difference from RandomIdGenerator is that it's not affected byrandom.seed(...).

Trace IDs are 128-bit integers.Span IDs are 64-bit integers.

IncrementalIdGeneratordataclass

IncrementalIdGenerator()

Bases:IdGenerator

Generate sequentially incrementing span/trace IDs for testing.

Trace IDs start at 1 and increment by 1 each time.Span IDs start at 1 and increment by 1 each time.

reset_trace_span_ids

reset_trace_span_ids()->None

Resets the trace and span ids.

Source code inlogfire/testing.py
44454647
defreset_trace_span_ids(self)->None:# pragma: no cover"""Resets the trace and span ids."""self.trace_id_counter=0self.span_id_counter=0

generate_span_id

generate_span_id()->int

Generates a span id.

Source code inlogfire/testing.py
495051525354
defgenerate_span_id(self)->int:"""Generates a span id."""self.span_id_counter+=1ifself.span_id_counter>2**64-1:# pragma: no branchraiseOverflowError('Span ID overflow')# pragma: no coverreturnself.span_id_counter

generate_trace_id

generate_trace_id()->int

Generates a trace id.

Source code inlogfire/testing.py
565758596061
defgenerate_trace_id(self)->int:"""Generates a trace id."""self.trace_id_counter+=1ifself.trace_id_counter>2**128-1:# pragma: no branchraiseOverflowError('Trace ID overflow')# pragma: no coverreturnself.trace_id_counter

TimeGenerator

TimeGenerator(ns_time:int=0)

Generate incrementing timestamps for testing.

Timestamps are in nanoseconds, start at 1_000_000_000, and increment by 1_000_000_000 (1 second) each time.

Source code inlogfire/testing.py
7172
def__init__(self,ns_time:int=0):self.ns_time=ns_time

CaptureLogfiredataclass

CaptureLogfire(exporter:TestExporter,metrics_reader:InMemoryMetricReader,log_exporter:TestLogExporter,)

A dataclass that holds a span exporter, log exporter, and metric reader.

This is used as the return type ofcapfire fixture.

exporterinstance-attribute

exporter:TestExporter

The span exporter.

metrics_readerinstance-attribute

metrics_reader:InMemoryMetricReader

The metric reader.

log_exporterinstance-attribute

log_exporter:TestLogExporter

The log exporter.

get_collected_metrics

get_collected_metrics()

Get the collected metrics as a list of dictionaries.

Source code inlogfire/testing.py
969798
defget_collected_metrics(self):"""Get the collected metrics as a list of dictionaries."""returnget_collected_metrics(self.metrics_reader)

capfire

capfire()->CaptureLogfire

A fixture that returns a CaptureLogfire instance.

Source code inlogfire/testing.py
101102103104105106107108109110111112113114115116117118119120
@pytest.fixturedefcapfire()->CaptureLogfire:"""A fixture that returns a CaptureLogfire instance."""exporter=TestExporter()metrics_reader=InMemoryMetricReader(preferred_temporality=METRICS_PREFERRED_TEMPORALITY)time_generator=TimeGenerator()log_exporter=TestLogExporter(time_generator)logfire.configure(send_to_logfire=False,console=False,advanced=logfire.AdvancedOptions(id_generator=IncrementalIdGenerator(),ns_timestamp_generator=time_generator,log_record_processors=[SimpleLogRecordProcessor(log_exporter)],),additional_span_processors=[SimpleSpanProcessor(exporter)],metrics=logfire.MetricsOptions(additional_readers=[metrics_reader]),)returnCaptureLogfire(exporter=exporter,metrics_reader=metrics_reader,log_exporter=log_exporter)
[8]ページ先頭
©2009-2025 Movatter.jp