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
|
import asyncio
from unittest.mock import MagicMock
import pytest
import tornado
from nbclient.util import run_hook, run_sync
# mypy: disable-error-code="no-untyped-call,no-untyped-def"
@run_sync
async def some_async_function():
await asyncio.sleep(0.01)
return 42
def test_nested_asyncio_with_existing_ioloop():
async def _test():
assert some_async_function() == 42
return asyncio.get_running_loop()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
event_loop = loop.run_until_complete(_test())
assert event_loop is loop
def test_nested_asyncio_with_no_ioloop():
asyncio.set_event_loop(None)
assert some_async_function() == 42
def test_nested_asyncio_with_tornado():
# This tests if tornado accepts the pure-Python Futures, see
# https://github.com/tornadoweb/tornado/issues/2753
asyncio.set_event_loop(asyncio.new_event_loop())
ioloop = tornado.ioloop.IOLoop.current()
async def some_async_function():
future: asyncio.Future[None] = asyncio.ensure_future(asyncio.sleep(0.1))
# the asyncio module, check if tornado likes it:
ioloop.add_future(future, lambda f: f.result())
await future
return 42
def some_sync_function():
return run_sync(some_async_function)()
async def run():
# calling some_async_function directly should work
assert await some_async_function() == 42
assert some_sync_function() == 42
ioloop.run_sync(run)
@pytest.mark.asyncio
async def test_run_hook_sync():
some_sync_function = MagicMock()
await run_hook(some_sync_function)
assert some_sync_function.call_count == 1
@pytest.mark.asyncio
async def test_run_hook_async():
hook = MagicMock(return_value=some_async_function())
await run_hook(hook)
assert hook.call_count == 1
|
