eventlet/tests/test__greenness.py

   1 """Test than modules in eventlet.green package are indeed green.
   2 To do that spawn a green server and then access it using a green socket.
   3 If either operation blocked the whole script would block and timeout.
   4 """
   5 import unittest
   6
   7 from eventlet.green import urllib2, BaseHTTPServer
   8 from eventlet import spawn, kill
   9
  10
  11 class QuietHandler(BaseHTTPServer.BaseHTTPRequestHandler):
  12     protocol_version = "HTTP/1.0"
  13
  14     def log_message(self, *args, **kw):
  15         pass
  16
  17
  18 def start_http_server():
  19     server_address = ('localhost', 0)
  20     httpd = BaseHTTPServer.HTTPServer(server_address, QuietHandler)
  21     sa = httpd.socket.getsockname()
  22     # print("Serving HTTP on", sa[0], "port", sa[1], "...")
  23     httpd.request_count = 0
  24
  25     def serve():
  26         # increment the request_count before handling the request because
  27         # the send() for the response blocks (or at least appeared to be)
  28         httpd.request_count += 1
  29         httpd.handle_request()
  30     return spawn(serve), httpd, sa[1]
  31
  32
  33 class TestGreenness(unittest.TestCase):
  34
  35     def setUp(self):
  36         self.gthread, self.server, self.port = start_http_server()
  37         # print('Spawned the server')
  38
  39     def tearDown(self):
  40         self.server.server_close()
  41         kill(self.gthread)
  42
  43     def test_urllib2(self):
  44         self.assertEqual(self.server.request_count, 0)
  45         try:
  46             urllib2.urlopen('http://127.0.0.1:%s' % self.port)
  47             assert False, 'should not get there'
  48         except urllib2.HTTPError as ex:
  49             assert ex.code == 501, repr(ex)
  50         self.assertEqual(self.server.request_count, 1)
  51
  52 if __name__ == '__main__':
  53     unittest.main()