Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use mocked_send method in locust

Best Python code snippet using locust

test_runners.py

Source:test_runners.py

...18        def __init__(self, host, port):19            pass20        21        @classmethod22        def mocked_send(cls, message):23            cls.queue.put(message.serialize())24            sleep(0)25        26        def recv(self):27            results = self.queue.get()28            return Message.unserialize(results)29        30        def send(self, message):31            self.outbox.append(message.serialize())32    33    return MockedRpcServer34class TestMasterRunner(LocustTestCase):35    def setUp(self):36        global_stats.reset_all()37        self._slave_report_event_handlers = [h for h in events.slave_report._handlers]38        parser, _, _ = parse_options()39        args = [40            "--clients", "10",41            "--hatch-rate", "10"42        ]43        opts, _ = parser.parse_args(args)44        self.options = opts45        46    def tearDown(self):47        events.slave_report._handlers = self._slave_report_event_handlers48    49    def test_slave_connect(self):50        import mock51        52        class MyTestLocust(Locust):53            pass54        55        with mock.patch("locust.rpc.rpc.Server", mocked_rpc_server()) as server:56            master = MasterLocustRunner(MyTestLocust, self.options)57            server.mocked_send(Message("client_ready", None, "zeh_fake_client1"))58            self.assertEqual(1, len(master.clients))59            self.assertTrue("zeh_fake_client1" in master.clients, "Could not find fake client in master instance's clients dict")60            server.mocked_send(Message("client_ready", None, "zeh_fake_client2"))61            server.mocked_send(Message("client_ready", None, "zeh_fake_client3"))62            server.mocked_send(Message("client_ready", None, "zeh_fake_client4"))63            self.assertEqual(4, len(master.clients))64            65            server.mocked_send(Message("quit", None, "zeh_fake_client3"))66            self.assertEqual(3, len(master.clients))67    68    def test_slave_stats_report_median(self):69        import mock70        71        class MyTestLocust(Locust):72            pass73        74        with mock.patch("locust.rpc.rpc.Server", mocked_rpc_server()) as server:75            master = MasterLocustRunner(MyTestLocust, self.options)76            server.mocked_send(Message("client_ready", None, "fake_client"))77            78            master.stats.get("/", "GET").log(100, 23455)79            master.stats.get("/", "GET").log(800, 23455)80            master.stats.get("/", "GET").log(700, 23455)81            82            data = {"user_count":1}83            events.report_to_master.fire(client_id="fake_client", data=data)84            master.stats.clear_all()85            86            server.mocked_send(Message("stats", data, "fake_client"))87            s = master.stats.get("/", "GET")88            self.assertEqual(700, s.median_response_time)89    90    def test_master_total_stats(self):91        import mock92        93        class MyTestLocust(Locust):94            pass95        96        with mock.patch("locust.rpc.rpc.Server", mocked_rpc_server()) as server:97            master = MasterLocustRunner(MyTestLocust, self.options)98            server.mocked_send(Message("client_ready", None, "fake_client"))99            stats = RequestStats()100            stats.log_request("GET", "/1", 100, 3546)101            stats.log_request("GET", "/1", 800, 56743)102            stats2 = RequestStats()103            stats2.log_request("GET", "/2", 700, 2201)104            server.mocked_send(Message("stats", {105                "stats":stats.serialize_stats(), 106                "stats_total": stats.total.serialize(),107                "errors":stats.serialize_errors(),108                "user_count": 1,109            }, "fake_client"))110            server.mocked_send(Message("stats", {111                "stats":stats2.serialize_stats(), 112                "stats_total": stats2.total.serialize(),113                "errors":stats2.serialize_errors(),114                "user_count": 2,115            }, "fake_client"))116            self.assertEqual(700, master.stats.total.median_response_time)117    118    def test_master_current_response_times(self):119        import mock120        121        class MyTestLocust(Locust):122            pass123        124        start_time = 1125        with mock.patch("time.time") as mocked_time:126            mocked_time.return_value = start_time127            global_stats.reset_all()128            with mock.patch("locust.rpc.rpc.Server", mocked_rpc_server()) as server:129                master = MasterLocustRunner(MyTestLocust, self.options)130                mocked_time.return_value += 1131                server.mocked_send(Message("client_ready", None, "fake_client"))132                stats = RequestStats()133                stats.log_request("GET", "/1", 100, 3546)134                stats.log_request("GET", "/1", 800, 56743)135                server.mocked_send(Message("stats", {136                    "stats":stats.serialize_stats(),137                    "stats_total": stats.total.get_stripped_report(),138                    "errors":stats.serialize_errors(),139                    "user_count": 1,140                }, "fake_client"))141                mocked_time.return_value += 1142                stats2 = RequestStats()143                stats2.log_request("GET", "/2", 400, 2201)144                server.mocked_send(Message("stats", {145                    "stats":stats2.serialize_stats(),146                    "stats_total": stats2.total.get_stripped_report(),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        182        timeout = gevent.Timeout(2.0)183        timeout.start()184        185        try:186            runner.start_hatching(0, 1, wait=True)187            runner.greenlet.join()188        except gevent.Timeout:189            self.fail("Got Timeout exception. A locust seems to have been spawned, even though 0 was specified.")190        finally:191            timeout.cancel()192    193    def test_spawn_uneven_locusts(self):194        """195        Tests that we can accurately spawn a certain number of locusts, even if it's not an 196        even number of the connected slaves197        """198        import mock199        200        class MyTestLocust(Locust):201            pass202        203        with mock.patch("locust.rpc.rpc.Server", mocked_rpc_server()) as server:204            master = MasterLocustRunner(MyTestLocust, self.options)205            for i in range(5):206                server.mocked_send(Message("client_ready", None, "fake_client%i" % i))207            208            master.start_hatching(7, 7)209            self.assertEqual(5, len(server.outbox))210            211            num_clients = 0212            for msg in server.outbox:213                num_clients += Message.unserialize(msg).data["num_clients"]214            215            self.assertEqual(7, num_clients, "Total number of locusts that would have been spawned is not 7")216    217    def test_spawn_fewer_locusts_than_slaves(self):218        import mock219        220        class MyTestLocust(Locust):221            pass222        223        with mock.patch("locust.rpc.rpc.Server", mocked_rpc_server()) as server:224            master = MasterLocustRunner(MyTestLocust, self.options)225            for i in range(5):226                server.mocked_send(Message("client_ready", None, "fake_client%i" % i))227            228            master.start_hatching(2, 2)229            self.assertEqual(5, len(server.outbox))230            231            num_clients = 0232            for msg in server.outbox:233                num_clients += Message.unserialize(msg).data["num_clients"]234            235            self.assertEqual(2, num_clients, "Total number of locusts that would have been spawned is not 2")236    237    def test_exception_in_task(self):238        class HeyAnException(Exception):239            pass240        ...

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.