How to use the baseplate.Baseplate 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 / integration / thrift_tests.py View on Github external
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()
github reddit / baseplate.py / tests / integration / thrift_tests.py View on Github external
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")
github reddit / baseplate.py / tests / integration / tracing_tests.py View on Github external
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)
github reddit / baseplate.py / tests / unit / core_tests.py View on Github external
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)
        )
github reddit / baseplate.py / tests / integration / pyramid_tests.py View on Github external
}
            },
            "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)
github reddit / baseplate.py / tests / integration / tracing_tests.py View on Github external
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)
github reddit / baseplate.py / tests / unit / core_tests.py View on Github external
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, [])
github reddit / baseplate.py / docs / tutorial / chapter4 / sql.py View on Github external
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()