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
|
# Copyright 2014 MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test Motor's asyncio test utilities."""
import asyncio
import concurrent.futures
import contextlib
import io
import os
import unittest
from test.asyncio_tests import AsyncIOTestCase, asyncio_test
def run_test_case(case, suppress_output=True):
suite = unittest.defaultTestLoader.loadTestsFromTestCase(case)
stream = io.StringIO() if suppress_output else None
runner = unittest.TextTestRunner(stream=stream)
return runner.run(suite)
@contextlib.contextmanager
def set_environ(name, value):
old_value = os.environ.get(name)
os.environ[name] = value
try:
yield
finally:
if old_value is None:
del os.environ[name]
else:
os.environ[name] = old_value
class TestAsyncIOTests(unittest.TestCase):
def test_basic(self):
class Test(AsyncIOTestCase):
@asyncio_test
async def test(self):
pass
result = run_test_case(Test)
self.assertEqual(1, result.testsRun)
self.assertEqual(0, len(result.errors))
def test_decorator_with_no_args(self):
class TestPasses(AsyncIOTestCase):
@asyncio_test
async def test_decorated_with_no_args(self):
pass
result = run_test_case(TestPasses)
self.assertEqual(0, len(result.errors))
class TestFails(AsyncIOTestCase):
@asyncio_test()
async def test_decorated_with_no_args(self):
self.fail()
result = run_test_case(TestFails)
self.assertEqual(1, len(result.failures))
def test_timeout_passed_as_positional(self):
with self.assertRaises(TypeError):
class _(AsyncIOTestCase):
# Should be "timeout=10".
@asyncio_test(10)
def test_decorated_with_no_args(self):
pass
def test_timeout(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(None)
self.addCleanup(self.loop.close)
self.addCleanup(setattr, self, "loop", None)
class Test(AsyncIOTestCase):
@asyncio_test(timeout=0.01)
async def test_that_is_too_slow(self):
await self.middle()
async def middle(self):
await self.inner()
async def inner(self):
await asyncio.sleep(1)
with set_environ("ASYNC_TEST_TIMEOUT", "0"):
result = run_test_case(Test)
self.assertEqual(1, len(result.errors))
case, text = result.errors[0]
self.assertTrue("TimeoutError" in text)
def test_timeout_environment_variable(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(None)
self.addCleanup(self.loop.close)
self.addCleanup(setattr, self, "loop", None)
@asyncio_test
async def default_timeout(self):
await asyncio.sleep(0.1)
with set_environ("ASYNC_TEST_TIMEOUT", "0.2"):
# No error, sleeps for 0.1 seconds and the timeout is 0.2 seconds.
default_timeout(self)
@asyncio_test(timeout=0.1)
async def custom_timeout(self):
await asyncio.sleep(0.2)
with set_environ("ASYNC_TEST_TIMEOUT", "0"):
# No error, default timeout of 5 seconds overrides '0'.
default_timeout(self)
with set_environ("ASYNC_TEST_TIMEOUT", "0"):
if hasattr(asyncio, "exceptions"):
with self.assertRaises(asyncio.exceptions.TimeoutError):
custom_timeout(self)
else:
with self.assertRaises(concurrent.futures.TimeoutError):
custom_timeout(self)
with set_environ("ASYNC_TEST_TIMEOUT", "1"):
# No error, 1-second timeout from environment overrides custom
# timeout of 0.1 seconds.
custom_timeout(self)
def test_failure(self):
class Test(AsyncIOTestCase):
@asyncio_test
async def test_that_fails(self):
await self.middle()
async def middle(self):
await self.inner()
async def inner(self):
assert False, "expected error"
result = run_test_case(Test)
self.assertEqual(1, len(result.failures))
case, text = result.failures[0]
self.assertFalse("CancelledError" in text)
self.assertTrue("AssertionError" in text)
self.assertTrue("expected error" in text)
# The traceback shows where the coroutine raised.
self.assertTrue("test_that_fails" in text)
self.assertTrue("middle" in text)
self.assertTrue("inner" in text)
def test_undecorated(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(None)
self.addCleanup(self.loop.close)
self.addCleanup(setattr, self, "loop", None)
class Test(AsyncIOTestCase):
async def test_that_should_be_decorated(self):
await asyncio.sleep(0.01)
result = run_test_case(Test)
self.assertEqual(1, len(result.errors))
case, text = result.errors[0]
self.assertFalse("CancelledError" in text)
self.assertTrue("TypeError" in text)
self.assertTrue("should be decorated with @asyncio_test" in text)
def test_other_return(self):
class Test(AsyncIOTestCase):
def test_other_return(self):
return 42
result = run_test_case(Test)
self.assertEqual(len(result.errors), 1)
case, text = result.errors[0]
self.assertIn("Return value from test method ignored", text)
if __name__ == "__main__":
unittest.main()
|
