testcase.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. import sys
  2. import socket
  3. import threading
  4. from nose.plugins.skip import SkipTest
  5. from tornado import ioloop, web
  6. from dummyserver.server import (
  7. SocketServerThread,
  8. run_tornado_app,
  9. run_loop_in_thread,
  10. DEFAULT_CERTS,
  11. )
  12. from dummyserver.handlers import TestingApp
  13. from dummyserver.proxy import ProxyHandler
  14. if sys.version_info >= (2, 7):
  15. import unittest
  16. else:
  17. import unittest2 as unittest
  18. def consume_socket(sock, chunks=65536):
  19. while not sock.recv(chunks).endswith(b'\r\n\r\n'):
  20. pass
  21. class SocketDummyServerTestCase(unittest.TestCase):
  22. """
  23. A simple socket-based server is created for this class that is good for
  24. exactly one request.
  25. """
  26. scheme = 'http'
  27. host = 'localhost'
  28. @classmethod
  29. def _start_server(cls, socket_handler):
  30. ready_event = threading.Event()
  31. cls.server_thread = SocketServerThread(socket_handler=socket_handler,
  32. ready_event=ready_event,
  33. host=cls.host)
  34. cls.server_thread.start()
  35. ready_event.wait(5)
  36. if not ready_event.is_set():
  37. raise Exception("most likely failed to start server")
  38. cls.port = cls.server_thread.port
  39. @classmethod
  40. def start_response_handler(cls, response, num=1, block_send=None):
  41. ready_event = threading.Event()
  42. def socket_handler(listener):
  43. for _ in range(num):
  44. ready_event.set()
  45. sock = listener.accept()[0]
  46. consume_socket(sock)
  47. if block_send:
  48. block_send.wait()
  49. block_send.clear()
  50. sock.send(response)
  51. sock.close()
  52. cls._start_server(socket_handler)
  53. return ready_event
  54. @classmethod
  55. def start_basic_handler(cls, **kw):
  56. return cls.start_response_handler(
  57. b'HTTP/1.1 200 OK\r\n'
  58. b'Content-Length: 0\r\n'
  59. b'\r\n', **kw)
  60. @classmethod
  61. def tearDownClass(cls):
  62. if hasattr(cls, 'server_thread'):
  63. cls.server_thread.join(0.1)
  64. class IPV4SocketDummyServerTestCase(SocketDummyServerTestCase):
  65. @classmethod
  66. def _start_server(cls, socket_handler):
  67. ready_event = threading.Event()
  68. cls.server_thread = SocketServerThread(socket_handler=socket_handler,
  69. ready_event=ready_event,
  70. host=cls.host)
  71. cls.server_thread.USE_IPV6 = False
  72. cls.server_thread.start()
  73. ready_event.wait(5)
  74. if not ready_event.is_set():
  75. raise Exception("most likely failed to start server")
  76. cls.port = cls.server_thread.port
  77. class HTTPDummyServerTestCase(unittest.TestCase):
  78. """ A simple HTTP server that runs when your test class runs
  79. Have your unittest class inherit from this one, and then a simple server
  80. will start when your tests run, and automatically shut down when they
  81. complete. For examples of what test requests you can send to the server,
  82. see the TestingApp in dummyserver/handlers.py.
  83. """
  84. scheme = 'http'
  85. host = 'localhost'
  86. host_alt = '127.0.0.1' # Some tests need two hosts
  87. certs = DEFAULT_CERTS
  88. @classmethod
  89. def _start_server(cls):
  90. cls.io_loop = ioloop.IOLoop()
  91. app = web.Application([(r".*", TestingApp)])
  92. cls.server, cls.port = run_tornado_app(app, cls.io_loop, cls.certs,
  93. cls.scheme, cls.host)
  94. cls.server_thread = run_loop_in_thread(cls.io_loop)
  95. @classmethod
  96. def _stop_server(cls):
  97. cls.io_loop.add_callback(cls.server.stop)
  98. cls.io_loop.add_callback(cls.io_loop.stop)
  99. cls.server_thread.join()
  100. @classmethod
  101. def setUpClass(cls):
  102. cls._start_server()
  103. @classmethod
  104. def tearDownClass(cls):
  105. cls._stop_server()
  106. class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
  107. scheme = 'https'
  108. host = 'localhost'
  109. certs = DEFAULT_CERTS
  110. class IPV6HTTPSDummyServerTestCase(HTTPSDummyServerTestCase):
  111. host = '::1'
  112. @classmethod
  113. def setUpClass(cls):
  114. if not socket.has_ipv6:
  115. raise SkipTest('IPv6 not available')
  116. else:
  117. super(IPV6HTTPSDummyServerTestCase, cls).setUpClass()
  118. class HTTPDummyProxyTestCase(unittest.TestCase):
  119. http_host = 'localhost'
  120. http_host_alt = '127.0.0.1'
  121. https_host = 'localhost'
  122. https_host_alt = '127.0.0.1'
  123. https_certs = DEFAULT_CERTS
  124. proxy_host = 'localhost'
  125. proxy_host_alt = '127.0.0.1'
  126. @classmethod
  127. def setUpClass(cls):
  128. cls.io_loop = ioloop.IOLoop()
  129. app = web.Application([(r'.*', TestingApp)])
  130. cls.http_server, cls.http_port = run_tornado_app(
  131. app, cls.io_loop, None, 'http', cls.http_host)
  132. app = web.Application([(r'.*', TestingApp)])
  133. cls.https_server, cls.https_port = run_tornado_app(
  134. app, cls.io_loop, cls.https_certs, 'https', cls.http_host)
  135. app = web.Application([(r'.*', ProxyHandler)])
  136. cls.proxy_server, cls.proxy_port = run_tornado_app(
  137. app, cls.io_loop, None, 'http', cls.proxy_host)
  138. cls.server_thread = run_loop_in_thread(cls.io_loop)
  139. @classmethod
  140. def tearDownClass(cls):
  141. cls.io_loop.add_callback(cls.http_server.stop)
  142. cls.io_loop.add_callback(cls.https_server.stop)
  143. cls.io_loop.add_callback(cls.proxy_server.stop)
  144. cls.io_loop.add_callback(cls.io_loop.stop)
  145. cls.server_thread.join()
  146. class IPv6HTTPDummyServerTestCase(HTTPDummyServerTestCase):
  147. host = '::1'
  148. @classmethod
  149. def setUpClass(cls):
  150. if not socket.has_ipv6:
  151. raise SkipTest('IPv6 not available')
  152. else:
  153. super(IPv6HTTPDummyServerTestCase, cls).setUpClass()
  154. class IPv6HTTPDummyProxyTestCase(HTTPDummyProxyTestCase):
  155. http_host = 'localhost'
  156. http_host_alt = '127.0.0.1'
  157. https_host = 'localhost'
  158. https_host_alt = '127.0.0.1'
  159. https_certs = DEFAULT_CERTS
  160. proxy_host = '::1'
  161. proxy_host_alt = '127.0.0.1'