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