gistfile1.py

"""This script was created to test which wsgi server handle correctly
absolute request URI in HTTP request (check rfc2616 5.1.2).

http://www.w3.org/Protocols/rfc2616/rfc2616-sec5.html

Results (12/12/2013):

+-----------------+----------------+
|Framework        |   Result       |
|                 | (OK/NOT OK)    |
+-----------------+----------------+
|cherrypy(3.2.4)  |     OK         |
+-----------------+----------------+
|eventlet(0.9.16) |     NOT OK     |
+-----------------+----------------+
|gevent(1.0)      |     NOT OK     |
+-----------------+----------------+
|paste(1.7.5.1)   |     OK         |
+-----------------+----------------+
|tornado(3.1.1)   |     NOT OK     |
+-----------------+----------------+
|twisted(13.2.0)  |     NOT OK     |
+-----------------+----------------+
|werkzeug(0.9.4)  |     OK         |
+-----------------+----------------+
|wsgiref(0.1.2)   |     NOT OK     |
+-----------------+----------------+
| apache(2.2.22)  |                |
| mod_wsgi(3.3)   |     OK         |
+-----------------+----------------+

NOTE: Apache test was done manually.

Dependencies:

  $ sudo pip install paste twisted tornado cherrypy eventlet gevent werkzeug

NOTE (Python 3.x):

For Python 3 some library doesn't yet support it, specially paste, other
than this both cherrypy and werkzeug work as expected.

"""
import os
import sys
import time
import signal
try:
    import httplib
except ImportError: # python 3.x
    import http.client as httplib
import multiprocessing


PY3 = sys.version_info[0] == 3
PORT = 8080


def _run_test(server_serve, args=()):
    "Return True if success else False."
    print("Starting server ...")
    p = multiprocessing.Process(
        target=server_serve, args=args
    )
    p.start()
    time.sleep(3)   # Wait for the server to become listening.
    print("Sending request ...")
    success = send_request()
    os.kill(p.pid, signal.SIGKILL)
    if not success:
        print("FAIL: WSGI server didn't interpret PATH_INFO as relative")
    return success


def dummy_app(env, start_response):
    """Dummy WSGI app that return PATH_INFO extract from environ."""
    start_response('200 OK', [('Content-Type', 'text/plain')])
    data = env['PATH_INFO']
    if PY3:
        data = bytes(data, 'UTF-8')
    return [data]


def send_request():
    """Send request with absolute Request URI."""
    conn = httplib.HTTPConnection("127.0.0.1", PORT)
    conn.putrequest("GET", "http://127.0.0.1:%s/something/" % PORT)
    conn.putheader("HOST", "127.0.0.1:%s" % PORT)
    conn.endheaders()
    conn.send("")
    resp = conn.getresponse()
    if PY3:
        return resp.read() == b"/something/"
    return resp.read() == "/something/"


def test_eventlet():
    try:
        from eventlet import wsgi
        import eventlet
    except ImportError as ex:
        print("SKIP Eventlet due to: %s" % ex)
        return
    return _run_test(wsgi.server, args=(eventlet.listen(('', PORT)), dummy_app))


def test_cherrypy():
    try:
        from cherrypy import wsgiserver
    except ImportError as ex:
        print("SKIP Cherrypy due to: %s" % ex)
        return
    server = wsgiserver.CherryPyWSGIServer(('0.0.0.0', PORT), dummy_app)
    return _run_test(server.start)


def test_gevent():
    try:
        from gevent import pywsgi
    except ImportError as ex:
        print("SKIP Gevent due to: %s" % ex)
        return
    server = pywsgi.WSGIServer(('', PORT), dummy_app)
    return _run_test(server.serve_forever)


def test_wsgiref():
    from wsgiref.simple_server import make_server

    server = make_server('', PORT, dummy_app)
    return _run_test(server.serve_forever)


def test_werkzeug():
    try:
        from werkzeug.serving import run_simple
    except ImportError as ex:
        print("SKIP Werkzeug due to: %s" % ex)
        return
    return _run_test(run_simple, args=('', PORT, dummy_app))


def test_paste():
    try:
        from paste import httpserver
    except ImportError as ex:
        print("SKIP Paste due to: %s" % ex)
        return
    return _run_test(httpserver.serve, args=(dummy_app, '0.0.0.0', PORT))


def test_tornado():
    try:
        import tornado.httpserver
        import tornado.ioloop
        import tornado.wsgi
    except ImportError as ex:
        print("SKIP Tornado due to: %s" % ex)
        return
    container = tornado.wsgi.WSGIContainer(dummy_app)
    http_server = tornado.httpserver.HTTPServer(container)
    http_server.listen(PORT)

    return _run_test(tornado.ioloop.IOLoop.instance().start)


def test_twisted():
    try:
        from twisted.web.server import Site
        from twisted.web.wsgi import WSGIResource
        from twisted.internet import reactor
    except ImportError as ex:
        print("SKIP Twisted due to: %s" % ex)
        return

    resource = WSGIResource(reactor, reactor.getThreadPool(), dummy_app)
    reactor.listenTCP(PORT, Site(resource))

    return _run_test(reactor.run)


def _with_version(pkg_name):
    try:
        import pkg_resources
    except ImportError:
        return pkg_name
    else:
        try:
            version = pkg_resources.get_distribution(pkg_name).version
            return "%s(%s)" % (pkg_name, version)
        except pkg_resources.DistributionNotFound:
            return pkg_name


def main(argv=sys.argv[1:]):
    """This script can be called to limit the tests to run like this:

        $ script.py eventlet gevent

    """
    global PORT
    worked = []
    failed = []
    skipped = []
    module_attr = globals().copy()
    for func_name in module_attr:
        if func_name.startswith('test'):
            framework = func_name.split('_')[1]
            if argv and framework not in argv:
                continue
            print("********* %s **********" % framework.upper())
            success = globals()[func_name]()
            framework = _with_version(framework)
            if success is None:
                skipped.append(framework)
            elif success:
                worked.append(framework)
            else:
                failed.append(framework)
            # Increment port so there will be no socket already in use errors.
            PORT += 1

    print('\n' * 2)
    print('Success: %s' % ', '.join(worked or ['None']))
    print('Failed: %s' % ', '.join(failed or ['None']))
    if skipped:
        print('Skipped: %s' % ', '.join(skipped))


if __name__ == "__main__":
    main()