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

How to use _download_csv_response method in locust

Best Python code snippet using locust

web.py

Source:web.py

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

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.