Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use spawn_users method in locust

Best Python code snippet using locust

test_spawn_link_network.py

Source:test_spawn_link_network.py

...12    root_output = await self.root_io_tree(13        machine=self.machine_ids[0], dataset_program_config=messages.data.demo_dataset_program_config())14    self.root_output = root_output15    await self.demo.run_for(ms=200)16    await self.spawn_users(root_output, n_users=n_output_leaves)17    await self.spawn_users(18        root_input,19        n_users=n_input_leaves,20        add_user=True,21        send_after=4000,22        ave_inter_message_time_ms=500,23        send_messages_for_ms=3000)24    # Need to wait for the new users to be fully connected.25    await self.demo.run_for(ms=1000)26    self.root_link = self.demo.connect_trees_with_sum_network(root_input, root_output, machine=self.machine_ids[0])27    await self.demo.run_for(ms=8000)28    if n_input_leaves > 0:29      assert self.demo.total_simulated_amount > 10 # Smoke test that sends were in fact simulated30    else:31      assert self.demo.total_simulated_amount == 032    output_leaves = self.demo.get_leaves(root_output)33    self.output_leaves = output_leaves34    assert len(output_leaves) == n_output_leaves35    self.demo.system.get_senders(root_output)36    self.demo.system.get_receivers(root_input)37    for leaf in output_leaves:38      assert self.demo.total_simulated_amount == self.demo.system.get_output_state(leaf)39  @pytest.mark.asyncio40  async def test_dns(self, cloud_demo):41    self.demo = cloud_demo42    self.machine_ids = await self.demo.new_machine_controllers(43        1, base_config=self.base_config(), random_seed='test_dns')44    root_input = await self.root_io_tree(45        machine=self.machine_ids[0], dataset_program_config=messages.data.demo_dataset_program_config())46    await self.demo.run_for(ms=6000)47    await self.spawn_users(root_input, n_users=2)48    domain_name = 'www.distzerotesting.com'49    self.demo.system.route_dns(root_input, domain_name)50    await self.demo.run_for(ms=2000)51  @pytest.mark.asyncio52  async def test_spawn_small_small(self, demo):53    self.demo = demo54    self.machine_ids = await demo.new_machine_controllers(55        1, base_config=self.base_config(), random_seed='test_spawn_small_small')56    await self._connect_and_test_io_trees(n_input_leaves=1, n_output_leaves=1)57  @pytest.mark.asyncio58  async def test_spawn_small_large(self, demo):59    self.demo = demo60    self.machine_ids = await demo.new_machine_controllers(61        4, base_config=self.base_config(), random_seed='test_spawn_small_large')62    await self._connect_and_test_io_trees(n_input_leaves=1, n_output_leaves=10)63  @pytest.mark.asyncio64  async def test_spawn_large_small(self, demo):65    self.demo = demo66    self.machine_ids = await demo.new_machine_controllers(67        1, base_config=self.base_config(), random_seed='test_spawn_large_small')68    await self._connect_and_test_io_trees(n_input_leaves=10, n_output_leaves=1)69  @pytest.mark.asyncio70  async def test_spawn_large_large(self, demo):71    self.demo = demo72    self.machine_ids = await demo.new_machine_controllers(73        1, base_config=self.base_config(), random_seed='test_spawn_large_large')74    await self._connect_and_test_io_trees(n_input_leaves=10, n_output_leaves=10)75  @pytest.mark.asyncio76  async def test_spawn_empty_small(self, demo):77    self.demo = demo78    self.machine_ids = await demo.new_machine_controllers(79        1, base_config=self.base_config(), random_seed='test_spawn_large_large')80    await self._connect_and_test_io_trees(n_input_leaves=0, n_output_leaves=3)81  @pytest.mark.asyncio82  async def test_spawn_small_empty(self, demo):83    self.demo = demo84    self.machine_ids = await demo.new_machine_controllers(85        1, base_config=self.base_config(), random_seed='test_spawn_large_large')86    await self._connect_and_test_io_trees(n_input_leaves=3, n_output_leaves=0)87  @pytest.mark.asyncio88  async def test_spawn_small_very_large(self, demo):89    self.demo = demo90    self.machine_ids = await demo.new_machine_controllers(91        1, base_config=self.base_config(), random_seed='test_spawn_small_very_large')92    await self._connect_and_test_io_trees(n_input_leaves=1, n_output_leaves=20)93  @pytest.mark.parametrize(94      'name,start_inputs,start_outputs,new_inputs,new_outputs,ending_input_height,ending_output_height,sender_limit',95      [96          ('grow_input', 2, 2, 5, 0, 2, 2, 10), # Just add inputs97          ('grow_output', 2, 1, 0, 5, 2, 2, 10), # Just add outputs98          ('bump_input_once', 2, 2, 10, 0, 3, 2, 10), # Add enough inputs that the tree bumps its height99          ('bump_input_twice', 2, 2, 29, 0, 4, 2, 10), # Add enough inputs that the tree bumps its height twice100          # FIXME(KK): Test and handle hourglass scenarios101          #('cause_hourglass', 2, 2, 15, 0, 3, 2, 4), # Restrict sender_limit to cause hourglass operations102      ])103  @pytest.mark.asyncio104  async def test_grow_trees(self, demo, name, start_inputs, start_outputs, new_inputs, new_outputs, ending_input_height,105                            ending_output_height, sender_limit):106    self.demo = demo107    config = self.base_config()108    # Make sure not to cause any hourglasses109    config['system_config']['SUM_NODE_SENDER_LIMIT'] = sender_limit110    self.machine_ids = await demo.new_machine_controllers(111        1, base_config=config, random_seed='test_add_leaves_after_spawn')112    await self._connect_and_test_io_trees(n_input_leaves=start_inputs, n_output_leaves=start_outputs)113    await demo.run_for(ms=7000)114    await self.spawn_users(self.root_output, n_users=new_outputs)115    await demo.run_for(ms=1000)116    await self.spawn_users(117        self.root_input,118        n_users=new_inputs,119        add_user=True,120        send_after=0,121        ave_inter_message_time_ms=500,122        send_messages_for_ms=3000)123    await demo.run_for(ms=5000)124    self.output_leaves = self.demo.get_leaves(self.root_output)125    self.input_leaves = self.demo.get_leaves(self.root_input)126    assert start_outputs + new_outputs == len(self.output_leaves)127    await demo.run_for(ms=8000)128    for leaf in self.output_leaves:129      assert self.demo.total_simulated_amount == self.demo.system.get_output_state(leaf)130      senders = self.demo.system.get_senders(leaf)...

locust_use_as_lib.py

Source:locust_use_as_lib.py

...5from locust.log import setup_logging6from locust.stats import stats_printer, stats_history, StatsCSVFileWriter7from .locust_sequential_tasks import UserGlobal8global_user = 09def spawn_users(env):10    global global_user11    while True:12        env.runner.start(env.runner.user_count + 1, spawn_rate=50)13        global_user += 114        next_arrival = random.expovariate(1 / 50)15        gevent.sleep(next_arrival / 1000.0)16def start_locust_master():17    setup_logging("INFO", None)18    logging.info('*******************')19    logging.info('{}'.format("Starting Manager"))20    logging.info('*******************')21    env = Environment(user_classes=[UserGlobal], host="http://192.168.2.12:32677")22    env.create_local_runner()23    # env.create_master_runner(master_bind_host="192.168.2.12", master_bind_port=5557)...

events_demo.py

Source:events_demo.py

1from locust import HttpUser, task, constant, SequentialTaskSet2from locust import events3import logging4@events.spawning_complete.add_listener5def spawn_users(user_count, **kwargs):6    print("Spawned ... ", user_count, " users.")7@events.request_success.add_listener8def send_notification(**kwargs):9    print("Sending the success notification")10@events.request_failure.add_listener11def send_notification(**kwargs):12    print("Sending the failed notification")13@events.quitting.add_listener14def sla(environment, **kwargs):15    if environment.stats.total.fail_ratio > 0.01:16        logging.error("Test failed due to failure ratio > 0.01%")17        environment.process_exit_code = 118        print(environment.process_exit_code)19    else:...

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.