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

How to use stop_and_optionally_quit method in locust

Best Python code snippet using locust

main.py

Source:main.py

...264    environment.events.init.fire(environment=environment, runner=runner, web_ui=web_ui)265    if web_ui:266        web_ui.start()267        main_greenlet = web_ui.greenlet268    def stop_and_optionally_quit():269        if options.autostart:270            logger.info("--run-time limit reached, stopping test")271            runner.stop()272            if options.autoquit != -1:273                logger.debug(f"Autoquit time limit set to {options.autoquit} seconds")274                time.sleep(options.autoquit)275                logger.info("--autoquit time reached, shutting down")276                runner.quit()277                if web_ui:278                    web_ui.stop()279            else:280                logger.info("--autoquit not specified, leaving web ui running indefinitely")281        else:  # --headless run282            logger.info("--run-time limit reached. Stopping Locust")283            runner.quit()284    def spawn_run_time_quit_greenlet():285        gevent.spawn_later(options.run_time, stop_and_optionally_quit).link_exception(greenlet_exception_handler)286    headless_master_greenlet = None287    stats_printer_greenlet = None288    if not options.only_summary and (options.print_stats or (options.headless and not options.worker)):289        # spawn stats printing greenlet290        stats_printer_greenlet = gevent.spawn(stats_printer(runner.stats))291        stats_printer_greenlet.link_exception(greenlet_exception_handler)292    gevent.spawn(stats_history, runner)293    def start_automatic_run():294        if options.master:295            # wait for worker nodes to connect296            start_time = time.monotonic()297            while len(runner.clients.ready) < options.expect_workers:298                if options.expect_workers_max_wait and options.expect_workers_max_wait < time.monotonic() - start_time:299                    logger.error("Gave up waiting for workers to connect.")300                    runner.quit()301                    sys.exit(1)302                logging.info(303                    "Waiting for workers to be ready, %s of %s connected",304                    len(runner.clients.ready),305                    options.expect_workers,306                )307                # TODO: Handle KeyboardInterrupt and send quit signal to workers that are started.308                #       Right now, if the user sends a ctrl+c, the master will not gracefully309                #       shutdown resulting in all the already started workers to stay active.310                time.sleep(1)311        if not options.worker:312            # apply headless mode defaults313            if options.num_users is None:314                options.num_users = 1315            if options.spawn_rate is None:316                options.spawn_rate = 1317            # start the test318            if environment.shape_class:319                if options.run_time:320                    sys.stderr.write("It makes no sense to combine --run-time and LoadShapes. Bailing out.\n")321                    sys.exit(1)322                environment.runner.start_shape()323                environment.runner.shape_greenlet.join()324                stop_and_optionally_quit()325            else:326                headless_master_greenlet = gevent.spawn(runner.start, options.num_users, options.spawn_rate)327                headless_master_greenlet.link_exception(greenlet_exception_handler)328        if options.run_time:329            logger.info(f"Run time limit set to {options.run_time} seconds")330            spawn_run_time_quit_greenlet()331        elif not options.worker and not environment.shape_class:332            logger.info("No run time limit set, use CTRL+C to interrupt")333    if options.csv_prefix:334        gevent.spawn(stats_csv_writer.stats_writer).link_exception(greenlet_exception_handler)335    if options.headless:336        start_automatic_run()337    input_listener_greenlet = None338    if not options.worker:...

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.