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      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