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

How to use _requests_data_rows method in locust

Best Python code snippet using locust

stats.py

Source:stats.py

...713        )714    def requests_csv(self, csv_writer):715        """Write requests csv with header and data rows."""716        csv_writer.writerow(self.requests_csv_columns)717        self._requests_data_rows(csv_writer)718    def _requests_data_rows(self, csv_writer):719        """Write requests csv data row, excluding header."""720        stats = self.environment.stats721        for stats_entry in chain(sort_stats(stats.entries), [stats.total]):722            csv_writer.writerow(723                chain(724                    [725                        stats_entry.method,726                        stats_entry.name,727                        stats_entry.num_requests,728                        stats_entry.num_failures,729                        stats_entry.median_response_time,730                        stats_entry.avg_response_time,731                        stats_entry.min_response_time or 0,732                        stats_entry.max_response_time,733                        stats_entry.avg_content_length,734                        stats_entry.total_rps,735                        stats_entry.total_fail_per_sec,736                    ],737                    self._percentile_fields(stats_entry),738                )739            )740    def failures_csv(self, csv_writer):741        csv_writer.writerow(self.failures_columns)742        self._failures_data_rows(csv_writer)743    def _failures_data_rows(self, csv_writer):744        for stats_error in sort_stats(self.environment.stats.errors):745            csv_writer.writerow(746                [747                    stats_error.method,748                    stats_error.name,749                    stats_error.error,750                    stats_error.occurrences,751                ]752            )753    def exceptions_csv(self, csv_writer):754        csv_writer.writerow(self.exceptions_columns)755        self._exceptions_data_rows(csv_writer)756    def _exceptions_data_rows(self, csv_writer):757        for exc in self.environment.runner.exceptions.values():758            csv_writer.writerow([exc["count"], exc["msg"], exc["traceback"], ", ".join(exc["nodes"])])759class StatsCSVFileWriter(StatsCSV):760    """Write statistics to to CSV files"""761    def __init__(self, environment, percentiles_to_report, base_filepath, full_history=False):762        super().__init__(environment, percentiles_to_report)763        self.base_filepath = base_filepath764        self.full_history = full_history765        self.requests_csv_filehandle = open(self.base_filepath + "_stats.csv", "w")766        self.requests_csv_writer = csv.writer(self.requests_csv_filehandle)767        self.stats_history_csv_filehandle = open(self.stats_history_file_name(), "w")768        self.stats_history_csv_writer = csv.writer(self.stats_history_csv_filehandle)769        self.failures_csv_filehandle = open(self.base_filepath + "_failures.csv", "w")770        self.failures_csv_writer = csv.writer(self.failures_csv_filehandle)771        self.failures_csv_data_start = 0772        self.exceptions_csv_filehandle = open(self.base_filepath + "_exceptions.csv", "w")773        self.exceptions_csv_writer = csv.writer(self.exceptions_csv_filehandle)774        self.exceptions_csv_data_start = 0775        self.stats_history_csv_columns = [776            "Timestamp",777            "User Count",778            "Type",779            "Name",780            "Requests/s",781            "Failures/s",782            *get_readable_percentiles(self.percentiles_to_report),783            "Total Request Count",784            "Total Failure Count",785            "Total Median Response Time",786            "Total Average Response Time",787            "Total Min Response Time",788            "Total Max Response Time",789            "Total Average Content Size",790        ]791    def __call__(self):792        self.stats_writer()793    def stats_writer(self):794        """Writes all the csv files for the locust run."""795        # Write header row for all files and save position for non-append files796        self.requests_csv_writer.writerow(self.requests_csv_columns)797        requests_csv_data_start = self.requests_csv_filehandle.tell()798        self.stats_history_csv_writer.writerow(self.stats_history_csv_columns)799        self.failures_csv_writer.writerow(self.failures_columns)800        self.failures_csv_data_start = self.failures_csv_filehandle.tell()801        self.exceptions_csv_writer.writerow(self.exceptions_columns)802        self.exceptions_csv_data_start = self.exceptions_csv_filehandle.tell()803        # Continuously write date rows for all files804        last_flush_time = 0805        while True:806            now = time.time()807            self.requests_csv_filehandle.seek(requests_csv_data_start)808            self._requests_data_rows(self.requests_csv_writer)809            self.requests_csv_filehandle.truncate()810            self._stats_history_data_rows(self.stats_history_csv_writer, now)811            self.failures_csv_filehandle.seek(self.failures_csv_data_start)812            self._failures_data_rows(self.failures_csv_writer)813            self.failures_csv_filehandle.truncate()814            self.exceptions_csv_filehandle.seek((self.exceptions_csv_data_start))815            self._exceptions_data_rows(self.exceptions_csv_writer)816            self.exceptions_csv_filehandle.truncate()817            if now - last_flush_time > CSV_STATS_FLUSH_INTERVAL_SEC:818                self.requests_flush()819                self.stats_history_flush()820                self.failures_flush()821                self.exceptions_flush()822                last_flush_time = now...

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.