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_constructor_explicit(self):
span_id = 'test_span_id'
span_name = 'test_span_name'
parent_span = mock.Mock()
start_time = utils.to_iso_str()
end_time = utils.to_iso_str()
attributes = {
'http.status_code': '200',
'component': 'HTTP load balancer',
}
annotations = [mock.Mock()]
message_events = [mock.Mock()]
links = [mock.Mock()]
stack_trace = mock.Mock()
status = mock.Mock()
context_tracer = mock.Mock()
span = self._make_one(
name=span_name,
parent_span=parent_span,
attributes=attributes,
def test_constructor_explicit(self):
span_id = 'test_span_id'
span_name = 'test_span_name'
parent_span = mock.Mock()
start_time = utils.to_iso_str()
end_time = utils.to_iso_str()
attributes = {
'http.status_code': '200',
'component': 'HTTP load balancer',
}
annotations = [mock.Mock()]
message_events = [mock.Mock()]
links = [mock.Mock()]
stack_trace = mock.Mock()
status = mock.Mock()
context_tracer = mock.Mock()
span = self._make_one(
name=span_name,
parent_span=parent_span,
attributes=attributes,
start_time=start_time,
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,
)
with self.assertRaises(AttributeError):
span_data.name = 'child'
with self.assertRaises(AttributeError):
span_data.new_attr = 'a'
def test_record_with_none_context(self):
measure = mock.Mock(spec=measure_module.MeasureInt)
sum_aggregation = aggregation_module.SumAggregation()
view = view_module.View("test_view", "description", ['key1', 'key2'],
measure, sum_aggregation)
start_time = datetime.utcnow()
end_time = datetime.utcnow()
view_data = view_data_module.ViewData(
view=view, start_time=start_time, end_time=end_time)
time = utils.to_iso_str()
value = 4
view_data.record(
context=None, value=value, timestamp=time, attachments=None)
tag_values = view_data.get_tag_values(
tags={}, columns=view.columns)
tuple_vals = tuple(tag_values)
self.assertEqual([None, None], tag_values)
self.assertTrue(tuple_vals in view_data.tag_value_aggregation_data_map)
sum_data = view_data.tag_value_aggregation_data_map.get(tuple_vals)
self.assertEqual(4, sum_data.sum_data)
view = view_module.View(
name=name,
description=description,
columns=columns,
measure=measure,
aggregation=distribution_aggregation)
start_time = datetime.utcnow()
attachments = {"One": "one", "Two": "two"}
end_time = datetime.utcnow()
view_data = view_data_module.ViewData(
view=view, start_time=start_time, end_time=end_time)
context = mock.Mock
context.map = {'key1': 'val1', 'key2': 'val2'}
time = utils.to_iso_str()
value = 1
view_data.record(
context=context,
value=value,
timestamp=time,
attachments=attachments)
tag_values = view_data.get_tag_values(
tags=context.map, columns=view.columns)
tuple_vals = tuple(tag_values)
self.assertEqual(['val1', 'val2'], tag_values)
self.assertIsNotNone(view_data.tag_value_aggregation_data_map)
self.assertTrue(tuple_vals in view_data.tag_value_aggregation_data_map)
self.assertIsNotNone(
view_data.tag_value_aggregation_data_map[tuple_vals])
self.assertIsNone(
def test_record_with_missing_key_in_context(self):
measure = mock.Mock(spec=measure_module.MeasureInt)
sum_aggregation = aggregation_module.SumAggregation()
view = view_module.View("test_view", "description", ['key1', 'key2'],
measure, sum_aggregation)
start_time = datetime.utcnow()
end_time = datetime.utcnow()
view_data = view_data_module.ViewData(
view=view, start_time=start_time, end_time=end_time)
context = mock.Mock()
context.map = {
'key1': 'val1',
'key3': 'val3'
} # key2 is not in the context.
time = utils.to_iso_str()
value = 4
view_data.record(
context=context, value=value, timestamp=time, attachments=None)
tag_values = view_data.get_tag_values(
tags=context.map, columns=view.columns)
tuple_vals = tuple(tag_values)
self.assertEqual(['val1', None], tag_values)
self.assertTrue(tuple_vals in view_data.tag_value_aggregation_data_map)
sum_data = view_data.tag_value_aggregation_data_map.get(tuple_vals)
self.assertEqual(4, sum_data.sum_data)
view_data.record(
context=context, value=value, timestamp=time, attachments=None)
tag_values = view_data.get_tag_values(
tags=context.map, columns=view.columns)
tuple_vals = tuple(tag_values)
self.assertEqual(['val1', 'val2'], tag_values)
self.assertIsNotNone(view_data.tag_value_aggregation_data_map)
self.assertTrue(tuple_vals in view_data.tag_value_aggregation_data_map)
self.assertIsNotNone(
view_data.tag_value_aggregation_data_map[tuple_vals])
sum_data = view_data.tag_value_aggregation_data_map.get(tuple_vals)
self.assertEqual(1, sum_data.sum_data)
context_2 = mock.Mock()
context_2.map = {'key1': 'val3', 'key2': 'val2'}
time_2 = utils.to_iso_str()
value_2 = 2
view_data.record(
context=context_2,
value=value_2,
timestamp=time_2,
attachments=None)
tag_values_2 = view_data.get_tag_values(
tags=context_2.map, columns=view.columns)
tuple_vals_2 = tuple(tag_values_2)
self.assertEqual(['val3', 'val2'], tag_values_2)
self.assertTrue(
tuple_vals_2 in view_data.tag_value_aggregation_data_map)
sum_data_2 = view_data.tag_value_aggregation_data_map.get(tuple_vals_2)
self.assertEqual(2, sum_data_2.sum_data)
time_3 = utils.to_iso_str()
def __init__(self, timestamp, description, attributes=None):
self.timestamp = utils.to_iso_str(timestamp)
self.description = description
self.attributes = attributes
"measurements, refusing to record")
return
for measure, value in self.measurement_map.items():
if value < 0:
self._invalid = True
logger.warning("Dropping values, value to record must be "
"non-negative")
logger.info("Measure '{}' has negative value ({}), refusing "
"to record measurements from {}"
.format(measure.name, value, self))
return
self.measure_to_view_map.record(
tags=tags,
measurement_map=self.measurement_map,
timestamp=utils.to_iso_str(),
attachments=self.attachments
)