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

88.41

/ringcentral/platform/platform.py

#!/usr/bin/env python
# encoding: utf-8

import sys
try:
    from urllib.parse import urlparse
except ImportError:
    from urlparse import urlparse
from observable import Observable
from functools import reduce
from .auth import Auth
from .events import Events
from ..core import base64encode

ACCOUNT_ID = '~'
ACCOUNT_PREFIX = '/account/'
URL_PREFIX = '/restapi'
TOKEN_ENDPOINT = '/restapi/oauth/token'
REVOKE_ENDPOINT = '/restapi/oauth/revoke'
AUTHORIZE_ENDPOINT = '/restapi/oauth/authorize'
API_VERSION = 'v1.0'
ACCESS_TOKEN_TTL = 3600  # 60 minutes
REFRESH_TOKEN_TTL = 604800  # 1 week
KNOWN_PREFIXES = [
    URL_PREFIX,
    '/rcvideo',
    '/analytics',
    '/ai',
    '/scim'
]


class Platform(Observable):
    def __init__(self, client, key='', secret='', server='', name='', version='', redirect_uri='',
                 known_prefixes=None):

        Observable.__init__(self)

        self._server = server
        self._key = key
        self._name = name if name else 'Unnamed'
        self._version = version if version else '0.0.0'
        self._redirect_uri = redirect_uri
        self._secret = secret
        self._client = client
        self._auth = Auth()
        self._account = ACCOUNT_ID
        self._known_prefixes = known_prefixes if known_prefixes else KNOWN_PREFIXES
        self._userAgent = ((self._name + ('/' + self._version if self._version else '') + ' ') if self._name else '') + \
                          sys.platform + '/VERSION' + ' ' + \
                          'PYTHON/VERSION ' + \
                          'RCPYTHONSDK/VERSION'

    def auth(self):
        return self._auth

    def create_url(self, url, add_server=False, add_method=None, add_token=False):
        built_url = ''
        has_http = url.startswith('http://') or url.startswith('https://')

        if add_server and not has_http:
            built_url += self._server

        if not reduce(lambda res, prefix: res if res else url.find(prefix) == 0, self._known_prefixes, False) and not has_http:
            built_url += URL_PREFIX + '/' + API_VERSION

        if url.find(ACCOUNT_PREFIX) >= 0:
            built_url = built_url.replace(ACCOUNT_PREFIX + ACCOUNT_ID, ACCOUNT_PREFIX + self._account)

        built_url += url

        if add_method:
            built_url += ('&' if built_url.find('?') >= 0 else '?') + '_method=' + add_method

        if add_token:
            built_url += ('&' if built_url.find('?') >= 0 else '?') + 'access_token=' + self._auth.access_token()

        return built_url

    def logged_in(self):
        try:
            return self._auth.access_token_valid() or self.refresh()
        except:
            return False

    def login_url(self, redirect_uri, state='', challenge='', challenge_method='S256'):
        built_url = self.create_url( AUTHORIZE_ENDPOINT, add_server=True )
        built_url += '?response_type=code&client_id=' + self._key + '&redirect_uri=' + urllib.parse.quote(redirect_uri)
        if state:
            built_url += '&state=' + urllib.parse.quote(state)
        if challenge:
            built_url += '&code_challenge=' + urllib.parse.quote(challenge) + '&code_challenge_method=' + challenge_method
        return built_url
        
    def login(self, username='', extension='', password='', code='', redirect_uri='', jwt='', verifier=''):
        try:
            if not code and not username and not password and not jwt:
                raise Exception('Either code, or username with password, or jwt has to be provided')
            if not code and not jwt:
                body = {
                    'grant_type': 'password',
                    'username': username,
                    'password': password,
                    'access_token_ttl': ACCESS_TOKEN_TTL,
                    'refresh_token_ttl': REFRESH_TOKEN_TTL
                }
                if extension:
                    body['extension'] = extension
            elif jwt:
                body = {
                    'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
                    'assertion': jwt
                }
            else:
                body = {
                    'grant_type': 'authorization_code',
                    'redirect_uri': redirect_uri if redirect_uri else self._redirect_uri,
                    'code': code
                }
                if verifier:
                    body['code_verifier'] = verifier

            built_url = self.create_url( TOKEN_ENDPOINT, add_server=True )
            response = self._request_token( built_url, body=body)
            self._auth.set_data(response.json_dict())
            self.trigger(Events.loginSuccess, response)
            return response
        except Exception as e:
            self.trigger(Events.loginError, e)
            raise e

    def refresh(self):
        try:
            if not self._auth.refresh_token_valid():
                raise Exception('Refresh token has expired')
            response = self._request_token(TOKEN_ENDPOINT, body={
                'grant_type': 'refresh_token',
                'refresh_token': self._auth.refresh_token(),
                'access_token_ttl': ACCESS_TOKEN_TTL,
                'refresh_token_ttl': REFRESH_TOKEN_TTL
            })
            self._auth.set_data(response.json_dict())
            self.trigger(Events.refreshSuccess, response)
            return response
        except Exception as e:
            self.trigger(Events.refreshError, e)
            raise e

    def logout(self):
        try:
            response = self._request_token(REVOKE_ENDPOINT, body={
                'token': self._auth.access_token()
            })
            self._auth.reset()
            self.trigger(Events.logoutSuccess, response)
            return response
        except Exception as e:
            self.trigger(Events.logoutError, e)
            raise e

    def inflate_request(self, request, skip_auth_check=False):
        if not skip_auth_check:
            self._ensure_authentication()
            request.headers['Authorization'] = self._auth_header()

        request.headers['User-Agent'] = self._userAgent
        request.headers['X-User-Agent'] = self._userAgent
        request.url = self.create_url(request.url, add_server=True)

        return request

    def send_request(self, request, skip_auth_check=False):
        return self._client.send(self.inflate_request(request, skip_auth_check=skip_auth_check))

    def get(self, url, query_params=None, headers=None, skip_auth_check=False):
        request = self._client.create_request('GET', url, query_params=query_params, headers=headers)
        return self.send_request(request, skip_auth_check=skip_auth_check)

    def post(self, url, body=None, query_params=None, headers=None, skip_auth_check=False):
        request = self._client.create_request('POST', url, query_params=query_params, headers=headers, body=body)
        return self.send_request(request, skip_auth_check=skip_auth_check)

    def put(self, url, body=None, query_params=None, headers=None, skip_auth_check=False):
        request = self._client.create_request('PUT', url, query_params=query_params, headers=headers, body=body)
        return self.send_request(request, skip_auth_check=skip_auth_check)

    def delete(self, url, query_params=None, headers=None, skip_auth_check=False):
        request = self._client.create_request('DELETE', url, query_params=query_params, headers=headers)
        return self.send_request(request, skip_auth_check=skip_auth_check)

    def _request_token(self, path='', body=None):
        headers = {
            'Authorization': 'Basic ' + self._api_key(),
            'Content-Type': 'application/x-www-form-urlencoded'
        }
        request = self._client.create_request('POST', path, body=body, headers=headers)
        return self.send_request(request, skip_auth_check=True)

    def _api_key(self):
        return base64encode(self._key + ':' + self._secret)

    def _auth_header(self):
        return self._auth.token_type() + ' ' + self._auth.access_token()

    def _ensure_authentication(self):
        if not self._auth.access_token_valid():
            self.refresh()

1	#!/usr/bin/env python
2	# encoding: utf-8
3
4	import sys	2✔
5	try:	2✔
6	from urllib.parse import urlparse	2✔
7	except ImportError:	1✔
8	from urlparse import urlparse	1✔
9	from observable import Observable	2✔
10	from functools import reduce	2✔
11	from .auth import Auth	2✔
12	from .events import Events	2✔
13	from ..core import base64encode	2✔
14
15	ACCOUNT_ID = '~'	2✔
16	ACCOUNT_PREFIX = '/account/'	2✔
17	URL_PREFIX = '/restapi'	2✔
18	TOKEN_ENDPOINT = '/restapi/oauth/token'	2✔
19	REVOKE_ENDPOINT = '/restapi/oauth/revoke'	2✔
20	AUTHORIZE_ENDPOINT = '/restapi/oauth/authorize'	2✔
21	API_VERSION = 'v1.0'	2✔
22	ACCESS_TOKEN_TTL = 3600 # 60 minutes	2✔
23	REFRESH_TOKEN_TTL = 604800 # 1 week	2✔
24	KNOWN_PREFIXES = [	2✔
25	URL_PREFIX,
26	'/rcvideo',
27	'/analytics',
28	'/ai',
29	'/scim'
30	]
31
32
33	class Platform(Observable):	2✔
34	def __init__(self, client, key='', secret='', server='', name='', version='', redirect_uri='',	2✔
35	known_prefixes=None):
36
37	Observable.__init__(self)	2✔
38
39	self._server = server	2✔
40	self._key = key	2✔
41	self._name = name if name else 'Unnamed'	2✔
42	self._version = version if version else '0.0.0'	2✔
43	self._redirect_uri = redirect_uri	2✔
44	self._secret = secret	2✔
45	self._client = client	2✔
46	self._auth = Auth()	2✔
47	self._account = ACCOUNT_ID	2✔
48	self._known_prefixes = known_prefixes if known_prefixes else KNOWN_PREFIXES	2✔
49	self._userAgent = ((self._name + ('/' + self._version if self._version else '') + ' ') if self._name else '') + \	2✔
50	sys.platform + '/VERSION' + ' ' + \
51	'PYTHON/VERSION ' + \
52	'RCPYTHONSDK/VERSION'
53
54	def auth(self):	2✔
55	return self._auth	2✔
56
57	def create_url(self, url, add_server=False, add_method=None, add_token=False):	2✔
58	built_url = ''	2✔
59	has_http = url.startswith('http://') or url.startswith('https://')	2✔
60
61	if add_server and not has_http:	2✔
62	built_url += self._server	2✔
63
64	if not reduce(lambda res, prefix: res if res else url.find(prefix) == 0, self._known_prefixes, False) and not has_http:	2✔
65	built_url += URL_PREFIX + '/' + API_VERSION	2✔
66
67	if url.find(ACCOUNT_PREFIX) >= 0:	2✔
68	built_url = built_url.replace(ACCOUNT_PREFIX + ACCOUNT_ID, ACCOUNT_PREFIX + self._account)	2✔
69
70	built_url += url	2✔
71
72	if add_method:	2✔
73	built_url += ('&' if built_url.find('?') >= 0 else '?') + '_method=' + add_method	2✔
74
75	if add_token:	2✔
76	built_url += ('&' if built_url.find('?') >= 0 else '?') + 'access_token=' + self._auth.access_token()	2✔
77
78	return built_url	2✔
79
80	def logged_in(self):	2✔
81	try:	2✔
82	return self._auth.access_token_valid() or self.refresh()	2✔
83	except:	2✔
84	return False	2✔
85
86	def login_url(self, redirect_uri, state='', challenge='', challenge_method='S256'):	2✔
87	built_url = self.create_url( AUTHORIZE_ENDPOINT, add_server=True )	×
88	built_url += '?response_type=code&client_id=' + self._key + '&redirect_uri=' + urllib.parse.quote(redirect_uri)	×
89	if state:	×
90	built_url += '&state=' + urllib.parse.quote(state)	×
91	if challenge:	×
92	built_url += '&code_challenge=' + urllib.parse.quote(challenge) + '&code_challenge_method=' + challenge_method	×
93	return built_url	×
94
95	def login(self, username='', extension='', password='', code='', redirect_uri='', jwt='', verifier=''):	2✔
96	try:	2✔
97	if not code and not username and not password and not jwt:	2✔
98	raise Exception('Either code, or username with password, or jwt has to be provided')	2✔
99	if not code and not jwt:	2✔
100	body = {	2✔
101	'grant_type': 'password',
102	'username': username,
103	'password': password,
104	'access_token_ttl': ACCESS_TOKEN_TTL,
105	'refresh_token_ttl': REFRESH_TOKEN_TTL
106	}
107	if extension:	2✔
108	body['extension'] = extension	×
109	elif jwt:	2✔
110	body = {	×
111	'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
112	'assertion': jwt
113	}
114	else:
115	body = {	2✔
116	'grant_type': 'authorization_code',
117	'redirect_uri': redirect_uri if redirect_uri else self._redirect_uri,
118	'code': code
119	}
120	if verifier:	2✔
121	body['code_verifier'] = verifier	×
122
123	built_url = self.create_url( TOKEN_ENDPOINT, add_server=True )	2✔
124	response = self._request_token( built_url, body=body)	2✔
125	self._auth.set_data(response.json_dict())	2✔
126	self.trigger(Events.loginSuccess, response)	2✔
127	return response	2✔
128	except Exception as e:	2✔
129	self.trigger(Events.loginError, e)	2✔
130	raise e	2✔
131
132	def refresh(self):	2✔
133	try:	2✔
134	if not self._auth.refresh_token_valid():	2✔
135	raise Exception('Refresh token has expired')	2✔
136	response = self._request_token(TOKEN_ENDPOINT, body={	2✔
137	'grant_type': 'refresh_token',
138	'refresh_token': self._auth.refresh_token(),
139	'access_token_ttl': ACCESS_TOKEN_TTL,
140	'refresh_token_ttl': REFRESH_TOKEN_TTL
141	})
142	self._auth.set_data(response.json_dict())	2✔
143	self.trigger(Events.refreshSuccess, response)	2✔
144	return response	2✔
145	except Exception as e:	2✔
146	self.trigger(Events.refreshError, e)	2✔
147	raise e	2✔
148
149	def logout(self):	2✔
150	try:	2✔
151	response = self._request_token(REVOKE_ENDPOINT, body={	2✔
152	'token': self._auth.access_token()
153	})
154	self._auth.reset()	2✔
155	self.trigger(Events.logoutSuccess, response)	2✔
156	return response	2✔
157	except Exception as e:	×
158	self.trigger(Events.logoutError, e)	×
159	raise e	×
160
161	def inflate_request(self, request, skip_auth_check=False):	2✔
162	if not skip_auth_check:	2✔
163	self._ensure_authentication()	2✔
164	request.headers['Authorization'] = self._auth_header()	2✔
165
166	request.headers['User-Agent'] = self._userAgent	2✔
167	request.headers['X-User-Agent'] = self._userAgent	2✔
168	request.url = self.create_url(request.url, add_server=True)	2✔
169
170	return request	2✔
171
172	def send_request(self, request, skip_auth_check=False):	2✔
173	return self._client.send(self.inflate_request(request, skip_auth_check=skip_auth_check))	2✔
174
175	def get(self, url, query_params=None, headers=None, skip_auth_check=False):	2✔
176	request = self._client.create_request('GET', url, query_params=query_params, headers=headers)	×
177	return self.send_request(request, skip_auth_check=skip_auth_check)	×
178
179	def post(self, url, body=None, query_params=None, headers=None, skip_auth_check=False):	2✔
180	request = self._client.create_request('POST', url, query_params=query_params, headers=headers, body=body)	2✔
181	return self.send_request(request, skip_auth_check=skip_auth_check)	2✔
182
183	def put(self, url, body=None, query_params=None, headers=None, skip_auth_check=False):	2✔
184	request = self._client.create_request('PUT', url, query_params=query_params, headers=headers, body=body)	2✔
185	return self.send_request(request, skip_auth_check=skip_auth_check)	2✔
186
187	def delete(self, url, query_params=None, headers=None, skip_auth_check=False):	2✔
188	request = self._client.create_request('DELETE', url, query_params=query_params, headers=headers)	2✔
189	return self.send_request(request, skip_auth_check=skip_auth_check)	2✔
190
191	def _request_token(self, path='', body=None):	2✔
192	headers = {	2✔
193	'Authorization': 'Basic ' + self._api_key(),
194	'Content-Type': 'application/x-www-form-urlencoded'
195	}
196	request = self._client.create_request('POST', path, body=body, headers=headers)	2✔
197	return self.send_request(request, skip_auth_check=True)	2✔
198
199	def _api_key(self):	2✔
200	return base64encode(self._key + ':' + self._secret)	2✔
201
202	def _auth_header(self):	2✔
203	return self._auth.token_type() + ' ' + self._auth.access_token()	2✔
204
205	def _ensure_authentication(self):	2✔
206	if not self._auth.access_token_valid():	2✔
207	self.refresh()	×

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