From e451a31ed7a0d5fa268a2065906c6fb803e6ae26 Mon Sep 17 00:00:00 2001
From: Martin Fischer <martin@push-f.com>
Date: Sat, 22 Mar 2025 11:47:54 +0100
Subject: initial commit
---
test_logformat.py | 222 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 222 insertions(+)
create mode 100755 test_logformat.py
(limited to 'test_logformat.py')
diff --git a/test_logformat.py b/test_logformat.py
new file mode 100755
index 0000000..cbcc14d
--- /dev/null
+++ b/test_logformat.py
@@ -0,0 +1,222 @@
+#!/usr/bin/env python3
+import asyncio
+import contextlib
+import datetime
+import functools
+import io
+import logging
+import textwrap
+import unittest
+import unittest.mock
+
+import logformat
+
+
+def set_up(level=logging.DEBUG):
+ stream = io.StringIO()
+ logfmt_handler = logging.StreamHandler(stream)
+ logfmt_handler.setFormatter(logformat.LogfmtFormatter())
+ logging.basicConfig(handlers=[logfmt_handler], level=level, force=True)
+ return stream
+
+
+# We're not using pytest here because pytest messes with the logging config,
+# setting it's own handlers, apparently even when disabling the pytest logging
+# plugin with `-p no:logging`.
+test_suite = unittest.TestSuite()
+
+
+def test(test):
+ @functools.wraps(test)
+ def wrapped_test():
+ t = unittest.TestCase()
+ with unittest.mock.patch("time.time_ns", return_value=1742634901036965331):
+ test(t)
+
+ test_suite.addTest(unittest.FunctionTestCase(wrapped_test))
+
+
+@test
+def test_logformat(t: unittest.TestCase):
+ stream = set_up()
+ logger = logformat.get_logger()
+ logger.debug("what's happening")
+ logger.info("something's happening")
+ logger.warning("this seems off")
+ logger.error("oh no")
+ t.assertEqual(
+ stream.getvalue(),
+ textwrap.dedent(
+ '''
+ time=2025-03-22T09:15:01Z level=debug msg="what's happening"
+ time=2025-03-22T09:15:01Z level=info msg="something's happening"
+ time=2025-03-22T09:15:01Z level=warning msg="this seems off"
+ time=2025-03-22T09:15:01Z level=error msg="oh no"
+ '''
+ ).lstrip(),
+ )
+
+
+@test
+def test_escaping(t: unittest.TestCase):
+ stream = set_up()
+ logger = logformat.get_logger()
+ logger.warning(" \t\n\x00='\"‽")
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=warning msg=" \\t\\n\\u0000=\'\\"‽"\n',
+ )
+
+
+@test
+def test_fmtlogger_extra_uses_str(t: unittest.TestCase):
+ class Foo:
+ def __str__(self):
+ return "hello!"
+
+ stream = set_up()
+ logger = logformat.get_logger()
+ logger.warning("test", value=Foo())
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=warning msg=test value=hello!\n',
+ )
+
+
+@test
+def test_fmtlogger_extra_none(t: unittest.TestCase):
+ stream = set_up()
+ logger = logformat.get_logger()
+ logger.warning("test", helpful_info=None)
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=warning msg=test helpful_info=null\n',
+ )
+
+
+@test
+def test_fmtlogger_extra_bool(t: unittest.TestCase):
+ stream = set_up()
+ logger = logformat.get_logger()
+ logger.warning("test", off_by_one=True)
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=warning msg=test off_by_one=true\n',
+ )
+
+
+@test
+def test_fmtlogger_extra_datetime(t: unittest.TestCase):
+ stream = set_up()
+ logger = logformat.get_logger()
+ logger.warning(
+ "test",
+ expiry=datetime.datetime(2025, 3, 22, 15, 30, 00, 00, tzinfo=datetime.UTC),
+ )
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=warning msg=test expiry=2025-03-22T15:30:00Z\n',
+ )
+
+
+@test
+def test_stdlib_logger(t: unittest.TestCase):
+ stream = set_up()
+ logger = logging.getLogger()
+ logger.debug("what's happening %s", 1234)
+ logger.info("something's happening")
+ logger.warning("this seems off")
+ logger.error("oh no")
+ logger.critical("aaaaaaaaah!")
+ t.assertEqual(
+ stream.getvalue(),
+ textwrap.dedent(
+ '''
+ time=2025-03-22T09:15:01Z level=debug msg="what's happening 1234"
+ time=2025-03-22T09:15:01Z level=info msg="something's happening"
+ time=2025-03-22T09:15:01Z level=warning msg="this seems off"
+ time=2025-03-22T09:15:01Z level=error msg="oh no"
+ time=2025-03-22T09:15:01Z level=critical msg=aaaaaaaaah!
+ '''
+ ).lstrip(),
+ )
+
+
+@test
+def test_log_uncaught_sync_returns(t: unittest.TestCase):
+ stream = set_up()
+
+ logger = logformat.get_logger()
+
+ @logger.log_uncaught
+ def run():
+ return 42
+
+ t.assertEqual(run(), 42)
+ t.assertEqual(stream.getvalue(), "")
+
+
+@test
+def test_log_uncaught_sync_raises(t: unittest.TestCase):
+ stream = set_up()
+
+ logger = logformat.get_logger()
+
+ @logger.log_uncaught
+ def run():
+ 5 / 0
+
+ with t.assertRaises(SystemExit), stub_traceback():
+ run()
+
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=error msg="unexpected exception: division by zero" traceback="stub traceback"\n',
+ )
+
+
+@test
+def test_log_uncaught_async_returns(t: unittest.TestCase):
+ stream = set_up(level=logging.INFO)
+
+ logger = logformat.get_logger()
+
+ @logger.log_uncaught
+ async def run():
+ return 42
+
+ t.assertEqual(asyncio.run(run()), 42)
+ t.assertEqual(stream.getvalue(), "")
+
+
+@test
+def test_log_uncaught_async_raises(t: unittest.TestCase):
+ stream = set_up(level=logging.INFO)
+
+ logger = logformat.get_logger()
+
+ @logger.log_uncaught
+ async def run():
+ 5 / 0
+
+ with t.assertRaises(SystemExit), stub_traceback():
+ asyncio.run(run())
+
+ t.assertEqual(
+ stream.getvalue(),
+ 'time=2025-03-22T09:15:01Z level=error msg="unexpected exception: division by zero" task=Task-4 traceback="stub traceback"\n',
+ )
+
+
+@contextlib.contextmanager
+def stub_traceback():
+ def stub_print_exception(*args, **kwargs):
+ kwargs['file'].write('stub traceback')
+
+ with unittest.mock.patch("traceback.print_exception", stub_print_exception):
+ yield
+
+
+if __name__ == "__main__":
+ runner = unittest.TextTestRunner()
+ runner.run(test_suite)
--
cgit v1.2.3