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
|
# 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 asyncio
import inspect
import secrets
import pytest
import structlog
from structlog.contextvars import (
_CONTEXT_VARS,
bind_contextvars,
bound_contextvars,
clear_contextvars,
get_contextvars,
get_merged_contextvars,
merge_contextvars,
reset_contextvars,
unbind_contextvars,
)
@pytest.fixture(autouse=True)
def _clear_contextvars():
"""
Make sure all tests start with a clean slate.
"""
clear_contextvars()
class TestContextvars:
async def test_bind(self):
"""
Binding a variable causes it to be included in the result of
merge_contextvars.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1)
return merge_contextvars(None, None, {"b": 2})
assert {"a": 1, "b": 2} == await event_loop.create_task(coro())
async def test_multiple_binds(self):
"""
Multiple calls to bind_contextvars accumulate values instead of
replacing them. But they override redefined ones.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1, c=3)
bind_contextvars(c=333, d=4)
return merge_contextvars(None, None, {"b": 2})
assert {
"a": 1,
"b": 2,
"c": 333,
"d": 4,
} == await event_loop.create_task(coro())
async def test_reset(self):
"""
reset_contextvars allows resetting contexvars to
previously-set values.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1)
assert {"a": 1} == get_contextvars()
await event_loop.create_task(nested_coro())
async def nested_coro():
tokens = bind_contextvars(a=2, b=3)
assert {"a": 2, "b": 3} == get_contextvars()
reset_contextvars(**tokens)
assert {"a": 1} == get_contextvars()
await event_loop.create_task(coro())
async def test_nested_async_bind(self):
"""
Context is passed correctly between "nested" concurrent operations.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1)
return await event_loop.create_task(nested_coro())
async def nested_coro():
bind_contextvars(c=3)
return merge_contextvars(None, None, {"b": 2})
assert {"a": 1, "b": 2, "c": 3} == await event_loop.create_task(coro())
async def test_merge_works_without_bind(self):
"""
merge_contextvars returns values as normal even when there has
been no previous calls to bind_contextvars.
"""
event_loop = asyncio.get_running_loop()
async def coro():
return merge_contextvars(None, None, {"b": 2})
assert {"b": 2} == await event_loop.create_task(coro())
async def test_merge_overrides_bind(self):
"""
Variables included in merge_contextvars override previously
bound variables.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1)
return merge_contextvars(None, None, {"a": 111, "b": 2})
assert {"a": 111, "b": 2} == await event_loop.create_task(coro())
async def test_clear(self):
"""
The context-local context can be cleared, causing any previously bound
variables to not be included in merge_contextvars's result.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1)
clear_contextvars()
return merge_contextvars(None, None, {"b": 2})
assert {"b": 2} == await event_loop.create_task(coro())
async def test_clear_without_bind(self):
"""
The context-local context can be cleared, causing any previously bound
variables to not be included in merge_contextvars's result.
"""
event_loop = asyncio.get_running_loop()
async def coro():
clear_contextvars()
return merge_contextvars(None, None, {})
assert {} == await event_loop.create_task(coro())
async def test_unbind(self):
"""
Unbinding a previously bound variable causes it to be removed from the
result of merge_contextvars.
"""
event_loop = asyncio.get_running_loop()
async def coro():
bind_contextvars(a=1)
unbind_contextvars("a")
return merge_contextvars(None, None, {"b": 2})
assert {"b": 2} == await event_loop.create_task(coro())
async def test_unbind_not_bound(self):
"""
Unbinding a not bound variable causes doesn't raise an exception.
"""
event_loop = asyncio.get_running_loop()
async def coro():
# Since unbinding means "setting to Ellipsis", we have to make
# some effort to ensure that the ContextVar never existed.
unbind_contextvars("a" + secrets.token_hex())
return merge_contextvars(None, None, {"b": 2})
assert {"b": 2} == await event_loop.create_task(coro())
async def test_parallel_binds(self):
"""
Binding a variable causes it to be included in the result of
merge_contextvars.
"""
event_loop = asyncio.get_running_loop()
coro1_bind = asyncio.Event()
coro2_bind = asyncio.Event()
bind_contextvars(c=3)
async def coro1():
bind_contextvars(a=1)
coro1_bind.set()
await coro2_bind.wait()
return merge_contextvars(None, None, {"b": 2})
async def coro2():
bind_contextvars(a=2)
await coro1_bind.wait()
coro2_bind.set()
return merge_contextvars(None, None, {"b": 2})
coro1_task = event_loop.create_task(coro1())
coro2_task = event_loop.create_task(coro2())
assert {"a": 1, "b": 2, "c": 3} == await coro1_task
assert {"a": 2, "b": 2, "c": 3} == await coro2_task
def test_get_only_gets_structlog_without_deleted(self):
"""
get_contextvars returns only the structlog-specific key-values with
the prefix removed. Deleted keys (= Ellipsis) are ignored.
"""
bind_contextvars(a=1, b=2)
unbind_contextvars("b")
_CONTEXT_VARS["foo"] = "bar"
assert {"a": 1} == get_contextvars()
def test_get_merged_merges_context(self):
"""
get_merged_contextvars merges a bound context into the copy.
"""
bind_contextvars(x=1)
log = structlog.get_logger().bind(y=2)
assert {"x": 1, "y": 2} == get_merged_contextvars(log)
class TestBoundContextvars:
def test_cleanup(self):
"""
Bindings are cleaned up
"""
with bound_contextvars(x=42, y="foo"):
assert {"x": 42, "y": "foo"} == get_contextvars()
assert {} == get_contextvars()
def test_cleanup_conflict(self):
"""
Overwritten keys are restored after the clean up
"""
bind_contextvars(x="original", z="unrelated")
with bound_contextvars(x=42, y="foo"):
assert {"x": 42, "y": "foo", "z": "unrelated"} == get_contextvars()
assert {"x": "original", "z": "unrelated"} == get_contextvars()
def test_preserve_independent_bind(self):
"""
New bindings inside bound_contextvars are preserved after the clean up
"""
with bound_contextvars(x=42):
bind_contextvars(y="foo")
assert {"x": 42, "y": "foo"} == get_contextvars()
assert {"y": "foo"} == get_contextvars()
def test_nesting_works(self):
"""
bound_contextvars binds and unbinds even when nested
"""
with bound_contextvars(l1=1):
assert {"l1": 1} == get_contextvars()
with bound_contextvars(l2=2):
assert {"l1": 1, "l2": 2} == get_contextvars()
assert {"l1": 1} == get_contextvars()
assert {} == get_contextvars()
def test_as_decorator(self):
"""
bound_contextvars can be used as a decorator and it preserves the
name, signature and documentation of the wrapped function.
"""
@bound_contextvars(x=42)
def wrapped(arg1):
"""Wrapped documentation"""
bind_contextvars(y=arg1)
assert {"x": 42, "y": arg1} == get_contextvars()
wrapped(23)
assert "wrapped" == wrapped.__name__
assert "(arg1)" == str(inspect.signature(wrapped))
assert "Wrapped documentation" == wrapped.__doc__
|
