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