9863694749

Committed 09 Jul 2024 08:24PM UTC coverage: 98.413% (-0.08%) from 98.493%

Build # 9863694749

Build Type

push

github

Committed by

hhertoghe

Commit Message

Rework tests

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

20 existing lines in 2 files now uncovered.

3348 of 3402 relevant lines covered (98.41%)

0.98 hits per line

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

85.19

/chift/api/client.py

import http.client as httplib
import json
from datetime import datetime

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

from chift.api import exceptions


class ChiftAuth(requests.auth.AuthBase):
    def __init__(self, client_id, client_secret, account_id, url_base, env_id, test_client):
        self.client_id = client_id
        self.client_secret = client_secret
        self.account_id = account_id
        self.url_base = url_base
        self.env_id = env_id
        self.test_client = test_client
        self.request_engine = self.test_client or requests

        self.access_token = None
        self.exires_at = None

    def __call__(self, request):
        request.headers.update(self.get_auth_header())
        return request

    def get_auth_header(self):
        return {"Authorization": f"Bearer {self.get_access_token()}"}

    def _parse_token(self, token):
        self.access_token = token.get("access_token")
        self.exires_at = datetime.fromtimestamp(token.get("expires_on"))

    def get_access_token(self):
        if self.access_token:
            if datetime.now() < self.exires_at:
                return self.access_token

        payload = {
            "clientId": self.client_id,
            "clientSecret": self.client_secret,
            "accountId": self.account_id,
        }
        if self.env_id:
            payload["envId"] = self.env_id
        
        response = self.request_engine.post(self.url_base + "/token", json=payload)

        if not response.status_code == httplib.OK:
            raise exceptions.ChiftException(
                f"Error while authenticating '{response.status_code}': {response.text}"
            )

        self._parse_token(response.json())

        return self.access_token


class ChiftClient:
    session = None
    access_token = None
    auth = None
    consumer_id = None
    connection_id = None
    related_chain_execution_id = None

    __instance = None
    __use_global = False

    def __new__(cls, **kwargs):
        """
        __use_global=True for singleton usage
        """
        if ChiftClient.__use_global:
            if ChiftClient.__instance is None:
                ChiftClient.__instance = object.__new__(cls)
            instance = ChiftClient.__instance
        else:
            instance = object.__new__(cls)

        return instance

    def __init__(self, consumer_id=None, **kwargs):
        from chift import (
            account_id,
            client_id,
            client_secret,
            related_chain_execution_id,
            url_base
        )

        self.consumer_id = consumer_id
        self.client_id = kwargs.get("client_id") or client_id
        self.client_secret = kwargs.get("client_secret") or client_secret
        self.account_id = kwargs.get("account_id") or account_id
        self.url_base = kwargs.get("url_base") or url_base
        self.env_id = kwargs.get("env_id")
        self.test_client = kwargs.get("test_client")
        self.related_chain_execution_id = (
            kwargs.get("related_chain_execution_id") or related_chain_execution_id
        )
        self.max_retries = kwargs.get("max_retries")
        self._start_session()

    def _start_session(self):
        if self.test_client:
            self.url_base = "" # set to empty string to avoid url_base in the request
            self.test_auth = ChiftAuth(
                self.client_id,
                self.client_secret,
                self.account_id,
                self.url_base,
                self.env_id,
                self.test_client
            )
        elif not self.session:
            self.session = requests.Session()
            self.session.auth = ChiftAuth(
                self.client_id,
                self.client_secret,
                self.account_id,
                self.url_base,
                self.env_id,
                None
            )

            if self.max_retries:
                retries = Retry(
                    total=int(self.max_retries),
                    backoff_factor=0.1,
                    status_forcelist=[502],
                )
                self.session.mount("http://", HTTPAdapter(max_retries=retries))
                self.session.mount("https://", HTTPAdapter(max_retries=retries))

    def make_request(
        self, request_type, url, data=None, content_type="application/json", params=None
    ):
        if not params:
            params = {}

        if isinstance(data, str):
            data = json.loads(data)

        headers = {
            "Content-Type": content_type,
            "Accept": "application/json",
            "User-Agent": "chift-python-sdk library",
        }

        if self.connection_id:
            headers["x-chift-connectionid"] = self.connection_id

        if self.related_chain_execution_id:
            headers["x-chift-relatedchainexecutionid"] = self.related_chain_execution_id

        try:
            req = self.process_request(
                request_type, url, headers=headers, params=params, json=data
            )
        except requests.exceptions.RetryError as e:
            raise exceptions.ChiftException(
                f"After {self.max_retries} retries, the request failed."
            )

        if req.status_code == httplib.UNAUTHORIZED:
            raise exceptions.ChiftException(
                "Application authentication failed",
                error_code=req.status_code,
                detail=req.text,
            )

        if req.status_code == 204:
            return True

        try:
            result = req.json()
        except:
            raise exceptions.ChiftException(f"Error parsing json response: {req.text}")

        if isinstance(result, dict) and result.get("status") == "error":
            raise exceptions.ChiftException(
                result.get("message"),
                error_code=result.get("error_code"),
                detail=result.get("detail"),
            )
        elif not req.status_code == httplib.OK:
            raise exceptions.ChiftException(
                f"Error returned with status code '{req.status_code}': {req.text}"
            )
        else:
            return result

    def process_request(
        self, request_type, url_path, headers=None, params=None, json=None
    ):
        engine = self.test_client or self.session
        if self.test_client:
            headers.update(self.test_auth.get_auth_header())
        return engine.request(
            request_type,
            self.url_base + url_path,
            headers=headers,
            params=params,
            json=json,
        )

    def get(self, *args, **kwargs):
        return self.make_request("GET", *args, **kwargs)

    def post(self, *args, **kwargs):
        return self.make_request("POST", *args, **kwargs)

    def patch(self, *args, **kwargs):
        return self.make_request("PATCH", *args, **kwargs)

    def delete(self, *args, **kwargs):
        return self.make_request("DELETE", *args, **kwargs)

    def path_builder(self, ordered_paths):
        if ordered_paths:
            return "/" + "/".join(
                [str(path).strip("/") for path in ordered_paths if path]
            )

    def delete_one(
        self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.delete(url_path, params=params)

    def get_one(
        self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.get(url_path, params=params)

    def get_all(self, chift_vertical, chift_model, params=None, extra_path=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                extra_path,
            ]
        )

        return self.get(url_path, params=params)

    def post_one(self, chift_vertical, chift_model, data, extra_path=None, params=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                extra_path,
            ]
        )

        return self.post(url_path, data=data, params=params)

    def update_one(
        self, chift_vertical, chift_model, chift_id, data, extra_path=None, params=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.patch(url_path, data=data, params=params)

1	import http.client as httplib	1✔
2	import json	1✔
3	from datetime import datetime	1✔
4
5	import requests	1✔
6	from requests.adapters import HTTPAdapter	1✔
7	from requests.packages.urllib3.util.retry import Retry	1✔
8
9	from chift.api import exceptions	1✔
10
11
12	class ChiftAuth(requests.auth.AuthBase):	1✔
13	def __init__(self, client_id, client_secret, account_id, url_base, env_id, test_client):	1✔
14	self.client_id = client_id	1✔
15	self.client_secret = client_secret	1✔
16	self.account_id = account_id	1✔
17	self.url_base = url_base	1✔
18	self.env_id = env_id	1✔
19	self.test_client = test_client	1✔
20	self.request_engine = self.test_client or requests	1✔
21
22	self.access_token = None	1✔
23	self.exires_at = None	1✔
24
25	def __call__(self, request):	1✔
26	request.headers.update(self.get_auth_header())	1✔
27	return request	1✔
28
29	def get_auth_header(self):	1✔
30	return {"Authorization": f"Bearer {self.get_access_token()}"}	1✔
31
32	def _parse_token(self, token):	1✔
33	self.access_token = token.get("access_token")	1✔
34	self.exires_at = datetime.fromtimestamp(token.get("expires_on"))	1✔
35
36	def get_access_token(self):	1✔
37	if self.access_token:	1✔
38	if datetime.now() < self.exires_at:	1✔
39	return self.access_token	1✔
40
41	payload = {	1✔
42	"clientId": self.client_id,
43	"clientSecret": self.client_secret,
44	"accountId": self.account_id,
45	}
46	if self.env_id:	1✔
UNCOV 47	payload["envId"] = self.env_id	×
48
49	response = self.request_engine.post(self.url_base + "/token", json=payload)	1✔
50
51	if not response.status_code == httplib.OK:	1✔
UNCOV 52	raise exceptions.ChiftException(	×
53	f"Error while authenticating '{response.status_code}': {response.text}"
54	)
55
56	self._parse_token(response.json())	1✔
57
58	return self.access_token	1✔
59
60
61	class ChiftClient:	1✔
62	session = None	1✔
63	access_token = None	1✔
64	auth = None	1✔
65	consumer_id = None	1✔
66	connection_id = None	1✔
67	related_chain_execution_id = None	1✔
68
69	__instance = None	1✔
70	__use_global = False	1✔
71
72	def __new__(cls, **kwargs):	1✔
73	"""
74	__use_global=True for singleton usage
75	"""
76	if ChiftClient.__use_global:	1✔
UNCOV 77	if ChiftClient.__instance is None:	×
UNCOV 78	ChiftClient.__instance = object.__new__(cls)	×
UNCOV 79	instance = ChiftClient.__instance	×
80	else:
81	instance = object.__new__(cls)	1✔
82
83	return instance	1✔
84
85	def __init__(self, consumer_id=None, **kwargs):	1✔
86	from chift import (	1✔
87	account_id,
88	client_id,
89	client_secret,
90	related_chain_execution_id,
91	url_base
92	)
93
94	self.consumer_id = consumer_id	1✔
95	self.client_id = kwargs.get("client_id") or client_id	1✔
96	self.client_secret = kwargs.get("client_secret") or client_secret	1✔
97	self.account_id = kwargs.get("account_id") or account_id	1✔
98	self.url_base = kwargs.get("url_base") or url_base	1✔
99	self.env_id = kwargs.get("env_id")	1✔
100	self.test_client = kwargs.get("test_client")	1✔
101	self.related_chain_execution_id = (	1✔
102	kwargs.get("related_chain_execution_id") or related_chain_execution_id
103	)
104	self.max_retries = kwargs.get("max_retries")	1✔
105	self._start_session()	1✔
106
107	def _start_session(self):	1✔
108	if self.test_client:	1✔
UNCOV 109	self.url_base = "" # set to empty string to avoid url_base in the request	×
UNCOV 110	self.test_auth = ChiftAuth(	×
111	self.client_id,
112	self.client_secret,
113	self.account_id,
114	self.url_base,
115	self.env_id,
116	self.test_client
117	)
118	elif not self.session:	1✔
119	self.session = requests.Session()	1✔
120	self.session.auth = ChiftAuth(	1✔
121	self.client_id,
122	self.client_secret,
123	self.account_id,
124	self.url_base,
125	self.env_id,
126	None
127	)
128
129	if self.max_retries:	1✔
130	retries = Retry(	×
131	total=int(self.max_retries),
132	backoff_factor=0.1,
133	status_forcelist=[502],
134	)
UNCOV 135	self.session.mount("http://", HTTPAdapter(max_retries=retries))	×
UNCOV 136	self.session.mount("https://", HTTPAdapter(max_retries=retries))	×
137
138	def make_request(	1✔
139	self, request_type, url, data=None, content_type="application/json", params=None
140	):
141	if not params:	1✔
142	params = {}	1✔
143
144	if isinstance(data, str):	1✔
UNCOV 145	data = json.loads(data)	×
146
147	headers = {	1✔
148	"Content-Type": content_type,
149	"Accept": "application/json",
150	"User-Agent": "chift-python-sdk library",
151	}
152
153	if self.connection_id:	1✔
154	headers["x-chift-connectionid"] = self.connection_id	1✔
155
156	if self.related_chain_execution_id:	1✔
UNCOV 157	headers["x-chift-relatedchainexecutionid"] = self.related_chain_execution_id	×
158
159	try:	1✔
160	req = self.process_request(	1✔
161	request_type, url, headers=headers, params=params, json=data
162	)
UNCOV 163	except requests.exceptions.RetryError as e:	×
UNCOV 164	raise exceptions.ChiftException(	×
165	f"After {self.max_retries} retries, the request failed."
166	)
167
168	if req.status_code == httplib.UNAUTHORIZED:	1✔
UNCOV 169	raise exceptions.ChiftException(	×
170	"Application authentication failed",
171	error_code=req.status_code,
172	detail=req.text,
173	)
174
175	if req.status_code == 204:	1✔
UNCOV 176	return True	×
177
178	try:	1✔
179	result = req.json()	1✔
UNCOV 180	except:	×
UNCOV 181	raise exceptions.ChiftException(f"Error parsing json response: {req.text}")	×
182
183	if isinstance(result, dict) and result.get("status") == "error":	1✔
184	raise exceptions.ChiftException(	1✔
185	result.get("message"),
186	error_code=result.get("error_code"),
187	detail=result.get("detail"),
188	)
189	elif not req.status_code == httplib.OK:	1✔
UNCOV 190	raise exceptions.ChiftException(	×
191	f"Error returned with status code '{req.status_code}': {req.text}"
192	)
193	else:
194	return result	1✔
195
196	def process_request(	1✔
197	self, request_type, url_path, headers=None, params=None, json=None
198	):
199	engine = self.test_client or self.session	1✔
200	if self.test_client:	1✔
UNCOV 201	headers.update(self.test_auth.get_auth_header())	×
202	return engine.request(	1✔
203	request_type,
204	self.url_base + url_path,
205	headers=headers,
206	params=params,
207	json=json,
208	)
209
210	def get(self, args, *kwargs):	1✔
211	return self.make_request("GET", args, *kwargs)	1✔
212
213	def post(self, args, *kwargs):	1✔
214	return self.make_request("POST", args, *kwargs)	1✔
215
216	def patch(self, args, *kwargs):	1✔
217	return self.make_request("PATCH", args, *kwargs)	1✔
218
219	def delete(self, args, *kwargs):	1✔
220	return self.make_request("DELETE", args, *kwargs)	1✔
221
222	def path_builder(self, ordered_paths):	1✔
223	if ordered_paths:	1✔
224	return "/" + "/".join(	1✔
225	[str(path).strip("/") for path in ordered_paths if path]
226	)
227
228	def delete_one(	1✔
229	self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
230	):
231	url_path = self.path_builder(	1✔
232	[
233	"consumers" if self.consumer_id else None,
234	self.consumer_id,
235	chift_vertical,
236	chift_model,
237	chift_id,
238	extra_path,
239	]
240	)
241
242	return self.delete(url_path, params=params)	1✔
243
244	def get_one(	1✔
245	self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
246	):
247	url_path = self.path_builder(	1✔
248	[
249	"consumers" if self.consumer_id else None,
250	self.consumer_id,
251	chift_vertical,
252	chift_model,
253	chift_id,
254	extra_path,
255	]
256	)
257
258	return self.get(url_path, params=params)	1✔
259
260	def get_all(self, chift_vertical, chift_model, params=None, extra_path=None):	1✔
261	url_path = self.path_builder(	1✔
262	[
263	"consumers" if self.consumer_id else None,
264	self.consumer_id,
265	chift_vertical,
266	chift_model,
267	extra_path,
268	]
269	)
270
271	return self.get(url_path, params=params)	1✔
272
273	def post_one(self, chift_vertical, chift_model, data, extra_path=None, params=None):	1✔
274	url_path = self.path_builder(	1✔
275	[
276	"consumers" if self.consumer_id else None,
277	self.consumer_id,
278	chift_vertical,
279	chift_model,
280	extra_path,
281	]
282	)
283
284	return self.post(url_path, data=data, params=params)	1✔
285
286	def update_one(	1✔
287	self, chift_vertical, chift_model, chift_id, data, extra_path=None, params=None
288	):
289	url_path = self.path_builder(	1✔
290	[
291	"consumers" if self.consumer_id else None,
292	self.consumer_id,
293	chift_vertical,
294	chift_model,
295	chift_id,
296	extra_path,
297	]
298	)
299
300	return self.patch(url_path, data=data, params=params)	1✔

chift-oneapi / chift-python-sdk / 9863694749

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