blob: 2c9caae6e7763ebb0005dcf18e71a643e75dc282 (
plain)
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
|
import eventlet
import tests
threading = eventlet.patcher.original('threading')
try:
import asyncio
except ImportError:
import trollius as asyncio
try:
get_ident = threading.get_ident # Python 3
except AttributeError:
get_ident = threading._get_ident # Python 2
class ThreadTests(tests.TestCase):
def test_ident(self):
result = {'ident': None}
def work():
result['ident'] = get_ident()
fut = self.loop.run_in_executor(None, work)
self.loop.run_until_complete(fut)
# ensure that work() was executed in a different thread
work_ident = result['ident']
self.assertIsNotNone(work_ident)
self.assertNotEqual(work_ident, get_ident())
def test_run_twice(self):
result = []
def work():
result.append("run")
fut = self.loop.run_in_executor(None, work)
self.loop.run_until_complete(fut)
self.assertEqual(result, ["run"])
# ensure that run_in_executor() can be called twice
fut = self.loop.run_in_executor(None, work)
self.loop.run_until_complete(fut)
self.assertEqual(result, ["run", "run"])
def test_policy(self):
result = {'loop': 'not set'} # sentinel, different than None
def work():
try:
result['loop'] = asyncio.get_event_loop()
except AssertionError as exc:
result['loop'] = exc
# get_event_loop() must return None in a different thread
fut = self.loop.run_in_executor(None, work)
self.loop.run_until_complete(fut)
self.assertIsInstance(result['loop'], AssertionError)
if __name__ == '__main__':
import unittest
unittest.main()
|
