• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import asyncio
2import unittest
3from test import support
4
5
6class MyException(Exception):
7    pass
8
9
10def tearDownModule():
11    asyncio.set_event_loop_policy(None)
12
13
14class TestAsyncCase(unittest.TestCase):
15    maxDiff = None
16
17    def tearDown(self):
18        # Ensure that IsolatedAsyncioTestCase instances are destroyed before
19        # starting a new event loop
20        support.gc_collect()
21
22    def test_full_cycle(self):
23        class Test(unittest.IsolatedAsyncioTestCase):
24            def setUp(self):
25                self.assertEqual(events, [])
26                events.append('setUp')
27
28            async def asyncSetUp(self):
29                self.assertEqual(events, ['setUp'])
30                events.append('asyncSetUp')
31                self.addAsyncCleanup(self.on_cleanup1)
32
33            async def test_func(self):
34                self.assertEqual(events, ['setUp',
35                                          'asyncSetUp'])
36                events.append('test')
37                self.addAsyncCleanup(self.on_cleanup2)
38
39            async def asyncTearDown(self):
40                self.assertEqual(events, ['setUp',
41                                          'asyncSetUp',
42                                          'test'])
43                events.append('asyncTearDown')
44
45            def tearDown(self):
46                self.assertEqual(events, ['setUp',
47                                          'asyncSetUp',
48                                          'test',
49                                          'asyncTearDown'])
50                events.append('tearDown')
51
52            async def on_cleanup1(self):
53                self.assertEqual(events, ['setUp',
54                                          'asyncSetUp',
55                                          'test',
56                                          'asyncTearDown',
57                                          'tearDown',
58                                          'cleanup2'])
59                events.append('cleanup1')
60
61            async def on_cleanup2(self):
62                self.assertEqual(events, ['setUp',
63                                          'asyncSetUp',
64                                          'test',
65                                          'asyncTearDown',
66                                          'tearDown'])
67                events.append('cleanup2')
68
69        events = []
70        test = Test("test_func")
71        result = test.run()
72        self.assertEqual(result.errors, [])
73        self.assertEqual(result.failures, [])
74        expected = ['setUp', 'asyncSetUp', 'test',
75                    'asyncTearDown', 'tearDown', 'cleanup2', 'cleanup1']
76        self.assertEqual(events, expected)
77
78        events = []
79        test = Test("test_func")
80        test.debug()
81        self.assertEqual(events, expected)
82        test.doCleanups()
83        self.assertEqual(events, expected)
84
85    def test_exception_in_setup(self):
86        class Test(unittest.IsolatedAsyncioTestCase):
87            async def asyncSetUp(self):
88                events.append('asyncSetUp')
89                self.addAsyncCleanup(self.on_cleanup)
90                raise MyException()
91
92            async def test_func(self):
93                events.append('test')
94
95            async def asyncTearDown(self):
96                events.append('asyncTearDown')
97
98            async def on_cleanup(self):
99                events.append('cleanup')
100
101
102        events = []
103        test = Test("test_func")
104        result = test.run()
105        self.assertEqual(events, ['asyncSetUp', 'cleanup'])
106        self.assertIs(result.errors[0][0], test)
107        self.assertIn('MyException', result.errors[0][1])
108
109        events = []
110        test = Test("test_func")
111        try:
112            test.debug()
113        except MyException:
114            pass
115        else:
116            self.fail('Expected a MyException exception')
117        self.assertEqual(events, ['asyncSetUp'])
118        test.doCleanups()
119        self.assertEqual(events, ['asyncSetUp', 'cleanup'])
120
121    def test_exception_in_test(self):
122        class Test(unittest.IsolatedAsyncioTestCase):
123            async def asyncSetUp(self):
124                events.append('asyncSetUp')
125
126            async def test_func(self):
127                events.append('test')
128                self.addAsyncCleanup(self.on_cleanup)
129                raise MyException()
130
131            async def asyncTearDown(self):
132                events.append('asyncTearDown')
133
134            async def on_cleanup(self):
135                events.append('cleanup')
136
137        events = []
138        test = Test("test_func")
139        result = test.run()
140        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
141        self.assertIs(result.errors[0][0], test)
142        self.assertIn('MyException', result.errors[0][1])
143
144        events = []
145        test = Test("test_func")
146        try:
147            test.debug()
148        except MyException:
149            pass
150        else:
151            self.fail('Expected a MyException exception')
152        self.assertEqual(events, ['asyncSetUp', 'test'])
153        test.doCleanups()
154        self.assertEqual(events, ['asyncSetUp', 'test', 'cleanup'])
155
156    def test_exception_in_tear_down(self):
157        class Test(unittest.IsolatedAsyncioTestCase):
158            async def asyncSetUp(self):
159                events.append('asyncSetUp')
160
161            async def test_func(self):
162                events.append('test')
163                self.addAsyncCleanup(self.on_cleanup)
164
165            async def asyncTearDown(self):
166                events.append('asyncTearDown')
167                raise MyException()
168
169            async def on_cleanup(self):
170                events.append('cleanup')
171
172        events = []
173        test = Test("test_func")
174        result = test.run()
175        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
176        self.assertIs(result.errors[0][0], test)
177        self.assertIn('MyException', result.errors[0][1])
178
179        events = []
180        test = Test("test_func")
181        try:
182            test.debug()
183        except MyException:
184            pass
185        else:
186            self.fail('Expected a MyException exception')
187        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown'])
188        test.doCleanups()
189        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
190
191    def test_exception_in_tear_clean_up(self):
192        class Test(unittest.IsolatedAsyncioTestCase):
193            async def asyncSetUp(self):
194                events.append('asyncSetUp')
195
196            async def test_func(self):
197                events.append('test')
198                self.addAsyncCleanup(self.on_cleanup1)
199                self.addAsyncCleanup(self.on_cleanup2)
200
201            async def asyncTearDown(self):
202                events.append('asyncTearDown')
203
204            async def on_cleanup1(self):
205                events.append('cleanup1')
206                raise MyException('some error')
207
208            async def on_cleanup2(self):
209                events.append('cleanup2')
210                raise MyException('other error')
211
212        events = []
213        test = Test("test_func")
214        result = test.run()
215        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup2', 'cleanup1'])
216        self.assertIs(result.errors[0][0], test)
217        self.assertIn('MyException: other error', result.errors[0][1])
218        self.assertIn('MyException: some error', result.errors[1][1])
219
220        events = []
221        test = Test("test_func")
222        try:
223            test.debug()
224        except MyException:
225            pass
226        else:
227            self.fail('Expected a MyException exception')
228        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup2'])
229        test.doCleanups()
230        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup2', 'cleanup1'])
231
232    def test_cleanups_interleave_order(self):
233        events = []
234
235        class Test(unittest.IsolatedAsyncioTestCase):
236            async def test_func(self):
237                self.addAsyncCleanup(self.on_sync_cleanup, 1)
238                self.addAsyncCleanup(self.on_async_cleanup, 2)
239                self.addAsyncCleanup(self.on_sync_cleanup, 3)
240                self.addAsyncCleanup(self.on_async_cleanup, 4)
241
242            async def on_sync_cleanup(self, val):
243                events.append(f'sync_cleanup {val}')
244
245            async def on_async_cleanup(self, val):
246                events.append(f'async_cleanup {val}')
247
248        test = Test("test_func")
249        test.run()
250        self.assertEqual(events, ['async_cleanup 4',
251                                  'sync_cleanup 3',
252                                  'async_cleanup 2',
253                                  'sync_cleanup 1'])
254
255    def test_base_exception_from_async_method(self):
256        events = []
257        class Test(unittest.IsolatedAsyncioTestCase):
258            async def test_base(self):
259                events.append("test_base")
260                raise BaseException()
261                events.append("not it")
262
263            async def test_no_err(self):
264                events.append("test_no_err")
265
266            async def test_cancel(self):
267                raise asyncio.CancelledError()
268
269        test = Test("test_base")
270        output = test.run()
271        self.assertFalse(output.wasSuccessful())
272
273        test = Test("test_no_err")
274        test.run()
275        self.assertEqual(events, ['test_base', 'test_no_err'])
276
277        test = Test("test_cancel")
278        output = test.run()
279        self.assertFalse(output.wasSuccessful())
280
281    def test_cancellation_hanging_tasks(self):
282        cancelled = False
283        class Test(unittest.IsolatedAsyncioTestCase):
284            async def test_leaking_task(self):
285                async def coro():
286                    nonlocal cancelled
287                    try:
288                        await asyncio.sleep(1)
289                    except asyncio.CancelledError:
290                        cancelled = True
291                        raise
292
293                # Leave this running in the background
294                asyncio.create_task(coro())
295
296        test = Test("test_leaking_task")
297        output = test.run()
298        self.assertTrue(cancelled)
299
300    def test_debug_cleanup_same_loop(self):
301        class Test(unittest.IsolatedAsyncioTestCase):
302            async def asyncSetUp(self):
303                async def coro():
304                    await asyncio.sleep(0)
305                fut = asyncio.ensure_future(coro())
306                self.addAsyncCleanup(self.cleanup, fut)
307                events.append('asyncSetUp')
308
309            async def test_func(self):
310                events.append('test')
311                raise MyException()
312
313            async def asyncTearDown(self):
314                events.append('asyncTearDown')
315
316            async def cleanup(self, fut):
317                try:
318                    # Raises an exception if in different loop
319                    await asyncio.wait([fut])
320                    events.append('cleanup')
321                except:
322                    import traceback
323                    traceback.print_exc()
324                    raise
325
326        events = []
327        test = Test("test_func")
328        result = test.run()
329        self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
330        self.assertIn('MyException', result.errors[0][1])
331
332        events = []
333        test = Test("test_func")
334        try:
335            test.debug()
336        except MyException:
337            pass
338        else:
339            self.fail('Expected a MyException exception')
340        self.assertEqual(events, ['asyncSetUp', 'test'])
341        test.doCleanups()
342        self.assertEqual(events, ['asyncSetUp', 'test', 'cleanup'])
343
344
345if __name__ == "__main__":
346    unittest.main()
347