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.assertEqual(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.assertEqual(result.testsRun, 1)
  26         self.assertEqual(len(result.failures), 0)
  27         self.assertEqual(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.assertEqual(result.testsRun, 1)
  36         self.assertEqual(len(result.failures), 0)
  37         self.assertEqual(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.assertEqual(result.testsRun, 1)
  46         self.assertEqual(len(result.failures), 0)
  47         self.assertEqual(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.assertEqual(result.testsRun, 1)
  56         self.assertEqual(len(result.failures), 0)
  57         self.assertEqual(len(result.errors), 0)
  58         self.assertEqual(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.assertEqual(result.testsRun, 1)
  82         self.assertEqual(len(result.failures), 0)
  83         self.assertEqual(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.assertEqual(result.testsRun, 1)
 106
 107     def test_passGenerated(self):
 108         result = self.runTest('test_passGenerated')
 109         self.failUnless(result.wasSuccessful())
 110         self.assertEqual(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.assertEqual(result.testsRun, 1)
 117         self.assertEqual(len(result.failures), 1)
 118
 119     def test_failureInCallback(self):
 120         result = self.runTest('test_failureInCallback')
 121         self.failIf(result.wasSuccessful())
 122         self.assertEqual(result.testsRun, 1)
 123         self.assertEqual(len(result.failures), 1)
 124
 125     def test_errorInCallback(self):
 126         result = self.runTest('test_errorInCallback')
 127         self.failIf(result.wasSuccessful())
 128         self.assertEqual(result.testsRun, 1)
 129         self.assertEqual(len(result.errors), 1)
 130
 131     def test_skip(self):
 132         result = self.runTest('test_skip')
 133         self.failUnless(result.wasSuccessful())
 134         self.assertEqual(result.testsRun, 1)
 135         self.assertEqual(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.assertEqual(result.testsRun, 1)
 142         self.assertEqual(len(result.errors), 0)
 143         self.assertEqual(len(result.failures), 0)
 144         self.assertEqual(len(result.expectedFailures), 1)
 145
 146     def test_thread(self):
 147         result = self.runTest('test_thread')
 148         self.assertEqual(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.assertEqual(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.assertEqual(result.testsRun, 1)
 164
 165     def test_passDefault(self):
 166         result = self.runTest('test_passDefault')
 167         self.failUnless(result.wasSuccessful())
 168         self.assertEqual(result.testsRun, 1)
 169
 170     def test_timeout(self):
 171         result = self.runTest('test_timeout')
 172         self.failIf(result.wasSuccessful())
 173         self.assertEqual(result.testsRun, 1)
 174         self.assertEqual(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.assertEqual(result.testsRun, 1)
 181         self.assertEqual(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.assertEqual(result.testsRun, 1)
 188         self.assertEqual(len(result.skips), 1)
 189
 190     def test_todo(self):
 191         result = self.runTest('test_expectedFailure')
 192         self.failUnless(result.wasSuccessful())
 193         self.assertEqual(result.testsRun, 1)
 194         self.assertEqual(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.assertEqual(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.assertEqual(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])