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

How to use _clean_query_value method in gabbi

Best Python code snippet using gabbi_python

case.py

Source:case.py

...134        """Compare the response with expected data."""135        self._test_status(self.test_data['status'], self.response['status'])136        for handler in self.response_handlers:137            handler(self)138    def _clean_query_value(self, value):139        """Clean up a single query from query_parameters."""140        value = self.replace_template(value)141        # stringify ints in Python version independent fashion142        value = '%s' % value143        value = value.encode('UTF-8')144        return value145    def _environ_replace(self, message):146        """Replace an indicator in a message with the environment value."""147        value = re.sub(self._replacer_regex('ENVIRON'),148                       self._environ_replacer, message)149        if value == "False":150            return False151        if value == "True":152            return True153        return value154    @staticmethod155    def _environ_replacer(match):156        """Replace a regex match with an environment value.157        Let KeyError raise if variable not present.158        """159        environ_name = match.group('arg')160        return os.environ[environ_name]161    @staticmethod162    def extract_json_path_value(data, path):163        """Extract the value at JSON Path path from the data.164        The input data is a Python datastructre, not a JSON string.165        """166        path_expr = json_parser.parse(path)167        matches = [match.value for match in path_expr.find(data)]168        try:169            return matches[0]170        except IndexError:171            raise ValueError(172                "JSONPath '%s' failed to match on data: '%s'" % (path, data))173    def _headers_replace(self, message):174        """Replace a header indicator in a message with that headers value from175        the prior request.176        """177        return re.sub(self._replacer_regex('HEADERS'),178                      self._header_replacer, message)179    def _header_replacer(self, match):180        """Replace a regex match with the value of a prior header."""181        header_key = match.group('arg')182        return self.prior.response[header_key.lower()]183    def _json_replacer(self, match):184        """Replace a regex match with the value of a JSON Path."""185        path = match.group('arg')186        return str(self.extract_json_path_value(self.prior.json_data, path))187    def _location_replace(self, message):188        """Replace $LOCATION in a message.189        With the location header from the prior request.190        """191        return message.replace('$LOCATION', self.prior.location)192    def _load_data_file(self, filename):193        """Read a file from the current test directory."""194        path = os.path.join(self.test_directory, os.path.basename(filename))195        with open(path, mode='rb') as data_file:196            return data_file.read()197    def _netloc_replace(self, message):198        """Replace $NETLOC with the current host and port."""199        netloc = self.netloc200        if self.prefix:201            netloc = '%s%s' % (netloc, self.prefix)202        return message.replace('$NETLOC', netloc)203    def _parse_url(self, url):204        """Create a url from test data.205        If provided with a full URL, just return that. If SSL is requested206        set the scheme appropriately.207        Scheme and netloc are saved for later use in comparisons.208        """209        query_params = self.test_data['query_parameters']210        ssl = self.test_data['ssl']211        parsed_url = urlparse.urlsplit(url)212        if not parsed_url.scheme:213            full_url = utils.create_url(url, self.host, port=self.port,214                                        prefix=self.prefix, ssl=ssl)215            # parse again to set updated netloc and scheme216            parsed_url = urlparse.urlsplit(full_url)217        self.scheme = parsed_url.scheme218        self.netloc = parsed_url.netloc219        if query_params:220            query_string = self._update_query_params(parsed_url.query,221                                                     query_params)222        else:223            query_string = parsed_url.query224        return urlparse.urlunsplit((parsed_url.scheme, parsed_url.netloc,225                                    parsed_url.path, query_string, ''))226    @staticmethod227    def _replacer_regex(key):228        """Compose a regular expression for test template variables."""229        return r"\$%s\[(?P<quote>['\"])(?P<arg>.+?)(?P=quote)\]" % key230    def _response_replace(self, message):231        """Replace a JSON Path from the prior request with a value."""232        return re.sub(self._replacer_regex('RESPONSE'),233                      self._json_replacer, message)234    def _run_request(self, url, method, headers, body):235        """Run the http request and decode output.236        The call to make the request will catch a WSGIAppError from237        wsgi_intercept so that the real traceback from a catastrophic238        error in the intercepted app can be examined.239        """240        if 'user-agent' not in (key.lower() for key in headers):241            headers['user-agent'] = "gabbi/%s (Python httplib2)" % __version__242        try:243            response, content = self.http.request(244                url,245                method=method,246                headers=headers,247                body=body248            )249        except wsgi_intercept.WSGIAppError as exc:250            # Extract and re-raise the wrapped exception.251            six.reraise(exc.exception_type, exc.exception_value,252                        exc.traceback)253        # Set headers and location attributes for follow on requests254        self.response = response255        if 'location' in response:256            self.location = response['location']257        # Decode and store response258        decoded_output = utils.decode_response_content(response, content)259        self.content_type = response.get('content-type', '').lower()260        if (decoded_output and261                ('application/json' in self.content_type or262                 '+json' in self.content_type)):263            self.json_data = json.loads(decoded_output)264        else:265            self.json_data = None266        self.output = decoded_output267    def _run_test(self):268        """Make an HTTP request and compare the response with expectations."""269        test = self.test_data270        base_url = self.replace_template(test['url'])271        full_url = self._parse_url(base_url)272        method = test['method'].upper()273        headers = test['request_headers']274        for name in headers:275            headers[name] = self.replace_template(headers[name])276        if test['data']:277            body = self._test_data_to_string(278                test['data'], headers.get('content-type', ''))279        else:280            body = ''281        # Reset follow_redirects with every go.282        self.http.follow_redirects = False283        if test['redirects']:284            self.http.follow_redirects = True285        if test['poll']:286            count = test['poll'].get('count', 1)287            delay = test['poll'].get('delay', 1)288            failure = None289            while count:290                try:291                    self._run_request(full_url, method, headers, body)292                    self._assert_response()293                    failure = None294                    break295                except (AssertionError, utils.ConnectionRefused) as exc:296                    failure = exc297                count -= 1298                time.sleep(delay)299            if failure:300                raise failure301        else:302            self._run_request(full_url, method, headers, body)303            self._assert_response()304    def _scheme_replace(self, message):305        """Replace $SCHEME with the current protocol."""306        return message.replace('$SCHEME', self.scheme)307    def _test_data_to_string(self, data, content_type):308        """Turn the request data into a string.309        If the data is not binary, replace template strings.310        """311        if isinstance(data, str):312            if data.startswith('<@'):313                info = self._load_data_file(data.replace('<@', '', 1))314                if utils.not_binary(content_type):315                    try:316                        info = str(info, 'UTF-8')317                    except TypeError:318                        info = info.encode('UTF-8')319                    data = info320                else:321                    return info322        else:323            data = json.dumps(data)324        return self.replace_template(data)325    def _test_status(self, expected_status, observed_status):326        """Confirm we got the expected status.327        If the status contains one or more || then it is treated as a328        list of acceptable statuses.329        """330        expected_status = str(expected_status)331        if '||' in expected_status:332            statii = [stat.strip() for stat in expected_status.split('||')]333        else:334            statii = [expected_status.strip()]335        self.assert_in_or_print_output(observed_status, statii)336    def _update_query_params(self, original_query_string, query_params):337        """Update a query string from query_params dict.338        An OrderedDict is used to allow easier testing and greater339        predictability when doing query updates.340        """341        encoded_query_params = OrderedDict()342        for param, value in query_params.items():343            # isinstance used because we can iter a string344            if isinstance(value, list):345                encoded_query_params[param] = [346                    self._clean_query_value(subvalue)347                    for subvalue in value]348            else:349                encoded_query_params[param] = (350                    self._clean_query_value(value))351        query_string = urlparse.urlencode(352            encoded_query_params, doseq=True)353        if original_query_string:354            query_string = '&'.join([original_query_string, query_string])355        return query_string356    def assert_in_or_print_output(self, expected, iterable):357        """Assert the iterable contains expected or print some output.358        If the output is long, it is limited by either GABBI_MAX_CHARS_OUTPUT359        in the environment or the MAX_CHARS_OUTPUT constant.360        """361        if utils.not_binary(self.content_type):362            if expected in iterable:363                return364            if self.json_data:...

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.