tests/test__twistedutil.py

   1 from tests import requires_twisted
   2 import unittest
   3 try:
   4     from twisted.internet import reactor
   5     from twisted.internet.error import DNSLookupError
   6     from twisted.internet import defer
   7     from twisted.python.failure import Failure
   8     from eventlet.twistedutil import block_on
   9 except ImportError:
  10     pass
  11
  12
  13 class Test(unittest.TestCase):
  14     @requires_twisted
  15     def test_block_on_success(self):
  16         from twisted.internet import reactor
  17         d = reactor.resolver.getHostByName('www.google.com')
  18         ip = block_on(d)
  19         assert len(ip.split('.')) == 4, ip
  20         ip2 = block_on(d)
  21         assert ip == ip2, (ip, ip2)
  22
  23     @requires_twisted
  24     def test_block_on_fail(self):
  25         from twisted.internet import reactor
  26         d = reactor.resolver.getHostByName('xxx')
  27         self.assertRaises(DNSLookupError, block_on, d)
  28
  29     @requires_twisted
  30     def test_block_on_already_succeed(self):
  31         d = defer.succeed('hey corotwine')
  32         res = block_on(d)
  33         assert res == 'hey corotwine', repr(res)
  34
  35     @requires_twisted
  36     def test_block_on_already_failed(self):
  37         d = defer.fail(Failure(ZeroDivisionError()))
  38         self.assertRaises(ZeroDivisionError, block_on, d)
  39
  40 if __name__ == '__main__':
  41     unittest.main()
  42