How to use the baseplate.Span function in baseplate

To help you get started, we’ve selected a few baseplate examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github reddit / baseplate.py / tests / unit / observers / tracing_tests.py View on Github external
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])
github reddit / baseplate.py / tests / unit / observers / tracing_tests.py View on Github external
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))
github reddit / baseplate.py / tests / unit / observers / tracing_tests.py View on Github external
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))
github reddit / baseplate.py / tests / unit / observers / tracing_tests.py View on Github external
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))
github reddit / baseplate.py / tests / unit / core_tests.py View on Github external
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
github reddit / baseplate.py / tests / unit / observers / tracing_tests.py View on Github external
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
        )
github reddit / baseplate.py / tests / unit / observers / metrics_tests.py View on Github external
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)
github reddit / baseplate.py / tests / unit / observers / tracing_tests.py View on Github external
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))