KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use my_custom_route method in locust

Best Python code snippet using locust

web.py

Source:web.py

...36    37        from flask import request38        39        @web_ui.app.route("/my_custom_route")40        def my_custom_route():41            return "your IP is: %s" % request.remote_addr42    """43    44    greenlet = None45    """46    Greenlet of the running web server47    """48    49    server = None50    """Reference to the :class:`pyqsgi.WSGIServer` instance"""51    52    def __init__(self, environment, host, port, auth_credentials=None, tls_cert=None, tls_key=None):53        """54        Create WebUI instance and start running the web server in a separate greenlet (self.greenlet)55        56        Arguments:57        environment: Reference to the curren Locust Environment58        host: Host/interface that the web server should accept connections to59        port: Port that the web server should listen to60        auth_credentials:  If provided, it will enable basic auth with all the routes protected by default.61                           Should be supplied in the format: "user:pass".62        tls_cert: A path to a TLS certificate63        tls_key: A path to a TLS private key64        """65        environment.web_ui = self66        self.environment = environment67        self.host = host68        self.port = port69        self.tls_cert = tls_cert70        self.tls_key = tls_key71        app = Flask(__name__)72        self.app = app73        app.debug = True74        app.root_path = os.path.dirname(os.path.abspath(__file__))75        self.app.config["BASIC_AUTH_ENABLED"] = False76        self.auth = None77        self.greenlet = None78        if auth_credentials is not None:79            credentials = auth_credentials.split(':')80            if len(credentials) == 2:81                self.app.config["BASIC_AUTH_USERNAME"] = credentials[0]82                self.app.config["BASIC_AUTH_PASSWORD"] = credentials[1]83                self.app.config["BASIC_AUTH_ENABLED"] = True84                self.auth = BasicAuth()85                self.auth.init_app(self.app)86            else:87                raise AuthCredentialsError("Invalid auth_credentials. It should be a string in the following format: 'user.pass'")88        @app.route('/')89        @self.auth_required_if_enabled90        def index():91            if not environment.runner:92                return make_response("Error: Locust Environment does not have any runner", 500)93            94            is_distributed = isinstance(environment.runner, MasterRunner)95            if is_distributed:96                worker_count = environment.runner.worker_count97            else:98                worker_count = 099            100            override_host_warning = False101            if environment.host:102                host = environment.host103            elif environment.runner.user_classes:104                all_hosts = set([l.host for l in environment.runner.user_classes])105                if len(all_hosts) == 1:106                    host = list(all_hosts)[0]107                else:108                    # since we have mulitple User classes with different host attributes, we'll109                    # inform that specifying host will override the host for all User classes110                    override_host_warning = True111                    host = None112            else:113                host = None114            115            options = environment.parsed_options116            return render_template("index.html",117                state=environment.runner.state,118                is_distributed=is_distributed,119                user_count=environment.runner.user_count,120                version=version,121                host=host,122                override_host_warning=override_host_warning,123                num_users=options and options.num_users,124                hatch_rate=options and options.hatch_rate,125                step_users=options and options.step_users,126                step_time=options and options.step_time,127                worker_count=worker_count,128                is_step_load=environment.step_load,129            )130        131        @app.route('/swarm', methods=["POST"])132        @self.auth_required_if_enabled133        def swarm():134            assert request.method == "POST"135            user_count = int(request.form["user_count"])136            hatch_rate = float(request.form["hatch_rate"])137            if (request.form.get("host")):138                environment.host = str(request.form["host"])139        140            if environment.step_load:141                step_user_count = int(request.form["step_user_count"])142                step_duration = parse_timespan(str(request.form["step_duration"]))143                environment.runner.start_stepload(user_count, hatch_rate, step_user_count, step_duration)144                return jsonify({'success': True, 'message': 'Swarming started in Step Load Mode', 'host': environment.host})145            146            environment.runner.start(user_count, hatch_rate)147            return jsonify({'success': True, 'message': 'Swarming started', 'host': environment.host})148        149        @app.route('/stop')150        @self.auth_required_if_enabled151        def stop():152            environment.runner.stop()153            return jsonify({'success':True, 'message': 'Test stopped'})154        155        @app.route("/stats/reset")156        @self.auth_required_if_enabled157        def reset_stats():158            environment.runner.stats.reset_all()159            environment.runner.exceptions = {}160            return "ok"161            162        @app.route("/stats/requests/csv")163        @self.auth_required_if_enabled164        def request_stats_csv():165            response = make_response(requests_csv(self.environment.runner.stats))166            file_name = "requests_{0}.csv".format(time())167            disposition = "attachment;filename={0}".format(file_name)168            response.headers["Content-type"] = "text/csv"169            response.headers["Content-disposition"] = disposition170            return response171        172        @app.route("/stats/failures/csv")173        @self.auth_required_if_enabled174        def failures_stats_csv():175            response = make_response(failures_csv(self.environment.runner.stats))176            file_name = "failures_{0}.csv".format(time())177            disposition = "attachment;filename={0}".format(file_name)178            response.headers["Content-type"] = "text/csv"179            response.headers["Content-disposition"] = disposition180            return response181        182        @app.route('/stats/requests')183        @self.auth_required_if_enabled184        @memoize(timeout=DEFAULT_CACHE_TIME, dynamic_timeout=True)185        def request_stats():186            stats = []187        188            for s in chain(sort_stats(self.environment.runner.stats.entries), [environment.runner.stats.total]):189                stats.append({190                    "method": s.method,191                    "name": s.name,192                    "safe_name": escape(s.name, quote=False),193                    "num_requests": s.num_requests,194                    "num_failures": s.num_failures,195                    "avg_response_time": s.avg_response_time,196                    "min_response_time": 0 if s.min_response_time is None else proper_round(s.min_response_time),197                    "max_response_time": proper_round(s.max_response_time),198                    "current_rps": s.current_rps,199                    "current_fail_per_sec": s.current_fail_per_sec,200                    "median_response_time": s.median_response_time,201                    "ninetieth_response_time": s.get_response_time_percentile(0.9),202                    "avg_content_length": s.avg_content_length,203                })204            205            errors = []206            for e in environment.runner.errors.values():207                err_dict = e.to_dict()208                err_dict["name"] = escape(err_dict["name"])209                err_dict["error"] = escape(err_dict["error"])210                errors.append(err_dict)211            # Truncate the total number of stats and errors displayed since a large number of rows will cause the app212            # to render extremely slowly. Aggregate stats should be preserved.213            report = {"stats": stats[:500], "errors": errors[:500]}214            if len(stats) > 500:215                report["stats"] += [stats[-1]]216        217            if stats:218                report["total_rps"] = stats[len(stats)-1]["current_rps"]219                report["fail_ratio"] = environment.runner.stats.total.fail_ratio220                report["current_response_time_percentile_95"] = environment.runner.stats.total.get_current_response_time_percentile(0.95)221                report["current_response_time_percentile_50"] = environment.runner.stats.total.get_current_response_time_percentile(0.5)222            223            is_distributed = isinstance(environment.runner, MasterRunner)224            if is_distributed:225                workers = []226                for worker in environment.runner.clients.values():227                    workers.append({"id":worker.id, "state":worker.state, "user_count": worker.user_count, "cpu_usage":worker.cpu_usage})228        229                report["workers"] = workers230            231            report["state"] = environment.runner.state232            report["user_count"] = environment.runner.user_count233        234            return jsonify(report)235        236        @app.route("/exceptions")237        @self.auth_required_if_enabled238        def exceptions():239            return jsonify({240                'exceptions': [241                    {242                        "count": row["count"],243                        "msg": row["msg"],244                        "traceback": row["traceback"],245                        "nodes" : ", ".join(row["nodes"])246                    } for row in environment.runner.exceptions.values()247                ]248            })249        250        @app.route("/exceptions/csv")251        @self.auth_required_if_enabled252        def exceptions_csv():253            data = StringIO()254            writer = csv.writer(data)255            writer.writerow(["Count", "Message", "Traceback", "Nodes"])256            for exc in environment.runner.exceptions.values():257                nodes = ", ".join(exc["nodes"])258                writer.writerow([exc["count"], exc["msg"], exc["traceback"], nodes])259            260            data.seek(0)261            response = make_response(data.read())262            file_name = "exceptions_{0}.csv".format(time())263            disposition = "attachment;filename={0}".format(file_name)264            response.headers["Content-type"] = "text/csv"265            response.headers["Content-disposition"] = disposition266            return response267        268        # start the web server269        self.greenlet = gevent.spawn(self.start)270        self.greenlet.link_exception(greenlet_exception_handler)271    def start(self):272        if self.tls_cert and self.tls_key:273            self.server = pywsgi.WSGIServer((self.host, self.port), self.app, log=None, keyfile=self.tls_key, certfile=self.tls_cert)274        else:275            self.server = pywsgi.WSGIServer((self.host, self.port), self.app, log=None)276        self.server.serve_forever()277    278    def stop(self):279        """280        Stop the running web server281        """282        self.server.stop()283    def auth_required_if_enabled(self, view_func):284        """285        Decorator that can be used on custom route methods that will turn on Basic Auth 286        authentication if the ``--web-auth`` flag is used. Example::287        288            @web_ui.app.route("/my_custom_route")289            @web_ui.auth_required_if_enabled290            def my_custom_route():291                return "custom response"292        """293        @wraps(view_func)294        def wrapper(*args, **kwargs):295            if self.app.config["BASIC_AUTH_ENABLED"]:296                if self.auth.authenticate():297                    return view_func(*args, **kwargs)298                else:299                    return self.auth.challenge()300            else:301                return view_func(*args, **kwargs)...

locustfile.py

Source:locustfile.py

...13    host = 'http://localhost:8080'14@events.init.add_listener15def on_locust_init(web_ui, **kw):16    @web_ui.app.route("/my_custom_route")17    def my_custom_route():...

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.