twisted/trial/test/detests.py

   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