← Back to branch summary

~0x44/nova/bug838466

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/trial/test/test_deferred.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/trial/test/test_deferred.py
 
1
from twisted.internet import defer
 
2
from twisted.trial import unittest
 
3
from twisted.trial import runner, reporter, util
 
4
from twisted.trial.test import detests
 
5
 
 
6
 
 
7
class TestSetUp(unittest.TestCase):
 
8
    def _loadSuite(self, klass):
 
9
        loader = runner.TestLoader()
 
10
        r = reporter.TestResult()
 
11
        s = loader.loadClass(klass)
 
12
        return r, s
 
13
 
 
14
    def test_success(self):
 
15
        result, suite = self._loadSuite(detests.DeferredSetUpOK)
 
16
        suite(result)
 
17
        self.failUnless(result.wasSuccessful())
 
18
        self.failUnlessEqual(result.testsRun, 1)
 
19
 
 
20
    def test_fail(self):
 
21
        self.failIf(detests.DeferredSetUpFail.testCalled)
 
22
        result, suite = self._loadSuite(detests.DeferredSetUpFail)
 
23
        suite(result)
 
24
        self.failIf(result.wasSuccessful())
 
25
        self.failUnlessEqual(result.testsRun, 1)
 
26
        self.failUnlessEqual(len(result.failures), 0)
 
27
        self.failUnlessEqual(len(result.errors), 1)
 
28
        self.failIf(detests.DeferredSetUpFail.testCalled)
 
29
 
 
30
    def test_callbackFail(self):
 
31
        self.failIf(detests.DeferredSetUpCallbackFail.testCalled)
 
32
        result, suite = self._loadSuite(detests.DeferredSetUpCallbackFail)
 
33
        suite(result)
 
34
        self.failIf(result.wasSuccessful())
 
35
        self.failUnlessEqual(result.testsRun, 1)
 
36
        self.failUnlessEqual(len(result.failures), 0)
 
37
        self.failUnlessEqual(len(result.errors), 1)
 
38
        self.failIf(detests.DeferredSetUpCallbackFail.testCalled)
 
39
 
 
40
    def test_error(self):
 
41
        self.failIf(detests.DeferredSetUpError.testCalled)
 
42
        result, suite = self._loadSuite(detests.DeferredSetUpError)
 
43
        suite(result)
 
44
        self.failIf(result.wasSuccessful())
 
45
        self.failUnlessEqual(result.testsRun, 1)
 
46
        self.failUnlessEqual(len(result.failures), 0)
 
47
        self.failUnlessEqual(len(result.errors), 1)
 
48
        self.failIf(detests.DeferredSetUpError.testCalled)
 
49
 
 
50
    def test_skip(self):
 
51
        self.failIf(detests.DeferredSetUpSkip.testCalled)
 
52
        result, suite = self._loadSuite(detests.DeferredSetUpSkip)
 
53
        suite(result)
 
54
        self.failUnless(result.wasSuccessful())
 
55
        self.failUnlessEqual(result.testsRun, 1)
 
56
        self.failUnlessEqual(len(result.failures), 0)
 
57
        self.failUnlessEqual(len(result.errors), 0)
 
58
        self.failUnlessEqual(len(result.skips), 1)
 
59
        self.failIf(detests.DeferredSetUpSkip.testCalled)
 
60
 
 
61
 
 
62
class TestNeverFire(unittest.TestCase):
 
63
    def setUp(self):
 
64
        self._oldTimeout = util.DEFAULT_TIMEOUT_DURATION
 
65
        util.DEFAULT_TIMEOUT_DURATION = 0.1
 
66
 
 
67
    def tearDown(self):
 
68
        util.DEFAULT_TIMEOUT_DURATION = self._oldTimeout
 
69
 
 
70
    def _loadSuite(self, klass):
 
71
        loader = runner.TestLoader()
 
72
        r = reporter.TestResult()
 
73
        s = loader.loadClass(klass)
 
74
        return r, s
 
75
 
 
76
    def test_setUp(self):
 
77
        self.failIf(detests.DeferredSetUpNeverFire.testCalled)
 
78
        result, suite = self._loadSuite(detests.DeferredSetUpNeverFire)
 
79
        suite(result)
 
80
        self.failIf(result.wasSuccessful())
 
81
        self.failUnlessEqual(result.testsRun, 1)
 
82
        self.failUnlessEqual(len(result.failures), 0)
 
83
        self.failUnlessEqual(len(result.errors), 1)
 
84
        self.failIf(detests.DeferredSetUpNeverFire.testCalled)
 
85
        self.failUnless(result.errors[0][1].check(defer.TimeoutError))
 
86
 
 
87
 
 
88
class TestTester(unittest.TestCase):
 
89
    def getTest(self, name):
 
90
        raise NotImplementedError("must override me")
 
91
 
 
92
    def runTest(self, name):
 
93
        result = reporter.TestResult()
 
94
        self.getTest(name).run(result)
 
95
        return result
 
96
 
 
97
 
 
98
class TestDeferred(TestTester):
 
99
    def getTest(self, name):
 
100
        return detests.DeferredTests(name)
 
101
 
 
102
    def test_pass(self):
 
103
        result = self.runTest('test_pass')
 
104
        self.failUnless(result.wasSuccessful())
 
105
        self.failUnlessEqual(result.testsRun, 1)
 
106
 
 
107
    def test_passGenerated(self):
 
108
        result = self.runTest('test_passGenerated')
 
109
        self.failUnless(result.wasSuccessful())
 
110
        self.failUnlessEqual(result.testsRun, 1)
 
111
        self.failUnless(detests.DeferredTests.touched)
 
112
 
 
113
    def test_fail(self):
 
114
        result = self.runTest('test_fail')
 
115
        self.failIf(result.wasSuccessful())
 
116
        self.failUnlessEqual(result.testsRun, 1)
 
117
        self.failUnlessEqual(len(result.failures), 1)
 
118
 
 
119
    def test_failureInCallback(self):
 
120
        result = self.runTest('test_failureInCallback')
 
121
        self.failIf(result.wasSuccessful())
 
122
        self.failUnlessEqual(result.testsRun, 1)
 
123
        self.failUnlessEqual(len(result.failures), 1)
 
124
 
 
125
    def test_errorInCallback(self):
 
126
        result = self.runTest('test_errorInCallback')
 
127
        self.failIf(result.wasSuccessful())
 
128
        self.failUnlessEqual(result.testsRun, 1)
 
129
        self.failUnlessEqual(len(result.errors), 1)
 
130
 
 
131
    def test_skip(self):
 
132
        result = self.runTest('test_skip')
 
133
        self.failUnless(result.wasSuccessful())
 
134
        self.failUnlessEqual(result.testsRun, 1)
 
135
        self.failUnlessEqual(len(result.skips), 1)
 
136
        self.failIf(detests.DeferredTests.touched)
 
137
 
 
138
    def test_todo(self):
 
139
        result = self.runTest('test_expectedFailure')
 
140
        self.failUnless(result.wasSuccessful())
 
141
        self.failUnlessEqual(result.testsRun, 1)
 
142
        self.failUnlessEqual(len(result.errors), 0)
 
143
        self.failUnlessEqual(len(result.failures), 0)
 
144
        self.failUnlessEqual(len(result.expectedFailures), 1)
 
145
 
 
146
    def test_thread(self):
 
147
        result = self.runTest('test_thread')
 
148
        self.failUnlessEqual(result.testsRun, 1)
 
149
        self.failUnless(result.wasSuccessful(), result.errors)
 
150
 
 
151
 
 
152
class TestTimeout(TestTester):
 
153
    def getTest(self, name):
 
154
        return detests.TimeoutTests(name)
 
155
 
 
156
    def _wasTimeout(self, error):
 
157
        self.failUnlessEqual(error.check(defer.TimeoutError),
 
158
                             defer.TimeoutError)
 
159
 
 
160
    def test_pass(self):
 
161
        result = self.runTest('test_pass')
 
162
        self.failUnless(result.wasSuccessful())
 
163
        self.failUnlessEqual(result.testsRun, 1)
 
164
 
 
165
    def test_passDefault(self):
 
166
        result = self.runTest('test_passDefault')
 
167
        self.failUnless(result.wasSuccessful())
 
168
        self.failUnlessEqual(result.testsRun, 1)
 
169
 
 
170
    def test_timeout(self):
 
171
        result = self.runTest('test_timeout')
 
172
        self.failIf(result.wasSuccessful())
 
173
        self.failUnlessEqual(result.testsRun, 1)
 
174
        self.failUnlessEqual(len(result.errors), 1)
 
175
        self._wasTimeout(result.errors[0][1])
 
176
 
 
177
    def test_timeoutZero(self):
 
178
        result = self.runTest('test_timeoutZero')
 
179
        self.failIf(result.wasSuccessful())
 
180
        self.failUnlessEqual(result.testsRun, 1)
 
181
        self.failUnlessEqual(len(result.errors), 1)
 
182
        self._wasTimeout(result.errors[0][1])
 
183
 
 
184
    def test_skip(self):
 
185
        result = self.runTest('test_skip')
 
186
        self.failUnless(result.wasSuccessful())
 
187
        self.failUnlessEqual(result.testsRun, 1)
 
188
        self.failUnlessEqual(len(result.skips), 1)
 
189
 
 
190
    def test_todo(self):
 
191
        result = self.runTest('test_expectedFailure')
 
192
        self.failUnless(result.wasSuccessful())
 
193
        self.failUnlessEqual(result.testsRun, 1)
 
194
        self.failUnlessEqual(len(result.expectedFailures), 1)
 
195
        self._wasTimeout(result.expectedFailures[0][1])
 
196
 
 
197
    def test_errorPropagation(self):
 
198
        result = self.runTest('test_errorPropagation')
 
199
        self.failIf(result.wasSuccessful())
 
200
        self.failUnlessEqual(result.testsRun, 1)
 
201
        self._wasTimeout(detests.TimeoutTests.timedOut)
 
202
 
 
203
    def test_classTimeout(self):
 
204
        loader = runner.TestLoader()
 
205
        suite = loader.loadClass(detests.TestClassTimeoutAttribute)
 
206
        result = reporter.TestResult()
 
207
        suite.run(result)
 
208
        self.failUnlessEqual(len(result.errors), 1)
 
209
        self._wasTimeout(result.errors[0][1])
 
210
 
 
211
    def test_callbackReturnsNonCallingDeferred(self):
 
212
        #hacky timeout
 
213
        # raises KeyboardInterrupt because Trial sucks
 
214
        from twisted.internet import reactor
 
215
        call = reactor.callLater(2, reactor.crash)
 
216
        result = self.runTest('test_calledButNeverCallback')
 
217
        if call.active():
 
218
            call.cancel()
 
219
        self.failIf(result.wasSuccessful())
 
220
        self._wasTimeout(result.errors[0][1])