← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/test/test_task.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2007-01-17 14:52:35 UTC
  • mfrom: (1.1.5 upstream) (2.1.2 etch)
  • Revision ID: james.westby@ubuntu.com-20070117145235-btmig6qfmqfen0om
Tags: 2.5.0-0ubuntu1
New upstream version, compatible with python2.5.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/test/test_task.py
1
1
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
2
2
# See LICENSE for details.
3
3
 
4
 
 
5
4
from twisted.trial import unittest
6
5
 
7
 
from twisted.internet import task, reactor, defer
 
6
from twisted.internet import interfaces, task, reactor, defer, error
 
7
 
 
8
# Be compatible with any jerks who used our private stuff
 
9
Clock = task.Clock
8
10
 
9
11
from twisted.python import failure
10
12
 
16
18
 
17
19
 
18
20
 
19
 
class FakeDelayedCall(object):
20
 
    def __init__(self, when, clock, what, a, kw):
21
 
        self.clock = clock
22
 
        self.when = when
23
 
        self.what = what
24
 
        self.a = a
25
 
        self.kw = kw
26
 
 
27
 
 
28
 
    def __call__(self):
29
 
        return self.what(*self.a, **self.kw)
30
 
 
31
 
 
32
 
    def __repr__(self):
33
 
        return "<FakeDelayedCall of %r>" % (self.what,)
34
 
 
35
 
 
36
 
    def cancel(self):
37
 
        self.clock.calls.remove((self.when, self))
38
 
 
39
 
 
40
 
 
41
 
class Clock(object):
42
 
    rightNow = 0.0
43
 
 
44
 
    def __init__(self):
45
 
        self.calls = []
46
 
 
47
 
    def seconds(self):
48
 
        return self.rightNow
49
 
 
50
 
    def callLater(self, when, what, *a, **kw):
51
 
        self.calls.append((self.seconds() + when, FakeDelayedCall(self.seconds() + when, self, what, a, kw)))
52
 
        return self.calls[-1][1]
53
 
 
54
 
    def adjust(self, amount):
55
 
        self.rightNow += amount
56
 
 
57
 
    def runUntilCurrent(self):
58
 
        while self.calls and self.calls[0][0] <= self.seconds():
59
 
            when, call = self.calls.pop(0)
60
 
            call()
61
 
 
62
 
    def pump(self, timings):
63
 
        timings = list(timings)
64
 
        timings.reverse()
65
 
        self.calls.sort()
66
 
        while timings:
67
 
            self.adjust(timings.pop())
68
 
            self.runUntilCurrent()
69
 
 
70
 
 
71
 
 
72
21
class TestException(Exception):
73
22
    pass
74
23
 
75
24
 
76
25
 
 
26
class ClockTestCase(unittest.TestCase):
 
27
    """
 
28
    Test the non-wallclock based clock implementation.
 
29
    """
 
30
    def testSeconds(self):
 
31
        """
 
32
        Test that the L{seconds} method of the fake clock returns fake time.
 
33
        """
 
34
        c = task.Clock()
 
35
        self.assertEquals(c.seconds(), 0)
 
36
 
 
37
 
 
38
    def testCallLater(self):
 
39
        """
 
40
        Test that calls can be scheduled for later with the fake clock and
 
41
        hands back an L{IDelayedCall}.
 
42
        """
 
43
        c = task.Clock()
 
44
        call = c.callLater(1, lambda a, b: None, 1, b=2)
 
45
        self.failUnless(interfaces.IDelayedCall.providedBy(call))
 
46
        self.assertEquals(call.getTime(), 1)
 
47
        self.failUnless(call.active())
 
48
 
 
49
 
 
50
    def testCallLaterCancelled(self):
 
51
        """
 
52
        Test that calls can be cancelled.
 
53
        """
 
54
        c = task.Clock()
 
55
        call = c.callLater(1, lambda a, b: None, 1, b=2)
 
56
        call.cancel()
 
57
        self.failIf(call.active())
 
58
 
 
59
 
 
60
    def testAdvance(self):
 
61
        """
 
62
        Test that advancing the clock will fire some calls.
 
63
        """
 
64
        events = []
 
65
        c = task.Clock()
 
66
        call = c.callLater(2, lambda: events.append(None))
 
67
        c.advance(1)
 
68
        self.assertEquals(events, [])
 
69
        c.advance(1)
 
70
        self.assertEquals(events, [None])
 
71
        self.failIf(call.active())
 
72
 
 
73
 
 
74
    def testAdvanceCancel(self):
 
75
        """
 
76
        Test attemping to cancel the call in a callback.
 
77
 
 
78
        AlreadyCalled should be raised, not for example a ValueError from
 
79
        removing the call from Clock.calls. This requires call.called to be
 
80
        set before the callback is called.
 
81
        """
 
82
        c = task.Clock()
 
83
        def cb():
 
84
            self.assertRaises(error.AlreadyCalled, call.cancel)
 
85
        call = c.callLater(1, cb)
 
86
        c.advance(1)
 
87
 
 
88
 
 
89
    def testCallLaterDelayed(self):
 
90
        """
 
91
        Test that calls can be delayed.
 
92
        """
 
93
        events = []
 
94
        c = task.Clock()
 
95
        call = c.callLater(1, lambda a, b: events.append((a, b)), 1, b=2)
 
96
        call.delay(1)
 
97
        self.assertEquals(call.getTime(), 2)
 
98
        c.advance(1.5)
 
99
        self.assertEquals(events, [])
 
100
        c.advance(1.0)
 
101
        self.assertEquals(events, [(1, 2)])
 
102
 
 
103
 
 
104
    def testCallLaterResetLater(self):
 
105
        """
 
106
        Test that calls can have their time reset to a later time.
 
107
        """
 
108
        events = []
 
109
        c = task.Clock()
 
110
        call = c.callLater(2, lambda a, b: events.append((a, b)), 1, b=2)
 
111
        c.advance(1)
 
112
        call.reset(3)
 
113
        self.assertEquals(call.getTime(), 4)
 
114
        c.advance(2)
 
115
        self.assertEquals(events, [])
 
116
        c.advance(1)
 
117
        self.assertEquals(events, [(1, 2)])
 
118
 
 
119
 
 
120
    def testCallLaterResetSooner(self):
 
121
        """
 
122
        Test that calls can have their time reset to an earlier time.
 
123
        """
 
124
        events = []
 
125
        c = task.Clock()
 
126
        call = c.callLater(4, lambda a, b: events.append((a, b)), 1, b=2)
 
127
        call.reset(3)
 
128
        self.assertEquals(call.getTime(), 3)
 
129
        c.advance(3)
 
130
        self.assertEquals(events, [(1, 2)])
 
131
 
 
132
 
 
133
 
77
134
class LoopTestCase(unittest.TestCase):
78
135
    def testBasicFunction(self):
79
136
        # Arrange to have time advanced enough so that our function is
82
139
        # happens before any time has elapsed.
83
140
        timings = [0.05, 0.1, 0.1]
84
141
 
85
 
        clock = Clock()
 
142
        clock = task.Clock()
86
143
 
87
144
        L = []
88
145
        def foo(a, b, c=None, d=None):
117
174
    def testDelayedStart(self):
118
175
        timings = [0.05, 0.1, 0.1]
119
176
 
120
 
        clock = Clock()
 
177
        clock = task.Clock()
121
178
 
122
179
        L = []
123
180
        lc = TestableLoopingCall(clock, L.append, None)
149
206
        def foo():
150
207
            ran.append(None)
151
208
 
152
 
        clock = Clock()
 
209
        clock = task.Clock()
153
210
        lc = TestableLoopingCall(clock, foo)
154
211
        d = lc.start(delay, now=False)
155
212
        lc.stop()
222
279
        # Tests if the callable isn't scheduled again before the returned
223
280
        # deferred has fired.
224
281
        timings = [0.2, 0.8]
225
 
        clock = Clock()
 
282
        clock = task.Clock()
226
283
 
227
284
        def foo():
228
285
            d = defer.Deferred()
238
295
    def testFailurePropagation(self):
239
296
        # Tests if the failure of the errback of the deferred returned by the
240
297
        # callable is propagated to the lc errback.
241
 
        # 
 
298
        #
242
299
        # To make sure this test does not hang trial when LoopingCall does not
243
300
        # wait for the callable's deferred, it also checks there are no
244
301
        # calls in the clock's callLater queue.
245
302
        timings = [0.3]
246
 
        clock = Clock()
 
303
        clock = task.Clock()
247
304
 
248
305
        def foo():
249
306
            d = defer.Deferred()