1
# Copyright (c) 2008 Divmod. See LICENSE for details.
4
Tests for L{epsilon.react}.
7
from twisted.internet.defer import Deferred, succeed, fail
8
from twisted.internet.task import Clock
9
from twisted.trial.unittest import TestCase
11
from epsilon.react import react
14
class _FakeReactor(object):
16
A fake implementation of L{IReactorCore}.
21
self.callLater = self._clock.callLater
22
self.seconds = self._clock.seconds
23
self.getDelayedCalls = self._clock.getDelayedCalls
24
self._whenRunning = []
25
self._shutdownTriggers = {'before': [], 'during': []}
28
def callWhenRunning(self, callable):
32
self._whenRunning.append(callable)
35
def addSystemEventTrigger(self, phase, event, callable, *args):
36
assert phase in ('before', 'during')
37
assert event == 'shutdown'
38
self._shutdownTriggers[phase].append((callable, args))
43
Call timed events until there are no more or the reactor is stopped.
45
@raise RuntimeError: When no timed events are left and the reactor is
49
whenRunning = self._whenRunning
50
self._whenRunning = None
51
for callable in whenRunning:
54
calls = self.getDelayedCalls()
56
raise RuntimeError("No DelayedCalls left")
57
self._clock.advance(calls[0].getTime() - self.seconds())
58
shutdownTriggers = self._shutdownTriggers
59
self._shutdownTriggers = None
60
for (trigger, args) in shutdownTriggers['before'] + shutdownTriggers['during']:
72
class ReactTests(TestCase):
74
Tests for L{epsilon.react.react}.
76
def test_runsUntilAsyncCallback(self):
78
L{react} runs the reactor until the L{Deferred} returned by the
79
function it is passed is called back, then stops it.
84
reactor.callLater(1, timePassed.append, True)
85
reactor.callLater(2, finished.callback, None)
89
self.assertEqual(timePassed, [True])
90
self.assertEqual(r.seconds(), 2)
93
def test_runsUntilSyncCallback(self):
95
L{react} returns quickly if the L{Deferred} returned by the function it
96
is passed has already been called back at the time it is returned.
102
self.assertEqual(r.seconds(), 0)
105
def test_runsUntilAsyncErrback(self):
107
L{react} runs the reactor until the L{Deferred} returned by the
108
function it is passed is errbacked, then it stops the reactor and
111
class ExpectedException(Exception):
115
finished = Deferred()
116
reactor.callLater(1, finished.errback, ExpectedException())
120
errors = self.flushLoggedErrors(ExpectedException)
121
self.assertEqual(len(errors), 1)
124
def test_runsUntilSyncErrback(self):
126
L{react} returns quickly if the L{Deferred} returned by the function it
127
is passed has already been errbacked at the time it is returned.
129
class ExpectedException(Exception):
133
return fail(ExpectedException())
136
self.assertEqual(r.seconds(), 0)
137
errors = self.flushLoggedErrors(ExpectedException)
138
self.assertEqual(len(errors), 1)
141
def test_singleStopCallback(self):
143
L{react} doesn't try to stop the reactor if the L{Deferred} the
144
function it is passed is called back after the reactor has already been
148
reactor.callLater(1, reactor.stop)
149
finished = Deferred()
150
reactor.addSystemEventTrigger(
151
'during', 'shutdown', finished.callback, None)
155
self.assertEqual(r.seconds(), 1)
158
def test_singleStopErrback(self):
160
L{react} doesn't try to stop the reactor if the L{Deferred} the
161
function it is passed is errbacked after the reactor has already been
164
class ExpectedException(Exception):
168
reactor.callLater(1, reactor.stop)
169
finished = Deferred()
170
reactor.addSystemEventTrigger(
171
'during', 'shutdown', finished.errback, ExpectedException())
175
self.assertEqual(r.seconds(), 1)
176
errors = self.flushLoggedErrors(ExpectedException)
177
self.assertEqual(len(errors), 1)
180
def test_arguments(self):
182
L{react} passes the elements of the list it is passed as positional
183
arguments to the function it is passed.
186
def main(reactor, x, y, z):
187
args.extend((x, y, z))
190
react(r, main, [1, 2, 3])
191
self.assertEqual(args, [1, 2, 3])
added
removed
Epsilon/epsilon/test/test_react.py
