KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use current_rps method in locust

Best Python code snippet using locust

test_predicted_team.py

Source:test_predicted_team.py

1from calculations import predicted_team2import server3from unittest import mock4class TestPredictedTeamCalc:5    def setup_method(self, method):6        self.test_server = server.Server()7        self.test_calc = predicted_team.PredictedTeamCalc(self.test_server)8        self.predicted_aim = [9            {10                'match_number': 1,11                'alliance_color_is_red': True,12                'predicted_rp1': 1.0,13                'predicted_rp2': 1.0,14                'predicted_score': 60.4,15            },16            {17                'match_number': 1,18                'alliance_color_is_red': False,19                'predicted_rp1': 0.0,20                'predicted_rp2': 0.0,21                'predicted_score': 56.8,22            },23            {24                'match_number': 2,25                'alliance_color_is_red': True,26                'predicted_rp1': 0.0,27                'predicted_rp2': 1.0,28                'predicted_score': 162.3,29            },30            {31                'match_number': 2,32                'alliance_color_is_red': False,33                'predicted_rp1': 1.0,34                'predicted_rp2': 0.0,35                'predicted_score': 100.2,36            },37            {38                'match_number': 3,39                'alliance_color_is_red': True,40                'predicted_rp1': 1.0,41                'predicted_rp2': 1.0,42                'predicted_score': 82.0,43            },44            {45                'match_number': 3,46                'alliance_color_is_red': False,47                'predicted_rp1': 1.0,48                'predicted_rp2': 0.0,49                'predicted_score': 90.5,50            },51            {52                'match_number': 4,53                'alliance_color_is_red': False,54                'predicted_rp1': 1.0,55                'predicted_rp2': 0.0,56                'predicted_score': 90.5,57            },58        ]59        self.aim_list = [60            {'match_number': 1, 'alliance_color': 'R', 'team_list': [1678, 1533, 7229]},61            {'match_number': 1, 'alliance_color': 'B', 'team_list': [254, 971, 1323]},62            {'match_number': 2, 'alliance_color': 'R', 'team_list': [2056, 1114, 7179]},63            {'match_number': 2, 'alliance_color': 'B', 'team_list': [1678, 971, 7229]},64            {'match_number': 3, 'alliance_color': 'R', 'team_list': [2056, 254, 1323]},65            {'match_number': 3, 'alliance_color': 'B', 'team_list': [1533, 1114, 7179]},66            {'match_number': 4, 'alliance_color': 'R', 'team_list': [4414, 1690, 973]},67        ]68        self.ranking_data = {69            'rankings': [70                {'extra_stats': [26], 'matches_played': 11, 'rank': 1, 'team_key': 'frc1678'},71                {'extra_stats': [25], 'matches_played': 11, 'rank': 2, 'team_key': 'frc1533'},72                {'extra_stats': [24], 'matches_played': 11, 'rank': 3, 'team_key': 'frc7229'},73                {'extra_stats': [23], 'matches_played': 11, 'rank': 4, 'team_key': 'frc254'},74                {'extra_stats': [22], 'matches_played': 11, 'rank': 5, 'team_key': 'frc971'},75                {'extra_stats': [21], 'matches_played': 11, 'rank': 6, 'team_key': 'frc1323'},76                {'extra_stats': [20], 'matches_played': 11, 'rank': 7, 'team_key': 'frc2056'},77                {'extra_stats': [19], 'matches_played': 11, 'rank': 8, 'team_key': 'frc1114'},78                {'extra_stats': [18], 'matches_played': 11, 'rank': 9, 'team_key': 'frc7179'},79            ]80        }81        self.predicted_alliance_rps = {82            1: {'R': 4, 'B': 0},83            2: {'R': 3, 'B': 1},84            3: {'R': 2, 'B': 3},85        }86        self.updates = [87            {'team_number': 1678, 'predicted_rps': 5},88            {'team_number': 1533, 'predicted_rps': 7},89            {'team_number': 7229, 'predicted_rps': 5},90            {'team_number': 254, 'predicted_rps': 2},91            {'team_number': 971, 'predicted_rps': 1},92            {'team_number': 1323, 'predicted_rps': 2},93            {'team_number': 2056, 'predicted_rps': 5},94            {'team_number': 1114, 'predicted_rps': 6},95            {'team_number': 7179, 'predicted_rps': 6},96        ]97        self.expected_results = [98            {99                'team_number': 1678,100                'predicted_rps': 5,101                'predicted_rank': 4,102                'current_rank': 1,103                'current_rps': 26,104                'current_avg_rps': 26 / 11,105            },106            {107                'team_number': 1533,108                'predicted_rps': 7,109                'predicted_rank': 1,110                'current_rank': 2,111                'current_rps': 25,112                'current_avg_rps': 25 / 11,113            },114            {115                'team_number': 7229,116                'predicted_rps': 5,117                'predicted_rank': 5,118                'current_rank': 3,119                'current_rps': 24,120                'current_avg_rps': 24 / 11,121            },122            {123                'team_number': 254,124                'predicted_rps': 2,125                'predicted_rank': 7,126                'current_rank': 4,127                'current_rps': 23,128                'current_avg_rps': 23 / 11,129            },130            {131                'team_number': 971,132                'predicted_rps': 1,133                'predicted_rank': 9,134                'current_rank': 5,135                'current_rps': 22,136                'current_avg_rps': 22 / 11,137            },138            {139                'team_number': 1323,140                'predicted_rps': 2,141                'predicted_rank': 8,142                'current_rank': 6,143                'current_rps': 21,144                'current_avg_rps': 21 / 11,145            },146            {147                'team_number': 2056,148                'predicted_rps': 5,149                'predicted_rank': 6,150                'current_rank': 7,151                'current_rps': 20,152                'current_avg_rps': 20 / 11,153            },154            {155                'team_number': 1114,156                'predicted_rps': 6,157                'predicted_rank': 2,158                'current_rank': 8,159                'current_rps': 19,160                'current_avg_rps': 19 / 11,161            },162            {163                'team_number': 7179,164                'predicted_rps': 6,165                'predicted_rank': 3,166                'current_rank': 9,167                'current_rps': 18,168                'current_avg_rps': 18 / 11,169            },170        ]171        self.teams = [1678, 1533, 7229, 254, 971, 1323, 2056, 1114, 7179]172    def test_calculate_current_values(self):173        current_values = self.test_calc.calculate_current_values(174            self.ranking_data['rankings'], 1678175        )176        assert current_values['current_rank'] == 1177        assert current_values['current_rps'] == 26178        assert current_values['current_avg_rps'] == 26 / 11179    def test_calculate_predicted_alliance_rps(self):180        with mock.patch('utils.log_warning') as log_mock:181            predicted_alliance_rps = self.test_calc.calculate_predicted_alliance_rps(182                self.predicted_aim183            )184            log_mock.assert_called_with('Incomplete AIM data for Match 4')185        assert predicted_alliance_rps[1]['R'] == 4186        assert predicted_alliance_rps[1]['B'] == 0187        assert predicted_alliance_rps[2]['R'] == 3188        assert predicted_alliance_rps[2]['B'] == 1189        assert predicted_alliance_rps[3]['R'] == 2190        assert predicted_alliance_rps[3]['B'] == 3191    def test_predicted_team_rps(self):192        assert self.test_calc.calculate_predicted_team_rps(193            1678, self.aim_list, self.predicted_alliance_rps) == 5194        assert self.test_calc.calculate_predicted_team_rps(195            1533, self.aim_list, self.predicted_alliance_rps) == 7196        assert self.test_calc.calculate_predicted_team_rps(197            7229, self.aim_list, self.predicted_alliance_rps) == 5198        assert self.test_calc.calculate_predicted_team_rps(199            254, self.aim_list, self.predicted_alliance_rps) == 2200        assert self.test_calc.calculate_predicted_team_rps(201            971, self.aim_list, self.predicted_alliance_rps) == 1202        assert self.test_calc.calculate_predicted_team_rps(203            1323, self.aim_list, self.predicted_alliance_rps) == 2204        assert self.test_calc.calculate_predicted_team_rps(205            2056, self.aim_list, self.predicted_alliance_rps) == 5206        assert self.test_calc.calculate_predicted_team_rps(207            1114, self.aim_list, self.predicted_alliance_rps) == 6208        assert self.test_calc.calculate_predicted_team_rps(209            7179, self.aim_list, self.predicted_alliance_rps) == 6210    211        with mock.patch('utils.log_warning') as log_mock:212            self.test_calc.calculate_predicted_team_rps(213                4414, self.aim_list, self.predicted_alliance_rps214            )215            log_mock.assert_called_with('Missing predicted RPs for Match 4')216    def test_calculate_predicted_ranks(self):217        updates = self.test_calc.calculate_predicted_ranks(self.updates, self.aim_list)218        for update in updates:219            for result in self.expected_results:220                if update['team_number'] == result['team_number']:221                    assert update['predicted_rank'] == result['predicted_rank']222    def test_update_predicted_team(self):223        with mock.patch(224            'data_transfer.tba_communicator.tba_request', return_value=self.ranking_data225        ), mock.patch(226            'calculations.predicted_team.PredictedTeamCalc._get_aim_list',227            return_value=self.aim_list,228        ), mock.patch(229            'calculations.predicted_team.PredictedTeamCalc._get_teams_list', return_value=self.teams230        ):231            assert self.test_calc.update_predicted_team(self.predicted_aim) == self.expected_results232    def test_run(self):233        self.test_server.db.insert_documents('predicted_aim', self.predicted_aim)234        with mock.patch(235            'data_transfer.tba_communicator.tba_request', return_value=self.ranking_data236        ), mock.patch(237            'calculations.predicted_team.PredictedTeamCalc._get_aim_list',238            return_value=self.aim_list,239        ), mock.patch(240            'calculations.predicted_team.PredictedTeamCalc._get_teams_list', return_value=self.teams241        ):242            self.test_calc.run()243        result = self.test_server.db.find('predicted_team')244        assert len(result) == 9245        for document in result:246            del document['_id']247            assert document in self.expected_results248            # Removes the matching expected result to protect against duplicates from the calculation...

demo_scenario.py

Source:demo_scenario.py

1import locust2import os, sys, random3from locust import TaskSet, Locust, HttpLocust, task, events, runners4PACKAGE_PARENT = '..'5SCRIPT_DIR = os.path.dirname(os.path.realpath(os.path.join(os.getcwd(), os.path.expanduser(__file__))))6sys.path.append(os.path.normpath(os.path.join(SCRIPT_DIR, PACKAGE_PARENT)))7sys.path.append(os.path.normpath(os.path.join(SCRIPT_DIR, "../..")))8from hooks.listeners import *9cwd = os.getcwd()10json_path = "{}/test_data".format(cwd)11base_uri = "http://localhost:3000"12# Add listeners for Stress Tests quiting13events.request_success += my_response_time_handler14events.request_failure += my_error_handler15events.request_success += my_requests_number_handler16events.quitting += validate_results17# RPS listeners WORKING ONLY WITH ONE NODE and MASTER mode18# events.report_to_master += on_report_to_master19# events.slave_report += on_slave_report20# events.slave_report += on_slave_report_latency_handler21class UserScenario(TaskSet):22    uuid = 123    post_id = 124    def setup(self):25        print("Setup Data!")26        self.uuid = self.client.post("/users", open("{}/user.json".format(json_path))).json()['id']27        self.post_id = self.client.post("/posts", {"title": 'foo', "body": 'bar', "userId": self.uuid}).json()['id']28    def teardown(self):29        self.client.delete("/users/{}".format(self.uuid))30        self.client.delete('/posts/{}'.format(self.client.get("/posts").json()[-1]['id']))31    @task(3)32    def get_publications(self):33        self.client.get("/posts")34    @task(2)35    def get_post(self):36        self.client.get("/posts/1")37    @task(1)38    def create_post(self):39        self.client.post("/posts", data={"title": 'foo', "body": 'bar', "userId": self.uuid})40    @task(1)41    def update_post(self):42        self.client.put("/posts/{}".format(self.post_id),43                        {"id": self.post_id, "title": 'foo', "body": 'bar', "userId": self.uuid})44    @task(1)45    def patch_post(self):46        self.client.patch("/posts/{}".format(self.post_id), {"body": "bar."})47class LoadTests(HttpLocust):48    host = base_uri49    task_set = UserScenario50    # min_wait = 100051    # max_wait = 100052    wait_function = lambda self: self.fixed_rps_wait_function(500)53    # wait_function = lambda t: 900 if runners.global_stats.total.current_rps < 100 else 110054    def __init__(self):55        super(LoadTests, self).__init__()56        self.my_wait = 100057    def fixed_rps_wait_function(self, desired_rps):58        # Will increase and decrease tasks wait time in range of 99.8 - 100.7 rps59        current_rps = runners.global_stats.total.current_rps60        if current_rps < desired_rps - 0.2:61            # the minimum wait is 10 ms62            if self.my_wait > 10:63                self.my_wait -= 1064        elif current_rps > desired_rps + 0.7:65            self.my_wait += 1066        # print("Current RPS: {}".format(current_rps))67        # print("Default wait is: {}".format(self.my_wait))...

rps.py

Source:rps.py

1from pyalgotrade import technical2from pyalgotrade import dataseries3import numpy as np4class RPSEventWindow(technical.EventWindow):5    def __init__(self, period, ma_period, changeToReturn):6        assert(period > 0)7        technical.EventWindow.__init__(self, period)8        self.__value = None9        self.__ma_value = None10        self.__period = period11        self.__ma_period = ma_period12        self.__value_list_for_ma = []13        self.__changeToReturn = changeToReturn14    def onNewValue(self, dateTime, value):15        technical.EventWindow.onNewValue(self, dateTime, value)16        if self.windowFull():17             if self.__changeToReturn:18                current_return = np.log(value /self.getValues()[-2])19                high_return = np.max(np.diff(np.log(self.getValues())))20                low_return = np.min(np.diff(np.log(self.getValues())))21                current_rps = (current_return - low_return)/(high_return - low_return)22             else:23                current_rps = (value - np.min(self.getValues())) /(np.max(self.getValues()) - np.min(self.getValues()))24             if self.__ma_period is not None:25                self.__value_list_for_ma.append(current_rps)26                self.__value = np.mean(self.__value_list_for_ma[-self.__ma_period:]) if len(self.__value_list_for_ma) >= self.__ma_period else None27             else:28                self.__value = current_rps29    def getValue(self):30        return self.__value31class RPS(technical.EventBasedFilter):32    def __init__(self, dataSeries, period=250, ma_period = 10,changeToReturn=False, maxLen=dataseries.DEFAULT_MAX_LEN):...

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.