import argparse
import ast
from contextlib import contextmanager
import math
from multiprocessing import Pool
import random
import subprocess
import sys
from timeit import default_timer
# This is a crappy script to feed the server with somewhat random arithmetic
# expressions.
# It's throwaway, hence the poor quality.
DEFAULT_CLIENT_PATH = 'client'
DEFAULT_HOST = 'localhost'
DEFAULT_PORT = 18000
class Client:
def __init__(self, path=DEFAULT_CLIENT_PATH, host=DEFAULT_HOST, port=DEFAULT_PORT):
self._path = path
self._host = host
self._port = port
def get_command_line(self):
return [self._path, '--host', self._host, '--port', str(self._port)]
@contextmanager
def timer(description):
start = default_timer()
yield
duration = default_timer() - start
print(f"{description}: {duration}")
def run_client(i, client, stdin):
with timer(f"Invocation #{i}"):
cmd = client.get_command_line()
return subprocess.run(cmd, text=True, input=stdin, capture_output=True)
OPERATORS = '+', '-', '*', '/'
MIN_OPERATORS = 10
MAX_OPERATORS = 1000
MIN_NUMBER = -10e10
MAX_NUMBER = 10e10
def random_operator():
return OPERATORS[random.randrange(len(OPERATORS))]
def random_number():
return random.randint(MIN_NUMBER, MAX_NUMBER)
def gen_expression():
numof_operators = random.randrange(MIN_OPERATORS, MAX_OPERATORS + 1)
expression = ''
for i in range(numof_operators):
expression += f"{random_number()} {random_operator()} "
expression += str(random_number())
return expression
def gen_expressions(n):
for i in range(n):
yield gen_expression()
def run_stress_test(args):
client = Client(args.client, args.host, args.port)
expressions = list(gen_expressions(args.expressions))
stdin = '\n'.join(expressions)
expected_output = [eval(expr) for expr in expressions]
with Pool(args.processes) as pool:
results = pool.starmap(run_client, [(i, client, stdin) for i in range(args.processes)])
assert results
# It's assumed every invocation gives the same output, yikes.
actual_output = list(map(float, results[0].stdout.split('\n\n')[:-1]))
assert len(expected_output) == len(actual_output)
for i in range(len(expected_output)):
if math.isclose(expected_output[i], actual_output[i]):
continue
print(f"Expression: {expressions[i]}")
print(f"Expected output: {expected_output[i]}")
print(f"Actual output: {actual_output[i]}")
def parse_args(argv=None):
if argv is None:
argv = sys.argv[1:]
parser = argparse.ArgumentParser()
parser.add_argument('--host', '-H', metavar='HOST',
default=DEFAULT_HOST,
help='set host')
parser.add_argument('--port', '-p', metavar='PORT',
type=int, default=DEFAULT_PORT,
help='set port')
parser.add_argument('--processes', '-n', metavar='N',
type=int, default=1,
help='set number of processes')
parser.add_argument('--expressions', '-e', metavar='N',
type=int, default=1,
help='set number of expressions')
parser.add_argument('--client', '-c', metavar='PATH',
default=DEFAULT_CLIENT_PATH,
help='set path to client.exe')
args = parser.parse_args(argv)
# Bleh
assert args.processes > 0
assert args.expressions > 0
return args
def main(argv=None):
args = parse_args(argv)
run_stress_test(args)
if __name__ == '__main__':
main()
