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
|
"""
Test toro.Event.
Adapted from Gevent's lock_tests.py.
"""
from datetime import timedelta
import time
from tornado import gen
from tornado.testing import gen_test, AsyncTestCase
import toro
from test import assert_raises
class TestEvent(AsyncTestCase):
def test_str(self):
event = toro.Event()
self.assertTrue('clear' in str(event))
self.assertFalse('set' in str(event))
event.set()
self.assertFalse('clear' in str(event))
self.assertTrue('set' in str(event))
@gen.coroutine
def _test_event(self, n):
e = toro.Event()
futures = []
for i in range(n):
futures.append(e.wait())
e.set()
e.clear()
yield futures
@gen_test
def test_event_1(self):
yield self._test_event(1)
@gen_test
def test_event_100(self):
yield self._test_event(100)
@gen_test
def test_event_timeout(self):
e = toro.Event()
st = time.time()
with assert_raises(toro.Timeout):
yield e.wait(deadline=timedelta(seconds=0.1))
duration = time.time() - st
self.assertAlmostEqual(0.1, duration, places=1)
# After a timed-out waiter, normal operation works
st = time.time()
self.io_loop.add_timeout(st + 0.1, e.set)
result = yield e.wait(deadline=timedelta(seconds=1))
duration = time.time() - st
self.assertAlmostEqual(0.1, duration, places=1)
self.assertEqual(None, result)
@gen_test
def test_event_nowait(self):
e = toro.Event()
e.set()
self.assertEqual(True, e.is_set())
yield e.wait()
|
