/
test__timeout.py
159 lines (134 loc) · 5.06 KB
/
test__timeout.py
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
import gevent.testing as greentest
import gevent
from gevent.hub import get_hub
import sys
SHOULD_EXPIRE = 0.01
if not greentest.RUNNING_ON_CI:
SHOULD_NOT_EXPIRE = SHOULD_EXPIRE * 2.0
else:
SHOULD_NOT_EXPIRE = SHOULD_EXPIRE * 20.0
class TestDirectRaise(greentest.TestCase):
switch_expected = False
def test_direct_raise_class(self):
try:
raise gevent.Timeout
except gevent.Timeout as t:
assert not t.pending, repr(t)
def test_direct_raise_instance(self):
timeout = gevent.Timeout()
try:
raise timeout
except gevent.Timeout as t:
assert timeout is t, (timeout, t)
assert not t.pending, repr(t)
class Test(greentest.TestCase):
def _test(self, timeout, close):
try:
get_hub().switch()
self.fail('Must raise Timeout')
except gevent.Timeout as ex:
if ex is not timeout:
raise
if close:
ex.close()
return ex
def _check_expires(self, timeout):
timeout.start()
self._test(timeout, False)
# Restart
timeout.start()
return self._test(timeout, True)
def test_expires(self):
timeout = gevent.Timeout(SHOULD_EXPIRE)
self._check_expires(timeout)
def test_expires_false(self):
# A False exception value only matters to a
# context manager
timeout = gevent.Timeout(SHOULD_EXPIRE, False)
self._check_expires(timeout)
def test_expires_str(self):
# str values are accepted but not documented; they change
# the message
timeout = gevent.Timeout(SHOULD_EXPIRE, 'XXX')
ex = self._check_expires(timeout)
self.assertTrue(str(ex).endswith('XXX'))
def assert_type_err(self, ex):
# PyPy3 uses 'exceptions must derive', everyone else uses "exceptions must be"
self.assertTrue("exceptions must be" in str(ex) or "exceptions must derive" in str(ex), str(ex))
def test_expires_non_exception(self):
timeout = gevent.Timeout(SHOULD_EXPIRE, object())
timeout.start()
try:
get_hub().switch()
self.fail("Most raise TypeError")
except TypeError as ex:
self.assert_type_err(ex)
timeout.close()
class OldStyle:
pass
timeout = gevent.Timeout(SHOULD_EXPIRE, OldStyle) # Type
timeout.start()
try:
get_hub().switch()
self.fail("Must raise OldStyle")
except TypeError as ex:
self.assertTrue(greentest.PY3, "Py3 raises a TypeError for non-BaseExceptions")
self.assert_type_err(ex)
except:
self.assertTrue(greentest.PY2, "Old style classes can only be raised on Py2")
t = sys.exc_info()[0]
self.assertEqual(t, OldStyle)
timeout.close()
timeout = gevent.Timeout(SHOULD_EXPIRE, OldStyle()) # instance
timeout.start()
try:
get_hub().switch()
self.fail("Must raise OldStyle")
except TypeError as ex:
self.assertTrue(greentest.PY3, "Py3 raises a TypeError for non-BaseExceptions")
self.assert_type_err(ex)
except:
self.assertTrue(greentest.PY2, "Old style classes can only be raised on Py2")
t = sys.exc_info()[0]
self.assertEqual(t, OldStyle)
timeout.close()
def _check_context_manager_expires(self, timeout, raises=True):
try:
with timeout:
get_hub().switch()
except gevent.Timeout as ex:
if ex is not timeout:
raise
return ex
if raises:
self.fail("Must raise Timeout")
def test_context_manager(self):
timeout = gevent.Timeout(SHOULD_EXPIRE)
self._check_context_manager_expires(timeout)
def test_context_manager_false(self):
# Suppress the exception
timeout = gevent.Timeout(SHOULD_EXPIRE, False)
self._check_context_manager_expires(timeout, raises=False)
self.assertTrue(str(timeout).endswith('(silent)'), str(timeout))
def test_context_manager_str(self):
timeout = gevent.Timeout(SHOULD_EXPIRE, 'XXX')
ex = self._check_context_manager_expires(timeout)
self.assertTrue(str(ex).endswith('XXX'), str(ex))
def test_cancel(self):
timeout = gevent.Timeout(SHOULD_EXPIRE)
timeout.start()
timeout.cancel()
gevent.sleep(SHOULD_NOT_EXPIRE)
self.assertFalse(timeout.pending, timeout)
timeout.close()
@greentest.ignores_leakcheck
def test_with_timeout(self):
with self.assertRaises(gevent.Timeout):
gevent.with_timeout(SHOULD_EXPIRE, gevent.sleep, SHOULD_NOT_EXPIRE)
X = object()
r = gevent.with_timeout(SHOULD_EXPIRE, gevent.sleep, SHOULD_NOT_EXPIRE, timeout_value=X)
self.assertIs(r, X)
r = gevent.with_timeout(SHOULD_NOT_EXPIRE, gevent.sleep, SHOULD_EXPIRE, timeout_value=X)
self.assertIsNone(r)
if __name__ == '__main__':
greentest.main()