How to use _report_request method in locust

Best Python code snippet using locust

webdriver.py

Source:webdriver.py

...154            if self._manual_result is True:155                self.request_meta["exception"] = None156            elif isinstance(self._manual_result, Exception):157                self.request_meta["exception"] = self._manual_result158            self._report_request()159            return exc is None160        # no manual result fired - set stop time as current time161        self.request_meta["response_time"] = (time.perf_counter() - self.request_meta["start_time"]) * 1000162        if exc:163            # Override the default rendering of exceptions to avoid printing complete stack traces.164            if isinstance(value, TimeoutException):165                self.request_meta["exception"] = "Timeout waiting for Webdriver:" + value.msg166                self._report_request()167            elif isinstance(value, WebDriverException):168                self.request_meta["exception"] = "Webdriver Exception:" + value.msg169                self._report_request()170            else:171                # we want other unknown exceptions to be raised172                return False173        else:174            self._report_request()175        return True176    def _report_request(self):177        # if URL is None, update with current url178        if not self.request_meta["url"]:179            self.request_meta["url"] = self.client.current_url180        self._request_event.fire(**self.request_meta)181    def success(self):182        """183        Report the response as successful184        Example::185            with self.request(name="helpful_name") as request:186                request.client.get("https://example.com/")187                title = request.client.find_element(By.CSS_SELECTOR, "body > div > h1")188                if title.text == "Example Domain":189                    request.success()190                else:...

ProjectViews.py

Source:ProjectViews.py

...181			self._report_request = FSZSQLMethod(self,self._report_request_ZSQLFile)182			self._report_request.connection = self.connection183		184		# log and execute report request185		logSQL = self._report_request(projectId= self.getProject_db_id(),186						tablePkey=self.getTablePkey(),187						tableName=self.getTableName(),188						fieldsList=fieldsList,189						src__=1)190		self.log = self.log + "\n" + str(fieldsList)+ "\n" + logSQL+"\n"191		192		reportRequestResults = self._report_request(projectId= self.getProject_db_id(),193						tablePkey=self.getTablePkey(),194						tableName=self.getTableName(),195						fieldsList=fieldsList)196		197		return self.getReportDictionary(fieldsList,reportRequestResults)198	199	_reportDictionary = {}200	def getReportDictionary(self,fieldsList,reportRequestResults):201		""" builds the report dictionary from the report request results202				this method may be low-level-user-defined203		204		THIS getReportDictionary METHOD :205			returns the latest value of each field for this project206		"""...

api.py

Source:api.py

...28        return self._request(url, stream=stream)29    def _content_request(self, resource, stream=True):30        url = self.url_content_base + resource31        return self._request(url, stream=stream)32    def _report_request(self, resource, stream=True):33        url = self.url_report_base + resource34        return self._request(url, stream=stream)35    def get(self, url, stream=True):36        """Allows for embedded report URLs to be easily downloaded."""37        return self._request(url, stream=stream)38    def get_environment(self):39        logging.debug("getting environment details.")40        r = self._api_request("environment")41        r.raise_for_status()42        return json.dumps(r.json(), indent=2, sort_keys=True)43    def get_user(self):44        logging.debug("getting user details.")45        r = self._api_request("user")46        r.raise_for_status()47        return json.dumps(r.json(), indent=2, sort_keys=True)48    def get_history(self):49        """Get analysis history for this account."""50        r = self._api_request("analysis")51        r.raise_for_status()52        return json.dumps(r.json(), indent=2, sort_keys=True)53    def get_report(self, task, write_path=False):54        """Get an analysis report by it's task id. return json unless write_path."""55        r = self._api_request(f"analysis/{task}")56        r.raise_for_status()57        if write_path:58            try:59                with open(write_path, "w") as f:60                    f.write(json.dumps(r.json(), indent=2, sort_keys=True))61                if os.path.exists(write_path):62                    logging.info(f"Wrote {write_path}")63                    return True64                else:65                    return False66            except Exception as e:67                logging.error(f"{e} : r.text")68                return False69        return json.dumps(r.json(), indent=2, sort_keys=True)70    def get_report_iocs(self, task, write_path=False):71        logging.debug(f"Downloading IOCs for {task}")72        r = self._report_request(f"report/{task}/ioc/json")73        r.raise_for_status74        if write_path:75            try:76                with open(write_path, "w") as f:77                    f.write(json.dumps(r.json(), indent=2, sort_keys=True))78                if os.path.exists(write_path):79                    logging.info(f"Wrote {write_path}")80                    return True81                else:82                    return False83            except Exception as e:84                logging.error(f"{e} : r.text")85                return False86        return json.dumps(r.json(), indent=2, sort_keys=True)87    def get_report_summary(self, task, write_path=False):88        logging.debug(f"Downloading IOCs for {task}")89        r = self._report_request(f"report/{task}/summary/json")90        r.raise_for_status91        if write_path:92            try:93                with open(write_path, "w") as f:94                    f.write(json.dumps(r.json(), indent=2, sort_keys=True))95                if os.path.exists(write_path):96                    logging.info(f"Wrote {write_path}")97                    return True98                else:99                    return False100            except Exception as e:101                logging.error(f"{e} : r.text")102                return False103        return json.dumps(r.json(), indent=2, sort_keys=True)...

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.