How to use _clean_query_value method in gabbi

Best Python code snippet using gabbi_python

case.py

Source:case.py Github

copy

Full Screen

...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:...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run gabbi automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful