ringcentral / ringcentral-python / 139

pending completion

Build # 139

Build Type

push

travis-ci-com

Committed by

DaKingKong

Commit Message

increase version number

Run Details

550 of 615 relevant lines covered (89.43%)

1.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.18

/ringcentral/http/api_response.py

# !/usr/bin/env python
# encoding: utf-8
import json
import requests
from email.feedparser import FeedParser
from .json_object import *
from ..core import is_third


class ApiResponse:
    def __init__(self, request=None, response=None):
        self._request = request
        self._response = response

    def ok(self):
        return self._response.ok

    def raw(self):
        return self._response.raw

    def body(self):
        return self._response.content

    def text(self):
        return self._response.text

    def json_dict(self):
        if not self._is_content_type('application/json'):
            raise Exception('Response is not JSON')

        return self._response.json()

    def json(self):
        return unfold(self._response.json())

    def multipart(self):
        if not self._is_content_type('multipart/mixed'):
            raise Exception('Response is not Batch (Multipart)')

        parts = self._break_into_parts()

        if len(parts) < 1:
            raise Exception("Malformed Batch Response (not enough parts)")  # sic! not specific extension

        statuses = json.loads(parts[0].get_payload())

        if len(statuses["response"]) != len(parts) - 1:
            raise Exception("Malformed Batch Response (not-consistent number of parts)")

        responses = []

        for response, payload in zip(statuses["response"], parts[1:]):
            res = create_response(payload, response['status'])

            responses.append(ApiResponse(response=res))

        return responses

    def error(self):

        if self._response is None or self.ok():
            return None

        message = 'HTTP' + str(self._response.status_code)  # TODO Status text

        try:
            data = self.json_dict()

            if 'message' in data:
                message = data['message']
            elif 'error_description' in data:
                message = data['error_description']
            elif 'description' in data:
                message = data['description']

        except Exception as e:
            message = message + ' (and additional error happened during JSON parse: ' + e.message + ')'

        return message

    def request(self):
        return self._request

    def response(self):
        return self._response

    def _is_content_type(self, content_type):
        return self._get_content_type().lower().find(content_type) >= 0

    def _get_content_type(self):
        return self._response.headers.get('Content-Type') or ''

    def _break_into_parts(self):
        p = FeedParser()

        for h in self._response.headers:
            p.feed(h + ':' + self._response.headers[h] + "\r\n")

        p.feed("\r\n")
        p.feed(self.text())

        msg = p.close()

        parts = msg.get_payload()

        return parts


def create_response(payload, status):
    res = requests.Response()
    res.headers = dict(payload)
    if is_third():
        res._content = bytes(payload.get_payload(), 'utf8')
        res.encoding = 'utf8'
    else:
        res._content = str(payload.get_payload())
    res.status_code = status
    return res

1	# !/usr/bin/env python
2	# encoding: utf-8
3	import json	2✔
4	import requests	2✔
5	from email.feedparser import FeedParser	2✔
6	from .json_object import *	2✔
7	from ..core import is_third	2✔
8
9
10	class ApiResponse:	2✔
11	def __init__(self, request=None, response=None):	2✔
12	self._request = request	2✔
13	self._response = response	2✔
14
15	def ok(self):	2✔
16	return self._response.ok	2✔
17
18	def raw(self):	2✔
19	return self._response.raw	×
20
21	def body(self):	2✔
22	return self._response.content	×
23
24	def text(self):	2✔
25	return self._response.text	2✔
26
27	def json_dict(self):	2✔
28	if not self._is_content_type('application/json'):	2✔
29	raise Exception('Response is not JSON')	×
30
31	return self._response.json()	2✔
32
33	def json(self):	2✔
34	return unfold(self._response.json())	2✔
35
36	def multipart(self):	2✔
37	if not self._is_content_type('multipart/mixed'):	2✔
38	raise Exception('Response is not Batch (Multipart)')	×
39
40	parts = self._break_into_parts()	2✔
41
42	if len(parts) < 1:	2✔
43	raise Exception("Malformed Batch Response (not enough parts)") # sic! not specific extension	×
44
45	statuses = json.loads(parts[0].get_payload())	2✔
46
47	if len(statuses["response"]) != len(parts) - 1:	2✔
48	raise Exception("Malformed Batch Response (not-consistent number of parts)")	×
49
50	responses = []	2✔
51
52	for response, payload in zip(statuses["response"], parts[1:]):	2✔
53	res = create_response(payload, response['status'])	2✔
54
55	responses.append(ApiResponse(response=res))	2✔
56
57	return responses	2✔
58
59	def error(self):	2✔
60
61	if self._response is None or self.ok():	2✔
62	return None	×
63
64	message = 'HTTP' + str(self._response.status_code) # TODO Status text	2✔
65
66	try:	2✔
67	data = self.json_dict()	2✔
68
69	if 'message' in data:	2✔
70	message = data['message']	2✔
71	elif 'error_description' in data:	2✔
72	message = data['error_description']	2✔
73	elif 'description' in data:	2✔
74	message = data['description']	2✔
75
76	except Exception as e:	×
77	message = message + ' (and additional error happened during JSON parse: ' + e.message + ')'	×
78
79	return message	2✔
80
81	def request(self):	2✔
82	return self._request	×
83
84	def response(self):	2✔
85	return self._response	2✔
86
87	def _is_content_type(self, content_type):	2✔
88	return self._get_content_type().lower().find(content_type) >= 0	2✔
89
90	def _get_content_type(self):	2✔
91	return self._response.headers.get('Content-Type') or ''	2✔
92
93	def _break_into_parts(self):	2✔
94	p = FeedParser()	2✔
95
96	for h in self._response.headers:	2✔
97	p.feed(h + ':' + self._response.headers[h] + "\r\n")	2✔
98
99	p.feed("\r\n")	2✔
100	p.feed(self.text())	2✔
101
102	msg = p.close()	2✔
103
104	parts = msg.get_payload()	2✔
105
106	return parts	2✔
107
108
109	def create_response(payload, status):	2✔
110	res = requests.Response()	2✔
111	res.headers = dict(payload)	2✔
112	if is_third():	2✔
113	res._content = bytes(payload.get_payload(), 'utf8')	1✔
114	res.encoding = 'utf8'	1✔
115	else:
116	res._content = str(payload.get_payload())	1✔
117	res.status_code = status	2✔
118	return res	2✔

ringcentral / ringcentral-python / 139

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous