Package lxml :: Package tests :: Module dummy_http_server
[hide private]
[frames] | no frames]

Source Code for Module lxml.tests.dummy_http_server

 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      # Python 3 
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) # yield control to 'with' body 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 # Python 3 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
53 54 -class _RequestHandler(wsgiserver.WSGIRequestHandler):
55 - def get_stderr(self):
56 # don't write to stderr 57 return sys.stdout
58
59 - def log_message(self, format, *args):
60 # message = "wsmock(%s) %s" % (self.address_string(), format % args) 61 pass # don't log messages
62
63 64 -def build_web_server(app, port, host=None):
65 server = wsgiserver.make_server( 66 host or '', port, app, 67 server_class=WebServer, 68 handler_class=_RequestHandler) 69 return server
70
71 72 -class HTTPRequestCollector(object):
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