More unit tests
This commit is contained in:
parent
e280496acb
commit
f8dc115192
|
@ -2,7 +2,8 @@ import datetime
|
||||||
import unittest
|
import unittest
|
||||||
from freezegun import freeze_time
|
from freezegun import freeze_time
|
||||||
from unittest.mock import patch, MagicMock
|
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):
|
class TestZohoToken(unittest.TestCase):
|
||||||
"""Test the ZohoToken class"""
|
"""Test the ZohoToken class"""
|
||||||
|
@ -13,11 +14,12 @@ class TestZohoToken(unittest.TestCase):
|
||||||
mock_response = MagicMock()
|
mock_response = MagicMock()
|
||||||
mock_response.read.return_value.decode.return_value = '{"access_token": "mock_access_token", "expires_in": 3600}'
|
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
|
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
|
# 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')
|
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')
|
@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_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
|
mock_request.urlopen.return_value.__enter__.return_value = mock_response
|
||||||
|
|
||||||
auth = ZohoToken('mock_client_id', 'mock_client_secret', 'mock_refresh_token')
|
token = ZohoToken('mock_client_id', 'mock_client_secret', 'mock_refresh_token')
|
||||||
self.assertEqual(auth.expires_at, datetime.datetime(2009, 6, 15, 10, 0, 0))
|
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}'
|
mock_response.read.return_value.decode.return_value = '{"access_token": "updated_mock_access_token", "expires_in": 3600}'
|
||||||
|
|
||||||
frozen_time.tick(datetime.timedelta(seconds=3599))
|
frozen_time.tick(datetime.timedelta(seconds=3599))
|
||||||
token = auth.get_token()
|
self.assertEqual(token.get_token(), 'mock_access_token')
|
||||||
self.assertEqual(token, 'mock_access_token')
|
|
||||||
frozen_time.tick(datetime.timedelta(seconds=1))
|
frozen_time.tick(datetime.timedelta(seconds=1))
|
||||||
token = auth.get_token()
|
self.assertEqual(token.get_token(), 'updated_mock_access_token')
|
||||||
self.assertEqual(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__':
|
if __name__ == '__main__':
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|
10
zoho_auth.py
10
zoho_auth.py
|
@ -1,5 +1,6 @@
|
||||||
import datetime
|
import datetime
|
||||||
import json
|
import json
|
||||||
|
from http import HTTPStatus
|
||||||
from urllib import request
|
from urllib import request
|
||||||
from urllib.parse import urlencode
|
from urllib.parse import urlencode
|
||||||
from urllib.error import HTTPError
|
from urllib.error import HTTPError
|
||||||
|
@ -11,7 +12,6 @@ class ZohoToken:
|
||||||
self.refresh_token = refresh_token
|
self.refresh_token = refresh_token
|
||||||
self._access_token = None
|
self._access_token = None
|
||||||
self.expires_at = None
|
self.expires_at = None
|
||||||
self.update_token()
|
|
||||||
|
|
||||||
def update_token(self):
|
def update_token(self):
|
||||||
url = "https://accounts.zoho.com/oauth/v2/token?%s" % urlencode({
|
url = "https://accounts.zoho.com/oauth/v2/token?%s" % urlencode({
|
||||||
|
@ -32,8 +32,6 @@ class ZohoToken:
|
||||||
return self._access_token
|
return self._access_token
|
||||||
|
|
||||||
def authenticate(token, key):
|
def authenticate(token, key):
|
||||||
ret = False
|
|
||||||
|
|
||||||
url = "https://www.zohoapis.com/crm/v2/Contacts/search?%s" % urlencode({
|
url = "https://www.zohoapis.com/crm/v2/Contacts/search?%s" % urlencode({
|
||||||
"criteria": "(Key_ID:equals:%s)" % key,
|
"criteria": "(Key_ID:equals:%s)" % key,
|
||||||
})
|
})
|
||||||
|
@ -43,10 +41,8 @@ def authenticate(token, key):
|
||||||
print("Pinging server with key: %s (%s)" % (key, url))
|
print("Pinging server with key: %s (%s)" % (key, url))
|
||||||
try:
|
try:
|
||||||
with request.urlopen(req) as response:
|
with request.urlopen(req) as response:
|
||||||
response_body = response.read()
|
return response.status == HTTPStatus.OK
|
||||||
if len(response_body) > 0:
|
|
||||||
ret = True
|
|
||||||
except HTTPError as e:
|
except HTTPError as e:
|
||||||
print("Error pinging server: %s" % e)
|
print("Error pinging server: %s" % e)
|
||||||
|
|
||||||
return ret
|
return False
|
||||||
|
|
Loading…
Reference in New Issue