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

How to use update_template_args method in locust

Best Python code snippet using locust

web.py

Source:web.py

...101                raise AuthCredentialsError(102                    "Invalid auth_credentials. It should be a string in the following format: 'user.pass'"103                )104        if environment.runner:105            self.update_template_args()106        if not delayed_start:107            self.start()108        @app.route("/")109        @self.auth_required_if_enabled110        def index():111            if not environment.runner:112                return make_response("Error: Locust Environment does not have any runner", 500)113            self.update_template_args()114            return render_template("index.html", **self.template_args)115        @app.route("/swarm", methods=["POST"])116        @self.auth_required_if_enabled117        def swarm():118            assert request.method == "POST"119            if request.form.get("host"):120                # Replace < > to guard against XSS121                environment.host = str(request.form["host"]).replace("<", "").replace(">", "")122            if environment.shape_class:123                environment.runner.start_shape()124                return jsonify(125                    {"success": True, "message": "Swarming started using shape class", "host": environment.host}126                )127            user_count = int(request.form["user_count"])128            spawn_rate = float(request.form["spawn_rate"])129            environment.runner.start(user_count, spawn_rate)130            return jsonify({"success": True, "message": "Swarming started", "host": environment.host})131        @app.route("/stop")132        @self.auth_required_if_enabled133        def stop():134            environment.runner.stop()135            return jsonify({"success": True, "message": "Test stopped"})136        @app.route("/stats/reset")137        @self.auth_required_if_enabled138        def reset_stats():139            environment.events.reset_stats.fire()140            environment.runner.stats.reset_all()141            environment.runner.exceptions = {}142            return "ok"143        @app.route("/stats/report")144        @self.auth_required_if_enabled145        def stats_report():146            res = get_html_report(self.environment, show_download_link=not request.args.get("download"))147            if request.args.get("download"):148                res = app.make_response(res)149                res.headers["Content-Disposition"] = "attachment;filename=report_%s.html" % time()150            return res151        def _download_csv_suggest_file_name(suggest_filename_prefix):152            """Generate csv file download attachment filename suggestion.153            Arguments:154            suggest_filename_prefix: Prefix of the filename to suggest for saving the download. Will be appended with timestamp.155            """156            return f"{suggest_filename_prefix}_{time()}.csv"157        def _download_csv_response(csv_data, filename_prefix):158            """Generate csv file download response with 'csv_data'.159            Arguments:160            csv_data: CSV header and data rows.161            filename_prefix: Prefix of the filename to suggest for saving the download. Will be appended with timestamp.162            """163            response = make_response(csv_data)164            response.headers["Content-type"] = "text/csv"165            response.headers[166                "Content-disposition"167            ] = f"attachment;filename={_download_csv_suggest_file_name(filename_prefix)}"168            return response169        @app.route("/stats/requests/csv")170        @self.auth_required_if_enabled171        def request_stats_csv():172            data = StringIO()173            writer = csv.writer(data)174            self.stats_csv_writer.requests_csv(writer)175            return _download_csv_response(data.getvalue(), "requests")176        @app.route("/stats/requests_full_history/csv")177        @self.auth_required_if_enabled178        def request_stats_full_history_csv():179            options = self.environment.parsed_options180            if options and options.stats_history_enabled:181                return send_file(182                    os.path.abspath(self.stats_csv_writer.stats_history_file_name()),183                    mimetype="text/csv",184                    as_attachment=True,185                    attachment_filename=_download_csv_suggest_file_name("requests_full_history"),186                    add_etags=True,187                    cache_timeout=None,188                    conditional=True,189                    last_modified=None,190                )191            return make_response("Error: Server was not started with option to generate full history.", 404)192        @app.route("/stats/failures/csv")193        @self.auth_required_if_enabled194        def failures_stats_csv():195            data = StringIO()196            writer = csv.writer(data)197            self.stats_csv_writer.failures_csv(writer)198            return _download_csv_response(data.getvalue(), "failures")199        @app.route("/stats/requests")200        @self.auth_required_if_enabled201        @memoize(timeout=DEFAULT_CACHE_TIME, dynamic_timeout=True)202        def request_stats():203            stats = []204            for s in chain(sort_stats(self.environment.runner.stats.entries), [environment.runner.stats.total]):205                stats.append(206                    {207                        "method": s.method,208                        "name": s.name,209                        "safe_name": escape(s.name, quote=False),210                        "num_requests": s.num_requests,211                        "num_failures": s.num_failures,212                        "avg_response_time": s.avg_response_time,213                        "min_response_time": 0 if s.min_response_time is None else proper_round(s.min_response_time),214                        "max_response_time": proper_round(s.max_response_time),215                        "current_rps": s.current_rps,216                        "current_fail_per_sec": s.current_fail_per_sec,217                        "median_response_time": s.median_response_time,218                        "ninetieth_response_time": s.get_response_time_percentile(0.9),219                        "avg_content_length": s.avg_content_length,220                    }221                )222            errors = []223            for e in environment.runner.errors.values():224                err_dict = e.to_dict()225                err_dict["name"] = escape(err_dict["name"])226                err_dict["error"] = escape(err_dict["error"])227                errors.append(err_dict)228            # Truncate the total number of stats and errors displayed since a large number of rows will cause the app229            # to render extremely slowly. Aggregate stats should be preserved.230            report = {"stats": stats[:500], "errors": errors[:500]}231            if len(stats) > 500:232                report["stats"] += [stats[-1]]233            if stats:234                report["total_rps"] = stats[len(stats) - 1]["current_rps"]235                report["fail_ratio"] = environment.runner.stats.total.fail_ratio236                report[237                    "current_response_time_percentile_95"238                ] = environment.runner.stats.total.get_current_response_time_percentile(0.95)239                report[240                    "current_response_time_percentile_50"241                ] = environment.runner.stats.total.get_current_response_time_percentile(0.5)242            is_distributed = isinstance(environment.runner, MasterRunner)243            if is_distributed:244                workers = []245                for worker in environment.runner.clients.values():246                    workers.append(247                        {248                            "id": worker.id,249                            "state": worker.state,250                            "user_count": worker.user_count,251                            "cpu_usage": worker.cpu_usage,252                        }253                    )254                report["workers"] = workers255            report["state"] = environment.runner.state256            report["user_count"] = environment.runner.user_count257            return jsonify(report)258        @app.route("/exceptions")259        @self.auth_required_if_enabled260        def exceptions():261            return jsonify(262                {263                    "exceptions": [264                        {265                            "count": row["count"],266                            "msg": row["msg"],267                            "traceback": row["traceback"],268                            "nodes": ", ".join(row["nodes"]),269                        }270                        for row in environment.runner.exceptions.values()271                    ]272                }273            )274        @app.route("/exceptions/csv")275        @self.auth_required_if_enabled276        def exceptions_csv():277            data = StringIO()278            writer = csv.writer(data)279            self.stats_csv_writer.exceptions_csv(writer)280            return _download_csv_response(data.getvalue(), "exceptions")281    def start(self):282        self.greenlet = gevent.spawn(self.start_server)283        self.greenlet.link_exception(greenlet_exception_handler)284    def start_server(self):285        if self.tls_cert and self.tls_key:286            self.server = pywsgi.WSGIServer(287                (self.host, self.port), self.app, log=None, keyfile=self.tls_key, certfile=self.tls_cert288            )289        else:290            self.server = pywsgi.WSGIServer((self.host, self.port), self.app, log=None)291        self.server.serve_forever()292    def stop(self):293        """294        Stop the running web server295        """296        self.server.stop()297    def auth_required_if_enabled(self, view_func):298        """299        Decorator that can be used on custom route methods that will turn on Basic Auth300        authentication if the ``--web-auth`` flag is used. Example::301            @web_ui.app.route("/my_custom_route")302            @web_ui.auth_required_if_enabled303            def my_custom_route():304                return "custom response"305        """306        @wraps(view_func)307        def wrapper(*args, **kwargs):308            if self.app.config["BASIC_AUTH_ENABLED"]:309                if self.auth.authenticate():310                    return view_func(*args, **kwargs)311                else:312                    return self.auth.challenge()313            else:314                return view_func(*args, **kwargs)315        return wrapper316    def update_template_args(self):317        override_host_warning = False318        if self.environment.host:319            host = self.environment.host320        elif self.environment.runner.user_classes:321            all_hosts = set([l.host for l in self.environment.runner.user_classes])322            if len(all_hosts) == 1:323                host = list(all_hosts)[0]324            else:325                # since we have multiple User classes with different host attributes, we'll326                # inform that specifying host will override the host for all User classes327                override_host_warning = True328                host = None329        else:330            host = None...

webutils.py

Source:webutils.py

...128	if not args:129		return False130	131	if has_app:132		args = update_template_args(page_name, args)133	134	return args	135def get_doc_fields(page_name):136	doc_type, doc_name = get_source_doc(page_name)137	if not doc_type:138		return False139	140	obj = webnotes.get_obj(doc_type, doc_name, with_children=True)141	if hasattr(obj, 'prepare_template_args'):142		obj.prepare_template_args()143	args = obj.doc.fields144	args['template'] = get_generators()[doc_type]["template"]145	args['obj'] = obj146	args['int'] = int...

github_page.py

Source:github_page.py

...24    Returns None25    """26    @project.route("/project-repo")27    def github_page() -> str:28        web_ui.update_template_args()29        return render_template("github-page.html", **web_ui.template_args)...

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.