diff --git a/test_zoho_auth.py b/test_zoho_auth.py index 2660fc4..29b163c 100644 --- a/test_zoho_auth.py +++ b/test_zoho_auth.py @@ -2,7 +2,8 @@ import datetime import unittest from freezegun import freeze_time from unittest.mock import patch, MagicMock -from zoho_auth import ZohoToken +from urllib.error import HTTPError +from zoho_auth import ZohoToken, authenticate class TestZohoToken(unittest.TestCase): """Test the ZohoToken class""" @@ -13,11 +14,12 @@ class TestZohoToken(unittest.TestCase): mock_response = MagicMock() mock_response.read.return_value.decode.return_value = '{"access_token": "mock_access_token", "expires_in": 3600}' mock_request.urlopen.return_value.__enter__.return_value = mock_response + token = ZohoToken('mock_client_id', 'mock_client_secret', 'mock_refresh_token') - auth = ZohoToken('mock_client_id', 'mock_client_secret', 'mock_refresh_token') + token.update_token() # Assert that the token was fetched - self.assertEqual(auth.get_token(), 'mock_access_token') + self.assertEqual(token.get_token(), 'mock_access_token') mock_request.Request.assert_called_with('https://accounts.zoho.com/oauth/v2/token?refresh_token=mock_refresh_token&client_id=mock_client_id&client_secret=mock_client_secret&grant_type=refresh_token', method='POST') @patch('zoho_auth.request') @@ -28,16 +30,61 @@ class TestZohoToken(unittest.TestCase): mock_response.read.return_value.decode.return_value = '{"access_token": "mock_access_token", "expires_in": 3600}' mock_request.urlopen.return_value.__enter__.return_value = mock_response - auth = ZohoToken('mock_client_id', 'mock_client_secret', 'mock_refresh_token') - self.assertEqual(auth.expires_at, datetime.datetime(2009, 6, 15, 10, 0, 0)) + token = ZohoToken('mock_client_id', 'mock_client_secret', 'mock_refresh_token') + token.update_token() + self.assertEqual(token.expires_at, datetime.datetime(2009, 6, 15, 10, 0, 0)) mock_response.read.return_value.decode.return_value = '{"access_token": "updated_mock_access_token", "expires_in": 3600}' + frozen_time.tick(datetime.timedelta(seconds=3599)) - token = auth.get_token() - self.assertEqual(token, 'mock_access_token') + self.assertEqual(token.get_token(), 'mock_access_token') frozen_time.tick(datetime.timedelta(seconds=1)) - token = auth.get_token() - self.assertEqual(token, 'updated_mock_access_token') + self.assertEqual(token.get_token(), 'updated_mock_access_token') + + @patch('zoho_auth.request') + def test_authenticate_calls_service(self, mock_request): + """Test that the authenticate function works""" + token = MagicMock() + token.get_token.return_value = 'mock_access_token' + + mock_response = MagicMock() + mock_response.status = 200 + mock_request.urlopen.return_value.__enter__.return_value = mock_response + + ret_val = authenticate(token, 'ABCDEF1234567890') + + mock_request.Request.assert_called_with('https://www.zohoapis.com/crm/v2/Contacts/search?criteria=%28Key_ID%3Aequals%3AABCDEF1234567890%29', method='GET', headers={ + 'Authorization': 'Zoho-oauthtoken mock_access_token', + }) + self.assertTrue(ret_val) + + @patch('zoho_auth.request') + def test_authenticate_fails_bad_key(self, mock_request): + """Test that the authenticate function works""" + token = MagicMock() + token.get_token.return_value = 'mock_access_token' + + mock_response = MagicMock() + mock_response.status = 204 + mock_request.urlopen.return_value.__enter__.return_value = mock_response + + ret_val = authenticate(token, 'ABCDEF1234567890') + + self.assertFalse(ret_val) + + + @patch('zoho_auth.request') + def test_authenticate_fails_when_server_is_down(self, mock_request): + """Test that the authenticate function works""" + token = MagicMock() + token.get_token.return_value = 'mock_access_token' + + error = HTTPError('https://www.youtube.com/watch?v=dQw4w9WgXcQ', 403, 'Forbidden', {}, None) + mock_request.urlopen.return_value.__enter__.side_effect = error + + ret_val = authenticate(token, 'ABCDEF1234567890') + + self.assertFalse(ret_val) if __name__ == '__main__': unittest.main() diff --git a/zoho_auth.py b/zoho_auth.py index 4417f2a..9a244ea 100644 --- a/zoho_auth.py +++ b/zoho_auth.py @@ -1,5 +1,6 @@ import datetime import json +from http import HTTPStatus from urllib import request from urllib.parse import urlencode from urllib.error import HTTPError @@ -11,7 +12,6 @@ class ZohoToken: self.refresh_token = refresh_token self._access_token = None self.expires_at = None - self.update_token() def update_token(self): url = "https://accounts.zoho.com/oauth/v2/token?%s" % urlencode({ @@ -32,8 +32,6 @@ class ZohoToken: return self._access_token def authenticate(token, key): - ret = False - url = "https://www.zohoapis.com/crm/v2/Contacts/search?%s" % urlencode({ "criteria": "(Key_ID:equals:%s)" % key, }) @@ -43,10 +41,8 @@ def authenticate(token, key): print("Pinging server with key: %s (%s)" % (key, url)) try: with request.urlopen(req) as response: - response_body = response.read() - if len(response_body) > 0: - ret = True + return response.status == HTTPStatus.OK except HTTPError as e: print("Error pinging server: %s" % e) - return ret + return False