CherryPy / cherrypy / test /

Script to simulate lots of concurrent requests to a site that uses sessions.
It then checks that the integrity of the session data has been kept


import cherrypy
import httplib
import sys
import thread
import time

if len(sys.argv) == 1:
    print """Usage: test_session_concurrency [storage_type]
            [storage_path (for file sessions)]
            [# of server threads] [# of client threads] [# of requests]"""
    print "Example 1: test_session_concurrency ram"
    print "Example 2: test_session_concurrency file /tmp"

storage_type = sys.argv[1]
if storage_type == 'file':
    storage_path = sys.argv[2]
    i = 3
    i = 2
    storage_path = 'dummy'

    server_thread_count = sys.argv[i]
except IndexError:
    server_thread_count = 10
    client_thread_count = sys.argv[i+1]
except IndexError:
    client_thread_count = 5
    request_count = sys.argv[i+2]
except IndexError:
    request_count = 30

# Server code
class Root:
    def index(self):
        # If you remove the "acquire_lock" call the assert at the end
        #   of this script will fail
        c = cherrypy.session.get('counter', 0) + 1
        cherrypy.session['counter'] = c
        return str(c) = True

    'server.environment': 'production',
    'server.log_to_screen': False,
    'server.thread_pool': server_thread_count,
    'session_filter.on': True,
    'session_filter.storage_type': storage_type,
    'session_filter.storage_path': storage_path,
cherrypy.root = Root()

# Client code
def run_client(cookie, request_count, data_dict, index):

    # Make other requests
    for i in xrange(request_count):
        conn = httplib.HTTPConnection('localhost:8080')
        conn.request("GET", "/", headers = {'Cookie': cookie})
        r = conn.getresponse()
        cookie = r.getheader('set-cookie').split(';')[0]
        data =
    data_dict[index] = int(data)

# Start server
thread.start_new_thread(cherrypy.server.start, ())

# Start client

# Make first request to get cookie
conn = httplib.HTTPConnection('localhost:8080')
conn.request("GET", "/")
r = conn.getresponse()
cookie = r.getheader('set-cookie').split(';')[0]
data =

data_dict = {}
# Simulate <request_count> concurrent requests from <client_thread_count>
# from the same client
for i in xrange(client_thread_count):
    data_dict[i] = 0
    thread.start_new_thread(run_client, (cookie, request_count, data_dict, i))
print "Please wait while test is running (default settings take about 30secs)"
while True:
    all_finished = True
    for data in data_dict.values():
        if data == 0:
            all_finished = False
    if all_finished:


m = max(data_dict.values())
expected_m = 1 + (client_thread_count * request_count)
if m != expected_m:
    print "Problem, max is %s instead of %s (data_dict: %s)" % (
        m, expected_m, data_dict)
    print "Everything OK"
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.