How to use get_current_response_time_percentile method in locust

Best Python code snippet using locust

locustfile.py

Source:locustfile.py

...30    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:31        #add time delay reduce the rate that the response time is checked and control variables are updated32        time.sleep(1)33        #If the locust runner is reporting values for the response time, i.e. integers34        if isinstance(environment.runner.stats.total.get_current_response_time_percentile(0.5), int) :35            #receive the response time from the locust runner and set to variables36            response_time_median = environment.runner.stats.total.get_current_response_time_percentile(0.5)37            response_time_95perc = environment.runner.stats.total.get_current_response_time_percentile(0.95)38            #If both the median and 95 percentile are above their respective thresholds39            if response_time_median > 1000 and response_time_95perc > 3000 :40                print(f"reducing load on server, both thresholds breached", flush=True)41                #set threshold_ratio to the multiplication of the ratios of response time to the thresholds42                threshold_ratio = (response_time_median/1000)*(response_time_95perc/3000)43                #set to True to indicate VM has been fully loaded44                first_fully_loaded = True45            #Else if just the median is above the threshold,46            elif response_time_median > 1000:47                print(f"reducing load on server, median threshold breached", flush=True)48                #set threshold_ratio to the ratio of response time to the threshold49                threshold_ratio = response_time_median/100050                #set to True to indicate VM has been fully loaded51                first_fully_loaded = True52            #Else if just the 95 percentile is above the threshold,53            elif response_time_95perc > 3000:54                print(f"reducing load on server, 95th percentile threshold breached", flush=True)55                #set threshold_ratio to the ratio of response time to the threshold56                threshold_ratio = response_time_95perc/300057                #set to True to indicate VM has been fully loaded58                first_fully_loaded = True59            #Else if neither the median of 95 percentile are above the thresholds60            else :61                #set threshold_ratio to the ratio of response time to the threshold62                threshold_ratio = response_time_median/100063#funtion to send regular reports of the metrics to Graphite via a socket connection64def send_report(environment):65    #initialise global variables inside function66    global user_count67    global run_time68    global records_start_time69    #create a socket connection and connect using the IP address of the Results Server70    sock = socket.socket()71    sock.connect( ("[RESULTS_IP]", 2003) )72    #Run while the state of the locust master's run state is not stopping, stopped or in post stop cleanup73    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:74        #add time delay to reduce the rate that the reports are sent to Graphite to avoid unnecessary network traffic75        time.sleep(0.5)76        #if the locust load test is running77        if environment.runner.state == STATE_RUNNING or environment.runner.state == STATE_SPAWNING:78            #checks for Nulls and assigns the median response time79            if environment.runner.stats.total.get_current_response_time_percentile(0.5) is None:80                response_time_median = 081            else:82                response_time_median = environment.runner.stats.total.get_current_response_time_percentile(0.5)83            #checks for Nulls and assigns the median response time84            if environment.runner.stats.total.get_current_response_time_percentile(0.95) is None:85                response_time_95perc = 086            else:87                response_time_95perc = environment.runner.stats.total.get_current_response_time_percentile(0.95)88            #checks if the num_reqs_per_sec dictionary in locust has a value for the current time with a 3 second delay to allow the dict to be written89            if round(time.time()-3) in environment.runner.stats.total.num_reqs_per_sec:90                #assigns the requests_per_second from 3 seconds ago91                requests_per_second = environment.runner.stats.total.num_reqs_per_sec[round(time.time()-3)]92            else :93                print("no requests per second recorded for this timeframe")94                requests_per_second = 095            #checks if the num_fail_per_sec dictionary in locust has a value for the current time with a 3 second delay to allow the dict to be written96            if round(time.time()-3) in environment.runner.stats.total.num_fail_per_sec:97                #assigns the failures_per_second from 3 seconds ago98                failures_per_second = environment.runner.stats.total.num_fail_per_sec[round(time.time()-3)]99            else :100                print("no failures per second recorded for this timeframe")101                failures_per_second = 0...

tasks.py

Source:tasks.py

...30    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:31        #add time delay reduce the rate that the response time is checked and control variables are updated32        time.sleep(1)33        #If the locust runner is reporting values for the response time, i.e. integers34        if isinstance(environment.runner.stats.total.get_current_response_time_percentile(0.5), int) :35            #receive the response time from the locust runner and set to variables36            response_time_median = environment.runner.stats.total.get_current_response_time_percentile(0.5)37            response_time_95perc = environment.runner.stats.total.get_current_response_time_percentile(0.95)38            #If both the median and 95 percentile are above their respective thresholds39            if response_time_median > 1000 and response_time_95perc > 3000 :40                print(f"reducing load on server, both thresholds breached", flush=True)41                #set threshold_ratio to the multiplication of the ratios of response time to the thresholds42                threshold_ratio = (response_time_median/1000)*(response_time_95perc/3000)43                #set to True to indicate VM has been fully loaded44                first_fully_loaded = True45            #Else if just the median is above the threshold,46            elif response_time_median > 1000:47                print(f"reducing load on server, median threshold breached", flush=True)48                #set threshold_ratio to the ratio of response time to the threshold49                threshold_ratio = response_time_median/100050                #set to True to indicate VM has been fully loaded51                first_fully_loaded = True52            #Else if just the 95 percentile is above the threshold,53            elif response_time_95perc > 3000:54                print(f"reducing load on server, 95th percentile threshold breached", flush=True)55                #set threshold_ratio to the ratio of response time to the threshold56                threshold_ratio = response_time_95perc/300057                #set to True to indicate VM has been fully loaded58                first_fully_loaded = True59            #Else if neither the median of 95 percentile are above the thresholds60            else :61                #set threshold_ratio to the ratio of response time to the threshold62                threshold_ratio = response_time_median/100063#funtion to send regular reports of the metrics to Graphite via a socket connection64def send_report(environment):65    #initialise global variables inside function66    global user_count67    global run_time68    global records_start_time69    #create a socket connection and connect using the IP address of the Results Server70    sock = socket.socket()71    sock.connect( ("34.71.225.20", 2003) )72    #Run while the state of the locust master's run state is not stopping, stopped or in post stop cleanup73    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:74        #add time delay to reduce the rate that the reports are sent to Graphite to avoid unnecessary network traffic75        time.sleep(0.5)76        #if the locust load test is running77        if environment.runner.state == STATE_RUNNING or environment.runner.state == STATE_SPAWNING:78            #checks for Nulls and assigns the median response time79            if environment.runner.stats.total.get_current_response_time_percentile(0.5) is None:80                response_time_median = 081            else:82                response_time_median = environment.runner.stats.total.get_current_response_time_percentile(0.5)83            #checks for Nulls and assigns the median response time84            if environment.runner.stats.total.get_current_response_time_percentile(0.95) is None:85                response_time_95perc = 086            else:87                response_time_95perc = environment.runner.stats.total.get_current_response_time_percentile(0.95)88            #checks if the num_reqs_per_sec dictionary in locust has a value for the current time with a 3 second delay to allow the dict to be written89            if round(time.time()-3) in environment.runner.stats.total.num_reqs_per_sec:90                #assigns the requests_per_second from 3 seconds ago91                requests_per_second = environment.runner.stats.total.num_reqs_per_sec[round(time.time()-3)]92            else :93                print("no requests per second recorded for this timeframe")94                requests_per_second = 095            #checks if the num_fail_per_sec dictionary in locust has a value for the current time with a 3 second delay to allow the dict to be written96            if round(time.time()-3) in environment.runner.stats.total.num_fail_per_sec:97                #assigns the failures_per_second from 3 seconds ago98                failures_per_second = environment.runner.stats.total.num_fail_per_sec[round(time.time()-3)]99            else :100                print("no failures per second recorded for this timeframe")101                failures_per_second = 0...

test_runners.py

Source:test_runners.py

...147                    "errors":stats2.serialize_errors(),148                    "user_count": 2,149                }, "fake_client"))150                mocked_time.return_value += 4151                self.assertEqual(400, master.stats.total.get_current_response_time_percentile(0.5))152                self.assertEqual(800, master.stats.total.get_current_response_time_percentile(0.95))153                154                # let 10 second pass, do some more requests, send it to the master and make155                # sure the current response time percentiles only accounts for these new requests156                mocked_time.return_value += 10157                stats.log_request("GET", "/1", 20, 1)158                stats.log_request("GET", "/1", 30, 1)159                stats.log_request("GET", "/1", 3000, 1)160                server.mocked_send(Message("stats", {161                    "stats":stats.serialize_stats(),162                    "stats_total": stats.total.get_stripped_report(),163                    "errors":stats.serialize_errors(),164                    "user_count": 2,165                }, "fake_client"))166                self.assertEqual(30, master.stats.total.get_current_response_time_percentile(0.5))167                self.assertEqual(3000, master.stats.total.get_current_response_time_percentile(0.95))168    169    def test_spawn_zero_locusts(self):170        class MyTaskSet(TaskSet):171            @task172            def my_task(self):173                pass174            175        class MyTestLocust(Locust):176            task_set = MyTaskSet177            min_wait = 100178            max_wait = 100179        180        runner = LocalLocustRunner([MyTestLocust], self.options)181        ...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.