From 877736faae61eace2fe256739f4d6fdaccb99b6d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Cruz?= Date: Thu, 28 Apr 2022 17:36:41 -0300 Subject: [PATCH 1/5] feat(base_request): creates the class wrapping get/post, adds user-agent header --- incognia/__init__.py | 3 ++- incognia/base_request.py | 56 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) create mode 100644 incognia/base_request.py diff --git a/incognia/__init__.py b/incognia/__init__.py index 7f3a6a2..464d643 100644 --- a/incognia/__init__.py +++ b/incognia/__init__.py @@ -12,4 +12,5 @@ 'feedback_events', 'json_util', 'models', - 'token_manager'] + 'token_manager', + 'base_request'] diff --git a/incognia/base_request.py b/incognia/base_request.py new file mode 100644 index 0000000..1155e1c --- /dev/null +++ b/incognia/base_request.py @@ -0,0 +1,56 @@ +import json +import platform +import sys +from typing import Final, Any, Union, Optional + +import requests + +from incognia.exceptions import IncogniaHTTPError + +_LIBRARY_VERSION: Final[str] = sys.modules['incognia'].__version__ +_OS_NAME: Final[str] = platform.system() +_OS_VERSION: Final[str] = platform.release() +_OS_ARCH: Final[str] = platform.architecture()[0] +_LANGUAGE_VERSION: Final[str] = platform.python_version() + +USER_AGENT_HEADER: Final[dict] = { + 'User-Agent': f'incognia-python/{_LIBRARY_VERSION} ({_OS_NAME} {_OS_VERSION} {_OS_ARCH}) Python/{_LANGUAGE_VERSION}' +} + +JSON_CONTENT_HEADER: Final[dict] = { + 'Content-Type': 'application/json' +} + + +class BaseRequest: + def __init__(self, timeout: float = 5.0): + self.__timeout: float = timeout + + def timeout(self) -> float: + return self.__timeout + + def post(self, url: Union[str, bytes], headers: Any = None, data: Any = None, params: Any = None, + auth: Optional[Any] = None) -> Optional[dict]: + headers = headers or {} + headers.update(USER_AGENT_HEADER) + + try: + response = requests.post(url=url, headers=headers, data=data, params=params, timeout=self.__timeout, + auth=auth) + response.raise_for_status() + return json.loads(response.content.decode('utf-8')) or None + + except requests.HTTPError as e: + raise IncogniaHTTPError(e) from None + + def get(self, url: Union[str, bytes], headers: Any = None, data: Any = None) -> Optional[dict]: + headers = headers or {} + headers.update(USER_AGENT_HEADER) + + try: + response = requests.get(url=url, headers=headers, data=data, timeout=self.__timeout) + response.raise_for_status() + return json.loads(response.content.decode('utf-8')) or None + + except requests.HTTPError as e: + raise IncogniaHTTPError(e) from None From ed9f06180e4b68e9680c3dcb91d3a9af0f29f3c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Cruz?= Date: Thu, 28 Apr 2022 17:38:19 -0300 Subject: [PATCH 2/5] test(test_base_request): adds initial tests to post and get routes --- tests/test_base_request.py | 77 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 tests/test_base_request.py diff --git a/tests/test_base_request.py b/tests/test_base_request.py new file mode 100644 index 0000000..40305ca --- /dev/null +++ b/tests/test_base_request.py @@ -0,0 +1,77 @@ +from typing import Final +from unittest import TestCase +from unittest.mock import patch, Mock + +import requests + +from incognia.base_request import BaseRequest, USER_AGENT_HEADER +from incognia.exceptions import IncogniaHTTPError +from incognia.json_util import encode + + +class TestBaseRequest(TestCase): + URL: Final[str] = 'https://some-valid-link.com' + JSON_RESPONSE: Final[dict] = { + 'first-key': 'first-value', + 'second-key': 'second-value' + } + OK_STATUS_CODE: Final[int] = 200 + CLIENT_ERROR_CODE: Final[int] = 400 + + @patch('requests.post') + def test_post_when_parameters_are_valid_should_return_a_valid_dict(self, mock_requests_post: Mock): + def get_mocked_response() -> requests.Response: + response = requests.Response() + response._content, response.status_code = encode(self.JSON_RESPONSE), self.OK_STATUS_CODE + return response + + mock_requests_post.configure_mock(return_value=get_mocked_response()) + + base_request = BaseRequest() + result = base_request.post(url=self.URL) + mock_requests_post.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, params=None, + timeout=base_request.timeout(), auth=None) + self.assertEqual(result, self.JSON_RESPONSE) + + @patch('requests.post') + def test_post_when_parameters_are_invalid_should_raise_an_IncogniaHTTPError(self, mock_requests_post: Mock): + def get_mocked_response() -> requests.Response: + response = requests.Response() + response._content, response.status_code = encode(self.JSON_RESPONSE), self.CLIENT_ERROR_CODE + return response + + mock_requests_post.configure_mock(return_value=get_mocked_response()) + + base_request = BaseRequest() + self.assertRaises(IncogniaHTTPError, base_request.post, url=self.URL) + mock_requests_post.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, params=None, + timeout=base_request.timeout(), auth=None) + + @patch('requests.get') + def test_get_when_parameters_are_valid_should_return_a_valid_dict(self, mock_requests_get: Mock): + def get_mocked_response() -> requests.Response: + response = requests.Response() + response._content, response.status_code = encode(self.JSON_RESPONSE), self.OK_STATUS_CODE + return response + + mock_requests_get.configure_mock(return_value=get_mocked_response()) + + base_request = BaseRequest() + result = base_request.get(url=self.URL) + mock_requests_get.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, + timeout=base_request.timeout()) + self.assertEqual(result, self.JSON_RESPONSE) + + @patch('requests.get') + def test_get_when_parameters_are_invalid_should_raise_an_IncogniaHTTPError(self, mock_requests_get: Mock): + def get_mocked_response() -> requests.Response: + response = requests.Response() + response._content, response.status_code = encode(self.JSON_RESPONSE), self.CLIENT_ERROR_CODE + return response + + mock_requests_get.configure_mock(return_value=get_mocked_response()) + + base_request = BaseRequest() + self.assertRaises(IncogniaHTTPError, base_request.get, url=self.URL) + mock_requests_get.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, + timeout=base_request.timeout()) From 36175f6daff9be4c1e8bf3c8fc1750995bd2bc7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Cruz?= Date: Thu, 28 Apr 2022 17:47:15 -0300 Subject: [PATCH 3/5] refactor(token_manager): replaces requests calls with BaseRequest --- incognia/token_manager.py | 34 +++++++++---------- tests/test_token_manager.py | 66 +++++++++++++------------------------ 2 files changed, 39 insertions(+), 61 deletions(-) diff --git a/incognia/token_manager.py b/incognia/token_manager.py index 70e39ca..b9af683 100644 --- a/incognia/token_manager.py +++ b/incognia/token_manager.py @@ -1,13 +1,14 @@ import base64 import datetime as dt -import json +from threading import Lock from typing import Final, Optional, NamedTuple -import requests - +from .base_request import BaseRequest from .endpoints import Endpoints from .exceptions import IncogniaHTTPError +_TOKEN_REFRESH_BEFORE_SECONDS: Final[int] = 10 + class TokenValues(NamedTuple): access_token: str @@ -15,13 +16,13 @@ class TokenValues(NamedTuple): class TokenManager: - TOKEN_REFRESH_BEFORE_SECONDS: Final[int] = 10 - def __init__(self, client_id: str, client_secret: str): self.__client_id: str = client_id self.__client_secret: str = client_secret self.__token_values: Optional[TokenValues] = None self.__expiration_time: Optional[dt.datetime] = None + self.__request: BaseRequest = BaseRequest() + self.__mutex: Lock = Lock() def __refresh_token(self) -> None: client_id, client_secret = self.__client_id, self.__client_secret @@ -30,26 +31,23 @@ def __refresh_token(self) -> None: headers = {'Authorization': f'Basic {client_id_and_secret_encoded}'} try: - response = requests.post(url=Endpoints.TOKEN, headers=headers, - auth=(client_id, client_secret)) - response.raise_for_status() - - parsed_response = json.loads(response.content.decode('utf-8')) - token_values = TokenValues(parsed_response['access_token'], - parsed_response['token_type']) - expiration_time = dt.datetime.now() + dt.timedelta( - seconds=int(parsed_response['expires_in'])) + response = self.__request.post(url=Endpoints.TOKEN, headers=headers, + auth=(client_id, client_secret)) + token_values = TokenValues(response['access_token'], response['token_type']) + expiration_time = dt.datetime.now() + dt.timedelta(seconds=int(response['expires_in'])) self.__token_values, self.__expiration_time = token_values, expiration_time - except requests.HTTPError as e: + except IncogniaHTTPError as e: raise IncogniaHTTPError(e) from None def __is_expired(self) -> bool: - return (self.__expiration_time - dt.datetime.now()).total_seconds() <= \ - self.TOKEN_REFRESH_BEFORE_SECONDS + return (self.__expiration_time - dt.datetime.now()).total_seconds() <= _TOKEN_REFRESH_BEFORE_SECONDS def get(self) -> TokenValues: + self.__mutex.acquire() if not self.__expiration_time or self.__is_expired(): self.__refresh_token() - return self.__token_values + token_values = self.__token_values + self.__mutex.release() + return token_values diff --git a/tests/test_token_manager.py b/tests/test_token_manager.py index c59eeac..64fdd6b 100644 --- a/tests/test_token_manager.py +++ b/tests/test_token_manager.py @@ -3,8 +3,7 @@ from unittest import TestCase from unittest.mock import Mock, patch -import requests - +from incognia.base_request import BaseRequest from incognia.endpoints import Endpoints from incognia.exceptions import IncogniaHTTPError from incognia.token_manager import TokenManager, TokenValues @@ -14,28 +13,26 @@ class TestTokenManager(TestCase): CLIENT_ID: Final[str] = 'ANY_ID' CLIENT_SECRET: Final[str] = 'ANY_SECRET' TOKEN_VALUES: Final[TokenValues] = TokenValues('ACCESS_TOKEN', 'TOKEN_TYPE') - JSON_POST_RESPONSE: Final[ascii] = b'{ "access_token": "ACCESS_TOKEN",' \ - b' "token_type": "TOKEN_TYPE", "expires_in": 900 }' - SHORT_EXPIRATION_TOKEN_VALUES: Final[TokenValues] = TokenValues('S_E_ACCESS_TOKEN', - 'S_E_TOKEN_TYPE') - SHORT_EXPIRATION_JSON_POST_RESPONSE: Final[ascii] = b'{ "access_token": "S_E_ACCESS_TOKEN",' \ - b' "token_type": "S_E_TOKEN_TYPE",' \ - b' "expires_in": 5 }' - CLIENT_ID_AND_SECRET_ENCODED: Final[str] = base64.urlsafe_b64encode( - f'{CLIENT_ID}:{CLIENT_SECRET}'.encode('ascii')).decode('utf-8') + JSON_POST_RESPONSE: Final[dict] = { + 'access_token': 'ACCESS_TOKEN', + 'token_type': 'TOKEN_TYPE', + 'expires_in': 900 + } + SHORT_EXPIRATION_TOKEN_VALUES: Final[TokenValues] = TokenValues('S_E_ACCESS_TOKEN', 'S_E_TOKEN_TYPE') + SHORT_EXPIRATION_JSON_POST_RESPONSE: Final[dict] = { + 'access_token': 'S_E_ACCESS_TOKEN', + 'token_type': 'S_E_TOKEN_TYPE', + 'expires_in': 5 + } + CLIENT_ID_AND_SECRET_ENCODED: Final[str] = base64.urlsafe_b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}' + .encode('ascii')).decode('utf-8') HEADERS: Final[dict] = {'Authorization': f'Basic {CLIENT_ID_AND_SECRET_ENCODED}'} OK_STATUS_CODE: Final[int] = 200 CLIENT_ERROR_CODE: Final[int] = 400 - @patch('requests.post') - def test_get_when_credentials_are_valid_should_return_a_valid_token(self, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response._content, response.status_code = self.JSON_POST_RESPONSE, self.OK_STATUS_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + @patch.object(BaseRequest, 'post') + def test_get_when_credentials_are_valid_should_return_a_valid_token(self, mock_requests_post: Mock): + mock_requests_post.configure_mock(return_value=self.JSON_POST_RESPONSE) token_manager = TokenManager(self.CLIENT_ID, self.CLIENT_SECRET) token_values = token_manager.get() @@ -43,48 +40,31 @@ def get_mocked_response() -> requests.Response: auth=(self.CLIENT_ID, self.CLIENT_SECRET)) self.assertEqual(token_values, self.TOKEN_VALUES) - @patch('requests.post') + @patch.object(BaseRequest, 'post') def test_get_when_credentials_are_invalid_should_raise_an_IncogniaHTTPError(self, mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response.status_code = self.CLIENT_ERROR_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + mock_requests_post.configure_mock(side_effect=IncogniaHTTPError) token_manager = TokenManager(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaHTTPError, token_manager.get) mock_requests_post.assert_called_with(url=Endpoints.TOKEN, headers=self.HEADERS, auth=(self.CLIENT_ID, self.CLIENT_SECRET)) - @patch('requests.post') + @patch.object(BaseRequest, 'post') def test_get_when_token_is_expired_should_return_a_new_valid_token(self, mock_requests_post: Mock): - token_manager = TokenManager(self.CLIENT_ID, self.CLIENT_SECRET) - - def get_first_mocked_response() -> requests.Response: - response = requests.Response() - response._content = self.SHORT_EXPIRATION_JSON_POST_RESPONSE - response.status_code = self.OK_STATUS_CODE - return response - - mock_requests_post.configure_mock(return_value=get_first_mocked_response()) + mock_requests_post.configure_mock(return_value=self.SHORT_EXPIRATION_JSON_POST_RESPONSE) + token_manager = TokenManager(self.CLIENT_ID, self.CLIENT_SECRET) first_token_values = token_manager.get() mock_requests_post.assert_called_with(url=Endpoints.TOKEN, headers=self.HEADERS, auth=(self.CLIENT_ID, self.CLIENT_SECRET)) - def get_second_mocked_response() -> requests.Response: - response = requests.Response() - response._content, response.status_code = self.JSON_POST_RESPONSE, self.OK_STATUS_CODE - return response - mock_requests_post.reset_mock() - mock_requests_post.configure_mock(return_value=get_second_mocked_response()) + mock_requests_post.configure_mock(return_value=self.JSON_POST_RESPONSE) second_token_values = token_manager.get() From acb8145f58a7f6bfc25088fb5478d1108cc89fc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Cruz?= Date: Thu, 28 Apr 2022 17:47:23 -0300 Subject: [PATCH 4/5] refactor(api): replaces requests calls with BaseRequest --- incognia/api.py | 54 +++---- tests/test_api.py | 348 ++++++++++++++++++---------------------------- 2 files changed, 154 insertions(+), 248 deletions(-) diff --git a/incognia/api.py b/incognia/api.py index c3e7dc0..b88dccf 100644 --- a/incognia/api.py +++ b/incognia/api.py @@ -1,8 +1,5 @@ import datetime as dt -import json -from typing import Final, Optional, List - -import requests +from typing import Optional, List from .datetime_util import total_milliseconds_since_epoch from .endpoints import Endpoints @@ -10,13 +7,13 @@ from .json_util import encode from .models import Coordinates, StructuredAddress, TransactionAddress, PaymentValue, PaymentMethod from .token_manager import TokenManager +from .base_request import BaseRequest, JSON_CONTENT_HEADER class IncogniaAPI: - JSON_CONTENT_HEADER: Final[dict] = {'Content-type': 'application/json'} - def __init__(self, client_id: str, client_secret: str): self.__token_manager = TokenManager(client_id, client_secret) + self.__request = BaseRequest() def __get_authorization_header(self) -> dict: access_token, token_type = self.__token_manager.get() @@ -32,7 +29,7 @@ def register_new_signup(self, try: headers = self.__get_authorization_header() - headers.update(self.JSON_CONTENT_HEADER) + headers.update(JSON_CONTENT_HEADER) body = { 'installation_id': installation_id, 'address_line': address_line, @@ -40,12 +37,9 @@ def register_new_signup(self, 'address_coordinates': address_coordinates } data = encode(body) - response = requests.post(Endpoints.SIGNUPS, headers=headers, data=data) - response.raise_for_status() - - return json.loads(response.content.decode('utf-8')) + return self.__request.post(Endpoints.SIGNUPS, headers=headers, data=data) - except requests.HTTPError as e: + except IncogniaHTTPError as e: raise IncogniaHTTPError(e) from None def get_signup_assessment(self, signup_id: str) -> dict: @@ -54,12 +48,9 @@ def get_signup_assessment(self, signup_id: str) -> dict: try: headers = self.__get_authorization_header() - response = requests.get(f'{Endpoints.SIGNUPS}/{signup_id}', headers=headers) - response.raise_for_status() + return self.__request.get(f'{Endpoints.SIGNUPS}/{signup_id}', headers=headers) - return json.loads(response.content.decode('utf-8')) - - except requests.HTTPError as e: + except IncogniaHTTPError as e: raise IncogniaHTTPError(e) from None def register_feedback(self, @@ -78,7 +69,7 @@ def register_feedback(self, try: headers = self.__get_authorization_header() - headers.update(self.JSON_CONTENT_HEADER) + headers.update(JSON_CONTENT_HEADER) body = { 'event': event, 'timestamp': total_milliseconds_since_epoch(timestamp), @@ -90,10 +81,9 @@ def register_feedback(self, 'installation_id': installation_id } data = encode(body) - response = requests.post(Endpoints.FEEDBACKS, headers=headers, data=data) - response.raise_for_status() + return self.__request.post(Endpoints.FEEDBACKS, headers=headers, data=data) - except requests.HTTPError as e: + except IncogniaHTTPError as e: raise IncogniaHTTPError(e) from None def register_payment(self, @@ -111,7 +101,7 @@ def register_payment(self, try: headers = self.__get_authorization_header() - headers.update(self.JSON_CONTENT_HEADER) + headers.update(JSON_CONTENT_HEADER) params = None if evaluate is None else {'eval': evaluate} body = { 'type': 'payment', @@ -123,13 +113,10 @@ def register_payment(self, 'payment_methods': payment_methods } data = encode(body) - response = requests.post(Endpoints.TRANSACTIONS, headers=headers, params=params, - data=data) - response.raise_for_status() - - return json.loads(response.content.decode('utf-8')) + return self.__request.post(Endpoints.TRANSACTIONS, headers=headers, params=params, + data=data) - except requests.HTTPError as e: + except IncogniaHTTPError as e: raise IncogniaHTTPError(e) from None def register_login(self, @@ -144,7 +131,7 @@ def register_login(self, try: headers = self.__get_authorization_header() - headers.update(self.JSON_CONTENT_HEADER) + headers.update(JSON_CONTENT_HEADER) params = None if evaluate is None else {'eval': evaluate} body = { 'type': 'login', @@ -153,11 +140,8 @@ def register_login(self, 'external_id': external_id } data = encode(body) - response = requests.post(Endpoints.TRANSACTIONS, headers=headers, params=params, - data=data) - response.raise_for_status() - - return json.loads(response.content.decode('utf-8')) + return self.__request.post(Endpoints.TRANSACTIONS, headers=headers, params=params, + data=data) - except requests.HTTPError as e: + except IncogniaHTTPError as e: raise IncogniaHTTPError(e) from None diff --git a/tests/test_api.py b/tests/test_api.py index a37cd81..8b70102 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,14 +1,13 @@ import datetime as dt -import json from typing import Final from unittest import TestCase from unittest.mock import patch, Mock -import requests - from incognia.api import IncogniaAPI +from incognia.base_request import BaseRequest from incognia.endpoints import Endpoints from incognia.exceptions import IncogniaHTTPError, IncogniaError +from incognia.json_util import encode from incognia.token_manager import TokenValues, TokenManager @@ -21,376 +20,299 @@ class TestIncogniaAPI(TestCase): INVALID_ACCOUNT_ID: Final[str] = 'INVALID_ACCOUNT_ID' SIGNUP_ID: Final[str] = 'ANY_SIGNUP_ID' TOKEN_VALUES: Final[TokenValues] = TokenValues('ACCESS_TOKEN', 'TOKEN_TYPE') - JSON_RESPONSE: Final[ascii] = b'{ "id": "signup_identifier",' \ - b' "request_id": "request_identifier",' \ - b' "risk_assessment": "unknown_risk", "evidence": [] }' + JSON_RESPONSE: Final[dict] = { + 'id': 'signup_identifier', + 'request_id': 'request_identifier', + 'risk_assessment': 'unknown_risk', + 'evidence': [] + } AUTH_HEADER: Final[dict] = { 'Authorization': f'{TOKEN_VALUES.token_type} {TOKEN_VALUES.access_token}' } AUTH_AND_JSON_CONTENT_HEADERS: Final[dict] = { - 'Content-type': 'application/json', - 'Authorization': f'{TOKEN_VALUES.token_type} {TOKEN_VALUES.access_token}' + 'Authorization': f'{TOKEN_VALUES.token_type} {TOKEN_VALUES.access_token}', + 'Content-Type': 'application/json' } - REGISTER_SIGNUP_DATA: Final[ascii] = f'{{"installation_id": "{INSTALLATION_ID}"}}' \ - .encode('utf-8') + REGISTER_SIGNUP_DATA: Final[bytes] = encode({ + 'installation_id': f'{INSTALLATION_ID}' + }) OK_STATUS_CODE: Final[int] = 200 CLIENT_ERROR_CODE: Final[int] = 400 VALID_EVENT_FEEDBACK_TYPE: Final[str] = 'valid_event_feedback_type' INVALID_EVENT_FEEDBACK_TYPE: Final[str] = 'invalid_event_feedback_type' TIMESTAMP: Final[dt.datetime] = dt.datetime.utcfromtimestamp(0) - REGISTER_VALID_FEEDBACK_DATA: Final[ascii] = f'{{"event": "{VALID_EVENT_FEEDBACK_TYPE}",' \ - f' "timestamp": 0}}'.encode('utf-8') - REGISTER_INVALID_FEEDBACK_DATA: Final[ascii] = f'{{"event": "{INVALID_EVENT_FEEDBACK_TYPE}",' \ - f' "timestamp": 0}}'.encode('utf-8') - REGISTER_VALID_PAYMENT_DATA: Final[ascii] = '{"type": "payment",' \ - f' "installation_id": "{INSTALLATION_ID}",' \ - f' "account_id": "{ACCOUNT_ID}"}}'.encode('utf-8') - REGISTER_INVALID_PAYMENT_DATA: Final[ascii] = \ - '{"type": "payment",' \ - f' "installation_id": "{INVALID_INSTALLATION_ID}",' \ - f' "account_id": "{INVALID_ACCOUNT_ID}"}}'.encode('utf-8') - REGISTER_VALID_LOGIN_DATA: Final[ascii] = '{"type": "login",' \ - f' "installation_id": "{INSTALLATION_ID}",' \ - f' "account_id": "{ACCOUNT_ID}"}}'.encode('utf-8') - REGISTER_INVALID_LOGIN_DATA: Final[ascii] = \ - '{"type": "login",' \ - f' "installation_id": "{INVALID_INSTALLATION_ID}",' \ - f' "account_id": "{INVALID_ACCOUNT_ID}"}}'.encode('utf-8') + REGISTER_VALID_FEEDBACK_DATA: Final[bytes] = encode({ + 'event': f'{VALID_EVENT_FEEDBACK_TYPE}', + 'timestamp': 0 + }) + REGISTER_INVALID_FEEDBACK_DATA: Final[bytes] = encode({ + 'event': f'{INVALID_EVENT_FEEDBACK_TYPE}', + 'timestamp': 0 + }) + REGISTER_VALID_PAYMENT_DATA: Final[bytes] = encode({ + 'type': 'payment', + 'installation_id': f'{INSTALLATION_ID}', + 'account_id': f'{ACCOUNT_ID}' + }) + REGISTER_INVALID_PAYMENT_DATA: Final[bytes] = encode({ + 'type': 'payment', + 'installation_id': f'{INVALID_INSTALLATION_ID}', + 'account_id': f'{INVALID_ACCOUNT_ID}' + }) + REGISTER_VALID_LOGIN_DATA: Final[bytes] = encode({ + 'type': 'login', + 'installation_id': f'{INSTALLATION_ID}', + 'account_id': f'{ACCOUNT_ID}' + }) + REGISTER_INVALID_LOGIN_DATA: Final[bytes] = encode({ + 'type': 'login', + 'installation_id': f'{INVALID_INSTALLATION_ID}', + 'account_id': f'{INVALID_ACCOUNT_ID}' + }) DEFAULT_PARAMS: Final[None] = None - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_new_signup_when_installation_id_is_valid_should_return_a_valid_dict( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response._content, response.status_code = self.JSON_RESPONSE, self.OK_STATUS_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(return_value=self.JSON_RESPONSE) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - request_response = api.register_new_signup(installation_id=self.INSTALLATION_ID) + response = api.register_new_signup(installation_id=self.INSTALLATION_ID) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.SIGNUPS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - data=self.REGISTER_SIGNUP_DATA) + mock_base_request_post.assert_called_with(Endpoints.SIGNUPS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + data=self.REGISTER_SIGNUP_DATA) - self.assertEqual(request_response, json.loads(self.JSON_RESPONSE.decode('utf-8'))) + self.assertEqual(response, self.JSON_RESPONSE) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_new_signup_when_installation_id_is_invalid_should_raise_an_IncogniaHTTPError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response.status_code = self.CLIENT_ERROR_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(side_effect=IncogniaHTTPError) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaHTTPError, api.register_new_signup, self.INSTALLATION_ID) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.SIGNUPS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - data=self.REGISTER_SIGNUP_DATA) + mock_base_request_post.assert_called_with(Endpoints.SIGNUPS, headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + data=self.REGISTER_SIGNUP_DATA) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get') def test_register_new_signup_when_installation_id_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaError, api.register_new_signup, '') mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.get') + @patch.object(BaseRequest, 'get') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_get_signup_assessment_when_signup_id_is_valid_should_return_a_valid_dict( - self, - mock_token_manager_get: Mock, - mock_requests_get: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response._content, response.status_code = self.JSON_RESPONSE, self.OK_STATUS_CODE - return response + self, mock_token_manager_get: Mock, mock_base_request_get: Mock): - mock_requests_get.configure_mock(return_value=get_mocked_response()) + mock_base_request_get.configure_mock(return_value=self.JSON_RESPONSE) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) request_response = api.get_signup_assessment(signup_id=self.SIGNUP_ID) mock_token_manager_get.assert_called() - mock_requests_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', - headers=self.AUTH_HEADER) + mock_base_request_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', headers=self.AUTH_HEADER) - self.assertEqual(request_response, json.loads(self.JSON_RESPONSE.decode('utf-8'))) + self.assertEqual(request_response, self.JSON_RESPONSE) - @patch('requests.get') + @patch.object(BaseRequest, 'get') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_get_signup_assessment_when_signup_id_invalid_should_raise_an_IncogniaHTTPError( - self, - mock_token_manager_get: Mock, - mock_requests_get: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response.status_code = self.CLIENT_ERROR_CODE - return response - - mock_requests_get.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_get: Mock): + mock_base_request_get.configure_mock(side_effect=IncogniaHTTPError) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaHTTPError, api.get_signup_assessment, self.SIGNUP_ID) mock_token_manager_get.assert_called() - mock_requests_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', - headers=self.AUTH_HEADER) + mock_base_request_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', headers=self.AUTH_HEADER) - @patch('requests.get') + @patch.object(BaseRequest, 'get') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_get_signup_assessment_when_signup_id_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_get: Mock): + self, mock_token_manager_get: Mock, mock_base_request_get: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaError, api.get_signup_assessment, '') mock_token_manager_get.assert_not_called() - mock_requests_get.assert_not_called() + mock_base_request_get.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_feedback_when_required_fields_are_valid_should_work( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE, self.TIMESTAMP) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.FEEDBACKS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - data=self.REGISTER_VALID_FEEDBACK_DATA) + mock_base_request_post.assert_called_with(Endpoints.FEEDBACKS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + data=self.REGISTER_VALID_FEEDBACK_DATA) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_feedback_when_event_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaError, api.register_feedback, event='', timestamp=self.TIMESTAMP) mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_feedback_when_timestamp_is_none_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_feedback, - event=self.VALID_EVENT_FEEDBACK_TYPE, - timestamp=None) + self.assertRaises(IncogniaError, api.register_feedback, event=self.VALID_EVENT_FEEDBACK_TYPE, timestamp=None) mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_feedback_when_required_fields_are_invalid_should_raise_an_IncogniaHTTPError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response.status_code = self.CLIENT_ERROR_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(side_effect=IncogniaHTTPError) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaHTTPError, api.register_feedback, - event=self.INVALID_EVENT_FEEDBACK_TYPE, + self.assertRaises(IncogniaHTTPError, api.register_feedback, event=self.INVALID_EVENT_FEEDBACK_TYPE, timestamp=self.TIMESTAMP) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.FEEDBACKS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - data=self.REGISTER_INVALID_FEEDBACK_DATA) + mock_base_request_post.assert_called_with(Endpoints.FEEDBACKS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + data=self.REGISTER_INVALID_FEEDBACK_DATA) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_payment_when_required_fields_are_valid_should_work( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response._content, response.status_code = self.JSON_RESPONSE, self.OK_STATUS_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(return_value=self.JSON_RESPONSE) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) request_response = api.register_payment(self.INSTALLATION_ID, self.ACCOUNT_ID) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.TRANSACTIONS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - params=self.DEFAULT_PARAMS, - data=self.REGISTER_VALID_PAYMENT_DATA) + mock_base_request_post.assert_called_with(Endpoints.TRANSACTIONS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + params=self.DEFAULT_PARAMS, + data=self.REGISTER_VALID_PAYMENT_DATA) - self.assertEqual(request_response, json.loads(self.JSON_RESPONSE.decode('utf-8'))) + self.assertEqual(request_response, self.JSON_RESPONSE) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_payment_when_installation_id_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_payment, installation_id='', - account_id=self.ACCOUNT_ID) + self.assertRaises(IncogniaError, api.register_payment, installation_id='', account_id=self.ACCOUNT_ID) mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_payment_when_account_id_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) self.assertRaises(IncogniaError, api.register_payment, installation_id=self.INSTALLATION_ID, account_id='') mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_payment_when_required_fields_are_invalid_should_raise_an_IncogniaHTTPError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response.status_code = self.CLIENT_ERROR_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(side_effect=IncogniaHTTPError) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaHTTPError, api.register_payment, - installation_id=self.INVALID_INSTALLATION_ID, + self.assertRaises(IncogniaHTTPError, api.register_payment, installation_id=self.INVALID_INSTALLATION_ID, account_id=self.INVALID_ACCOUNT_ID) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.TRANSACTIONS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - params=self.DEFAULT_PARAMS, - data=self.REGISTER_INVALID_PAYMENT_DATA) + mock_base_request_post.assert_called_with(Endpoints.TRANSACTIONS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + params=self.DEFAULT_PARAMS, + data=self.REGISTER_INVALID_PAYMENT_DATA) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_login_when_required_fields_are_valid_should_work( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response._content, response.status_code = self.JSON_RESPONSE, self.OK_STATUS_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(return_value=self.JSON_RESPONSE) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) request_response = api.register_login(self.INSTALLATION_ID, self.ACCOUNT_ID) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.TRANSACTIONS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - params=self.DEFAULT_PARAMS, - data=self.REGISTER_VALID_LOGIN_DATA) + mock_base_request_post.assert_called_with(Endpoints.TRANSACTIONS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + params=self.DEFAULT_PARAMS, + data=self.REGISTER_VALID_LOGIN_DATA) - self.assertEqual(request_response, json.loads(self.JSON_RESPONSE.decode('utf-8'))) + self.assertEqual(request_response, self.JSON_RESPONSE) - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_login_when_installation_id_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_login, installation_id='', - account_id=self.ACCOUNT_ID) + self.assertRaises(IncogniaError, api.register_login, installation_id='', account_id=self.ACCOUNT_ID) mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_login_when_account_id_is_empty_should_raise_an_IncogniaError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_login, installation_id=self.INSTALLATION_ID, - account_id='') + self.assertRaises(IncogniaError, api.register_login, installation_id=self.INSTALLATION_ID, account_id='') mock_token_manager_get.assert_not_called() - mock_requests_post.assert_not_called() + mock_base_request_post.assert_not_called() - @patch('requests.post') + @patch.object(BaseRequest, 'post') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_register_login_when_required_fields_are_invalid_should_raise_an_IncogniaHTTPError( - self, - mock_token_manager_get: Mock, - mock_requests_post: Mock): - def get_mocked_response() -> requests.Response: - response = requests.Response() - response.status_code = self.CLIENT_ERROR_CODE - return response - - mock_requests_post.configure_mock(return_value=get_mocked_response()) + self, mock_token_manager_get: Mock, mock_base_request_post: Mock): + mock_base_request_post.configure_mock(side_effect=IncogniaHTTPError) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaHTTPError, api.register_login, - installation_id=self.INVALID_INSTALLATION_ID, + self.assertRaises(IncogniaHTTPError, api.register_login, installation_id=self.INVALID_INSTALLATION_ID, account_id=self.INVALID_ACCOUNT_ID) mock_token_manager_get.assert_called() - mock_requests_post.assert_called_with(Endpoints.TRANSACTIONS, - headers=self.AUTH_AND_JSON_CONTENT_HEADERS, - params=self.DEFAULT_PARAMS, - data=self.REGISTER_INVALID_LOGIN_DATA) + mock_base_request_post.assert_called_with(Endpoints.TRANSACTIONS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + params=self.DEFAULT_PARAMS, + data=self.REGISTER_INVALID_LOGIN_DATA) From 50949b4f9a8c40d2f60eadad4706b3e1d383d184 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Cruz?= Date: Thu, 28 Apr 2022 18:02:33 -0300 Subject: [PATCH 5/5] style(*): fixes column width --- incognia/base_request.py | 10 +++++++--- incognia/token_manager.py | 3 ++- tests/test_api.py | 31 ++++++++++++++++++++----------- tests/test_base_request.py | 30 ++++++++++++++++++++---------- tests/test_token_manager.py | 15 ++++++++------- 5 files changed, 57 insertions(+), 32 deletions(-) diff --git a/incognia/base_request.py b/incognia/base_request.py index 1155e1c..7d7d5d3 100644 --- a/incognia/base_request.py +++ b/incognia/base_request.py @@ -14,7 +14,9 @@ _LANGUAGE_VERSION: Final[str] = platform.python_version() USER_AGENT_HEADER: Final[dict] = { - 'User-Agent': f'incognia-python/{_LIBRARY_VERSION} ({_OS_NAME} {_OS_VERSION} {_OS_ARCH}) Python/{_LANGUAGE_VERSION}' + 'User-Agent': f'incognia-python/{_LIBRARY_VERSION}' + f' ({_OS_NAME} {_OS_VERSION} {_OS_ARCH})' + f' Python/{_LANGUAGE_VERSION}' } JSON_CONTENT_HEADER: Final[dict] = { @@ -29,13 +31,15 @@ def __init__(self, timeout: float = 5.0): def timeout(self) -> float: return self.__timeout - def post(self, url: Union[str, bytes], headers: Any = None, data: Any = None, params: Any = None, + def post(self, url: Union[str, bytes], headers: Any = None, data: Any = None, + params: Any = None, auth: Optional[Any] = None) -> Optional[dict]: headers = headers or {} headers.update(USER_AGENT_HEADER) try: - response = requests.post(url=url, headers=headers, data=data, params=params, timeout=self.__timeout, + response = requests.post(url=url, headers=headers, data=data, params=params, + timeout=self.__timeout, auth=auth) response.raise_for_status() return json.loads(response.content.decode('utf-8')) or None diff --git a/incognia/token_manager.py b/incognia/token_manager.py index b9af683..ccaa875 100644 --- a/incognia/token_manager.py +++ b/incognia/token_manager.py @@ -42,7 +42,8 @@ def __refresh_token(self) -> None: raise IncogniaHTTPError(e) from None def __is_expired(self) -> bool: - return (self.__expiration_time - dt.datetime.now()).total_seconds() <= _TOKEN_REFRESH_BEFORE_SECONDS + return (self.__expiration_time - dt.datetime.now()) \ + .total_seconds() <= _TOKEN_REFRESH_BEFORE_SECONDS def get(self) -> TokenValues: self.__mutex.acquire() diff --git a/tests/test_api.py b/tests/test_api.py index 8b70102..517a664 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -98,7 +98,8 @@ def test_register_new_signup_when_installation_id_is_invalid_should_raise_an_Inc self.assertRaises(IncogniaHTTPError, api.register_new_signup, self.INSTALLATION_ID) mock_token_manager_get.assert_called() - mock_base_request_post.assert_called_with(Endpoints.SIGNUPS, headers=self.AUTH_AND_JSON_CONTENT_HEADERS, + mock_base_request_post.assert_called_with(Endpoints.SIGNUPS, + headers=self.AUTH_AND_JSON_CONTENT_HEADERS, data=self.REGISTER_SIGNUP_DATA) @patch.object(BaseRequest, 'post') @@ -116,14 +117,14 @@ def test_register_new_signup_when_installation_id_is_empty_should_raise_an_Incog @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) def test_get_signup_assessment_when_signup_id_is_valid_should_return_a_valid_dict( self, mock_token_manager_get: Mock, mock_base_request_get: Mock): - mock_base_request_get.configure_mock(return_value=self.JSON_RESPONSE) api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) request_response = api.get_signup_assessment(signup_id=self.SIGNUP_ID) mock_token_manager_get.assert_called() - mock_base_request_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', headers=self.AUTH_HEADER) + mock_base_request_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', + headers=self.AUTH_HEADER) self.assertEqual(request_response, self.JSON_RESPONSE) @@ -138,7 +139,8 @@ def test_get_signup_assessment_when_signup_id_invalid_should_raise_an_IncogniaHT self.assertRaises(IncogniaHTTPError, api.get_signup_assessment, self.SIGNUP_ID) mock_token_manager_get.assert_called() - mock_base_request_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', headers=self.AUTH_HEADER) + mock_base_request_get.assert_called_with(f'{Endpoints.SIGNUPS}/{self.SIGNUP_ID}', + headers=self.AUTH_HEADER) @patch.object(BaseRequest, 'get') @patch.object(TokenManager, 'get', return_value=TOKEN_VALUES) @@ -181,7 +183,8 @@ def test_register_feedback_when_timestamp_is_none_should_raise_an_IncogniaError( self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_feedback, event=self.VALID_EVENT_FEEDBACK_TYPE, timestamp=None) + self.assertRaises(IncogniaError, api.register_feedback, + event=self.VALID_EVENT_FEEDBACK_TYPE, timestamp=None) mock_token_manager_get.assert_not_called() mock_base_request_post.assert_not_called() @@ -194,7 +197,8 @@ def test_register_feedback_when_required_fields_are_invalid_should_raise_an_Inco api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaHTTPError, api.register_feedback, event=self.INVALID_EVENT_FEEDBACK_TYPE, + self.assertRaises(IncogniaHTTPError, api.register_feedback, + event=self.INVALID_EVENT_FEEDBACK_TYPE, timestamp=self.TIMESTAMP) mock_token_manager_get.assert_called() @@ -226,7 +230,8 @@ def test_register_payment_when_installation_id_is_empty_should_raise_an_Incognia self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_payment, installation_id='', account_id=self.ACCOUNT_ID) + self.assertRaises(IncogniaError, api.register_payment, installation_id='', + account_id=self.ACCOUNT_ID) mock_token_manager_get.assert_not_called() mock_base_request_post.assert_not_called() @@ -251,7 +256,8 @@ def test_register_payment_when_required_fields_are_invalid_should_raise_an_Incog api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaHTTPError, api.register_payment, installation_id=self.INVALID_INSTALLATION_ID, + self.assertRaises(IncogniaHTTPError, api.register_payment, + installation_id=self.INVALID_INSTALLATION_ID, account_id=self.INVALID_ACCOUNT_ID) mock_token_manager_get.assert_called() @@ -284,7 +290,8 @@ def test_register_login_when_installation_id_is_empty_should_raise_an_IncogniaEr self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_login, installation_id='', account_id=self.ACCOUNT_ID) + self.assertRaises(IncogniaError, api.register_login, installation_id='', + account_id=self.ACCOUNT_ID) mock_token_manager_get.assert_not_called() mock_base_request_post.assert_not_called() @@ -295,7 +302,8 @@ def test_register_login_when_account_id_is_empty_should_raise_an_IncogniaError( self, mock_token_manager_get: Mock, mock_base_request_post: Mock): api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaError, api.register_login, installation_id=self.INSTALLATION_ID, account_id='') + self.assertRaises(IncogniaError, api.register_login, installation_id=self.INSTALLATION_ID, + account_id='') mock_token_manager_get.assert_not_called() mock_base_request_post.assert_not_called() @@ -308,7 +316,8 @@ def test_register_login_when_required_fields_are_invalid_should_raise_an_Incogni api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET) - self.assertRaises(IncogniaHTTPError, api.register_login, installation_id=self.INVALID_INSTALLATION_ID, + self.assertRaises(IncogniaHTTPError, api.register_login, + installation_id=self.INVALID_INSTALLATION_ID, account_id=self.INVALID_ACCOUNT_ID) mock_token_manager_get.assert_called() diff --git a/tests/test_base_request.py b/tests/test_base_request.py index 40305ca..e42b458 100644 --- a/tests/test_base_request.py +++ b/tests/test_base_request.py @@ -19,39 +19,47 @@ class TestBaseRequest(TestCase): CLIENT_ERROR_CODE: Final[int] = 400 @patch('requests.post') - def test_post_when_parameters_are_valid_should_return_a_valid_dict(self, mock_requests_post: Mock): + def test_post_when_parameters_are_valid_should_return_a_valid_dict( + self, mock_requests_post: Mock): def get_mocked_response() -> requests.Response: response = requests.Response() - response._content, response.status_code = encode(self.JSON_RESPONSE), self.OK_STATUS_CODE + response._content, response.status_code = encode( + self.JSON_RESPONSE), self.OK_STATUS_CODE return response mock_requests_post.configure_mock(return_value=get_mocked_response()) base_request = BaseRequest() result = base_request.post(url=self.URL) - mock_requests_post.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, params=None, + mock_requests_post.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, + params=None, timeout=base_request.timeout(), auth=None) self.assertEqual(result, self.JSON_RESPONSE) @patch('requests.post') - def test_post_when_parameters_are_invalid_should_raise_an_IncogniaHTTPError(self, mock_requests_post: Mock): + def test_post_when_parameters_are_invalid_should_raise_an_IncogniaHTTPError( + self, mock_requests_post: Mock): def get_mocked_response() -> requests.Response: response = requests.Response() - response._content, response.status_code = encode(self.JSON_RESPONSE), self.CLIENT_ERROR_CODE + response._content, response.status_code = encode( + self.JSON_RESPONSE), self.CLIENT_ERROR_CODE return response mock_requests_post.configure_mock(return_value=get_mocked_response()) base_request = BaseRequest() self.assertRaises(IncogniaHTTPError, base_request.post, url=self.URL) - mock_requests_post.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, params=None, + mock_requests_post.assert_called_with(url=self.URL, headers=USER_AGENT_HEADER, data=None, + params=None, timeout=base_request.timeout(), auth=None) @patch('requests.get') - def test_get_when_parameters_are_valid_should_return_a_valid_dict(self, mock_requests_get: Mock): + def test_get_when_parameters_are_valid_should_return_a_valid_dict(self, + mock_requests_get: Mock): def get_mocked_response() -> requests.Response: response = requests.Response() - response._content, response.status_code = encode(self.JSON_RESPONSE), self.OK_STATUS_CODE + response._content, response.status_code = encode( + self.JSON_RESPONSE), self.OK_STATUS_CODE return response mock_requests_get.configure_mock(return_value=get_mocked_response()) @@ -63,10 +71,12 @@ def get_mocked_response() -> requests.Response: self.assertEqual(result, self.JSON_RESPONSE) @patch('requests.get') - def test_get_when_parameters_are_invalid_should_raise_an_IncogniaHTTPError(self, mock_requests_get: Mock): + def test_get_when_parameters_are_invalid_should_raise_an_IncogniaHTTPError( + self, mock_requests_get: Mock): def get_mocked_response() -> requests.Response: response = requests.Response() - response._content, response.status_code = encode(self.JSON_RESPONSE), self.CLIENT_ERROR_CODE + response._content, response.status_code = encode( + self.JSON_RESPONSE), self.CLIENT_ERROR_CODE return response mock_requests_get.configure_mock(return_value=get_mocked_response()) diff --git a/tests/test_token_manager.py b/tests/test_token_manager.py index 64fdd6b..88f6e63 100644 --- a/tests/test_token_manager.py +++ b/tests/test_token_manager.py @@ -18,20 +18,22 @@ class TestTokenManager(TestCase): 'token_type': 'TOKEN_TYPE', 'expires_in': 900 } - SHORT_EXPIRATION_TOKEN_VALUES: Final[TokenValues] = TokenValues('S_E_ACCESS_TOKEN', 'S_E_TOKEN_TYPE') + SHORT_EXPIRATION_TOKEN_VALUES: Final[TokenValues] = \ + TokenValues('S_E_ACCESS_TOKEN', 'S_E_TOKEN_TYPE') SHORT_EXPIRATION_JSON_POST_RESPONSE: Final[dict] = { 'access_token': 'S_E_ACCESS_TOKEN', 'token_type': 'S_E_TOKEN_TYPE', 'expires_in': 5 } - CLIENT_ID_AND_SECRET_ENCODED: Final[str] = base64.urlsafe_b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}' - .encode('ascii')).decode('utf-8') + CLIENT_ID_AND_SECRET_ENCODED: Final[str] = base64.urlsafe_b64encode( + f'{CLIENT_ID}:{CLIENT_SECRET}'.encode('ascii')).decode('utf-8') HEADERS: Final[dict] = {'Authorization': f'Basic {CLIENT_ID_AND_SECRET_ENCODED}'} OK_STATUS_CODE: Final[int] = 200 CLIENT_ERROR_CODE: Final[int] = 400 @patch.object(BaseRequest, 'post') - def test_get_when_credentials_are_valid_should_return_a_valid_token(self, mock_requests_post: Mock): + def test_get_when_credentials_are_valid_should_return_a_valid_token(self, + mock_requests_post: Mock): mock_requests_post.configure_mock(return_value=self.JSON_POST_RESPONSE) token_manager = TokenManager(self.CLIENT_ID, self.CLIENT_SECRET) @@ -41,9 +43,8 @@ def test_get_when_credentials_are_valid_should_return_a_valid_token(self, mock_r self.assertEqual(token_values, self.TOKEN_VALUES) @patch.object(BaseRequest, 'post') - def test_get_when_credentials_are_invalid_should_raise_an_IncogniaHTTPError(self, - mock_requests_post: - Mock): + def test_get_when_credentials_are_invalid_should_raise_an_IncogniaHTTPError( + self, mock_requests_post: Mock): mock_requests_post.configure_mock(side_effect=IncogniaHTTPError) token_manager = TokenManager(self.CLIENT_ID, self.CLIENT_SECRET)