summaryrefslogtreecommitdiff
path: root/tests/timeout_test.py
blob: e33003f7346d6258166cec11e513092a045e8386 (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
import eventlet

import tests


DELAY = 0.01


class TestDirectRaise(tests.LimitedTestCase):
    def test_direct_raise_class(self):
        try:
            raise eventlet.Timeout
        except eventlet.Timeout as t:
            assert not t.pending, repr(t)

    def test_direct_raise_instance(self):
        tm = eventlet.Timeout()
        try:
            raise tm
        except eventlet.Timeout as t:
            assert tm is t, (tm, t)
            assert not t.pending, repr(t)

    def test_repr(self):
        # just verify these don't crash
        tm = eventlet.Timeout(1)
        eventlet.sleep(0)
        repr(tm)
        str(tm)
        tm.cancel()
        tm = eventlet.Timeout(None, RuntimeError)
        repr(tm)
        str(tm)
        tm = eventlet.Timeout(None, False)
        repr(tm)
        str(tm)


class TestWithTimeout(tests.LimitedTestCase):
    def test_with_timeout(self):
        self.assertRaises(eventlet.Timeout, eventlet.with_timeout, DELAY, eventlet.sleep, DELAY * 10)
        X = object()
        r = eventlet.with_timeout(DELAY, eventlet.sleep, DELAY * 10, timeout_value=X)
        assert r is X, (r, X)
        r = eventlet.with_timeout(DELAY * 10, eventlet.sleep, DELAY, timeout_value=X)
        assert r is None, r

    def test_with_outer_timer(self):
        def longer_timeout():
            # this should not catch the outer timeout's exception
            return eventlet.with_timeout(DELAY * 10, eventlet.sleep, DELAY * 20, timeout_value='b')
        self.assertRaises(
            eventlet.Timeout,
            eventlet.with_timeout,
            DELAY, longer_timeout)


def test_is_timeout_attribute():
    tests.check_is_timeout(eventlet.Timeout())