tcc/scripts/testes.py

import requests
import concurrent.futures
import time
import sys
import os
from graph import generateGraph

if len(sys.argv) != 2 or sys.argv[1] == '-h' or sys.argv[1] == '--help':
    print("Usage: python testes.py <name>")
    sys.exit(1)

FRAMEWORK_NAME = sys.argv[1]
URL_BASE = 'http://localhost:9080'
ENDPOINTS = [
    '/status/ok',
    '/image/load-image',
]

#num_requests = [10, 1000, 5000, 10_000, 50_000, 100_000, 500_000, 1_000_000]
num_requests = range(0, 50_000, 5000)

def send_request(url):
    success = False
    while not success:
        try:
            response = requests.get(url)
        except:
            continue
        success = response.status_code == 200
    return response.status_code

def getFileName(endpoint):
    endpoint = endpoint.replace('/', '_')
    return f"{FRAMEWORK_NAME}_{endpoint}.csv"

def record(filename, requests, reqpersec):
    with open(filename, "a") as file:
        file.write(f"{requests},{reqpersec}\n")

def run_tests(endpoint):
    filename = getFileName(endpoint)
    if os.path.exists(filename):
        os.remove(filename)

    for num_request in num_requests:
        if num_request <= 0: continue

        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            url = f'{URL_BASE}{endpoint}'

            start_time = time.time()

            futures = []
            #with requests.Session() as session:
            #    futures = [executor.submit(send_request, session, url) for _ in range(num_request)]
            futures = [executor.submit(send_request, url) for _ in range(num_request)]

            concurrent.futures.wait(futures)

            elapsed_time = time.time() - start_time

        print(f"{num_request}: {elapsed_time:.2f} seconds. {elapsed_time/num_request:.4f} seconds per request. {num_request/elapsed_time:.2f} requests per second.")
        record(filename, num_request, f"{num_request/elapsed_time:.2f}")
        generateGraph(filename, FRAMEWORK_NAME, endpoint)

        time.sleep(3)

for endpoint in ENDPOINTS:
    print(f"# {endpoint}")
    run_tests(endpoint)