1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
|
import threading
from contextlib import contextmanager
import pytest
from tornado import ioloop, web
from dummyserver.handlers import TestingApp
from dummyserver.proxy import ProxyHandler
from dummyserver.server import (
DEFAULT_CERTS,
HAS_IPV6,
SocketServerThread,
run_loop_in_thread,
run_tornado_app,
)
from urllib3.connection import HTTPConnection
def consume_socket(sock, chunks=65536):
consumed = bytearray()
while True:
b = sock.recv(chunks)
consumed += b
if b.endswith(b"\r\n\r\n"):
break
return consumed
class SocketDummyServerTestCase(object):
"""
A simple socket-based server is created for this class that is good for
exactly one request.
"""
scheme = "http"
host = "localhost"
@classmethod
def _start_server(cls, socket_handler):
ready_event = threading.Event()
cls.server_thread = SocketServerThread(
socket_handler=socket_handler, ready_event=ready_event, host=cls.host
)
cls.server_thread.start()
ready_event.wait(5)
if not ready_event.is_set():
raise Exception("most likely failed to start server")
cls.port = cls.server_thread.port
@classmethod
def start_response_handler(cls, response, num=1, block_send=None):
ready_event = threading.Event()
def socket_handler(listener):
for _ in range(num):
ready_event.set()
sock = listener.accept()[0]
consume_socket(sock)
if block_send:
block_send.wait()
block_send.clear()
sock.send(response)
sock.close()
cls._start_server(socket_handler)
return ready_event
@classmethod
def start_basic_handler(cls, **kw):
return cls.start_response_handler(
b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n" b"\r\n", **kw
)
@classmethod
def teardown_class(cls):
if hasattr(cls, "server_thread"):
cls.server_thread.join(0.1)
def assert_header_received(
self, received_headers, header_name, expected_value=None
):
header_name = header_name.encode("ascii")
if expected_value is not None:
expected_value = expected_value.encode("ascii")
header_titles = []
for header in received_headers:
key, value = header.split(b": ")
header_titles.append(key)
if key == header_name and expected_value is not None:
assert value == expected_value
assert header_name in header_titles
class IPV4SocketDummyServerTestCase(SocketDummyServerTestCase):
@classmethod
def _start_server(cls, socket_handler):
ready_event = threading.Event()
cls.server_thread = SocketServerThread(
socket_handler=socket_handler, ready_event=ready_event, host=cls.host
)
cls.server_thread.USE_IPV6 = False
cls.server_thread.start()
ready_event.wait(5)
if not ready_event.is_set():
raise Exception("most likely failed to start server")
cls.port = cls.server_thread.port
class HTTPDummyServerTestCase(object):
"""A simple HTTP server that runs when your test class runs
Have your test class inherit from this one, and then a simple server
will start when your tests run, and automatically shut down when they
complete. For examples of what test requests you can send to the server,
see the TestingApp in dummyserver/handlers.py.
"""
scheme = "http"
host = "localhost"
host_alt = "127.0.0.1" # Some tests need two hosts
certs = DEFAULT_CERTS
@classmethod
def _start_server(cls):
cls.io_loop = ioloop.IOLoop.current()
app = web.Application([(r".*", TestingApp)])
cls.server, cls.port = run_tornado_app(
app, cls.io_loop, cls.certs, cls.scheme, cls.host
)
cls.server_thread = run_loop_in_thread(cls.io_loop)
@classmethod
def _stop_server(cls):
cls.io_loop.add_callback(cls.server.stop)
cls.io_loop.add_callback(cls.io_loop.stop)
cls.server_thread.join()
@classmethod
def setup_class(cls):
cls._start_server()
@classmethod
def teardown_class(cls):
cls._stop_server()
class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
scheme = "https"
host = "localhost"
certs = DEFAULT_CERTS
class HTTPDummyProxyTestCase(object):
http_host = "localhost"
http_host_alt = "127.0.0.1"
https_host = "localhost"
https_host_alt = "127.0.0.1"
https_certs = DEFAULT_CERTS
proxy_host = "localhost"
proxy_host_alt = "127.0.0.1"
@classmethod
def setup_class(cls):
cls.io_loop = ioloop.IOLoop.current()
app = web.Application([(r".*", TestingApp)])
cls.http_server, cls.http_port = run_tornado_app(
app, cls.io_loop, None, "http", cls.http_host
)
app = web.Application([(r".*", TestingApp)])
cls.https_server, cls.https_port = run_tornado_app(
app, cls.io_loop, cls.https_certs, "https", cls.http_host
)
app = web.Application([(r".*", ProxyHandler)])
cls.proxy_server, cls.proxy_port = run_tornado_app(
app, cls.io_loop, None, "http", cls.proxy_host
)
upstream_ca_certs = cls.https_certs.get("ca_certs", None)
app = web.Application(
[(r".*", ProxyHandler)], upstream_ca_certs=upstream_ca_certs
)
cls.https_proxy_server, cls.https_proxy_port = run_tornado_app(
app, cls.io_loop, cls.https_certs, "https", cls.proxy_host
)
cls.server_thread = run_loop_in_thread(cls.io_loop)
@classmethod
def teardown_class(cls):
cls.io_loop.add_callback(cls.http_server.stop)
cls.io_loop.add_callback(cls.https_server.stop)
cls.io_loop.add_callback(cls.proxy_server.stop)
cls.io_loop.add_callback(cls.https_proxy_server.stop)
cls.io_loop.add_callback(cls.io_loop.stop)
cls.server_thread.join()
@pytest.mark.skipif(not HAS_IPV6, reason="IPv6 not available")
class IPv6HTTPDummyServerTestCase(HTTPDummyServerTestCase):
host = "::1"
@pytest.mark.skipif(not HAS_IPV6, reason="IPv6 not available")
class IPv6HTTPDummyProxyTestCase(HTTPDummyProxyTestCase):
http_host = "localhost"
http_host_alt = "127.0.0.1"
https_host = "localhost"
https_host_alt = "127.0.0.1"
https_certs = DEFAULT_CERTS
proxy_host = "::1"
proxy_host_alt = "127.0.0.1"
class ConnectionMarker(object):
"""
Marks an HTTP(S)Connection's socket after a request was made.
Helps a test server understand when a client finished a request,
without implementing a complete HTTP server.
"""
MARK_FORMAT = b"$#MARK%04x*!"
@classmethod
@contextmanager
def mark(cls, monkeypatch):
"""
Mark connections under in that context.
"""
orig_request = HTTPConnection.request
orig_request_chunked = HTTPConnection.request_chunked
def call_and_mark(target):
def part(self, *args, **kwargs):
result = target(self, *args, **kwargs)
self.sock.sendall(cls._get_socket_mark(self.sock, False))
return result
return part
with monkeypatch.context() as m:
m.setattr(HTTPConnection, "request", call_and_mark(orig_request))
m.setattr(
HTTPConnection, "request_chunked", call_and_mark(orig_request_chunked)
)
yield
@classmethod
def consume_request(cls, sock, chunks=65536):
"""
Consume a socket until after the HTTP request is sent.
"""
consumed = bytearray()
mark = cls._get_socket_mark(sock, True)
while True:
b = sock.recv(chunks)
if not b:
break
consumed += b
if consumed.endswith(mark):
break
return consumed
@classmethod
def _get_socket_mark(cls, sock, server):
if server:
port = sock.getpeername()[1]
else:
port = sock.getsockname()[1]
return cls.MARK_FORMAT % (port,)
|
