Home | History | Annotate | Download | only in debug
      1 # (c) 2005 Clark C. Evans
      2 # This module is part of the Python Paste Project and is released under
      3 # the MIT License: http://www.opensource.org/licenses/mit-license.php
      4 # This code was written with funding by http://prometheusresearch.com
      5 """
      6 WSGI Test Server
      7 
      8 This builds upon paste.util.baseserver to customize it for regressions
      9 where using raw_interactive won't do.
     10 
     11 
     12 """
     13 import time
     14 from paste.httpserver import *
     15 
     16 class WSGIRegressionServer(WSGIServer):
     17     """
     18     A threaded WSGIServer for use in regression testing.  To use this
     19     module, call serve(application, regression=True), and then call
     20     server.accept() to let it handle one request.  When finished, use
     21     server.stop() to shutdown the server. Note that all pending requests
     22     are processed before the server shuts down.
     23     """
     24     defaulttimeout = 10
     25     def __init__ (self, *args, **kwargs):
     26         WSGIServer.__init__(self, *args, **kwargs)
     27         self.stopping = []
     28         self.pending = []
     29         self.timeout = self.defaulttimeout
     30         # this is a local connection, be quick
     31         self.socket.settimeout(2)
     32     def serve_forever(self):
     33         from threading import Thread
     34         thread = Thread(target=self.serve_pending)
     35         thread.start()
     36     def reset_expires(self):
     37         if self.timeout:
     38             self.expires = time.time() + self.timeout
     39     def close_request(self, *args, **kwargs):
     40         WSGIServer.close_request(self, *args, **kwargs)
     41         self.pending.pop()
     42         self.reset_expires()
     43     def serve_pending(self):
     44         self.reset_expires()
     45         while not self.stopping or self.pending:
     46             now = time.time()
     47             if now > self.expires and self.timeout:
     48                 # note regression test doesn't handle exceptions in
     49                 # threads very well; so we just print and exit
     50                 print("\nWARNING: WSGIRegressionServer timeout exceeded\n")
     51                 break
     52             if self.pending:
     53                 self.handle_request()
     54             time.sleep(.1)
     55     def stop(self):
     56         """ stop the server (called from tester's thread) """
     57         self.stopping.append(True)
     58     def accept(self, count = 1):
     59         """ accept another request (called from tester's thread) """
     60         assert not self.stopping
     61         [self.pending.append(True) for x in range(count)]
     62 
     63 def serve(application, host=None, port=None, handler=None):
     64     server = WSGIRegressionServer(application, host, port, handler)
     65     print("serving on %s:%s" % server.server_address)
     66     server.serve_forever()
     67     return server
     68 
     69 if __name__ == '__main__':
     70     from six.moves.urllib.request import urlopen
     71     from paste.wsgilib import dump_environ
     72     server = serve(dump_environ)
     73     baseuri = ("http://%s:%s" % server.server_address)
     74 
     75     def fetch(path):
     76         # tell the server to humor exactly one more request
     77         server.accept(1)
     78         # not needed; but this is what you do if the server
     79         # may not respond in a resonable time period
     80         import socket
     81         socket.setdefaulttimeout(5)
     82         # build a uri, fetch and return
     83         return urlopen(baseuri + path).read()
     84 
     85     assert "PATH_INFO: /foo" in fetch("/foo")
     86     assert "PATH_INFO: /womble" in fetch("/womble")
     87 
     88     # ok, let's make one more final request...
     89     server.accept(1)
     90     # and then schedule a stop()
     91     server.stop()
     92     # and then... fetch it...
     93     urlopen(baseuri)
     94