Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
thread_patch = mock.patch("threading.Thread", autospec=True)
thread_patch.start()
self.addCleanup(thread_patch.stop)
configurator = Configurator()
configurator.add_route("example", "/example", request_method="GET")
configurator.add_view(example_application, route_name="example", renderer="json")
configurator.add_route("local_test", "/local_test", request_method="GET")
configurator.add_view(
local_parent_trace_within_context, route_name="local_test", renderer="json"
)
self.client = make_client("test-service")
self.observer = TraceBaseplateObserver(self.client)
self.baseplate = Baseplate()
self.baseplate.register(self.observer)
self.baseplate_configurator = BaseplateConfigurator(
self.baseplate, trust_trace_headers=True
)
configurator.include(self.baseplate_configurator.includeme)
app = configurator.make_wsgi_app()
self.local_span_ids = []
self.local_span_observers = []
self.test_app = webtest.TestApp(app)
def test_null_recorder_setup(self):
client = make_client("test-service", log_if_unconfigured=False)
baseplate_observer = TraceBaseplateObserver(client)
self.assertEqual(type(baseplate_observer.recorder), NullRecorder)
def test_register_server_span_observer(self):
client = make_client("test-service")
baseplate_observer = TraceBaseplateObserver(client)
context_mock = mock.Mock()
span = ServerSpan(
"test-id", "test-parent-id", "test-span-id", True, 0, "test", self.mock_context
)
baseplate_observer.on_server_span_created(context_mock, span)
self.assertEqual(len(span.observers), 1)
self.assertEqual(type(span.observers[0]), TraceServerSpanObserver)
def test_remote_recorder_setup(self):
client = make_client("test-service", tracing_endpoint=Endpoint("test:1111"))
baseplate_observer = TraceBaseplateObserver(client)
self.assertTrue(isinstance(baseplate_observer.recorder, RemoteRecorder))
def test_no_tracing_without_sampling(self):
client = make_client("test-service", sample_rate=0)
baseplate_observer = TraceBaseplateObserver(client)
context_mock = mock.Mock()
span = ServerSpan(
"test-id", "test-parent-id", "test-span-id", False, 0, "test", self.mock_context
)
baseplate_observer.on_server_span_created(context_mock, span)
self.assertEqual(len(span.observers), 0)
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))
def test_local_tracing_embedded(self):
with mock.patch.object(
TraceBaseplateObserver, "on_server_span_created", side_effect=self._register_server_mock
), mock.patch.object(
TraceServerSpanObserver, "on_child_span_created", side_effect=self._register_local_mock
), mock.patch.object(
TraceLocalSpanObserver, "on_child_span_created", side_effect=self._register_local_mock
):
self.test_app.get("/local_test")
# Verify that child span can be created within a local span context
# and parent IDs are inherited accordingly.
span = self.local_span_observers[-1]._serialize()
self.assertEqual(span["name"], "local-req")
self.assertEqual(len(span["annotations"]), 0)
self.assertEqual(span["parentId"], self.local_span_ids[-2])
def test_sets_hostname(self):
client = make_client("test-service")
baseplate_observer = TraceBaseplateObserver(client)
self.assertIsNotNone(baseplate_observer.hostname)
"""Collect and send span information for request tracing.
When configured, this will send tracing information automatically
collected by Baseplate to the configured distributed tracing service.
.. deprecated:: 1.0
Use :py:meth:`configure_observers` instead.
:param tracing_client: Tracing client to send request traces to.
"""
# pylint: disable=cyclic-import
from baseplate.observers.tracing import TraceBaseplateObserver
self.register(TraceBaseplateObserver(tracing_client))