Python load test exmaple

import datetime

import pytz

import threading

 

def functionToTest():

  # raise Exception("D")

  pass

 

class MyThread(threading.Thread):

  threadNum = None

  numCallsPerThread = None

  numErrors = None

  def __init__(self, threadNum, numCallsPerThread):

    super(MyThread, self).__init__()

    self.threadNum = threadNum

    self.numCallsPerThread = numCallsPerThread

    self.numErrors = 0

 

  def runSingle(self):

    try:

      functionToTest()

    except Exception as err:

      print("Error in thread " + str(self.threadNum) + "\n", err) # for the repr

      # print(str(err)) # for just the message

      # print(err.args) # the arguments that the exception has been called with.

      self.numErrors = self.numErrors + 1

 

  def run(self):

    for curTry in range(0, self.numCallsPerThread):

      self.runSingle()

 

print("Start")

 

numThreadsSTR = input("How many threads?")

numThreads = int(numThreadsSTR)

numCallsPerThreadSTR = input("How many calls in each thread?")

numCallsPerThread = int(numCallsPerThreadSTR)

 

starttime = datetime.datetime.now(pytz.timezone("UTC"))

 

threads = []

for threadNum in range(1, numThreads+1):

  threads.append(MyThread(

    threadNum=threadNum,

    numCallsPerThread=numCallsPerThread

  ))

 

print("Starting all the threads")

for curThread in threads:

  curThread.start()

 

for curThread in threads:

  curThread.join()

 

# Add up the number of errors

numErrors = 0

for curThread in threads:

  numErrors += curThread.numErrors

 

endtime = datetime.datetime.now(pytz.timezone("UTC"))

 

print("---------------------------------------------------------")

print("Treads:", numThreads)

print("Calls per thread:", numCallsPerThread)

print("Errors:", numErrors)

print("Duration (seconds):", '{:6.1f}'.format((endtime-starttime).total_seconds()))

print("End")