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 in Python 2.5+. 
 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 40 41 try: 42 from SocketServer import ThreadingMixIn 43 except ImportError: 44 # Python 3 45 from socketserver import ThreadingMixIn 46 47 import wsgiref.simple_server as wsgiserver
48 -class WebServer(wsgiserver.WSGIServer, ThreadingMixIn):
49 """A web server that starts a new thread for each request. 50 """
51
52 53 -class _RequestHandler(wsgiserver.WSGIRequestHandler):
54 - def get_stderr(self):
55 # don't write to stderr 56 return sys.stdout
57
58 - def log_message(self, format, *args):
59 # message = "wsmock(%s) %s" % (self.address_string(), format % args) 60 pass # don't log messages
61
62 63 -def build_web_server(app, port, host=None):
64 server = wsgiserver.make_server( 65 host or '', port, app, 66 server_class=WebServer, 67 handler_class=_RequestHandler) 68 return server
69
70 71 -class HTTPRequestCollector(object):
72 - def __init__(self, response_data, response_code=200, headers=()):
73 self.requests = [] 74 self.response_code = response_code 75 self.response_data = response_data 76 self.headers = list(headers or ())
77
78 - def __call__(self, environ, start_response):
79 self.requests.append(( 80 environ.get('PATH_INFO'), 81 urlparse.parse_qsl(environ.get('QUERY_STRING')))) 82 start_response('%s OK' % self.response_code, self.headers) 83 return [self.response_data]
84