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
|
import asyncio
import contextvars
import random
import threading
from sqlalchemy import exc
from sqlalchemy.testing import async_test
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_true
from sqlalchemy.util import await_fallback
from sqlalchemy.util import await_only
from sqlalchemy.util import greenlet_spawn
from sqlalchemy.util import queue
try:
from greenlet import greenlet
except ImportError:
greenlet = None
async def run1():
return 1
async def run2():
return 2
def go(*fns):
return sum(await_only(fn()) for fn in fns)
class TestAsyncioCompat(fixtures.TestBase):
__requires__ = ("greenlet",)
@async_test
async def test_ok(self):
eq_(await greenlet_spawn(go, run1, run2), 3)
@async_test
async def test_async_error(self):
async def err():
raise ValueError("an error")
with expect_raises_message(ValueError, "an error"):
await greenlet_spawn(go, run1, err)
@async_test
async def test_propagate_cancelled(self):
"""test #6652"""
cleanup = []
async def async_meth_raise():
raise asyncio.CancelledError()
def sync_meth():
try:
await_only(async_meth_raise())
except:
cleanup.append(True)
raise
async def run_w_cancel():
await greenlet_spawn(sync_meth)
with expect_raises(asyncio.CancelledError, check_context=False):
await run_w_cancel()
assert cleanup
@async_test
async def test_sync_error(self):
def go():
await_only(run1())
raise ValueError("sync error")
with expect_raises_message(ValueError, "sync error"):
await greenlet_spawn(go)
def test_await_fallback_no_greenlet(self):
to_await = run1()
await_fallback(to_await)
@async_test
async def test_await_only_no_greenlet(self):
to_await = run1()
with expect_raises_message(
exc.MissingGreenlet,
"greenlet_spawn has not been called; "
r"can't call await_only\(\) here.",
):
await_only(to_await)
# existing awaitable is done
with expect_raises(RuntimeError):
await greenlet_spawn(await_fallback, to_await)
# no warning for a new one...
to_await = run1()
await greenlet_spawn(await_fallback, to_await)
@async_test
async def test_await_fallback_error(self):
to_await = run1()
await to_await
async def inner_await():
nonlocal to_await
to_await = run1()
await_fallback(to_await)
def go():
await_fallback(inner_await())
with expect_raises_message(
exc.MissingGreenlet,
"greenlet_spawn has not been called and asyncio event loop",
):
await greenlet_spawn(go)
with expect_raises(RuntimeError):
await to_await
@async_test
async def test_await_only_error(self):
to_await = run1()
await to_await
async def inner_await():
nonlocal to_await
to_await = run1()
await_only(to_await)
def go():
await_only(inner_await())
with expect_raises_message(
exc.InvalidRequestError,
"greenlet_spawn has not been called; "
r"can't call await_only\(\) here.",
):
await greenlet_spawn(go)
with expect_raises(RuntimeError):
await to_await
@async_test
async def test_require_await(self):
def run():
return 1 + 1
assert (await greenlet_spawn(run)) == 2
with expect_raises_message(
exc.AwaitRequired,
"The current operation required an async execution but none was",
):
await greenlet_spawn(run, _require_await=True)
class TestAsyncAdaptedQueue(fixtures.TestBase):
__requires__ = ("greenlet",)
def test_lazy_init(self):
run = [False]
def thread_go(q):
def go():
q.get(timeout=0.1)
with expect_raises(queue.Empty):
asyncio.run(greenlet_spawn(go))
run[0] = True
t = threading.Thread(
target=thread_go, args=[queue.AsyncAdaptedQueue()]
)
t.start()
t.join()
is_true(run[0])
@async_test
async def test_error_other_loop(self):
run = [False]
def thread_go(q):
def go():
eq_(q.get(block=False), 1)
q.get(timeout=0.1)
with expect_raises_message(
RuntimeError, ".* to a different .*loop"
):
asyncio.run(greenlet_spawn(go))
run[0] = True
q = queue.AsyncAdaptedQueue()
def prime():
with expect_raises(queue.Empty):
q.get(timeout=0.1)
await greenlet_spawn(prime)
q.put_nowait(1)
t = threading.Thread(target=thread_go, args=[q])
t.start()
t.join()
is_true(run[0])
class GracefulNoGreenletTest(fixtures.TestBase):
__requires__ = ("no_greenlet",)
def test_await_only_graceful(self):
async def async_fn():
pass
with expect_raises_message(
ValueError,
"the greenlet library is required to use this "
"function. No module named 'greenlet'",
):
await_only(async_fn())
|
