root/galaxy-central/eggs/Paste-1.6-py2.6.egg/paste/debug/testserver.py

リビジョン 3, 3.3 KB (コミッタ: kohda, 14 年 前)

Install Unix tools  http://hannonlab.cshl.edu/galaxy_unix_tools/galaxy.html

行番号 
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    import urllib
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 urllib.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    urllib.urlopen(baseuri)
Note: リポジトリブラウザについてのヘルプは TracBrowser を参照してください。