Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_basic_spans_emit(self):
hex_encoder = codecs.getencoder('hex')
client = mock.Mock()
client.Export.return_value = iter([1])
span_data0 = span_data_module.SpanData(
name="name0",
context=span_context_module.SpanContext(
trace_id='0e0c63257de34c92bf9efcd03927272e'),
span_id='0e0c63257de34c92',
parent_span_id=None,
start_time=None,
end_time=None,
attributes=None,
child_span_count=None,
stack_trace=None,
time_events=None,
links=None,
status=None,
same_process_as_parent_span=None,
span_kind=0)
def test_emit(self, monitor_resource_mock):
trace_id = '6e0c63257de34c92bf9efcd03927272e'
span_datas = [
span_data_module.SpanData(
name='span',
context=span_context.SpanContext(trace_id=trace_id),
span_id='1111',
parent_span_id=None,
attributes=None,
start_time=None,
end_time=None,
child_span_count=None,
stack_trace=None,
time_events=None,
links=None,
status=None,
same_process_as_parent_span=None,
span_kind=0,
)
]
mock_exporter = mock.MagicMock()
tracer = tracer_module.Tracer(exporter=mock_exporter,
sampler=samplers.AlwaysOnSampler())
return_value = "test"
@tracer.trace_decorator()
def test_decorator():
return return_value
returned = test_decorator()
self.assertEqual(returned, return_value)
self.assertEqual(mock_exporter.export.call_count, 1)
exported_spandata = mock_exporter.export.call_args[0][0][0]
self.assertIsInstance(exported_spandata, span_data.SpanData)
self.assertEqual(exported_spandata.name, 'test_decorator')
def test_translate_time_events_invalid(self):
ts = datetime.utcnow() + timedelta(seconds=-10)
span_data = span_data_module.SpanData(
context=span_context_module.SpanContext(
trace_id='6e0c63257de34c92bf9efcd03927272e'),
span_id='6e0c63257de34c92',
time_events=[time_event_module.TimeEvent(timestamp=ts)],
span_kind=span_module.SpanKind.SERVER,
status=None,
start_time=None,
end_time=None,
child_span_count=None,
name=None,
parent_span_id=None,
attributes=None,
same_process_as_parent_span=False,
stack_trace=None,
links=None)
def test_translate_status(self):
span_data = span_data_module.SpanData(
context=span_context_module.SpanContext(
trace_id='6e0c63257de34c92bf9efcd03927272e'),
span_id='6e0c63257de34c92',
status=status_module.Status(
code=2, message='ERR', details='details'),
span_kind=span_module.SpanKind.SERVER,
start_time=None,
end_time=None,
child_span_count=None,
name=None,
parent_span_id=None,
attributes=None,
same_process_as_parent_span=False,
stack_trace=None,
time_events=None,
links=None)
name="name0",
context=span_context_module.SpanContext(
trace_id='0e0c63257de34c92bf9efcd03927272e'),
span_id='0e0c63257de34c92',
parent_span_id=None,
attributes=None,
start_time=None,
end_time=None,
child_span_count=None,
stack_trace=None,
time_events=None,
links=None,
status=None,
same_process_as_parent_span=None,
span_kind=0),
span_data_module.SpanData(
name="name1",
context=span_context_module.SpanContext(
trace_id='1e0c63257de34c92bf9efcd03927272e'),
span_id='1e0c63257de34c92',
parent_span_id=None,
attributes=None,
start_time=None,
end_time=None,
child_span_count=None,
stack_trace=None,
time_events=None,
links=None,
status=None,
same_process_as_parent_span=None,
span_kind=0)
]
def test_span_data_immutable(self):
span_data = span_data_module.SpanData(
name='root',
context=None,
span_id='6e0c63257de34c92',
parent_span_id='6e0c63257de34c93',
attributes={'key1': 'value1'},
start_time=utils.to_iso_str(),
end_time=utils.to_iso_str(),
stack_trace=None,
links=None,
status=None,
annotations=None,
message_events=None,
same_process_as_parent_span=None,
child_span_count=None,
span_kind=0,
)
def test_translate_server_span_kindt(self):
server_span_data = span_data_module.SpanData(
context=span_context_module.SpanContext(
trace_id='6e0c63257de34c92bf9efcd03927272e'),
span_id='6e0c63257de34c92',
span_kind=span_module.SpanKind.SERVER,
child_span_count=1,
start_time=None,
end_time=None,
name=None,
parent_span_id=None,
attributes=None,
same_process_as_parent_span=False,
stack_trace=None,
time_events=None,
links=None,
status=None)
name='child_span',
context=span_context.SpanContext(trace_id=trace_id),
span_id='6e0c63257de34c92',
parent_span_id='6e0c63257de34c93',
attributes={'test_key': 'test_value'},
start_time='2017-08-15T18:02:26.071158Z',
end_time='2017-08-15T18:02:36.071158Z',
child_span_count=None,
stack_trace=None,
time_events=None,
links=None,
status=None,
same_process_as_parent_span=None,
span_kind=0,
),
span_data_module.SpanData(
name='child_span',
context=span_context.SpanContext(trace_id=trace_id),
span_id='6e0c63257de34c92',
parent_span_id='6e0c63257de34c93',
attributes={'test_key': 1},
start_time='2017-08-15T18:02:26.071158Z',
end_time='2017-08-15T18:02:36.071158Z',
child_span_count=None,
stack_trace=None,
time_events=None,
links=None,
status=None,
same_process_as_parent_span=None,
span_kind=None,
),
]
def format_legacy_trace_json(span_datas):
"""Formats a list of SpanData tuples into the legacy 'trace' dictionary
format for backwards compatibility
:type span_datas: list of :class:
`~opencensus.trace.span_data.SpanData`
:param list of opencensus.trace.span_data.SpanData span_datas:
SpanData tuples to emit
:rtype: dict
:return: Legacy 'trace' dictionary representing given SpanData tuples
"""
if not span_datas:
return {}
top_span = span_datas[0]
assert isinstance(top_span, SpanData)
trace_id = top_span.context.trace_id if top_span.context is not None \
else None
assert trace_id is not None
return {
'traceId': trace_id,
'spans': [_format_legacy_span_json(sd) for sd in span_datas],
}