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
|
"""
Test toro.Lock.
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 make_callback, assert_raises, ContextManagerTestsMixin
# Adapted from Gevent's lock_tests.py.
class LockTests(AsyncTestCase):
def test_acquire_release(self):
lock = toro.Lock()
self.assertFalse(lock.locked())
self.assertTrue(lock.acquire())
self.assertTrue(lock.locked())
lock.release()
self.assertFalse(lock.locked())
@gen_test
def test_acquire_contended(self):
lock = toro.Lock()
self.assertTrue(lock.acquire())
N = 5
@gen.coroutine
def f():
yield lock.acquire()
lock.release()
futures = [f() for _ in range(N)]
lock.release()
yield futures
@gen_test
def test_reacquire(self):
# Lock needs to be released before re-acquiring.
lock = toro.Lock()
phase = []
@gen.coroutine
def f():
yield lock.acquire()
self.assertTrue(lock.locked())
phase.append(None)
yield lock.acquire()
self.assertTrue(lock.locked())
phase.append(None)
future = f()
while len(phase) == 0:
yield gen.Task(self.io_loop.add_callback)
self.assertEqual(len(phase), 1)
lock.release()
yield future
self.assertEqual(len(phase), 2)
# Not adapted from Gevent's tests, written just for Toro
class LockTests2(AsyncTestCase):
def test_str(self):
lock = toro.Lock()
# No errors in various states
str(lock)
lock.acquire()
str(lock)
@gen_test
def test_acquire_timeout(self):
lock = toro.Lock()
self.assertTrue(lock.acquire())
self.assertTrue(lock.locked())
st = time.time()
with assert_raises(toro.Timeout):
yield lock.acquire(deadline=timedelta(seconds=0.1))
duration = time.time() - st
self.assertAlmostEqual(0.1, duration, places=1)
self.assertTrue(lock.locked())
@gen_test
def test_acquire_callback(self):
lock = toro.Lock()
history = []
lock.acquire().add_done_callback(make_callback('acquire1', history))
lock.acquire().add_done_callback(make_callback('acquire2', history))
lock.release()
history.append('release')
self.assertEqual(['acquire1', 'acquire2', 'release'], history)
def test_multi_release(self):
lock = toro.Lock()
lock.acquire()
lock.release()
self.assertRaises(RuntimeError, lock.release)
class LockContextManagerTest(ContextManagerTestsMixin, AsyncTestCase):
toro_class = toro.Lock
|
