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
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
|
# SPDX-License-Identifier: MIT OR Apache-2.0
# This file is dual licensed under the terms of the Apache License, Version
# 2.0, and the MIT License. See the LICENSE file in the root of this
# repository for complete details.
import logging
import pickle
import pytest
from structlog import make_filtering_bound_logger
from structlog._log_levels import LEVEL_TO_NAME
from structlog._native import _nop
from structlog.contextvars import (
bind_contextvars,
clear_contextvars,
merge_contextvars,
)
@pytest.fixture(name="bl")
def _bl(cl):
return make_filtering_bound_logger(logging.INFO)(cl, [], {})
class TestNativeFilteringLogger:
def test_is_enabled_for(self, bl):
"""
is_enabled_for returns True if the log level is enabled.
"""
assert bl.is_enabled_for(20)
assert bl.is_enabled_for(logging.INFO)
assert not bl.is_enabled_for(19)
assert not bl.is_enabled_for(logging.DEBUG)
def test_get_effective_level(self, bl):
"""
get_effective_level returns the log level.
"""
assert 20 == logging.INFO == bl.get_effective_level()
def test_exact_level(self, bl, cl):
"""
if log level is exactly the min_level, log.
"""
bl.info("yep")
assert [("info", (), {"event": "yep"})] == cl.calls
async def test_async_exact_level(self, bl, cl):
"""
if log level is exactly the min_level, log.
"""
await bl.ainfo("yep")
assert [("info", (), {"event": "yep"})] == cl.calls
def test_one_below(self, bl, cl):
"""
if log level is below the min_level, don't log.
"""
bl.debug("nope")
assert [] == cl.calls
async def test_async_one_below(self, bl, cl):
"""
if log level is below the min_level, don't log.
"""
await bl.adebug("nope")
assert [] == cl.calls
def test_filtered_interp(self, bl, cl):
"""
Passing interpolation args works if the log entry is filtered out.
"""
bl.debug("hello %s!", "world")
assert [] == cl.calls
async def test_async_filtered_interp(self, bl, cl):
"""
Passing interpolation args works if the log entry is filtered out.
"""
await bl.adebug("hello %s!", "world")
assert [] == cl.calls
def test_no_args(self, bl, cl):
"""
If no args are passed, don't attempt interpolation.
See also #473
"""
bl.info(42)
assert 42 == cl.calls[0][2]["event"]
async def test_async_no_args(self, bl, cl):
"""
If no args are passed, don't attempt interpolation.
See also #473
"""
await bl.ainfo(42)
assert 42 == cl.calls[0][2]["event"]
def test_log_exact_level(self, bl, cl):
"""
if log level is exactly the min_level, log.
"""
bl.log(logging.INFO, "yep")
assert [("info", (), {"event": "yep"})] == cl.calls
async def test_alog_exact_level(self, bl, cl):
"""
if log level is exactly the min_level, log.
"""
await bl.alog(logging.INFO, "yep")
assert [("info", (), {"event": "yep"})] == cl.calls
def test_log_one_below(self, bl, cl):
"""
if log level is below the min_level, don't log.
"""
bl.log(logging.DEBUG, "nope")
assert [] == cl.calls
async def test_alog_one_below(self, bl, cl):
"""
if log level is below the min_level, don't log.
"""
await bl.alog(logging.DEBUG, "nope")
assert [] == cl.calls
async def test_alog_no_args(self, bl, cl):
"""
If no args are passed, interpolation is not attempted.
See also #473
"""
await bl.alog(logging.INFO, 42)
assert 42 == cl.calls[0][2]["event"]
def test_log_interp(self, bl, cl):
"""
Interpolation happens if args are passed.
"""
bl.log(logging.INFO, "answer is %d.", 42)
assert "answer is 42." == cl.calls[0][2]["event"]
async def test_alog_interp(self, bl, cl):
"""
Interpolation happens if args are passed.
"""
await bl.alog(logging.INFO, "answer is %d.", 42)
assert "answer is 42." == cl.calls[0][2]["event"]
def test_filter_bound_below_missing_event_string(self, bl):
"""
Missing event arg causes exception below min_level.
"""
with pytest.raises(TypeError) as exc_info:
bl.debug(missing="event string!")
assert exc_info.type is TypeError
message = "missing 1 required positional argument: 'event'"
assert message in exc_info.value.args[0]
def test_filter_bound_exact_missing_event_string(self, bl):
"""
Missing event arg causes exception even at min_level.
"""
with pytest.raises(TypeError) as exc_info:
bl.info(missing="event string!")
assert exc_info.type is TypeError
message = "missing 1 required positional argument: 'event'"
assert message in exc_info.value.args[0]
def test_exception(self, bl, cl):
"""
exception ensures that exc_info is set to True, unless it's already
set.
"""
bl.exception("boom")
assert [("error", (), {"event": "boom", "exc_info": True})] == cl.calls
async def test_async_exception(self, bl, cl):
"""
aexception sets exc_info to current exception info, if it's not already
set.
"""
try:
raise Exception("boom")
except Exception as e:
await bl.aexception("foo")
exc = e
assert 1 == len(cl.calls)
assert isinstance(cl.calls[0][2]["exc_info"], tuple)
assert exc == cl.calls[0][2]["exc_info"][1]
def test_exception_positional_args(self, bl, cl):
"""
exception allows for positional args
"""
bl.exception("%s %s", "boom", "bastic")
assert [
("error", (), {"event": "boom bastic", "exc_info": True})
] == cl.calls
async def test_aexception_positional_args(self, bl, cl):
"""
aexception allows for positional args
"""
await bl.aexception("%s %s", "boom", "bastic")
assert 1 == len(cl.calls)
assert "boom bastic" == cl.calls[0][2]["event"]
async def test_async_exception_true(self, bl, cl):
"""
aexception replaces exc_info with current exception info, if exc_info
is True.
"""
try:
raise Exception("boom")
except Exception as e:
await bl.aexception("foo", exc_info=True)
exc = e
assert 1 == len(cl.calls)
assert isinstance(cl.calls[0][2]["exc_info"], tuple)
assert exc is cl.calls[0][2]["exc_info"][1]
def test_exception_passed(self, bl, cl):
"""
exception if exc_info has a value, exception doesn't tamper with it.
"""
bl.exception("boom", exc_info=42)
assert [("error", (), {"event": "boom", "exc_info": 42})] == cl.calls
async def test_async_exception_passed(self, bl, cl):
"""
exception if exc_info has a value (other than True), exception doesn't
tamper with it.
"""
await bl.aexception("boom", exc_info=42)
assert [("error", (), {"event": "boom", "exc_info": 42})] == cl.calls
def test_exception_pass_exception(self, bl, cl):
"""
If an Exception is passed for the event, don't explode.
Not a documented feature, but a regression for some people. See #473.
"""
try:
raise Exception("foo")
except Exception as e:
bl.exception(e)
exc = e
assert exc is cl.calls[0][2]["event"]
@pytest.mark.parametrize("level", tuple(LEVEL_TO_NAME.keys()))
def test_pickle(self, level):
"""
FilteringBoundLogger are pickleable.
"""
bl = make_filtering_bound_logger(level)
assert bl == pickle.loads(pickle.dumps(bl))
def test_pos_args(self, bl, cl):
"""
Positional arguments are used for string interpolation.
"""
bl.info("hello %s -- %d!", "world", 42)
assert [("info", (), {"event": "hello world -- 42!"})] == cl.calls
async def test_async_pos_args(self, bl, cl):
"""
Positional arguments are used for string interpolation.
"""
await bl.ainfo("hello %s -- %d!", "world", 42)
assert [("info", (), {"event": "hello world -- 42!"})] == cl.calls
@pytest.mark.parametrize(
("meth", "args"),
[
("aexception", ("ev",)),
("ainfo", ("ev",)),
("alog", (logging.INFO, "ev")),
],
)
async def test_async_contextvars_merged(self, meth, args, cl):
"""
Contextvars are merged into the event dict.
"""
clear_contextvars()
bl = make_filtering_bound_logger(logging.INFO)(
cl, [merge_contextvars], {}
)
bind_contextvars(context_included="yep")
await getattr(bl, meth)(*args)
assert len(cl.calls) == 1
assert "context_included" in cl.calls[0].kwargs
def test_log_percent(self, bl, cl):
"""
As long as there's no positional args passed, logging % is possible.
"""
bl.info("hey %! %%!")
assert [("info", (), {"event": "hey %! %%!"})] == cl.calls
def test_log_level_str(self):
"""
*min_level* can be a string and the case doesn't matter.
"""
bl = make_filtering_bound_logger("wArNiNg")
assert bl.warning is not _nop
assert bl.info is _nop
|
