summaryrefslogtreecommitdiff
path: root/tests/timeout_test.py
diff options
context:
space:
mode:
authorRyan Williams <rdw@lindenlab.com>2010-02-16 13:13:21 -0800
committerRyan Williams <rdw@lindenlab.com>2010-02-16 13:13:21 -0800
commit7664905e5571de23bdff2fbda9f4d0562fa9a436 (patch)
tree416c7282ab09829ec73da699cbd08a25d05a1621 /tests/timeout_test.py
parentdf20576a6931278de34394c333423efd7d7b2e96 (diff)
downloadeventlet-7664905e5571de23bdff2fbda9f4d0562fa9a436.tar.gz
Adding timeout files, derr.
Diffstat (limited to 'tests/timeout_test.py')
-rw-r--r--tests/timeout_test.py54
1 files changed, 54 insertions, 0 deletions
diff --git a/tests/timeout_test.py b/tests/timeout_test.py
new file mode 100644
index 0000000..006ece7
--- /dev/null
+++ b/tests/timeout_test.py
@@ -0,0 +1,54 @@
+from tests import LimitedTestCase
+from eventlet import timeout
+from eventlet import greenthread
+DELAY = 0.01
+
+class TestDirectRaise(LimitedTestCase):
+ def test_direct_raise_class(self):
+ try:
+ raise timeout.Timeout
+ except timeout.Timeout, t:
+ assert not t.pending, repr(t)
+
+ def test_direct_raise_instance(self):
+ tm = timeout.Timeout()
+ try:
+ raise tm
+ except timeout.Timeout, t:
+ assert tm is t, (tm, t)
+ assert not t.pending, repr(t)
+
+ def test_repr(self):
+ # just verify these don't crash
+ tm = timeout.Timeout(1)
+ greenthread.sleep(0)
+ repr(tm)
+ str(tm)
+ tm.cancel()
+ tm = timeout.Timeout(None, RuntimeError)
+ repr(tm)
+ str(tm)
+ tm = timeout.Timeout(None, False)
+ repr(tm)
+ str(tm)
+
+class TestWithTimeout(LimitedTestCase):
+ def test_with_timeout(self):
+ self.assertRaises(timeout.Timeout, timeout.with_timeout, DELAY, greenthread.sleep, DELAY*10)
+ X = object()
+ r = timeout.with_timeout(DELAY, greenthread.sleep, DELAY*10, timeout_value=X)
+ self.assert_(r is X, (r, X))
+ r = timeout.with_timeout(DELAY*10, greenthread.sleep,
+ DELAY, timeout_value=X)
+ self.assert_(r is None, r)
+
+
+ def test_with_outer_timer(self):
+ def longer_timeout():
+ # this should not catch the outer timeout's exception
+ return timeout.with_timeout(DELAY * 10,
+ greenthread.sleep, DELAY * 20,
+ timeout_value='b')
+ self.assertRaises(timeout.Timeout,
+ timeout.with_timeout, DELAY, longer_timeout)
+ \ No newline at end of file