Logging handler to send logs to your OpenSearch cluster with bulk SSL. Forked from https://github.com/logzio/logzio-python-handler
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
175 lines
5.1 KiB
175 lines
5.1 KiB
import os
|
|
from unittest import TestCase
|
|
|
|
import logging
|
|
|
|
import sys
|
|
|
|
import re
|
|
from logzio.handler import LogzioHandler
|
|
|
|
|
|
class TestLogzioHandler(TestCase):
|
|
def setUp(self):
|
|
self.handler = LogzioHandler('moo')
|
|
|
|
def test_json(self):
|
|
formatter = logging.Formatter(
|
|
'{ "appname":"%(name)s", "functionName":"%(funcName)s", \"lineNo":"%(lineno)d", "severity":"%('
|
|
'levelname)s", "message":"%(message)s"}')
|
|
self.handler.setFormatter(formatter)
|
|
|
|
record = logging.LogRecord(
|
|
name='my-logger',
|
|
level=0,
|
|
pathname='handler_test.py',
|
|
lineno=10,
|
|
msg="this is a test: moo.",
|
|
args=[],
|
|
exc_info=None,
|
|
func='test_json'
|
|
)
|
|
|
|
formatted_message = self.handler.format_message(record)
|
|
formatted_message["@timestamp"] = None
|
|
|
|
self.assertDictEqual(
|
|
formatted_message,
|
|
{
|
|
'@timestamp': None,
|
|
'appname': 'my-logger',
|
|
'functionName': 'test_json',
|
|
'lineNo': '10',
|
|
'line_number': 10,
|
|
'log_level': 'NOTSET',
|
|
'logger': 'my-logger',
|
|
'message': 'this is a test: moo.',
|
|
'path_name': 'handler_test.py',
|
|
'severity': 'NOTSET',
|
|
'type': 'python'
|
|
}
|
|
)
|
|
|
|
def test_string(self):
|
|
record = logging.LogRecord(
|
|
name='my-logger',
|
|
level=0,
|
|
pathname='handler_test.py',
|
|
lineno=10,
|
|
msg="this is a test: moo.",
|
|
args=[],
|
|
exc_info=None,
|
|
func='test_json'
|
|
)
|
|
|
|
formatted_message = self.handler.format_message(record)
|
|
formatted_message["@timestamp"] = None
|
|
|
|
self.assertDictEqual(
|
|
formatted_message,
|
|
{
|
|
'@timestamp': None,
|
|
'line_number': 10,
|
|
'log_level': 'NOTSET',
|
|
'logger': 'my-logger',
|
|
'message': 'this is a test: moo.',
|
|
'path_name': 'handler_test.py',
|
|
'type': 'python'
|
|
}
|
|
)
|
|
|
|
def test_extra_formatting(self):
|
|
record = logging.LogRecord(
|
|
name='my-logger',
|
|
level=0,
|
|
pathname='handler_test.py',
|
|
lineno=10,
|
|
msg="this is a test: moo.",
|
|
args=[],
|
|
exc_info=None,
|
|
func='test_json'
|
|
)
|
|
|
|
record.__dict__["extra_key"] = "extra_value"
|
|
record.__dict__["module"] = "testing"
|
|
formatted_message = self.handler.format_message(record)
|
|
formatted_message["@timestamp"] = None
|
|
|
|
self.assertDictEqual(
|
|
formatted_message,
|
|
{
|
|
'@timestamp': None,
|
|
'line_number': 10,
|
|
'log_level': 'NOTSET',
|
|
'logger': 'my-logger',
|
|
'message': 'this is a test: moo.',
|
|
'path_name': 'handler_test.py',
|
|
'type': 'python',
|
|
'extra_key': 'extra_value'
|
|
}
|
|
)
|
|
|
|
def test_format_string_message(self):
|
|
record = logging.LogRecord(
|
|
name='my-logger',
|
|
level=0,
|
|
pathname='handler_test.py',
|
|
lineno=10,
|
|
msg="this is a test: %s.",
|
|
args=('moo',),
|
|
exc_info=None,
|
|
func='test_json'
|
|
)
|
|
|
|
formatted_message = self.handler.format_message(record)
|
|
formatted_message["@timestamp"] = None
|
|
|
|
self.assertDictEqual(
|
|
formatted_message,
|
|
{
|
|
'@timestamp': None,
|
|
'line_number': 10,
|
|
'log_level': 'NOTSET',
|
|
'logger': 'my-logger',
|
|
'message': 'this is a test: moo.',
|
|
'path_name': 'handler_test.py',
|
|
'type': 'python'
|
|
}
|
|
)
|
|
|
|
def test_exc(self):
|
|
try:
|
|
raise ValueError("oops.")
|
|
except:
|
|
exc_info = sys.exc_info()
|
|
|
|
record = logging.LogRecord(
|
|
name='my-logger',
|
|
level=0,
|
|
pathname='handler_test.py',
|
|
lineno=10,
|
|
msg='exception test:',
|
|
args=[],
|
|
exc_info=exc_info,
|
|
func='test_json'
|
|
)
|
|
|
|
formatted_message = self.handler.format_message(record)
|
|
formatted_message["@timestamp"] = None
|
|
|
|
formatted_message["exception"] = formatted_message["exception"].replace(os.path.abspath(__file__), "")
|
|
formatted_message["exception"] = re.sub(r", line \d+", "", formatted_message["exception"])
|
|
|
|
self.assertDictEqual(
|
|
{
|
|
'@timestamp': None,
|
|
'line_number': 10,
|
|
'log_level': 'NOTSET',
|
|
'logger': 'my-logger',
|
|
'message': 'exception test:',
|
|
'exception': 'Traceback (most recent call last):\n\n File "", in test_exc\n raise ValueError("oops.")\n\nValueError: oops.\n',
|
|
'path_name': 'handler_test.py',
|
|
'type': 'python'
|
|
},
|
|
formatted_message
|
|
)
|