How to use _percentile_fields method in locust

Best Python code snippet using locust

stats.py

Source:stats.py

...704            "Message",705            "Traceback",706            "Nodes",707        ]708    def _percentile_fields(self, stats_entry):709        return (710            [int(stats_entry.get_response_time_percentile(x) or 0) for x in self.percentiles_to_report]711            if stats_entry.num_requests712            else self.percentiles_na713        )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 = now823            gevent.sleep(CSV_STATS_INTERVAL_SEC)824    def _stats_history_data_rows(self, csv_writer, now):825        """826        Write CSV rows with the *current* stats. By default only includes the827        Aggregated stats entry, but if self.full_history is set to True, a row for each entry will828        will be included.829        Note that this method differs from the other methods as it appends time-stamped data to the file, whereas the other methods overwrites the data.830        """831        stats = self.environment.stats832        timestamp = int(now)833        stats_entries = []834        if self.full_history:835            stats_entries = sort_stats(stats.entries)836        for stats_entry in chain(stats_entries, [stats.total]):837            csv_writer.writerow(838                chain(839                    (840                        timestamp,841                        self.environment.runner.user_count,842                        stats_entry.method or "",843                        stats_entry.name,844                        f"{stats_entry.current_rps:2f}",845                        f"{stats_entry.current_fail_per_sec:2f}",846                    ),847                    self._percentile_fields(stats_entry),848                    (849                        stats_entry.num_requests,850                        stats_entry.num_failures,851                        stats_entry.median_response_time,852                        stats_entry.avg_response_time,853                        stats_entry.min_response_time or 0,854                        stats_entry.max_response_time,855                        stats_entry.avg_content_length,856                    ),857                )858            )859    def requests_flush(self):860        self.requests_csv_filehandle.flush()861    def stats_history_flush(self):...

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.