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