Simple Load Tester

load_test.py

import sys
import requests
import logging
import time
import math

from concurrent.futures import ThreadPoolExecutor

# Set up logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Read system arguments
if len(sys.argv) < 2:
    logger.info("Usage: python load_test.py <url> <num_requests>")
    sys.exit(1)

url = sys.argv[1]
n = int(sys.argv[2])

def send_request(url):
    t1 = time.time()
    response = requests.get(url)
    t2 = time.time()
    return response.status_code, t2 - t1

def send_async_request(url, workers=10):
    pass_count = 0
    fail_count = 0
    response_times = []

    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = [executor.submit(send_request, url) for _ in range(workers)]
        for future in futures:
            status_code, response_time = future.result()
            response_times.append(response_time)
            if status_code == 200:
                pass_count += 1
            else:
                fail_count += 1
    return pass_count, fail_count, response_times

def load_test(url, n):
    logger.info(f"Starting test with {n} requests to {url}")
    pass_count = 0
    fail_count = 0

    pass_count, fail_count, response_times = send_async_request(url, workers=n)


    logger.info("Test complete")
    logger.info(f"Pass count: {pass_count}")
    logger.info(f"Fail count: {fail_count}")
    logger.info(f"Pass rate: {pass_count / n * 100:.2f}%")
    logger.info(f"Fail rate: {fail_count / n * 100:.2f}%")
    logger.info(f"Max response time: {max(response_times):.2f} seconds")
    logger.info(f"Min response time: {min(response_times):.2f} seconds")
    logger.info(f"Average response time: {sum(response_times) / len(response_times):.2f} seconds")


    mean_response_time = sum(response_times) / len(response_times)
    std_dev = math.sqrt(sum((x - mean_response_time) ** 2 for x in response_times) / len(response_times))
    logger.info(f"Standard deviation of response time: {std_dev:.2f}")

    # Plot response times
    import matplotlib.pyplot as plt
    plt.plot(response_times)
    plt.xlabel("Request number")
    plt.ylabel("Response time (s)")
    plt.title("Response times")
    plt.show()
    

if __name__ == "__main__":
    load_test(url, n)