Quick note about stress testing on Google Appengine

In a follow up to my last post on GAE I have the following script that hammers a GAE application to make sure it does not fail under load. Note the following:

1. Change the url for the endpoint to your app.
2. Change the test to be appropriate for what should be returned from your app.
3. The test endpoint should be unique (see the fake query string) or google will cache the results and your app will only see about two requests a second.

"""
Stress tests a endpoint for a GAE application
"""
# rate is in requests per second
rate = 5
# test time is in seconds
test_time = 120
# url to hit has to be unique so that the request hits 
# the app not the cache
url = 'http://your-app.appspot.com/endpoint?test=%s'

test_string = "This string should be in the html response"

import time
import multiprocessing
import urllib
import random


def test():
    """
    The test function
    """
    url_open = urllib.urlopen(url%random.random())
    if test_string in url_open.read():
        pass
    else:
        print 'Failed'

if __name__ == '__main__':
    processes = []
    start = time.time()
    while time.time() <= start + test_time:
        p = multiprocessing.Process(target=test)
        p.start()
        processes.append(p)
        time.sleep(1.0 / rate)
    for p in processes:
        p.join()
    print 'Tested url %s times' % (test_time * rate)