# Written by Jeremy Karst 2025
# This software is licensed under the MIT License with Additional Terms.
# See LICENSE.md for more information.
# Built-in Dependencies
import time
import statistics
from concurrent.futures import ThreadPoolExecutor
import argparse
import random
# External Dependencies
import requests
from rich.console import Console
from rich.table import Table
from rich.progress import Progress
def make_request(base_url, session):
"""Make requests to test the full signature flow"""
start_time = time.time()
try:
# Fetch fonts
fonts_response = session.get(f"{base_url}/fonts")
fonts_response.raise_for_status()
available_fonts = list(fonts_response.json().keys())
# 1. Register a new signature
name = f"Test User {random.randint(100000000, 999999999)}"
register_data = {
'name': name,
'font': random.choice(available_fonts),
'timezone': '0',
'invert': 'false'
}
register_response = session.get(f"{base_url}/register", params=register_data)
register_response.raise_for_status()
signature_hash = register_response.text
register_time = time.time()
# 2. Verify the signature
verify_response = session.get(f"{base_url}/verify/{signature_hash}")
verify_response.raise_for_status()
verify_time = time.time()
# 3. Get the signature image
image_response = session.get(f"{base_url}/images/{signature_hash}.png")
image_response.raise_for_status()
image_time = time.time()
return {
'register_time': register_time - start_time,
'register_status': register_response.status_code,
'verify_time': verify_time - register_time,
'verify_status': verify_response.status_code,
'image_time': image_time - verify_time,
'image_status': image_response.status_code
}
except requests.RequestException as e:
print(f"Request failed: {e}")
return None
def run_benchmark(base_url, num_threads, requests_per_thread, progress=None):
"""Run benchmark with specified number of threads"""
session = requests.Session()
register_times = []
verify_times = []
image_times = []
errors = 0
def worker():
for _ in range(requests_per_thread):
result = make_request(base_url, session)
if result is None:
nonlocal errors
errors += 1
else:
register_times.append(result['register_time'])
verify_times.append(result['verify_time'])
image_times.append(result['image_time'])
if progress:
progress.update(task, advance=1)
# Create and start threads
with Progress() as progress:
task = progress.add_task(f"[cyan]Running {num_threads} threads...", total=num_threads * requests_per_thread)
with ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = [executor.submit(worker) for _ in range(num_threads)]
for future in futures:
future.result()
if not image_times:
return None
return {
'successful_register_requests': len(register_times),
'successful_verify_requests': len(verify_times),
'successful_image_requests': len(image_times),
'failed_requests': errors,
'median_register_response_time': statistics.median(register_times),
'median_verify_response_time': statistics.median(verify_times),
'median_image_response_time': statistics.median(image_times),
'min_response_time': min(register_times + verify_times + image_times),
'max_response_time': max(register_times + verify_times + image_times),
'total_time': sum(register_times + verify_times + image_times)
}
def main():
parser = argparse.ArgumentParser(description='Benchmark web server performance')
parser.add_argument('--url', default='http://localhost:8080', help='Base URL of the server')
parser.add_argument('--start-threads', type=int, default=1, help='Starting number of threads')
parser.add_argument('--max-threads', type=int, default=100, help='Maximum number of threads')
parser.add_argument('--requests-per-thread', type=int, default=20, help='Number of requests per thread')
args = parser.parse_args()
console = Console()
results_table = Table(show_header=True, header_style="bold cyan")
results_table.add_column("Threads")
results_table.add_column("Total Requests")
results_table.add_column("Success Rate")
results_table.add_column("Median Response (s)")
results_table.add_column("Req/sec")
console.print(f"\n[bold green]Starting benchmark against {args.url}[/bold green]")
console.print("[yellow]Note: Each request includes register, verify, and image generation[/yellow]\n")
# Test thread counts from start to max, doubling every step
assert args.start_threads > 0, "Starting threads must be greater than 0"
thread_counts = [args.start_threads]
while thread_counts[-1] < args.max_threads:
thread_counts.append(thread_counts[-1] * 2)
if thread_counts[-1] > args.max_threads:
thread_counts[-1] = args.max_threads
for num_threads in thread_counts:
console.print(f"[yellow]Testing with {num_threads} threads...[/yellow]")
start_time = time.time()
result = run_benchmark(args.url, num_threads, args.requests_per_thread)
end_time = time.time()
if result:
register_success_rate = (result['successful_register_requests'] / (num_threads * args.requests_per_thread)) * 100
verify_success_rate = (result['successful_verify_requests'] / (num_threads * args.requests_per_thread)) * 100
image_success_rate = (result['successful_image_requests'] / (num_threads * args.requests_per_thread)) * 100
requests_per_second = (result['successful_register_requests'] + result['successful_verify_requests'] + result['successful_image_requests']) / (end_time - start_time)
results_table.add_row(
str(num_threads),
str(num_threads * args.requests_per_thread * 3),
f"{register_success_rate:.1f}% / {verify_success_rate:.1f}% / {image_success_rate:.1f}%",
f"{result['median_register_response_time']:.3f} / {result['median_verify_response_time']:.3f} / {result['median_image_response_time']:.3f}",
f"{requests_per_second:.1f}"
)
else:
results_table.add_row(
str(num_threads),
str(num_threads * args.requests_per_thread),
"0%",
"N/A",
"N/A"
)
console.print("\n[bold]Benchmark Results:[/bold]")
console.print(results_table)
if __name__ == "__main__":
main()