Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def baseplate_thrift_client(endpoint, client_span_observer=None):
baseplate = Baseplate()
if client_span_observer:
class TestServerSpanObserver(ServerSpanObserver):
def on_child_span_created(self, span):
span.register(client_span_observer)
observer = TestServerSpanObserver()
class TestBaseplateObserver(BaseplateObserver):
def on_server_span_created(self, context, span):
span.register(observer)
baseplate.register(TestBaseplateObserver())
context = baseplate.make_context_object()
def serve_thrift(handler, server_span_observer=None):
# create baseplate root
baseplate = Baseplate()
if server_span_observer:
class TestBaseplateObserver(BaseplateObserver):
def on_server_span_created(self, context, server_span):
server_span.register(server_span_observer)
baseplate.register(TestBaseplateObserver())
# set up the server's processor
logger = mock.Mock(spec=logging.Logger)
edge_context_factory = make_edge_context_factory()
processor = TestService.Processor(handler)
processor = baseplateify_processor(processor, logger, baseplate, edge_context_factory)
# bind a server socket on an available port
server_bind_endpoint = config.Endpoint("127.0.0.1:0")
def test_configure_tracing_with_defaults_new_style(self):
baseplate = Baseplate()
self.assertEqual(0, len(baseplate.observers))
client = make_client("test")
baseplate.configure_tracing(client)
self.assertEqual(1, len(baseplate.observers))
tracing_observer = baseplate.observers[0]
self.assertEqual("test", tracing_observer.service_name)
def test_server_observer_made(self):
baseplate = Baseplate()
mock_context = baseplate.make_context_object()
mock_observer = mock.Mock(spec=BaseplateObserver)
baseplate.register(mock_observer)
server_span = baseplate.make_server_span(mock_context, "name", TraceInfo(1, 2, 3, None, 0))
self.assertEqual(baseplate.observers, [mock_observer])
self.assertEqual(mock_observer.on_server_span_created.call_count, 1)
self.assertEqual(
mock_observer.on_server_span_created.call_args, mock.call(mock_context, server_span)
)
}
},
"vault": {"token": "test", "url": "http://vault.example.com:8200/"},
}
secrets = SecretsStore("/secrets")
secrets._filewatcher = mock_filewatcher
self.observer = mock.Mock(spec=BaseplateObserver)
self.server_observer = mock.Mock(spec=ServerSpanObserver)
def _register_mock(context, server_span):
server_span.register(self.server_observer)
self.observer.on_server_span_created.side_effect = _register_mock
self.baseplate = Baseplate()
self.baseplate.register(self.observer)
self.baseplate_configurator = BaseplateConfigurator(
self.baseplate,
trust_trace_headers=True,
edge_context_factory=EdgeRequestContextFactory(secrets),
)
configurator.include(self.baseplate_configurator.includeme)
self.context_init_event_subscriber = mock.Mock()
configurator.add_subscriber(self.context_init_event_subscriber, ServerSpanInitialized)
app = configurator.make_wsgi_app()
self.test_app = webtest.TestApp(app)
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_server_observer(self):
baseplate = Baseplate()
mock_context = baseplate.make_context_object()
mock_observer = mock.Mock(spec=BaseplateObserver)
mock_observer.on_server_span_created.return_value = None
baseplate.register(mock_observer)
server_span = baseplate.make_server_span(mock_context, "name", TraceInfo(1, 2, 3, None, 0))
self.assertEqual(server_span.observers, [])
def make_wsgi_app(app_config):
baseplate = Baseplate(app_config)
baseplate.configure_observers()
baseplate.configure_context({"db": SQLAlchemySession()})
configurator = Configurator(settings=app_config)
configurator.include(BaseplateConfigurator(baseplate).includeme)
configurator.add_route("hello_world", "/", request_method="GET")
configurator.scan()
return configurator.make_wsgi_app()