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_null_recorder_flush(self):
span = Span("test-id", "test-parent-id", "test-span-id", None, 0, "test", self.mock_context)
self.recorder.flush_func([span])
def test_force_sampling(self):
span_with_debug_flag = Span(
"test-id", "test-parent", "test-span-id", True, 1, "test", self.mock_context
)
span_without_debug_flag = Span(
"test-id", "test-parent", "test-span-id", True, 0, "test", self.mock_context
)
self.assertTrue(TraceBaseplateObserver.force_sampling(span_with_debug_flag))
self.assertFalse(TraceBaseplateObserver.force_sampling(span_without_debug_flag))
def test_should_sample_utilizes_sampled_setting(self):
client = make_client("test-service", sample_rate=0)
baseplate_observer = TraceBaseplateObserver(client)
span_with_sampled_flag = Span(
"test-id", "test-parent", "test-span-id", True, 0, "test", self.mock_context
)
self.assertTrue(baseplate_observer.should_sample(span_with_sampled_flag))
def test_should_sample_utilizes_force_sampling(self):
client = make_client("test-service", sample_rate=0)
baseplate_observer = TraceBaseplateObserver(client)
span_with_forced = Span(
"test-id", "test-parent", "test-span-id", False, 1, "test", self.mock_context
)
span_without_forced = Span(
"test-id", "test-parent", "test-span-id", False, 0, "test", self.mock_context
)
self.assertTrue(baseplate_observer.should_sample(span_with_forced))
self.assertFalse(baseplate_observer.should_sample(span_without_forced))
def make_test_span(context=None, local=False):
if not context:
context = mock.Mock()
span = Span(1, 2, 3, None, 0, "name", context)
if local:
span = LocalSpan(1, 2, 3, None, 0, "name", context)
return span
def setUp(self):
super().setUp()
self.recorder = NullRecorder()
self.mock_context = mock.Mock()
self.span = Span(
"test-id", "test-parent-id", "test-span-id", None, 0, "test", self.mock_context
)
self.debug_span = Span(
"test-id", "test-parent-id", "test-span-id", None, 1, "test", self.mock_context
)
self.test_span_observer = TraceSpanObserver(
"test-service", "test-hostname", self.span, self.recorder
)
self.test_debug_span_observer = TraceSpanObserver(
"test-service", "test-hostname", self.debug_span, self.recorder
)
def test_metrics(self):
mock_timer = mock.Mock(spec=Timer)
mock_counter = mock.Mock(spec=Counter)
mock_batch = mock.Mock(spec=Batch)
mock_batch.timer.return_value = mock_timer
mock_batch.counter.return_value = mock_counter
mock_client_span = mock.Mock(spec=Span)
mock_client_span.name = "example"
observer = MetricsClientSpanObserver(mock_batch, mock_client_span)
self.assertEqual(mock_batch.timer.call_count, 1)
self.assertEqual(mock_batch.timer.call_args, mock.call("clients.example"))
observer.on_start()
self.assertEqual(mock_timer.start.call_count, 1)
observer.on_incr_tag("test", delta=1)
mock_counter.increment.assert_called()
mock_counter.reset_mock()
observer.on_finish(exc_info=None)
self.assertEqual(mock_timer.stop.call_count, 1)
self.assertEqual(mock_counter.increment.call_count, 1)
def test_should_sample_utilizes_sample_rate(self):
client = make_client("test-service", sample_rate=1)
baseplate_observer = TraceBaseplateObserver(client)
span = Span("test-id", "test-parent", "test-span-id", None, 0, "test", self.mock_context)
self.assertTrue(baseplate_observer.should_sample(span))
baseplate_observer.sample_rate = 0
self.assertFalse(baseplate_observer.should_sample(span))