1 """
2 Simple HTTP request dumper for tests.
3 """
4
5 import sys
6 from contextlib import contextmanager
7
8 try:
9 import urlparse
10 except ImportError:
11
12 import urllib.parse as urlparse
13
14
15 @contextmanager
16 -def webserver(app, port=0, host=None):
17 """Context manager entry point for the 'with' statement.
18
19 Pass 0 as port number to dynamically allocate a free port.
20
21 Usage:
22
23 with webserver(wsgi_app_function, 8080) as host_url:
24 do_ws_calls(host_url)
25 """
26 server = build_web_server(app, port, host or '127.0.0.1')
27 host, port = server.socket.getsockname()
28
29 import threading
30 thread = threading.Thread(target=server.serve_forever,
31 kwargs={'poll_interval': 0.5})
32 thread.setDaemon(True)
33 thread.start()
34 try:
35 yield 'http://%s:%s/' % (host, port)
36 finally:
37 server.shutdown()
38 server.server_close()
39 thread.join(timeout=1)
40
41
42 try:
43 from SocketServer import ThreadingMixIn
44 except ImportError:
45
46 from socketserver import ThreadingMixIn
47
48 import wsgiref.simple_server as wsgiserver
49 -class WebServer(wsgiserver.WSGIServer, ThreadingMixIn):
50 """A web server that starts a new thread for each request.
51 """
52
62
65 server = wsgiserver.make_server(
66 host or '', port, app,
67 server_class=WebServer,
68 handler_class=_RequestHandler)
69 return server
70
73 - def __init__(self, response_data, response_code=200, headers=()):
74 self.requests = []
75 self.response_code = response_code
76 self.response_data = response_data
77 self.headers = list(headers or ())
78
79 - def __call__(self, environ, start_response):
80 self.requests.append((
81 environ.get('PATH_INFO'),
82 urlparse.parse_qsl(environ.get('QUERY_STRING'))))
83 start_response('%s OK' % self.response_code, self.headers)
84 return [self.response_data]
85