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