Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init(initializers):
""" test constructor with invalid context, request_reader and response_reader """
Framework('django', **initializers)
def test_init_called_twice(mocker):
""" test cf_logging.init can be called only once """
framework = mocker.Mock(Framework)
cf_logging._SETUP_DONE = False
cf_logging.init(framework, level=logging.DEBUG)
cf_logging.init(framework, level=logging.DEBUG)
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())
def test_init_name(name):
""" test invalid 'name' provided to constructor """
Framework(name, CONTEXT, REQUEST_READER, RESPONSE_READER)
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')
response_reader = ResponseReader()
mocker.patch.object(response_reader, 'get_status_code', return_value='200')
mocker.patch.object(response_reader, 'get_response_size', return_value=0)
mocker.patch.object(response_reader, 'get_content_type',
return_value='text/plain')
_clean_log_env_vars()
global FRAMEWORK # pylint: disable=global-statement
FRAMEWORK = Framework('name', context, request_reader, response_reader)
yield
def init(cfl_framework=None, level=defaults.DEFAULT_LOGGING_LEVEL):
""" Initialize function. It sets up the logging library to output JSON
formatted messages.
Optional arguments framework to use and logging.level
"""
global FRAMEWORK # pylint: disable=global-statement
global _SETUP_DONE # pylint: disable=global-statement
if _SETUP_DONE:
raise RuntimeError('cf_logging already initialized')
if cfl_framework is not None and not isinstance(cfl_framework, Framework):
raise TypeError('expecting framework of type {}'.format(Framework.__name__))
_SETUP_DONE = True
FRAMEWORK = cfl_framework or JobFramework()
logging.setLoggerClass(CfLogger)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
root = logging.getLogger()
root.setLevel(level)
root.addHandler(handler)
def init(app, level=defaults.DEFAULT_LOGGING_LEVEL, username_key='username'):
""" Initializes logging in JSON format.
:param app: - Falcon application object
:param level: - valid log level from standard logging package (optional)
:param username_key: key used by the framework to get the username
out of the request user, set in the request context,
like `request.context.get('user').get(key)`
"""
if not isinstance(app, falcon.API):
raise TypeError('application should be instance of Falcon API')
framework = Framework(FALCON_FRAMEWORK_NAME, FalconContext(),
FalconRequestReader(username_key), FalconResponseReader())
cf_logging.init(framework, level)
def _init_framework(level):
logging.getLogger('werkzeug').disabled = True
framework = Framework(FLASK_FRAMEWORK_NAME,
FlaskContext(), FlaskRequestReader(), FlaskResponseReader())
cf_logging.init(framework, level)
def init(cfl_framework=None, level=defaults.DEFAULT_LOGGING_LEVEL):
""" Initialize function. It sets up the logging library to output JSON
formatted messages.
Optional arguments framework to use and logging.level
"""
global FRAMEWORK # pylint: disable=global-statement
global _SETUP_DONE # pylint: disable=global-statement
if _SETUP_DONE:
raise RuntimeError('cf_logging already initialized')
if cfl_framework is not None and not isinstance(cfl_framework, Framework):
raise TypeError('expecting framework of type {}'.format(Framework.__name__))
_SETUP_DONE = True
FRAMEWORK = cfl_framework or JobFramework()
logging.setLoggerClass(CfLogger)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
root = logging.getLogger()
root.setLevel(level)
root.addHandler(handler)