How to use maintenance_timeouts method in yandex-tank

Best Python code snippet using yandex-tank

client.py

Source:client.py

...65    def network_timeouts(self):66        return (self.network_timeout for _ in range(self.network_attempts - 1))67    def api_timeouts(self):68        return (self.api_timeout for _ in range(self.api_attempts - 1))69    def maintenance_timeouts(self):70        return (71            self.maintenance_timeout for _ in range(72                self.maintenance_attempts - 1))73    @staticmethod74    def filter_headers(headers):75        boring = ['X-Content-Security-Policy', 'Content-Security-Policy',76                  'Strict-Transport-Security', 'X-WebKit-CSP', 'Set-Cookie',77                  'X-DNS-Prefetch-Control', 'X-Frame-Options', 'P3P',78                  'X-Content-Type-Options', 'X-Download-Options',79                  'Surrogate-Control']80        for h in boring:81            if h in headers:82                del (headers[h])83        return headers84    def __send_single_request(self, req, trace=False):85        p = self.session.prepare_request(req)86        if trace:87            logger.debug("Making request: %s %s Headers: %s Body: %s",88                         p.method, p.url, p.headers, p.body)89        resp = self.session.send(p, timeout=self.connection_timeout)90        if trace:91            logger.debug("Got response in %ss: %s %s Headers: %s Body: %s",92                         resp.elapsed.total_seconds(), resp.reason,93                         resp.status_code, self.filter_headers(resp.headers),94                         resp.content)95        if resp.status_code in [500, 502, 503, 504]:96            raise self.NotAvailable(97                request="request: %s %s\n\tHeaders: %s\n\tBody: %s" %98                (p.method,99                 p.url,100                 p.headers,101                 p.body),102                response="Got response in %ss: %s %s\n\tHeaders: %s\n\tBody: %s" %103                (resp.elapsed.total_seconds(),104                 resp.reason,105                 resp.status_code,106                 self.filter_headers(107                    resp.headers),108                    resp.content))109        elif resp.status_code == 410:110            raise self.StoppedFromOnline111        elif resp.status_code == 423:112            raise self.UnderMaintenance113        else:114            resp.raise_for_status()115            return resp116    def __make_api_request(117            self,118            http_method,119            path,120            data=None,121            response_callback=lambda x: x,122            writer=False,123            trace=False,124            json=None,125            maintenance_timeouts=None,126            maintenance_msg=None):127        url = urljoin(self.base_url, path)128        if json:129            request = requests.Request(130                http_method, url, json=json, headers={'User-Agent': self.user_agent}, params=self.params)131        else:132            request = requests.Request(133                http_method, url, data=data, headers={'User-Agent': self.user_agent}, params=self.params)134        network_timeouts = self.network_timeouts()135        maintenance_timeouts = maintenance_timeouts or self.maintenance_timeouts()136        maintenance_msg = maintenance_msg or "%s is under maintenance" % (self._base_url)137        while True:138            try:139                response = self.__send_single_request(request, trace=trace)140                return response_callback(response)141            except (Timeout, ConnectionError):142                logger.warn(traceback.format_exc())143                try:144                    timeout = next(network_timeouts)145                    logger.warn(146                        "Network error, will retry in %ss..." %147                        timeout)148                    time.sleep(timeout)149                    continue150                except StopIteration:151                    raise self.NetworkError()152            except self.UnderMaintenance as e:153                try:154                    timeout = next(maintenance_timeouts)155                    logger.warn(maintenance_msg)156                    logger.warn("Retrying in %ss..." % timeout)157                    time.sleep(timeout)158                    continue159                except StopIteration:160                    raise e161    def __make_writer_request(162            self,163            params=None,164            json=None,165            http_method="POST",166            trace=False):167        '''168        Send request to writer service.169        '''170        request = requests.Request(171            http_method,172            self.writer_url,173            params=params,174            json=json,175            headers={176                'User-Agent': self.user_agent})177        network_timeouts = self.network_timeouts()178        maintenance_timeouts = self.maintenance_timeouts()179        while True:180            try:181                response = self.__send_single_request(request, trace=trace)182                return response183            except (Timeout, ConnectionError):184                logger.warn(traceback.format_exc())185                try:186                    timeout = next(network_timeouts)187                    logger.warn(188                        "Network error, will retry in %ss..." %189                        timeout)190                    time.sleep(timeout)191                    continue192                except StopIteration:...

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.