testcase.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210
  1. import threading
  2. import pytest
  3. from tornado import ioloop, web
  4. from dummyserver.server import (
  5. SocketServerThread,
  6. run_tornado_app,
  7. run_loop_in_thread,
  8. DEFAULT_CERTS,
  9. HAS_IPV6,
  10. )
  11. from dummyserver.handlers import TestingApp
  12. from dummyserver.proxy import ProxyHandler
  13. def consume_socket(sock, chunks=65536):
  14. consumed = bytearray()
  15. while True:
  16. b = sock.recv(chunks)
  17. consumed += b
  18. if b.endswith(b"\r\n\r\n"):
  19. break
  20. return consumed
  21. class SocketDummyServerTestCase(object):
  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(
  32. socket_handler=socket_handler, ready_event=ready_event, host=cls.host
  33. )
  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" b"Content-Length: 0\r\n" b"\r\n", **kw
  58. )
  59. @classmethod
  60. def teardown_class(cls):
  61. if hasattr(cls, "server_thread"):
  62. cls.server_thread.join(0.1)
  63. def assert_header_received(
  64. self, received_headers, header_name, expected_value=None
  65. ):
  66. header_name = header_name.encode("ascii")
  67. if expected_value is not None:
  68. expected_value = expected_value.encode("ascii")
  69. header_titles = []
  70. for header in received_headers:
  71. key, value = header.split(b": ")
  72. header_titles.append(key)
  73. if key == header_name and expected_value is not None:
  74. assert value == expected_value
  75. assert header_name in header_titles
  76. class IPV4SocketDummyServerTestCase(SocketDummyServerTestCase):
  77. @classmethod
  78. def _start_server(cls, socket_handler):
  79. ready_event = threading.Event()
  80. cls.server_thread = SocketServerThread(
  81. socket_handler=socket_handler, ready_event=ready_event, host=cls.host
  82. )
  83. cls.server_thread.USE_IPV6 = False
  84. cls.server_thread.start()
  85. ready_event.wait(5)
  86. if not ready_event.is_set():
  87. raise Exception("most likely failed to start server")
  88. cls.port = cls.server_thread.port
  89. class HTTPDummyServerTestCase(object):
  90. """ A simple HTTP server that runs when your test class runs
  91. Have your test class inherit from this one, and then a simple server
  92. will start when your tests run, and automatically shut down when they
  93. complete. For examples of what test requests you can send to the server,
  94. see the TestingApp in dummyserver/handlers.py.
  95. """
  96. scheme = "http"
  97. host = "localhost"
  98. host_alt = "127.0.0.1" # Some tests need two hosts
  99. certs = DEFAULT_CERTS
  100. @classmethod
  101. def _start_server(cls):
  102. cls.io_loop = ioloop.IOLoop.current()
  103. app = web.Application([(r".*", TestingApp)])
  104. cls.server, cls.port = run_tornado_app(
  105. app, cls.io_loop, cls.certs, cls.scheme, cls.host
  106. )
  107. cls.server_thread = run_loop_in_thread(cls.io_loop)
  108. @classmethod
  109. def _stop_server(cls):
  110. cls.io_loop.add_callback(cls.server.stop)
  111. cls.io_loop.add_callback(cls.io_loop.stop)
  112. cls.server_thread.join()
  113. @classmethod
  114. def setup_class(cls):
  115. cls._start_server()
  116. @classmethod
  117. def teardown_class(cls):
  118. cls._stop_server()
  119. class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
  120. scheme = "https"
  121. host = "localhost"
  122. certs = DEFAULT_CERTS
  123. class HTTPDummyProxyTestCase(object):
  124. http_host = "localhost"
  125. http_host_alt = "127.0.0.1"
  126. https_host = "localhost"
  127. https_host_alt = "127.0.0.1"
  128. https_certs = DEFAULT_CERTS
  129. proxy_host = "localhost"
  130. proxy_host_alt = "127.0.0.1"
  131. @classmethod
  132. def setup_class(cls):
  133. cls.io_loop = ioloop.IOLoop.current()
  134. app = web.Application([(r".*", TestingApp)])
  135. cls.http_server, cls.http_port = run_tornado_app(
  136. app, cls.io_loop, None, "http", cls.http_host
  137. )
  138. app = web.Application([(r".*", TestingApp)])
  139. cls.https_server, cls.https_port = run_tornado_app(
  140. app, cls.io_loop, cls.https_certs, "https", cls.http_host
  141. )
  142. app = web.Application([(r".*", ProxyHandler)])
  143. cls.proxy_server, cls.proxy_port = run_tornado_app(
  144. app, cls.io_loop, None, "http", cls.proxy_host
  145. )
  146. cls.server_thread = run_loop_in_thread(cls.io_loop)
  147. @classmethod
  148. def teardown_class(cls):
  149. cls.io_loop.add_callback(cls.http_server.stop)
  150. cls.io_loop.add_callback(cls.https_server.stop)
  151. cls.io_loop.add_callback(cls.proxy_server.stop)
  152. cls.io_loop.add_callback(cls.io_loop.stop)
  153. cls.server_thread.join()
  154. @pytest.mark.skipif(not HAS_IPV6, reason="IPv6 not available")
  155. class IPv6HTTPDummyServerTestCase(HTTPDummyServerTestCase):
  156. host = "::1"
  157. @pytest.mark.skipif(not HAS_IPV6, reason="IPv6 not available")
  158. class IPv6HTTPDummyProxyTestCase(HTTPDummyProxyTestCase):
  159. http_host = "localhost"
  160. http_host_alt = "127.0.0.1"
  161. https_host = "localhost"
  162. https_host_alt = "127.0.0.1"
  163. https_certs = DEFAULT_CERTS
  164. proxy_host = "::1"
  165. proxy_host_alt = "127.0.0.1"