multithreadTest.py
def runTest(self):
        """STRESS TEST - Random Request Combination"""
        sample = random.sample(range(n_clientThreads), 5)
        s = sum(sample)
        numRequests = [(i*n_clientThreads)//s for i in sample]
        # The number of client threads to create
        getThreads = numRequests[0]
        headThreads = numRequests[1]
        postThreads = numRequests[2]
        putThreads = numRequests[3]
        deleteThreads = numRequests[4]
        print(f"\nDispatching {getThreads} GET, {headThreads} HEAD, {postThreads} POST, {putThreads} PUT, {deleteThreads} DELETE request threads")
        request_threads = []
        try:
            def get_test():
                try:
                    r = requests.get(SERVER_URL + "/")
                    # print(f"Status : {r.status_code} {r.reason}")
                except:
                    print("Error in making request, maybe server queue is full")
            def head_test():
                try:
                    r = requests.head(SERVER_URL + "/")
                    # print(f"Status : {r.status_code} {r.reason}")
                except:
                    print("Error in making request, maybe server queue is full")
            def post_test():
                data = dict(
                    key1='TEST',
                    value1='TEST DATA'
                )
                try:
                    r = requests.post(SERVER_URL + "/test",
                        data=json.dumps(data),
                        headers={'content-type': 'application/json'}
                    )
                    # print(f"Status : {r.status_code} {r.reason}")
                except:
                    print("Error in making request, maybe server queue is full")
            def put_test(fileno):
                data = dict(
                    key1='TEST',
                    value1='TEST DATA'
                )
                try:
                    r = requests.put(SERVER_URL + f"/test/test{fileno}.json",
                        data=json.dumps(data),
                        headers={'content-type': 'application/json'}
                    )
                    # print(f"Status : {r.status_code} {r.reason}")
                except:
                    print("Error in making request, maybe server queue is full")
            def delete_test(fileno):
                data = dict(
                    key1='TEST',
                    value1='TEST DATA'
                )
                try:
                    r = requests.delete(SERVER_URL + f"/test/test{fileno}.json")
                    # print(f"Status : {r.status_code} {r.reason}")
                except:
                    print("Error in making request, maybe server queue is full")

            # Create threads for all of the requests and start them
            for i in range(getThreads):
                t = threading.Thread(target=get_test)
                request_threads.append(t)
                t.start()
            # for i in range(headThreads):
            #     t = threading.Thread(target=head_test)
            #     request_threads.append(t)
            #     t.start()
            for i in range(postThreads):
                t = threading.Thread(target=post_test)
                request_threads.append(t)
                t.start()
            for i in range(putThreads):
                t = threading.Thread(target=put_test, args=(i+1,))
                request_threads.append(t)
                t.start()
            for i in range(deleteThreads):
                t = threading.Thread(target=delete_test, args=(i+1,))
                request_threads.append(t)
                t.start()

            # Wait until all of the threads are complete
            for thread in request_threads:
                thread.join()
            
        except Exception as ex:
            print('Something went horribly wrong!', ex)
        finally:
            # Stop all running threads
            return
STRESS TEST - Random Request Combination
STRESS TEST - Random Request Combination
STRESS TEST - Random Request Combination
def runTest(self):
        sample = random.sample(range(n_clientThreads), 5)
        s = sum(sample)
        numRequests = [(i*n_clientThreads)//s for i in sample]
        getThreads = numRequests[0]
        headThreads = numRequests[1]
        postThreads = numRequests[2]
        putThreads = numRequests[3]
        deleteThreads = numRequests[4]
        print(f"\nDispatching {getThreads} GET, {headThreads} HEAD, {postThreads} POST, {putThreads} PUT, {deleteThreads} DELETE request threads")
        request_threads = []
        try:
            def get_test():
                try:
                    r = requests.get(SERVER_URL + "/")
                except:
                    print("Error in making request, maybe server queue is full")
            def head_test():
                try:
                    r = requests.head(SERVER_URL + "/")
                except:
                    print("Error in making request, maybe server queue is full")
            def post_test():
                data = dict(
                    key1='TEST',
                    value1='TEST DATA'
                )
                try:
                    r = requests.post(SERVER_URL + "/test",
                        data=json.dumps(data),
                        headers={'content-type': 'application/json'}
                    )
                except:
                    print("Error in making request, maybe server queue is full")
            def put_test(fileno):
                data = dict(
                    key1='TEST',
                    value1='TEST DATA'
                )
                try:
                    r = requests.put(SERVER_URL + f"/test/test{fileno}.json",
                        data=json.dumps(data),
                        headers={'content-type': 'application/json'}
                    )
                except:
                    print("Error in making request, maybe server queue is full")
            def delete_test(fileno):
                data = dict(
                    key1='TEST',
                    value1='TEST DATA'
                )
                try:
                    r = requests.delete(SERVER_URL + f"/test/test{fileno}.json")
                except:
                    print("Error in making request, maybe server queue is full")
            for i in range(getThreads):
                t = threading.Thread(target=get_test)
                request_threads.append(t)
                t.start()
            for i in range(postThreads):
                t = threading.Thread(target=post_test)
                request_threads.append(t)
                t.start()
            for i in range(putThreads):
                t = threading.Thread(target=put_test, args=(i+1,))
                request_threads.append(t)
                t.start()
            for i in range(deleteThreads):
                t = threading.Thread(target=delete_test, args=(i+1,))
                request_threads.append(t)
                t.start()
            for thread in request_threads:
                thread.join()
        except Exception as ex:
            print('Something went horribly wrong!', ex)
        finally:
            return
def runTest ( self ) : sample = random . sample ( range ( n_clientThreads ) , 5 ) s = sum ( sample ) numRequests = [ ( i * n_clientThreads ) // s for i in sample ] getThreads = numRequests [ 0 ] headThreads = numRequests [ 1 ] postThreads = numRequests [ 2 ] putThreads = numRequests [ 3 ] deleteThreads = numRequests [ 4 ] print ( f"\nDispatching {getThreads} GET, {headThreads} HEAD, {postThreads} POST, {putThreads} PUT, {deleteThreads} DELETE request threads" ) request_threads = [ ] try : def get_test ( ) : try : r = requests . get ( SERVER_URL + "/" ) except : print ( "Error in making request, maybe server queue is full" ) def head_test ( ) : try : r = requests . head ( SERVER_URL + "/" ) except : print ( "Error in making request, maybe server queue is full" ) def post_test ( ) : data = dict ( key1 = 'TEST' , value1 = 'TEST DATA' ) try : r = requests . post ( SERVER_URL + "/test" , data = json . dumps ( data ) , headers = { 'content-type' : 'application/json' } ) except : print ( "Error in making request, maybe server queue is full" ) def put_test ( fileno ) : data = dict ( key1 = 'TEST' , value1 = 'TEST DATA' ) try : r = requests . put ( SERVER_URL + f"/test/test{fileno}.json" , data = json . dumps ( data ) , headers = { 'content-type' : 'application/json' } ) except : print ( "Error in making request, maybe server queue is full" ) def delete_test ( fileno ) : data = dict ( key1 = 'TEST' , value1 = 'TEST DATA' ) try : r = requests . delete ( SERVER_URL + f"/test/test{fileno}.json" ) except : print ( "Error in making request, maybe server queue is full" ) for i in range ( getThreads ) : t = threading . Thread ( target = get_test ) request_threads . append ( t ) t . start ( ) for i in range ( postThreads ) : t = threading . Thread ( target = post_test ) request_threads . append ( t ) t . start ( ) for i in range ( putThreads ) : t = threading . Thread ( target = put_test , args = ( i + 1 , ) ) request_threads . append ( t ) t . start ( ) for i in range ( deleteThreads ) : t = threading . Thread ( target = delete_test , args = ( i + 1 , ) ) request_threads . append ( t ) t . start ( ) for thread in request_threads : thread . join ( ) except Exception as ex : print ( 'Something went horribly wrong!' , ex ) finally : return
STRESS TEST - Random Request Combination
STRESS TEST - Random Request Combination
"""STRESS TEST - Random Request Combination""" # The number of client threads to create # print(f"Status : {r.status_code} {r.reason}") # print(f"Status : {r.status_code} {r.reason}") # print(f"Status : {r.status_code} {r.reason}") # print(f"Status : {r.status_code} {r.reason}") # print(f"Status : {r.status_code} {r.reason}") # Create threads for all of the requests and start them # for i in range(headThreads): #     t = threading.Thread(target=head_test) #     request_threads.append(t) #     t.start() # Wait until all of the threads are complete # Stop all running threads
self