How to use execute_next_task method in locust

Best Python code snippet using locust

test_locust_class.py

Source:test_locust_class.py Github

copy

Full Screen

...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...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run locust automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful