OLD | NEW |
| (Empty) |
1 from __future__ import generators | |
2 from twisted.trial import unittest | |
3 from twisted.internet import defer, threads, reactor | |
4 | |
5 | |
6 class DeferredSetUpOK(unittest.TestCase): | |
7 def setUp(self): | |
8 d = defer.succeed('value') | |
9 d.addCallback(self._cb_setUpCalled) | |
10 return d | |
11 | |
12 def _cb_setUpCalled(self, ignored): | |
13 self._setUpCalled = True | |
14 | |
15 def test_ok(self): | |
16 self.failUnless(self._setUpCalled) | |
17 | |
18 | |
19 class DeferredSetUpFail(unittest.TestCase): | |
20 testCalled = False | |
21 | |
22 def setUp(self): | |
23 return defer.fail(unittest.FailTest('i fail')) | |
24 | |
25 def test_ok(self): | |
26 DeferredSetUpFail.testCalled = True | |
27 self.fail("I should not get called") | |
28 | |
29 | |
30 class DeferredSetUpCallbackFail(unittest.TestCase): | |
31 testCalled = False | |
32 | |
33 def setUp(self): | |
34 d = defer.succeed('value') | |
35 d.addCallback(self._cb_setUpCalled) | |
36 return d | |
37 | |
38 def _cb_setUpCalled(self, ignored): | |
39 self.fail('deliberate failure') | |
40 | |
41 def test_ok(self): | |
42 DeferredSetUpCallbackFail.testCalled = True | |
43 | |
44 | |
45 class DeferredSetUpError(unittest.TestCase): | |
46 testCalled = False | |
47 | |
48 def setUp(self): | |
49 return defer.fail(RuntimeError('deliberate error')) | |
50 | |
51 def test_ok(self): | |
52 DeferredSetUpError.testCalled = True | |
53 | |
54 | |
55 class DeferredSetUpNeverFire(unittest.TestCase): | |
56 testCalled = False | |
57 | |
58 def setUp(self): | |
59 return defer.Deferred() | |
60 | |
61 def test_ok(self): | |
62 DeferredSetUpNeverFire.testCalled = True | |
63 | |
64 | |
65 class DeferredSetUpSkip(unittest.TestCase): | |
66 testCalled = False | |
67 | |
68 def setUp(self): | |
69 d = defer.succeed('value') | |
70 d.addCallback(self._cb1) | |
71 return d | |
72 | |
73 def _cb1(self, ignored): | |
74 raise unittest.SkipTest("skip me") | |
75 | |
76 def test_ok(self): | |
77 DeferredSetUpSkip.testCalled = True | |
78 | |
79 | |
80 class DeferredTests(unittest.TestCase): | |
81 touched = False | |
82 | |
83 def _cb_fail(self, reason): | |
84 self.fail(reason) | |
85 | |
86 def _cb_error(self, reason): | |
87 raise RuntimeError(reason) | |
88 | |
89 def _cb_skip(self, reason): | |
90 raise unittest.SkipTest(reason) | |
91 | |
92 def _touchClass(self, ignored): | |
93 self.__class__.touched = True | |
94 | |
95 def setUp(self): | |
96 self.__class__.touched = False | |
97 | |
98 def test_pass(self): | |
99 return defer.succeed('success') | |
100 | |
101 def test_passGenerated(self): | |
102 self._touchClass(None) | |
103 yield None | |
104 test_passGenerated = defer.deferredGenerator(test_passGenerated) | |
105 | |
106 def test_fail(self): | |
107 return defer.fail(self.failureException('I fail')) | |
108 | |
109 def test_failureInCallback(self): | |
110 d = defer.succeed('fail') | |
111 d.addCallback(self._cb_fail) | |
112 return d | |
113 | |
114 def test_errorInCallback(self): | |
115 d = defer.succeed('error') | |
116 d.addCallback(self._cb_error) | |
117 return d | |
118 | |
119 def test_skip(self): | |
120 d = defer.succeed('skip') | |
121 d.addCallback(self._cb_skip) | |
122 d.addCallback(self._touchClass) | |
123 return d | |
124 | |
125 def test_thread(self): | |
126 return threads.deferToThread(lambda : None) | |
127 | |
128 def test_expectedFailure(self): | |
129 d = defer.succeed('todo') | |
130 d.addCallback(self._cb_error) | |
131 return d | |
132 test_expectedFailure.todo = "Expected failure" | |
133 | |
134 | |
135 class TimeoutTests(unittest.TestCase): | |
136 timedOut = None | |
137 | |
138 def test_pass(self): | |
139 d = defer.Deferred() | |
140 reactor.callLater(0, d.callback, 'hoorj!') | |
141 return d | |
142 test_pass.timeout = 2 | |
143 | |
144 def test_passDefault(self): | |
145 # test default timeout | |
146 d = defer.Deferred() | |
147 reactor.callLater(0, d.callback, 'hoorj!') | |
148 return d | |
149 | |
150 def test_timeout(self): | |
151 return defer.Deferred() | |
152 test_timeout.timeout = 0.1 | |
153 | |
154 def test_timeoutZero(self): | |
155 return defer.Deferred() | |
156 test_timeoutZero.timeout = 0 | |
157 | |
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" | |
162 | |
163 def test_skip(self): | |
164 return defer.Deferred() | |
165 test_skip.timeout = 0.1 | |
166 test_skip.skip = "i will get it right, eventually" | |
167 | |
168 def test_errorPropagation(self): | |
169 def timedOut(err): | |
170 self.__class__.timedOut = err | |
171 return err | |
172 d = defer.Deferred() | |
173 d.addErrback(timedOut) | |
174 return d | |
175 test_errorPropagation.timeout = 0.1 | |
176 | |
177 def test_calledButNeverCallback(self): | |
178 d = defer.Deferred() | |
179 def neverFire(r): | |
180 return defer.Deferred() | |
181 d.addCallback(neverFire) | |
182 d.callback(1) | |
183 return d | |
184 test_calledButNeverCallback.timeout = 0.1 | |
185 | |
186 | |
187 class TestClassTimeoutAttribute(unittest.TestCase): | |
188 timeout = 0.2 | |
189 | |
190 def setUp(self): | |
191 self.d = defer.Deferred() | |
192 | |
193 def testMethod(self): | |
194 self.methodCalled = True | |
195 return self.d | |
OLD | NEW |