How to use decode_response_content method in gabbi

Best Python code snippet using gabbi_python

transports.py

Source:transports.py Github

copy

Full Screen

...36 })37 def send(self, method, url, query_params=None, content=None, encoding=None):38 options = self.get_request_options(query_params, content, encoding)39 response = self.session.request(method, url, **options)40 result = self.decode_response_content(response)41 if response.status_code >= 400 and response.status_code <= 599:42 title = '%d %s' % (response.status_code, response.reason)43 raise exceptions.ErrorResponse(title, result)44 return result45 def get_request_options(self, query_params, content, encoding):46 """47 Returns a dictionary of keyword parameters to include when making48 the outgoing request.49 """50 options = {51 'headers': dict(self.headers)52 }53 if query_params:54 options['params'] = query_params55 if content is not None:56 if encoding == 'application/json':57 options['json'] = content58 elif encoding == 'multipart/form-data':59 data = {}60 files = ForceMultiPartDict()61 for key, value in content.items():62 if is_file(value):63 files[key] = value64 else:65 data[key] = value66 elif encoding == 'application/x-www-form-urlencoded':67 options['data'] = content68 elif encoding == 'application/octet-stream':69 if isinstance(content, File):70 options['data'] = content.content71 else:72 options['data'] = content73 upload_headers = self.get_upload_headers(content)74 options['headers'].update(upload_headers)75 return options76 def get_upload_headers(self, file_obj):77 """78 When a raw file upload is made, determine the Content-Type and79 Content-Disposition headers to use with the request.80 """81 name = guess_filename(file_obj)82 content_type = None83 content_disposition = None84 # Determine the content type of the upload.85 if getattr(file_obj, 'content_type', None):86 content_type = file_obj.content_type87 elif name:88 content_type, encoding = mimetypes.guess_type(name)89 # Determine the content disposition of the upload.90 if name:91 content_disposition = 'attachment; filename="%s"' % name92 return {93 'Content-Type': content_type or 'application/octet-stream',94 'Content-Disposition': content_disposition or 'attachment'95 }96 def decode_response_content(self, response):97 """98 Given an HTTP response, return the decoded data.99 """100 if not response.content:101 return None102 content_type = response.headers.get('content-type')103 codec = conneg.negotiate_content_type(self.decoders, content_type)104 options = {105 'base_url': response.url106 }107 if 'content-type' in response.headers:108 options['content_type'] = response.headers['content-type']109 if 'content-disposition' in response.headers:110 options['content_disposition'] = response.headers['content-disposition']...

Full Screen

Full Screen

requesthelper.py

Source:requesthelper.py Github

copy

Full Screen

1import logging2import time3from functools import wraps4from json import loads5from json.decoder import JSONDecodeError6from requests import request7from requests import exceptions as request_exception8#9logger = logging.getLogger(__name__)10def wrap_response(func):11 @wraps(func)12 def wrapper(method, url, **kwargs):13 request_info = '{} {} {}'.format(method.upper(), url, kwargs)14 logger.info(f'Prepare start request, request info: {request_info}')15 #16 result = None17 response = None18 decode_response_content = None19 try:20 request_at = time.time()21 response = func(method, url, **kwargs)22 decode_response_content = response.content.decode('utf-8')23 result = loads(decode_response_content)24 except request_exception.ContentDecodingError:25 logger.error(f'Request failed, response content deocde error. '26 f'request info: {request_info}, response content: {response.content}')27 except request_exception.ConnectionError:28 logger.error(f'Request failed, connect error. request info: {request_info}')29 except request_exception.ReadTimeout:30 logger.error(f'Request failed, read timout. request info: {request_info}')31 except request_exception.InvalidURL:32 logger.error(f'Request failed, invalid url. request info: {request_info}')33 except JSONDecodeError:34 logger.error(f'Request failed, json decode error. '35 f'request info: {request_info}, decode response content: {decode_response_content}')36 except:37 logger.exception('Request failed. request info: {}'.format(request_info))38 else:39 logger.info(f'Request succssfully, duration: {int(time.time() - request_at)}s. '40 f'request info: {request_info}, result: {result}')41 return result42 return wrapper43@wrap_response44def _make_request(method, url, **kwargs):45 response = request(method, url, **kwargs)46 return response47class RequestHelper:48 """49 Request helper50 """51 def __init__(self):52 self.default_headers = {"Content-Type": "application/json"}53 self.default_timeout = 254 def _set_default_timeout(self, kwargs: dict):55 if 'timeout' not in kwargs:56 kwargs['timeout'] = self.default_timeout57 def _set_default_headers(self, kwargs: dict):58 headers = kwargs.get('headers', {})59 if headers:60 for name, value in self.default_headers.items():61 if name not in headers:62 headers[name] = value63 else:64 kwargs['headers'] = self.default_headers65 def get(self, url, params=None, **kwargs):66 self._set_default_timeout(kwargs)67 return _make_request('get', url, params=params, **kwargs)68 def post(self, url, data=None, **kwargs):69 self._set_default_timeout(kwargs)70 self._set_default_headers(kwargs)71 return _make_request('post', url, data=data, **kwargs)...

Full Screen

Full Screen

monitoring_server.py

Source:monitoring_server.py Github

copy

Full Screen

...11api = Api(app)12api.add_resource(RegisteringREST, '/register')13class MessageHelper():14 @staticmethod15 def decode_response_content(response: Response) -> str:16 content = response.content17 content = content.decode('utf-8')18 return content19 @staticmethod20 def convert_response_to_dict(response: Response) -> Dict[str, Any]:21 content = MessageHelper.decode_response_content(response)22 content_array = content.split('\n')23 content_array = [24 element for element in content_array if '#' not in element and len(element) > 0]25 content_dict: Dict[str, Any] = {}26 for elem in content_array:27 key, value = elem.split(' ')28 content_dict[key] = value29 return content_dict30class Monitoring():31 def get_device_status(self, device: str = None):32 # get status of all monitored_devices33 if device is None:34 return device_handler.get_all_devices()35 return device_handler.get_device(device)...

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