Simple load tester that continuously sends GET requests to a list of URLs for a given duration of time. Requires python libraries click and requests.

loadtest.py

#!/usr/bin/env python

from __future__ import print_function, division

import click
import csv
from datetime import timedelta
from itertools import cycle
from queue import Queue
import requests
from threading import Thread
from time import sleep, time
import sys


def load_urls(urllistfile):
    return cycle(url.strip() for url in urllistfile)


def log_requests(queue, logfile):
    writer = csv.writer(logfile)
    writer.writerow(['Address', 'Start Time', 'Duration', 'Response Code'])

    while True:
        request_data = queue.get()
        if request_data is None:
            break
        writer.writerow(request_data)


def make_request(url, queue):
    start_time = time()
    response = requests.get(url)
    request_duration = time() - start_time

    queue.put([url, start_time, request_duration, response.status_code])


def do_load_test(duration, count, urllistfile, logfile):
    average_step = (duration / count).total_seconds()
    url_it = iter(load_urls(urllistfile))

    response_queue = Queue()
    logger_thread = Thread(target=lambda q=response_queue, f=logfile: log_requests(q, f))
    logger_thread.start()

    request_threads = []
    for _ in range(count):
        # start a new request in a thread
        request_func = lambda a=next(url_it), q=response_queue: make_request(a, q)
        thread = Thread(target=request_func)
        thread.start()
        request_threads.append(thread)

        # wait until starting the next request
        sleep(average_step)

    # wait for all the threads to finish
    for t in request_threads:
        t.join()
    logger_thread.join()


@click.command('loadtest')
@click.option('--hours', '-h', type=int, default=0)
@click.option('--minutes', '-m', type=int, default=0)
@click.option('--seconds', '-s', type=int, default=0)
@click.option('--numrequests', '-n', type=int, required=True,
              help='The number of requests to send; these will be spread over '
                   'the duration of the test.')
@click.option('--urls', '-i', type=click.File('rU'), default=sys.stdin,
              help='A file containing a list of URLs, one on each line. The '
                   'load tester will cycle through the addresses in this file. '
                   '[default: stdin].')
@click.option('--outfile', '-o', type=click.File('w'), default=sys.stdout,
              help='The file to dump the results. [default: stdout].')
def _loadtest(hours, minutes, seconds, numrequests, urls, outfile):
    '''
    Simple load tester that continuously sends GET requests to a list of URLs
    for a given duration of time.
    '''
    if not hours and not minutes and not seconds:
        raise click.UsageError('You must specify some non-zero duration of '
                               'HOURS, MINUTES, and/or SECONDS.')
    duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
    do_load_test(duration, numrequests, urls, outfile)


if __name__ == '__main__':
    _loadtest()