Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
""" Tests `core.framework` """
import pytest
from sap.cf_logging.core.context import Context
from sap.cf_logging.core.request_reader import RequestReader
from sap.cf_logging.core.response_reader import ResponseReader
from sap.cf_logging.core.framework import Framework
# pylint: disable=abstract-method
CONTEXT = Context()
REQUEST_READER = RequestReader()
RESPONSE_READER = ResponseReader()
@pytest.mark.parametrize('initializers', [
{'context': None, 'request_reader': REQUEST_READER, 'response_reader': RESPONSE_READER},
{'context': {}, 'request_reader': REQUEST_READER, 'response_reader': RESPONSE_READER},
{'context': CONTEXT, 'request_reader': None, 'response_reader': RESPONSE_READER},
{'context': CONTEXT, 'request_reader': {}, 'response_reader': RESPONSE_READER},
{'context': CONTEXT, 'request_reader': REQUEST_READER, 'response_reader': None},
{'context': CONTEXT, 'request_reader': REQUEST_READER, 'response_reader': {}}
])
@pytest.mark.xfail(raises=TypeError, strict=True)
def test_init(initializers):
""" test constructor with invalid context, request_reader and response_reader """
Framework('django', **initializers)
def before_each_setup_mocks(mocker):
context = Context()
mocker.patch.object(context, 'get', return_value=None)
request_reader = RequestReader()
request_reader.get_http_header = lambda self, key, default: 'some.host' \
if key == 'x-forwarded-for' else 'referer'
mocker.patch.object(request_reader, 'get_remote_user', return_value='user')
mocker.patch.object(request_reader, 'get_protocol', return_value='http')
mocker.patch.object(request_reader, 'get_content_length', return_value=0)
mocker.patch.object(request_reader, 'get_remote_ip',
return_value='1.2.3.4')
mocker.patch.object(request_reader, 'get_remote_port', return_value='1234')
mocker.patch.object(request_reader, 'get_path', return_value='/test/path')
mocker.patch.object(request_reader, 'get_method', return_value='GET')
def test_init_accept_inherited():
""" test Framework::init accepts inherited classes arguments """
class MyContext(Context): # pylint: disable=missing-docstring
pass
class MyRequestReader(RequestReader): # pylint: disable=missing-docstring
pass
class MyResponseReader(ResponseReader): # pylint: disable=missing-docstring
pass
Framework('name', MyContext(), MyRequestReader(), MyResponseReader())
"""
Django logging context - used by the logging package to keep
request specific data, needed for logging purposes.
For example correlation_id needs to be stored during request processing,
so all log entries contain it.
"""
from sap.cf_logging.core.context import Context
def _init_context(request):
if not hasattr(request, 'context'):
request.context = {}
class DjangoContext(Context):
""" Stores logging context in Django's request objecct """
def set(self, key, value, request):
if request is None:
return
_init_context(request)
request.context[key] = value
def get(self, key, request):
if request is None:
return None
_init_context(request)
return request.context.get(key) if request else None
def __init__(self, name, context, request_reader, response_reader):
if not name or not isinstance(name, STR_CLASS):
raise TypeError('Provided name is not valid string')
_check_instance(context, Context)
_check_instance(request_reader, RequestReader)
_check_instance(response_reader, ResponseReader)
self._name = name
self._context = context
self._request_reader = request_reader
self._response_reader = response_reader
""" Sanic logging context - used by the logging package to keep
request specific data during request processing for logging purposes.
For example correlation_id needs to be stored so all log entries contain it.
"""
from sap.cf_logging.core.context import Context
CONTEXT_NAME = 'cf_logger_context'
def _init_context(request):
if CONTEXT_NAME not in request:
request[CONTEXT_NAME] = {}
class SanicContext(Context):
""" Stores logging context in Sanic's request object """
def set(self, key, value, request):
if request is None:
return
_init_context(request)
request[CONTEXT_NAME][key] = value
def get(self, key, request):
if request is None:
return None
_init_context(request)
if key in request[CONTEXT_NAME]:
return request[CONTEXT_NAME][key]
return None
""" Falcon logging context - used by the logging package to keep
request specific data, needed for logging purposes.
For example correlation_id needs to be stored during request processing,
so all log entries contain it.
"""
from sap.cf_logging.core.context import Context
class FalconContext(Context):
""" Stores logging context in Falcon's request object"""
def set(self, key, value, request):
if request:
request.context[key] = value
def get(self, key, request):
return request.context.get(key) if request else None
""" Job logging context - used by the logging package to keep log data """
import threading
from sap.cf_logging.core.context import Context
class JobContext(Context, threading.local):
""" Stores logging context in dict """
def __init__(self):
super(JobContext, self).__init__()
self._mem_store = {}
def set(self, key, value, request):
self._mem_store[key] = value
def get(self, key, request):
return self._mem_store[key] if key in self._mem_store else None
""" Flask logging context - used by the logging package to keep
request specific data, needed for logging purposes.
For example correlation_id needs to be stored during request processing,
so all log entries contain it.
"""
from flask import g
from sap.cf_logging.core.context import Context
class FlaskContext(Context):
""" Stores logging context in Flask's request scope """
def set(self, key, value, request):
if g:
setattr(g, key, value)
def get(self, key, request):
return getattr(g, key, None) if g else None