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""" 6WSGI Test Server 7 8This builds upon paste.util.baseserver to customize it for regressions 9where using raw_interactive won't do. 10 11 12""" 13import time 14from paste.httpserver import * 15 16class 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 63def 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 69if __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