1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
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)
|
