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