Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use execute_next_task method in locust

Best Python code snippet using locust

test_locust_class.py

Source:test_locust_class.py

...76        class MyTasks(TaskSet):77            tasks = [t1, t2]78        taskset = MyTasks(self.locust)79        taskset.schedule_task(t1)80        taskset.execute_next_task()81        self.assertTrue(self.t1_executed)82        taskset.schedule_task(t2, args=["argument to t2"])83        taskset.execute_next_task()84        self.assertEqual("argument to t2", self.t2_arg)85    86    def test_schedule_task_with_kwargs(self):87        class MyTasks(TaskSet):88            @task89            def t1(self):90                self.t1_executed = True91            @task92            def t2(self, *args, **kwargs):93                self.t2_args = args94                self.t2_kwargs = kwargs95        loc = MyTasks(self.locust)96        loc.schedule_task(loc.t2, [42], {"test_kw":"hello"})97        loc.execute_next_task()98        self.assertEqual((42, ), loc.t2_args)99        self.assertEqual({"test_kw":"hello"}, loc.t2_kwargs)100        101        loc.schedule_task(loc.t2, args=[10, 4], kwargs={"arg1":1, "arg2":2})102        loc.execute_next_task()103        self.assertEqual((10, 4), loc.t2_args)104        self.assertEqual({"arg1":1, "arg2":2}, loc.t2_kwargs)105    106    def test_schedule_task_bound_method(self):107        class MyTasks(TaskSet):108            host = ""109            110            @task()111            def t1(self):112                self.t1_executed = True113                self.schedule_task(self.t2)114            def t2(self):115                self.t2_executed = True116        117        taskset = MyTasks(self.locust)118        taskset.schedule_task(taskset.get_next_task())119        taskset.execute_next_task()120        self.assertTrue(taskset.t1_executed)121        taskset.execute_next_task()122        self.assertTrue(taskset.t2_executed)123        124    125    def test_taskset_inheritance(self):126        def t1(l):127            pass128        class MyBaseTaskSet(TaskSet):129            tasks = [t1]130            host = ""131        class MySubTaskSet(MyBaseTaskSet):132            @task133            def t2(self):134                pass135        136        l = MySubTaskSet(self.locust)137        self.assertEqual(2, len(l.tasks))138        self.assertEqual([t1, six.get_unbound_function(MySubTaskSet.t2)], l.tasks)139    140    def test_task_decorator_with_or_without_argument(self):141        class MyTaskSet(TaskSet):142            @task143            def t1(self):144                pass145        taskset = MyTaskSet(self.locust)146        self.assertEqual(len(taskset.tasks), 1)147        148        class MyTaskSet2(TaskSet):149            @task()150            def t1(self):151                pass152        taskset = MyTaskSet2(self.locust)153        self.assertEqual(len(taskset.tasks), 1)154        155        class MyTaskSet3(TaskSet):156            @task(3)157            def t1(self):158                pass159        taskset = MyTaskSet3(self.locust)160        self.assertEqual(len(taskset.tasks), 3)161    162    def test_sub_taskset(self):163        class MySubTaskSet(TaskSet):164            min_wait = 1165            max_wait = 1166            @task()167            def a_task(self):168                self.locust.sub_locust_task_executed = True169                self.interrupt()170            171        class MyTaskSet(TaskSet):172            tasks = [MySubTaskSet]173        174        self.sub_locust_task_executed = False175        loc = MyTaskSet(self.locust)176        loc.schedule_task(loc.get_next_task())177        self.assertRaises(RescheduleTaskImmediately, lambda: loc.execute_next_task())178        self.assertTrue(self.locust.sub_locust_task_executed)179    180    def test_sub_taskset_tasks_decorator(self):181        class MyTaskSet(TaskSet):182            @task183            class MySubTaskSet(TaskSet):184                min_wait = 1185                max_wait = 1186                @task()187                def a_task(self):188                    self.locust.sub_locust_task_executed = True189                    self.interrupt()190        191        self.sub_locust_task_executed = False192        loc = MyTaskSet(self.locust)193        loc.schedule_task(loc.get_next_task())194        self.assertRaises(RescheduleTaskImmediately, lambda: loc.execute_next_task())195        self.assertTrue(self.locust.sub_locust_task_executed)196    197    def test_sub_taskset_arguments(self):198        class MySubTaskSet(TaskSet):199            min_wait = 1200            max_wait = 1201            @task()202            def a_task(self):203                self.locust.sub_taskset_args = self.args204                self.locust.sub_taskset_kwargs = self.kwargs205                self.interrupt()206        class MyTaskSet(TaskSet):207            sub_locust_args = None208            sub_locust_kwargs = None209            tasks = [MySubTaskSet]210        211        self.locust.sub_taskset_args = None212        self.locust.sub_taskset_kwargs = None213        214        loc = MyTaskSet(self.locust)215        loc.schedule_task(MySubTaskSet, args=[1,2,3], kwargs={"hello":"world"})216        self.assertRaises(RescheduleTaskImmediately, lambda: loc.execute_next_task())217        self.assertEqual((1,2,3), self.locust.sub_taskset_args)218        self.assertEqual({"hello":"world"}, self.locust.sub_taskset_kwargs)219    220    def test_interrupt_taskset_in_main_taskset(self):221        class MyTaskSet(TaskSet):222            @task223            def interrupted_task(self):224                raise InterruptTaskSet(reschedule=False)225        class MyLocust(Locust):226            host = "http://127.0.0.1"227            task_set = MyTaskSet228        229        class MyTaskSet2(TaskSet):230            @task...

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.