Source code in logfire/_internal/exporters/test.py
35363738
defexport(self,spans:Sequence[ReadableSpan])->SpanExportResult:"""Exports a batch of telemetry data."""self.exported_spans.extend(spans)returnSpanExportResult.SUCCESS
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)ifSpanAttributes.EXCEPTION_STACKTRACEinattributes:last_line=next(# pragma: no branchline.strip()forlineinreversed(cast(str,event.attributes[SpanAttributes.EXCEPTION_STACKTRACE]).split('\n'))ifline.strip())attributes[SpanAttributes.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')]
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 by random.seed(...).
Trace IDs are 128-bit integers.
Span IDs are 64-bit integers.
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
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
@pytest.fixturedefcapfire()->CaptureLogfire:"""A fixture that returns a CaptureLogfire instance."""exporter=TestExporter()metrics_reader=InMemoryMetricReader()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=[InMemoryMetricReader()]),)returnCaptureLogfire(exporter=exporter,metrics_reader=metrics_reader,log_exporter=log_exporter)