diff --git a/.gitignore b/.gitignore index 2ea5a6c..8a6a5c2 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ __pycache__ /.tox /.cache /JSON_log_formatter.egg-info +.mypy_cache/ diff --git a/json_log_formatter/__init__.py b/json_log_formatter/__init__.py index 6e0be3d..a8b6979 100644 --- a/json_log_formatter/__init__.py +++ b/json_log_formatter/__init__.py @@ -1,33 +1,36 @@ +from typing import Any + import logging -from decimal import Decimal from datetime import datetime, timezone - +from decimal import Decimal import json - -BUILTIN_ATTRS = { - 'args', - 'asctime', - 'created', - 'exc_info', - 'exc_text', - 'filename', - 'funcName', - 'levelname', - 'levelno', - 'lineno', - 'module', - 'msecs', - 'message', - 'msg', - 'name', - 'pathname', - 'process', - 'processName', - 'relativeCreated', - 'stack_info', - 'taskName', - 'thread', - 'threadName', +from types import ModuleType + + +BUILTIN_ATTRS: set[str] = { + "args", + "asctime", + "created", + "exc_info", + "exc_text", + "filename", + "funcName", + "levelname", + "levelno", + "lineno", + "module", + "msecs", + "message", + "msg", + "name", + "pathname", + "process", + "processName", + "relativeCreated", + "stack_info", + "taskName", + "thread", + "threadName", } @@ -58,13 +61,13 @@ class JSONFormatter(logging.Formatter): """ - json_lib = json + json_lib: ModuleType = json - def format(self, record): - message = record.getMessage() - extra = self.extra_from_record(record) - json_record = self.json_record(message, extra, record) - mutated_record = self.mutate_json_record(json_record) + def format(self, record: logging.LogRecord) -> Any | str: + message: str = record.getMessage() + extra: dict[str, Any] = self.extra_from_record(record) + json_record: dict[str, Any] = self.json_record(message, extra, record) + mutated_record: dict[str, Any] = self.mutate_json_record(json_record) # Backwards compatibility: Functions that overwrite this but don't # return a new value will return None because they modified the # argument passed in. @@ -72,7 +75,7 @@ def format(self, record): mutated_record = json_record return self.to_json(mutated_record) - def to_json(self, record): + def to_json(self, record: dict[str, Any]) -> Any | str: """Converts record dict to a JSON string. It makes best effort to serialize a record (represents an object as a string) @@ -93,9 +96,9 @@ def to_json(self, record): try: return self.json_lib.dumps(record) except (TypeError, ValueError, OverflowError): - return '{}' + return "{}" - def extra_from_record(self, record): + def extra_from_record(self, record: logging.LogRecord) -> dict[str, Any]: """Returns `extra` dict you passed to logger. The `extra` keyword argument is used to populate the `__dict__` of @@ -108,7 +111,9 @@ def extra_from_record(self, record): if attr_name not in BUILTIN_ATTRS } - def json_record(self, message, extra, record): + def json_record( + self, message: str, extra: dict[str, Any], record: logging.LogRecord + ) -> dict[str, Any]: """Prepares a JSON payload which will be logged. Override this method to change JSON log format. @@ -120,16 +125,16 @@ def json_record(self, message, extra, record): :return: Dictionary which will be passed to JSON lib. """ - extra['message'] = message - if 'time' not in extra: - extra['time'] = datetime.now(timezone.utc) + extra["message"] = message + if "time" not in extra: + extra["time"] = datetime.now(timezone.utc) if record.exc_info: - extra['exc_info'] = self.formatException(record.exc_info) + extra["exc_info"] = self.formatException(record.exc_info) return extra - def mutate_json_record(self, json_record): + def mutate_json_record(self, json_record: dict[str, Any]) -> dict[str, Any]: """Override it to convert fields of `json_record` to needed types. Default implementation converts `datetime` to string in ISO8601 format. @@ -142,7 +147,7 @@ def mutate_json_record(self, json_record): return json_record -def _json_serializable(obj): +def _json_serializable(obj: Any) -> Any: try: return obj.__dict__ except AttributeError: @@ -189,23 +194,29 @@ class VerboseJSONFormatter(JSONFormatter): https://docs.python.org/3/library/logging.html#logrecord-attributes. """ - def json_record(self, message, extra, record): - extra['filename'] = record.filename - extra['funcName'] = record.funcName - extra['levelname'] = record.levelname - extra['lineno'] = record.lineno - extra['module'] = record.module - extra['name'] = record.name - extra['pathname'] = record.pathname - extra['process'] = record.process - extra['processName'] = record.processName - if hasattr(record, 'stack_info'): - extra['stack_info'] = record.stack_info - else: - extra['stack_info'] = None - extra['thread'] = record.thread - extra['threadName'] = record.threadName - return super(VerboseJSONFormatter, self).json_record(message, extra, record) + + def json_record( + self, message: str, extra: dict[str, Any], record: logging.LogRecord + ) -> dict[str, Any]: + extra.update( + { + "filename": record.filename, + "funcName": record.funcName, + "levelname": record.levelname, + "lineno": record.lineno, + "module": record.module, + "name": record.name, + "pathname": record.pathname, + "process": record.process, + "processName": record.processName, + "stack_info": record.stack_info + if hasattr(record, "stack_info") + else None, + "thread": record.thread, + "threadName": record.threadName, + } + ) + return super().json_record(message, extra, record) class FlatJSONFormatter(JSONFormatter): @@ -230,10 +241,12 @@ class FlatJSONFormatter(JSONFormatter): """ - keep = (bool, int, float, Decimal, complex, str, datetime) + keep: tuple[type, ...] = (bool, int, float, Decimal, complex, str, datetime) - def json_record(self, message, extra, record): - extra = super(FlatJSONFormatter, self).json_record(message, extra, record) + def json_record( + self, message: str, extra: dict[str, Any], record: logging.LogRecord + ) -> dict[str, Any]: + extra = super().json_record(message, extra, record) return { k: v if v is None or isinstance(v, self.keep) else str(v) for k, v in extra.items() diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 0000000..b22b34a --- /dev/null +++ b/mypy.ini @@ -0,0 +1,13 @@ +[mypy] +# To recognize all the files +files = ['__init__.py', 'tests.py', 'setup.py' ] +strict = True +check_untyped_defs = True +disallow_untyped_defs = True +disallow_incomplete_defs = True +no_implicit_optional = True +warn_redundant_casts = True +warn_unused_ignores = True +show_error_codes = True +show_column_numbers = True +ignore_missing_imports = False diff --git a/pyproject.toml b/pyproject.toml index cfb46b2..ac9fd07 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,3 +26,6 @@ classifiers=[ [project.urls] repository = "https://github.com/marselester/json-log-formatter" + +[tool.mypy] +warn_unused_configs = true diff --git a/setup.py b/setup.py index 6cd6c89..6febf06 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,4 @@ -from setuptools import setup +from setuptools import setup # type: ignore setup( name='JSON-log-formatter', diff --git a/tests.py b/tests.py index b2012d7..75bf3ba 100644 --- a/tests.py +++ b/tests.py @@ -2,6 +2,7 @@ from decimal import Decimal from io import BytesIO import unittest +from typing import Any import logging import json import os.path @@ -9,11 +10,11 @@ from django.core.handlers.wsgi import WSGIRequest from django.conf import settings -import ujson -import simplejson +import ujson # type: ignore +import simplejson # type: ignore try: - from cStringIO import StringIO + from cStringIO import StringIO # type: ignore except ImportError: from io import StringIO @@ -22,69 +23,72 @@ log_buffer = StringIO() json_handler = logging.StreamHandler(log_buffer) -logger = logging.getLogger('test') +logger = logging.getLogger("test") logger.addHandler(json_handler) logger.setLevel(logging.DEBUG) -logging.propagate = False +logger.propagate = False DATETIME = datetime(2015, 9, 1, 6, 9, 42, 797203) -DATETIME_ISO = u'2015-09-01T06:09:42.797203' +DATETIME_ISO = "2015-09-01T06:09:42.797203" settings.configure(DEBUG=True) class TestCase(unittest.TestCase): - def tearDown(self): + def tearDown(self) -> None: log_buffer.seek(0) log_buffer.truncate() class JSONFormatterTest(TestCase): - def setUp(self): + def setUp(self) -> None: json_handler.setFormatter(JSONFormatter()) - def test_given_time_is_used_in_log_record(self): - logger.info('Sign up', extra={'time': DATETIME}) + def test_given_time_is_used_in_log_record(self) -> None: + logger.info("Sign up", extra={"time": DATETIME}) expected_time = '"time": "2015-09-01T06:09:42.797203"' self.assertIn(expected_time, log_buffer.getvalue()) - def test_current_time_is_used_by_default_in_log_record(self): - logger.info('Sign up', extra={'fizz': 'bazz'}) + def test_current_time_is_used_by_default_in_log_record(self) -> None: + logger.info("Sign up", extra={"fizz": "bazz"}) self.assertNotIn(DATETIME_ISO, log_buffer.getvalue()) - def test_message_and_time_are_in_json_record_when_extra_is_blank(self): - logger.info('Sign up') + def test_message_and_time_are_in_json_record_when_extra_is_blank(self) -> None: + logger.info("Sign up") json_record = json.loads(log_buffer.getvalue()) - expected_fields = set([ - 'message', - 'time', - ]) + expected_fields = set( + [ + "message", + "time", + ] + ) self.assertTrue(expected_fields.issubset(json_record)) - def test_message_and_time_and_extra_are_in_json_record_when_extra_is_provided(self): - logger.info('Sign up', extra={'fizz': 'bazz'}) + def test_message_and_time_and_extra_are_in_json_record_when_extra_is_provided( + self, + ) -> None: + logger.info("Sign up", extra={"fizz": "bazz"}) json_record = json.loads(log_buffer.getvalue()) - expected_fields = set([ - 'message', - 'time', - 'fizz', - ]) + expected_fields = set( + [ + "message", + "time", + "fizz", + ] + ) self.assertTrue(expected_fields.issubset(json_record)) - def test_exc_info_is_logged(self): + def test_exc_info_is_logged(self) -> None: try: - raise ValueError('something wrong') + raise ValueError("something wrong") except ValueError: - logger.error('Request failed', exc_info=True) + logger.error("Request failed", exc_info=True) json_record = json.loads(log_buffer.getvalue()) - self.assertIn( - 'Traceback (most recent call last)', - json_record['exc_info'] - ) + self.assertIn("Traceback (most recent call last)", json_record["exc_info"]) class MutatingFormatter(JSONFormatter): - def mutate_json_record(self, json_record): + def mutate_json_record(self, json_record: dict[str, Any]) -> dict[str, Any]: new_record = {} for k, v in json_record.items(): if isinstance(v, datetime): @@ -94,338 +98,375 @@ def mutate_json_record(self, json_record): class MutatingFormatterTest(TestCase): - def setUp(self): + def setUp(self) -> None: json_handler.setFormatter(MutatingFormatter()) - def test_new_record_accepted(self): - logger.info('Sign up', extra={'fizz': DATETIME}) + def test_new_record_accepted(self) -> None: + logger.info("Sign up", extra={"fizz": DATETIME}) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['fizz'], DATETIME_ISO) + self.assertEqual(json_record["fizz"], DATETIME_ISO) class JsonLibTest(TestCase): - def setUp(self): + def setUp(self) -> None: json_handler.setFormatter(JSONFormatter()) - def test_builtin_types_are_serialized(self): - logger.log(level=logging.ERROR, msg='Payment was sent', extra={ - 'first_name': 'bob', - 'amount': 0.00497265, - 'context': { - 'tags': ['fizz', 'bazz'], + def test_builtin_types_are_serialized(self) -> None: + logger.log( + level=logging.ERROR, + msg="Payment was sent", + extra={ + "first_name": "bob", + "amount": 0.00497265, + "context": { + "tags": ["fizz", "bazz"], + }, + "things": ("a", "b"), + "ok": True, + "none": None, }, - 'things': ('a', 'b'), - 'ok': True, - 'none': None, - }) + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['first_name'], 'bob') - self.assertEqual(json_record['amount'], 0.00497265) - self.assertEqual(json_record['context'], {'tags': ['fizz', 'bazz']}) - self.assertEqual(json_record['things'], ['a', 'b']) - self.assertEqual(json_record['ok'], True) - self.assertEqual(json_record['none'], None) - - def test_decimal_is_serialized_as_string(self): - logger.log(level=logging.ERROR, msg='Payment was sent', extra={ - 'amount': Decimal('0.00497265') - }) + self.assertEqual(json_record["first_name"], "bob") + self.assertEqual(json_record["amount"], 0.00497265) + self.assertEqual(json_record["context"], {"tags": ["fizz", "bazz"]}) + self.assertEqual(json_record["things"], ["a", "b"]) + self.assertEqual(json_record["ok"], True) + self.assertEqual(json_record["none"], None) + + def test_decimal_is_serialized_as_string(self) -> None: + logger.log( + level=logging.ERROR, + msg="Payment was sent", + extra={"amount": Decimal("0.00497265")}, + ) expected_amount = '"amount": "0.00497265"' self.assertIn(expected_amount, log_buffer.getvalue()) - def test_django_wsgi_request_is_serialized_as_dict(self): - request = WSGIRequest({ - 'PATH_INFO': 'bogus', - 'REQUEST_METHOD': 'bogus', - 'CONTENT_TYPE': 'text/html; charset=utf8', - 'wsgi.input': BytesIO(b''), - }) - - logger.log(level=logging.ERROR, msg='Django response error', extra={ - 'status_code': 500, - 'request': request - }) + def test_django_wsgi_request_is_serialized_as_dict(self) -> None: + request = WSGIRequest( + { + "PATH_INFO": "bogus", + "REQUEST_METHOD": "bogus", + "CONTENT_TYPE": "text/html; charset=utf8", + "wsgi.input": BytesIO(b""), + } + ) + + logger.log( + level=logging.ERROR, + msg="Django response error", + extra={"status_code": 500, "request": request}, + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['status_code'], 500) - self.assertEqual(json_record['request']['path'], '/bogus') - self.assertEqual(json_record['request']['method'], 'BOGUS') + self.assertEqual(json_record["status_code"], 500) + self.assertEqual(json_record["request"]["path"], "/bogus") + self.assertEqual(json_record["request"]["method"], "BOGUS") - def test_json_circular_reference_is_handled(self): - d = {} - d['circle'] = d - logger.info('Referer checking', extra=d) - self.assertEqual('{}\n', log_buffer.getvalue()) + def test_json_circular_reference_is_handled(self) -> None: + d: dict[str, object] = {} + d["circle"] = d + logger.info("Referer checking", extra=d) + self.assertEqual("{}\n", log_buffer.getvalue()) class UjsonLibTest(TestCase): - def setUp(self): + def setUp(self) -> None: formatter = JSONFormatter() formatter.json_lib = ujson json_handler.setFormatter(formatter) - def test_builtin_types_are_serialized(self): - logger.log(level=logging.ERROR, msg='Payment was sent', extra={ - 'first_name': 'bob', - 'amount': 0.00497265, - 'context': { - 'tags': ['fizz', 'bazz'], + def test_builtin_types_are_serialized(self) -> None: + logger.log( + level=logging.ERROR, + msg="Payment was sent", + extra={ + "first_name": "bob", + "amount": 0.00497265, + "context": { + "tags": ["fizz", "bazz"], + }, + "things": ("a", "b"), + "ok": True, + "none": None, }, - 'things': ('a', 'b'), - 'ok': True, - 'none': None, - }) + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['first_name'], 'bob') - self.assertEqual(json_record['amount'], 0.00497265) - self.assertEqual(json_record['context'], {'tags': ['fizz', 'bazz']}) - self.assertEqual(json_record['things'], ['a', 'b']) - self.assertEqual(json_record['ok'], True) - self.assertEqual(json_record['none'], None) - - def test_decimal_is_serialized_as_number(self): - logger.info('Payment was sent', extra={ - 'amount': Decimal('0.00497265') - }) + self.assertEqual(json_record["first_name"], "bob") + self.assertEqual(json_record["amount"], 0.00497265) + self.assertEqual(json_record["context"], {"tags": ["fizz", "bazz"]}) + self.assertEqual(json_record["things"], ["a", "b"]) + self.assertEqual(json_record["ok"], True) + self.assertEqual(json_record["none"], None) + + def test_decimal_is_serialized_as_number(self) -> None: + logger.info("Payment was sent", extra={"amount": Decimal("0.00497265")}) expected_amount = '"amount":0.00497265' self.assertIn(expected_amount, log_buffer.getvalue()) - def test_zero_expected_when_decimal_is_in_scientific_notation(self): - logger.info('Payment was sent', extra={ - 'amount': Decimal('0E-8') - }) + def test_zero_expected_when_decimal_is_in_scientific_notation(self) -> None: + logger.info("Payment was sent", extra={"amount": Decimal("0E-8")}) expected_amount = '"amount":0.0' self.assertIn(expected_amount, log_buffer.getvalue()) - def test_django_wsgi_request_is_serialized_as_empty_list(self): - request = WSGIRequest({ - 'PATH_INFO': 'bogus', - 'REQUEST_METHOD': 'bogus', - 'CONTENT_TYPE': 'text/html; charset=utf8', - 'wsgi.input': BytesIO(b''), - }) - - logger.log(level=logging.ERROR, msg='Django response error', extra={ - 'status_code': 500, - 'request': request - }) + def test_django_wsgi_request_is_serialized_as_empty_list(self) -> None: + request = WSGIRequest( + { + "PATH_INFO": "bogus", + "REQUEST_METHOD": "bogus", + "CONTENT_TYPE": "text/html; charset=utf8", + "wsgi.input": BytesIO(b""), + } + ) + + logger.log( + level=logging.ERROR, + msg="Django response error", + extra={"status_code": 500, "request": request}, + ) json_record = json.loads(log_buffer.getvalue()) - if 'status_code' in json_record: - self.assertEqual(json_record['status_code'], 500) - if 'request' in json_record: - self.assertEqual(json_record['request']['path'], '/bogus') - self.assertEqual(json_record['request']['method'], 'BOGUS') + if "status_code" in json_record: + self.assertEqual(json_record["status_code"], 500) + if "request" in json_record: + self.assertEqual(json_record["request"]["path"], "/bogus") + self.assertEqual(json_record["request"]["method"], "BOGUS") - def test_json_circular_reference_is_handled(self): - d = {} - d['circle'] = d - logger.info('Referer checking', extra=d) - self.assertEqual('{}\n', log_buffer.getvalue()) + def test_json_circular_reference_is_handled(self) -> None: + d: dict[str, object] = {} + d["circle"] = d + logger.info("Referer checking", extra=d) + self.assertEqual("{}\n", log_buffer.getvalue()) class SimplejsonLibTest(TestCase): - def setUp(self): + def setUp(self) -> None: formatter = JSONFormatter() formatter.json_lib = simplejson json_handler.setFormatter(formatter) - def test_builtin_types_are_serialized(self): - logger.log(level=logging.ERROR, msg='Payment was sent', extra={ - 'first_name': 'bob', - 'amount': 0.00497265, - 'context': { - 'tags': ['fizz', 'bazz'], + def test_builtin_types_are_serialized(self) -> None: + logger.log( + level=logging.ERROR, + msg="Payment was sent", + extra={ + "first_name": "bob", + "amount": 0.00497265, + "context": { + "tags": ["fizz", "bazz"], + }, + "things": ("a", "b"), + "ok": True, + "none": None, }, - 'things': ('a', 'b'), - 'ok': True, - 'none': None, - }) + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['first_name'], 'bob') - self.assertEqual(json_record['amount'], 0.00497265) - self.assertEqual(json_record['context'], {'tags': ['fizz', 'bazz']}) - self.assertEqual(json_record['things'], ['a', 'b']) - self.assertEqual(json_record['ok'], True) - self.assertEqual(json_record['none'], None) - - def test_decimal_is_serialized_as_number(self): - logger.info('Payment was sent', extra={ - 'amount': Decimal('0.00497265') - }) + self.assertEqual(json_record["first_name"], "bob") + self.assertEqual(json_record["amount"], 0.00497265) + self.assertEqual(json_record["context"], {"tags": ["fizz", "bazz"]}) + self.assertEqual(json_record["things"], ["a", "b"]) + self.assertEqual(json_record["ok"], True) + self.assertEqual(json_record["none"], None) + + def test_decimal_is_serialized_as_number(self) -> None: + logger.info("Payment was sent", extra={"amount": Decimal("0.00497265")}) expected_amount = '"amount": 0.00497265' self.assertIn(expected_amount, log_buffer.getvalue()) - def test_decimal_is_serialized_as_it_is_when_it_is_in_scientific_notation(self): - logger.info('Payment was sent', extra={ - 'amount': Decimal('0E-8') - }) + def test_decimal_is_serialized_as_it_is_when_it_is_in_scientific_notation( + self, + ) -> None: + logger.info("Payment was sent", extra={"amount": Decimal("0E-8")}) expected_amount = '"amount": 0E-8' self.assertIn(expected_amount, log_buffer.getvalue()) - def test_django_wsgi_request_is_serialized_as_dict(self): - request = WSGIRequest({ - 'PATH_INFO': 'bogus', - 'REQUEST_METHOD': 'bogus', - 'CONTENT_TYPE': 'text/html; charset=utf8', - 'wsgi.input': BytesIO(b''), - }) - - logger.log(level=logging.ERROR, msg='Django response error', extra={ - 'status_code': 500, - 'request': request - }) + def test_django_wsgi_request_is_serialized_as_dict(self) -> None: + request = WSGIRequest( + { + "PATH_INFO": "bogus", + "REQUEST_METHOD": "bogus", + "CONTENT_TYPE": "text/html; charset=utf8", + "wsgi.input": BytesIO(b""), + } + ) + + logger.log( + level=logging.ERROR, + msg="Django response error", + extra={"status_code": 500, "request": request}, + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['status_code'], 500) - self.assertEqual(json_record['request']['path'], '/bogus') - self.assertEqual(json_record['request']['method'], 'BOGUS') + self.assertEqual(json_record["status_code"], 500) + self.assertEqual(json_record["request"]["path"], "/bogus") + self.assertEqual(json_record["request"]["method"], "BOGUS") - def test_json_circular_reference_is_handled(self): - d = {} - d['circle'] = d - logger.info('Referer checking', extra=d) - self.assertEqual('{}\n', log_buffer.getvalue()) + def test_json_circular_reference_is_handled(self) -> None: + d: dict[str, object] = {} + d["circle"] = d + logger.info("Referer checking", extra=d) + self.assertEqual("{}\n", log_buffer.getvalue()) class VerboseJSONFormatterTest(TestCase): - def setUp(self): + def setUp(self) -> None: json_handler.setFormatter(VerboseJSONFormatter()) - def test_file_name_is_testspy(self): - logger.error('An error has occured') + def test_file_name_is_testspy(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['filename'], 'tests.py') + self.assertEqual(json_record["filename"], "tests.py") - def test_function_name(self): - logger.error('An error has occured') + def test_function_name(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['funcName'], 'test_function_name') + self.assertEqual(json_record["funcName"], "test_function_name") - def test_level_name_is_error(self): - logger.error('An error has occured') + def test_level_name_is_error(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['levelname'], 'ERROR') + self.assertEqual(json_record["levelname"], "ERROR") - def test_module_name_is_tests(self): - logger.error('An error has occured') + def test_module_name_is_tests(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['module'], 'tests') + self.assertEqual(json_record["module"], "tests") - def test_logger_name_is_test(self): - logger.error('An error has occured') + def test_logger_name_is_test(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['name'], 'test') + self.assertEqual(json_record["name"], "test") - def test_path_name_is_test(self): - logger.error('An error has occured') + def test_path_name_is_test(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertIn(os.path.basename(os.path.abspath('.')) + '/tests.py', json_record['pathname']) + self.assertIn( + os.path.basename(os.path.abspath(".")) + "/tests.py", + json_record["pathname"], + ) - def test_process_name_is_MainProcess(self): - logger.error('An error has occured') + def test_process_name_is_MainProcess(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['processName'], 'MainProcess') + self.assertEqual(json_record["processName"], "MainProcess") - def test_thread_name_is_MainThread(self): - logger.error('An error has occured') + def test_thread_name_is_MainThread(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['threadName'], 'MainThread') + self.assertEqual(json_record["threadName"], "MainThread") - def test_stack_info_is_none(self): - logger.error('An error has occured') + def test_stack_info_is_none(self) -> None: + logger.error("An error has occured") json_record = json.loads(log_buffer.getvalue()) - self.assertIsNone(json_record['stack_info']) + self.assertIsNone(json_record["stack_info"]) class FlatJSONFormatterTest(TestCase): - def setUp(self): + def setUp(self) -> None: json_handler.setFormatter(FlatJSONFormatter()) - def test_given_time_is_used_in_log_record(self): - logger.info('Sign up', extra={'time': DATETIME}) + def test_given_time_is_used_in_log_record(self) -> None: + logger.info("Sign up", extra={"time": DATETIME}) expected_time = '"time": "2015-09-01T06:09:42.797203"' self.assertIn(expected_time, log_buffer.getvalue()) - def test_current_time_is_used_by_default_in_log_record(self): - logger.info('Sign up', extra={'fizz': 'bazz'}) + def test_current_time_is_used_by_default_in_log_record(self) -> None: + logger.info("Sign up", extra={"fizz": "bazz"}) self.assertNotIn(DATETIME_ISO, log_buffer.getvalue()) - def test_message_and_time_are_in_json_record_when_extra_is_blank(self): - logger.info('Sign up') + def test_message_and_time_are_in_json_record_when_extra_is_blank(self) -> None: + logger.info("Sign up") json_record = json.loads(log_buffer.getvalue()) - expected_fields = set([ - 'message', - 'time', - ]) + expected_fields = set( + [ + "message", + "time", + ] + ) self.assertTrue(expected_fields.issubset(json_record)) - def test_message_and_time_and_extra_are_in_json_record_when_extra_is_provided(self): - logger.info('Sign up', extra={'fizz': 'bazz'}) + def test_message_and_time_and_extra_are_in_json_record_when_extra_is_provided( + self, + ) -> None: + logger.info("Sign up", extra={"fizz": "bazz"}) json_record = json.loads(log_buffer.getvalue()) - expected_fields = set([ - 'message', - 'time', - 'fizz', - ]) + expected_fields = set( + [ + "message", + "time", + "fizz", + ] + ) self.assertTrue(expected_fields.issubset(json_record)) - def test_exc_info_is_logged(self): + def test_exc_info_is_logged(self) -> None: try: - raise ValueError('something wrong') + raise ValueError("something wrong") except ValueError: - logger.error('Request failed', exc_info=True) + logger.error("Request failed", exc_info=True) json_record = json.loads(log_buffer.getvalue()) - self.assertIn( - 'Traceback (most recent call last)', - json_record['exc_info'] - ) - - def test_builtin_types_are_serialized(self): - logger.log(level=logging.ERROR, msg='Payment was sent', extra={ - 'first_name': 'bob', - 'amount': 0.00497265, - 'context': { - 'tags': ['fizz', 'bazz'], + self.assertIn("Traceback (most recent call last)", json_record["exc_info"]) + + def test_builtin_types_are_serialized(self) -> None: + logger.log( + level=logging.ERROR, + msg="Payment was sent", + extra={ + "first_name": "bob", + "amount": 0.00497265, + "context": { + "tags": ["fizz", "bazz"], + }, + "things": ("a", "b"), + "ok": True, + "none": None, }, - 'things': ('a', 'b'), - 'ok': True, - 'none': None, - }) + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['first_name'], 'bob') - self.assertEqual(json_record['amount'], 0.00497265) - self.assertEqual(json_record['context'], "{'tags': ['fizz', 'bazz']}") - self.assertEqual(json_record['things'], "('a', 'b')") - self.assertEqual(json_record['ok'], True) - self.assertEqual(json_record['none'], None) - - def test_decimal_is_serialized_as_string(self): - logger.log(level=logging.ERROR, msg='Payment was sent', extra={ - 'amount': Decimal('0.00497265') - }) + self.assertEqual(json_record["first_name"], "bob") + self.assertEqual(json_record["amount"], 0.00497265) + self.assertEqual(json_record["context"], "{'tags': ['fizz', 'bazz']}") + self.assertEqual(json_record["things"], "('a', 'b')") + self.assertEqual(json_record["ok"], True) + self.assertEqual(json_record["none"], None) + + def test_decimal_is_serialized_as_string(self) -> None: + logger.log( + level=logging.ERROR, + msg="Payment was sent", + extra={"amount": Decimal("0.00497265")}, + ) expected_amount = '"amount": "0.00497265"' self.assertIn(expected_amount, log_buffer.getvalue()) - def test_django_wsgi_request_is_serialized_as_dict(self): - request = WSGIRequest({ - 'PATH_INFO': 'bogus', - 'REQUEST_METHOD': 'bogus', - 'CONTENT_TYPE': 'text/html; charset=utf8', - 'wsgi.input': BytesIO(b''), - }) - - logger.log(level=logging.ERROR, msg='Django response error', extra={ - 'status_code': 500, - 'request': request, - 'dict': { - 'request': request, + def test_django_wsgi_request_is_serialized_as_dict(self) -> None: + request = WSGIRequest( + { + "PATH_INFO": "bogus", + "REQUEST_METHOD": "bogus", + "CONTENT_TYPE": "text/html; charset=utf8", + "wsgi.input": BytesIO(b""), + } + ) + + logger.log( + level=logging.ERROR, + msg="Django response error", + extra={ + "status_code": 500, + "request": request, + "dict": { + "request": request, + }, + "list": [request], }, - 'list': [request], - }) + ) json_record = json.loads(log_buffer.getvalue()) - self.assertEqual(json_record['status_code'], 500) - self.assertEqual(json_record['request'], "") - self.assertEqual(json_record['dict'], "{'request': }") - self.assertEqual(json_record['list'], "[]") + self.assertEqual(json_record["status_code"], 500) + self.assertEqual(json_record["request"], "") + self.assertEqual( + json_record["dict"], "{'request': }" + ) + self.assertEqual(json_record["list"], "[]")