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
|
import asyncio
import functools
import wrapt
def futurized(o):
''' Makes the given object to be awaitable.
:param any o: Object to wrap
:return: awaitable that resolves to provided object
:rtype: asyncio.Future
Anything passed to :code:`futurized` is wrapped in :code:`asyncio.Future`.
This makes it awaitable (can be run with :code:`await` or :code:`yield from`) as
a result of await it returns the original object.
If provided object is a Exception (or its sublcass) then the `Future` will raise it on await.
.. code-block:: python
fut = aiounittest.futurized('SOME TEXT')
ret = await fut
print(ret) # prints SOME TEXT
fut = aiounittest.futurized(Exception('Dummy error'))
ret = await fut # will raise the exception "dummy error"
The main goal is to use it with :code:`unittest.mock.Mock` (or :code:`MagicMock`) to
be able to mock awaitable functions (coroutines).
Consider the below code
.. code-block:: python
from asyncio import sleep
async def add(x, y):
await sleep(666)
return x + y
You rather don't want to wait 666 seconds, you've gotta mock that.
.. code-block:: python
from aiounittest import futurized, AsyncTestCase
from unittest.mock import Mock, patch
import dummy_math
class MyAddTest(AsyncTestCase):
async def test_add(self):
mock_sleep = Mock(return_value=futurized('whatever'))
patch('dummy_math.sleep', mock_sleep).start()
ret = await dummy_math.add(5, 6)
self.assertEqual(ret, 11)
mock_sleep.assert_called_once_with(666)
async def test_fail(self):
mock_sleep = Mock(return_value=futurized(Exception('whatever')))
patch('dummy_math.sleep', mock_sleep).start()
with self.assertRaises(Exception) as e:
await dummy_math.add(5, 6)
mock_sleep.assert_called_once_with(666)
'''
f = asyncio.Future()
if isinstance(o, Exception):
f.set_exception(o)
else:
f.set_result(o)
return f
def run_sync(func=None, loop=None):
''' Runs synchonously given function (coroutine)
:param callable func: function to run (mostly coroutine)
:param ioloop loop: event loop to use to run `func`
:type loop: event loop of None
By default the brand new event loop will be created (old closed). After completion, the loop will be closed and then recreated, set as default,
leaving asyncio clean.
**Note**: :code:`aiounittest.async_test` is an alias of :code:`aiounittest.helpers.run_sync`
Function can be used like a `pytest.mark.asyncio` (implementation differs),
but it's compatible with :code:`unittest.TestCase` class.
.. code-block:: python
import asyncio
import unittest
from aiounittest import async_test
async def add(x, y):
await asyncio.sleep(0.1)
return x + y
class MyAsyncTestDecorator(unittest.TestCase):
@async_test
async def test_async_add(self):
ret = await add(5, 6)
self.assertEqual(ret, 11)
.. note::
If the loop is provided, it won't be closed. It's up to you.
This function is also used internally by :code:`aiounittest.AsyncTestCase` to run coroutines.
'''
def get_brand_new_default_event_loop():
try:
old_loop = asyncio.get_event_loop()
if not old_loop.is_closed():
old_loop.close()
except RuntimeError:
# no default event loop, ignore exception
pass
_loop = asyncio.new_event_loop()
asyncio.set_event_loop(_loop)
return _loop
@wrapt.decorator
def decorator(wrapped, instance, args, kwargs):
nonlocal loop
use_default_event_loop = loop is None
if use_default_event_loop:
loop = get_brand_new_default_event_loop()
try:
ret = wrapped(*args, **kwargs)
future = asyncio.ensure_future(ret, loop=loop)
return loop.run_until_complete(future)
finally:
if use_default_event_loop:
# clean up
loop.close()
del loop
# again set a new (unstopped) event loop
get_brand_new_default_event_loop()
if func is None:
return decorator
else:
return decorator(func)
async_test = run_sync
|
