HTTP load testing utility in Python

README

virtualenv .venv --distribute
source .venv/bin/activate
pip install -r requirements.txt
python bench.py http://google.com # Sends 1 request to the googles

bench.py

import argparse
import requests
from concurrent.futures import ThreadPoolExecutor
import itertools
import time

def request_sender(options):
    auth = tuple(options.auth.split(':')) if options.auth else ()
    def send(url):
        resp = requests.post(url, auth=auth, data=options.post) if options.post else requests.get(url, auth=auth)
        resp.raise_for_status()
        resp.content
    return send

def send_requests(urls, options):
    executor = ThreadPoolExecutor(max_workers=options.concurrency)
    for i, __ in enumerate(executor.map(request_sender(options), urls)):
        if i and i % 100 == 0:
            print '{} requests sent'.format(i)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Sends a bunch of requests.')
    parser.add_argument('url', help='URL to send requests to')
    parser.add_argument('--concurrency', type=int, default=1, help='concurrency (default 1)')
    parser.add_argument('--requests', type=int, default=1, help='number of requests (default 1)')
    parser.add_argument('--auth', help='HTTP auth (username:password)')
    parser.add_argument('--post', help='POST data to send')
    args = parser.parse_args()
    start = time.time()
    send_requests(itertools.repeat(args.url, args.requests), args)
    elapsed = time.time() - start
    print 'Time taken for {} requests: {} seconds'.format(args.requests, elapsed)

requirements.txt

distribute==0.6.31
futures==2.1.3
requests==0.14.2
wsgiref==0.1.2