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_severity_critical(self):
handler = BugsnagHandler()
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.critical('The system is down')
logger.removeHandler(handler)
self.assertSentReportCount(1)
json_body = self.server.received[0]['json_body']
event = json_body['events'][0]
exception = event['exceptions'][0]
self.assertEqual('LogCRITICAL', exception['errorClass'])
self.assertEqual('error', event['severity'])
self.assertEqual(logging.CRITICAL,
event['metaData']['extra data']['levelno'])
self.assertEqual(u('CRITICAL'),
def test_custom_level(self):
handler = BugsnagHandler()
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.log(341, 'The system is down')
logger.removeHandler(handler)
self.assertSentReportCount(1)
json_body = self.server.received[0]['json_body']
event = json_body['events'][0]
exception = event['exceptions'][0]
self.assertEqual('LogLevel 341', exception['errorClass'])
def test_message(self):
handler = BugsnagHandler()
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.critical('The system is down')
logger.removeHandler(handler)
self.assertSentReportCount(1)
json_body = self.server.received[0]['json_body']
event = json_body['events'][0]
exception = event['exceptions'][0]
self.assertEqual('The system is down', exception['message'])
log.info('Git revision %r', app.config['GIT_REVISION'])
# Configure Bugsnag
if not app.config.get('TESTING') and app.config.get('BUGSNAG_API_KEY'):
import bugsnag
import bugsnag.flask
import bugsnag.handlers
bugsnag.configure(
api_key=app.config['BUGSNAG_API_KEY'],
project_root="/data/git/pillar/pillar",
revision=app.config['GIT_REVISION'],
)
bugsnag.flask.handle_exceptions(app)
bs_handler = bugsnag.handlers.BugsnagHandler()
bs_handler.setLevel(logging.ERROR)
log.addHandler(bs_handler)
else:
log.info('Bugsnag NOT configured.')
# Google Cloud project
try:
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = \
app.config['GCLOUD_APP_CREDENTIALS']
except KeyError:
raise SystemExit('GCLOUD_APP_CREDENTIALS configuration is missing')
# Storage backend (GCS)
try:
os.environ['GCLOUD_PROJECT'] = app.config['GCLOUD_PROJECT']
except KeyError:
def log_handler(self, extra_fields=None):
return BugsnagHandler(client=self, extra_fields=extra_fields)
def _initialize(self):
bugsnag.configure(api_key=self._bugsnag_key, project_root=self._app_dir)
logging.config.dictConfig(self.get_logging_config())
logger = logging.getLogger('antbs')
for logger_name in self._noisy_loggers:
_logger = logging.getLogger(logger_name)
_logger.setLevel(logging.ERROR)
bs_handler_found = [h for h in logger.handlers if isinstance(h, BugsnagHandler)]
if not bs_handler_found:
bugsnag_handler = BugsnagHandler()
bugsnag_handler.setLevel(logging.WARNING)
logger.addHandler(bugsnag_handler)
self.logger = logger
def __init__(self, api_key=None, client=None, extra_fields=None):
"""
Creates a new handler which sends records to Bugsnag
"""
super(BugsnagHandler, self).__init__()
self.client = client
self.custom_metadata_fields = extra_fields
self.callbacks = [self.extract_default_metadata,
self.extract_custom_metadata,
self.extract_severity]
if api_key is not None:
warnings.warn('api_key is deprecated in favor of using a client '
'to set the correct API key '
'and will be removed in a future release.',
DeprecationWarning)
def add_api_key(record, options):
options['api_key'] = api_key
self.add_callback(add_api_key)
logger.addHandler(stream_handler)
if cmd_line.log_server:
log_server = cmd_line.log_server
if log_server == 'bugsnag':
if not BUGSNAG_ERROR:
# Bugsnag logger
bugsnag_api = context_filter.api_key
if bugsnag_api is not None:
bugsnag.configure(
api_key=bugsnag_api,
app_version=info.CNCHI_VERSION,
project_root='/usr/share/cnchi/cnchi',
release_stage=info.CNCHI_RELEASE_STAGE)
bugsnag_handler = BugsnagHandler(api_key=bugsnag_api)
bugsnag_handler.setLevel(logging.WARNING)
bugsnag_handler.setFormatter(formatter)
bugsnag_handler.addFilter(context_filter.filter)
bugsnag.before_notify(context_filter.bugsnag_before_notify_callback)
logger.addHandler(bugsnag_handler)
logging.info(
"Sending Cnchi log messages to bugsnag server (using python-bugsnag).")
else:
logging.warning(
"Cannot read the bugsnag api key, logging to bugsnag is not possible.")
else:
logging.warning(BUGSNAG_ERROR)
else:
# Socket logger
socket_handler = logging.handlers.SocketHandler(
log_server,
def _initialize(self):
bugsnag.configure(api_key=self._bugsnag_key, project_root=self._app_dir)
logging.config.dictConfig(self.get_logging_config())
logger = logging.getLogger('antbs')
for logger_name in self._noisy_loggers:
_logger = logging.getLogger(logger_name)
_logger.setLevel(logging.ERROR)
bs_handler_found = [h for h in logger.handlers if isinstance(h, BugsnagHandler)]
if not bs_handler_found:
bugsnag_handler = BugsnagHandler()
bugsnag_handler.setLevel(logging.WARNING)
logger.addHandler(bugsnag_handler)
self.logger = logger