Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use decode_response_content method in gabbi

Best Python code snippet using gabbi_python

transports.py

Source:transports.py

...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']...

requesthelper.py

Source:requesthelper.py

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

monitoring_server.py

Source:monitoring_server.py

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

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.