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

How to use _log_response_time method in locust

Best Python code snippet using locust

stats.py

Source:stats.py

...129    def log(self, response_time, content_length):130        self.stats.num_requests += 1131        self.num_requests += 1132        self._log_time_of_request()133        self._log_response_time(response_time)134        # increase total content-length135        self.total_content_length += content_length136    def _log_time_of_request(self):137        t = int(time.time())138        self.num_reqs_per_sec[t] = self.num_reqs_per_sec.setdefault(t, 0) + 1139        self.last_request_timestamp = t140        self.stats.last_request_timestamp = t141    def _log_response_time(self, response_time):142        self.total_response_time += response_time143        if self.min_response_time is None:144            self.min_response_time = response_time145        self.min_response_time = min(self.min_response_time, response_time)146        self.max_response_time = max(self.max_response_time, response_time)147        # to avoid to much data that has to be transfered to the master node when148        # running in distributed mode, we save the response time rounded in a dict149        # so that 147 becomes 150, 3432 becomes 3400 and 58760 becomes 59000150        if response_time < 100:151            rounded_response_time = response_time152        elif response_time < 1000:153            rounded_response_time = int(round(response_time, -1))154        elif response_time < 10000:155            rounded_response_time = int(round(response_time, -2))...

transport.py

Source:transport.py

...257                    self.log_response_to_file(response)258                # Successful response259                if response.status_code == 200:260                    self._log_response_success(base_url, endpoint)261                    self._log_response_time(262                        base_url, endpoint, time.perf_counter() - start_time263                    )264                    if output_format == "json":265                        return response.json()266                    elif output_format == "content":267                        return response.content268                    elif output_format == "status_code":269                        return response.status_code270                    else:271                        return response272                # Cancel bad URLs273                # Logged status codes274                elif response.status_code in DO_NOT_REPEAT_STATUS_CODES:275                    self._log_response_fail(276                        base_url,277                        endpoint,278                        response.status_code,279                        response.content.decode(),280                    )281                    if raise_error:282                        raise APIError(response.content.decode())283                # Retry other errors284                else:285                    self._log_response_fail(286                        base_url,287                        endpoint,288                        response.status_code,289                        response.content.decode(),290                    )291                    continue292            except (ConnectionError) as error:293                # Print the error294                traceback.print_exc()295                self._log_connection_error(base_url, endpoint)296                time.sleep(seconds_between_retries)297                continue298            except JSONDecodeError as error:299                self._log_no_json(base_url, endpoint, response.status_code, response)300                return response301        return response302    async def make_async_http_request(303        self,304        endpoint: str,305        method: str = "GET",306        parameters: Optional[dict] = None,307        base_url: str = None,308        output_format=None,309        raise_error: bool = True,310    ):311        """312        Make an asynchronous HTTP request313        Parameters314        ----------315        endpoint: str316            The Relevance AI endpoint to access317        method_type: str318            Currently only support GET and POST requests319        raise_error: bool320            If True, error is raised rather than just logged.321        """322        parameters = {} if parameters is None else parameters323        self._last_used_endpoint = endpoint324        start_time = time.perf_counter()325        base_url = self.base_url if base_url is None else base_url  # type: ignore326        output_format = (327            self.config.get_option("api.output_format")328            if output_format is None329            else output_format330        )331        retries = int(self.config.get_option("retries.number_of_retries"))332        seconds_between_retries = int(333            self.config.get_option("retries.seconds_between_retries")334        )335        request_url = base_url + endpoint336        for _ in range(retries):337            self.logger.info(f"URL you are trying to access: {request_url}")338            try:339                if Transport._is_search_in_path(request_url):340                    self._log_search_to_dashboard(341                        method=method, parameters=parameters, endpoint=endpoint342                    )343                async with aiohttp.request(344                    method=method.upper(),345                    url=request_url,346                    headers=self.auth_header,347                    json=parameters if method.upper() == "POST" else {},348                    params=parameters if method.upper() == "GET" else {},349                ) as response:350                    if os.getenv("DEBUG_REQUESTS") == "TRUE":351                        self.log_response_to_file(response)352                    if response.status == 200:353                        self._log_response_success(base_url, endpoint)354                        self._log_response_time(355                            base_url, endpoint, time.perf_counter() - start_time356                        )357                        if output_format.lower() == "json":358                            return await response.json()359                        elif output_format.lower() == "content":360                            decoded_content = codecs.decode(361                                await response.content.read()362                            )363                            return decoded_content364                        elif output_format.lower() == "status_code":365                            return response.status366                        else:367                            return response368                    elif response.status in DO_NOT_REPEAT_STATUS_CODES:369                        # Cancel bad URLs370                        # Logged status codes371                        decoded_content = codecs.decode(await response.content.read())372                        self._log_response_fail(373                            base_url, endpoint, response.status, decoded_content374                        )375                        if raise_error:376                            raise APIError(decoded_content)377                    else:378                        # Retry other errors379                        decoded_content = codecs.decode(await response.content.read())380                        self._log_response_fail(381                            base_url, endpoint, response.status, decoded_content382                        )383            except aiohttp.ClientError as error:384                traceback.print_exc()385                self._log_connection_error(base_url, endpoint)386                time.sleep(seconds_between_retries)387                continue388            except JSONDecodeError as error:389                self._log_no_json(base_url, endpoint, response.status, response)390                return response391        return response392    def _log_response_success(self, base_url, endpoint):393        self.logger.success(f"Response success! ({base_url + endpoint})")394    def _log_response_time(self, base_url, endpoint, time):395        self.logger.debug(f"Request ran in {time} seconds ({base_url + endpoint})")396    def _log_response_fail(self, base_url, endpoint, status_code, content):397        self.logger.error(398            f"Response failed ({base_url + endpoint}) (Status: {status_code} Response: {content})"399        )400    def _log_connection_error(self, base_url, endpoint):401        self.logger.error(f"Connection error but re-trying. ({base_url + endpoint})")402    def _log_no_json(self, base_url, endpoint, status_code, content):403        self.logger.error(404            f"No JSON Available ({base_url + endpoint}) (Status: {status_code} Response: {content})"...

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.