1
from __future__ import generators
2
from twisted.trial import unittest
3
from twisted.internet import defer, threads, reactor
6
class DeferredSetUpOK(unittest.TestCase):
8
d = defer.succeed('value')
9
d.addCallback(self._cb_setUpCalled)
12
def _cb_setUpCalled(self, ignored):
13
self._setUpCalled = True
16
self.failUnless(self._setUpCalled)
19
class DeferredSetUpFail(unittest.TestCase):
23
return defer.fail(unittest.FailTest('i fail'))
26
DeferredSetUpFail.testCalled = True
27
self.fail("I should not get called")
30
class DeferredSetUpCallbackFail(unittest.TestCase):
34
d = defer.succeed('value')
35
d.addCallback(self._cb_setUpCalled)
38
def _cb_setUpCalled(self, ignored):
39
self.fail('deliberate failure')
42
DeferredSetUpCallbackFail.testCalled = True
45
class DeferredSetUpError(unittest.TestCase):
49
return defer.fail(RuntimeError('deliberate error'))
52
DeferredSetUpError.testCalled = True
55
class DeferredSetUpNeverFire(unittest.TestCase):
59
return defer.Deferred()
62
DeferredSetUpNeverFire.testCalled = True
65
class DeferredSetUpSkip(unittest.TestCase):
69
d = defer.succeed('value')
70
d.addCallback(self._cb1)
73
def _cb1(self, ignored):
74
raise unittest.SkipTest("skip me")
77
DeferredSetUpSkip.testCalled = True
80
class DeferredTests(unittest.TestCase):
83
def _cb_fail(self, reason):
86
def _cb_error(self, reason):
87
raise RuntimeError(reason)
89
def _cb_skip(self, reason):
90
raise unittest.SkipTest(reason)
92
def _touchClass(self, ignored):
93
self.__class__.touched = True
96
self.__class__.touched = False
99
return defer.succeed('success')
101
def test_passGenerated(self):
102
self._touchClass(None)
104
test_passGenerated = defer.deferredGenerator(test_passGenerated)
107
return defer.fail(self.failureException('I fail'))
109
def test_failureInCallback(self):
110
d = defer.succeed('fail')
111
d.addCallback(self._cb_fail)
114
def test_errorInCallback(self):
115
d = defer.succeed('error')
116
d.addCallback(self._cb_error)
120
d = defer.succeed('skip')
121
d.addCallback(self._cb_skip)
122
d.addCallback(self._touchClass)
125
def test_thread(self):
126
return threads.deferToThread(lambda : None)
128
def test_expectedFailure(self):
129
d = defer.succeed('todo')
130
d.addCallback(self._cb_error)
132
test_expectedFailure.todo = "Expected failure"
135
class TimeoutTests(unittest.TestCase):
140
reactor.callLater(0, d.callback, 'hoorj!')
142
test_pass.timeout = 2
144
def test_passDefault(self):
145
# test default timeout
147
reactor.callLater(0, d.callback, 'hoorj!')
150
def test_timeout(self):
151
return defer.Deferred()
152
test_timeout.timeout = 0.1
154
def test_timeoutZero(self):
155
return defer.Deferred()
156
test_timeoutZero.timeout = 0
158
def test_expectedFailure(self):
159
return defer.Deferred()
160
test_expectedFailure.timeout = 0.1
161
test_expectedFailure.todo = "i will get it right, eventually"
164
return defer.Deferred()
165
test_skip.timeout = 0.1
166
test_skip.skip = "i will get it right, eventually"
168
def test_errorPropagation(self):
170
self.__class__.timedOut = err
173
d.addErrback(timedOut)
175
test_errorPropagation.timeout = 0.1
177
def test_calledButNeverCallback(self):
180
return defer.Deferred()
181
d.addCallback(neverFire)
184
test_calledButNeverCallback.timeout = 0.1
187
class TestClassTimeoutAttribute(unittest.TestCase):
191
self.d = defer.Deferred()
193
def testMethod(self):
194
self.methodCalled = True