Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .travis.yml
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,6 @@ install:
- pip install .
# command to run tests
script:
- pip install coveralls
- pip install coveralls requests-mock
- coverage run --source=callhub -m unittest discover tests
- coveralls
13 changes: 12 additions & 1 deletion callhub/callhub.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,10 @@ def __init__(self, api_key=None, rate_limit=API_LIMIT):
self.bulk_create = sleep_and_retry(limits(**rate_limit["BULK_CREATE"])(self.bulk_create))

self.session.auth = CallHubAuth(api_key=api_key)
self.admin_email = self.get_admin_email()

def __repr__(self):
return "<CallHub admin: {}>".format(self.admin_email)

def _collect_fields(self, contacts):
""" Internal Function to get all fields used in a list of contacts """
Expand Down Expand Up @@ -74,9 +78,16 @@ def _assert_fields_exist(self, contacts):
"created in CallHub. Fields present in upload: {} Fields present in "
"account: {}".format(fields_in_contact, fields_in_callhub))

def get_admin_email(self):
response = self.session.get("https://api.callhub.io/v1/agents/").result()
if response.json().get("count"):
return response.json()["results"][0]["owner"][0]["username"]
else:
return "Cannot deduce admin account. No agent accounts (not even the default account) exist."

def agent_leaderboard(self, start, end):
params = {"start_date": start, "end_date": end}
response = self.session.get("https://api.callhub.io/v1/analytics/agent-leaderboard", params=params).result()
response = self.session.get("https://api.callhub.io/v1/analytics/agent-leaderboard/", params=params).result()
return response.json().get("plot_data")

def fields(self):
Expand Down
2 changes: 2 additions & 0 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
with open(os.path.join(here, "README.md"), mode="r") as f:
README = f.read()

tests_require = ["requests-mock"]
install_requires = ["requests==2.23.0", "ratelimit==2.2.1", "requests-futures==1.0.0"]

setup(
Expand All @@ -25,6 +26,7 @@
version=about["__version__"],
packages=["callhub"],
install_requires=install_requires,
tests_requre=tests_require,
python_requires=">=3.5",
keywords=["callhub", "api"],
license=about["__license__"],
Expand Down
29 changes: 19 additions & 10 deletions tests/tests_auth.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,30 @@
import os

import unittest
from unittest.mock import MagicMock
from requests_mock import Mocker
from callhub import CallHub


class TestInit(unittest.TestCase):
def create_callhub(self, api_key=None):
callhub = CallHub(api_key=api_key, rate_limit=False)
# Override all http methods with mocking so a poorly designed test can't mess with
callhub.session.get = MagicMock(returnvalue=None)
callhub.session.post = MagicMock(returnvalue=None)
callhub.session.put = MagicMock(returnvalue=None)
callhub.session.delete = MagicMock(returnvalue=None)
callhub.session.head = MagicMock(returnvalue=None)
callhub.session.options = MagicMock(returnvalue=None)
return True
with Mocker() as mock:
mock.get("https://api.callhub.io/v1/agents/",
status_code=200,
json={'count': 1,
'next': None,
'previous': None,
'results': [{'email': 'user@example.com',
'id': 1111111111111111111,
'owner': [{'url': 'https://api.callhub.io/v1/users/0/',
'username': 'admin@example.com'}],
'teams': [],
'username': 'defaultuser'}]
},
complete_qs=True,
)

self.callhub = CallHub(api_key=api_key, rate_limit=False)
return True

def setUp(self):
os.environ["CALLHUB_API_KEY"] = "123456789ABCDEF"
Expand Down
204 changes: 109 additions & 95 deletions tests/tests_mock.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
from callhub import CallHub
import time
import math
from requests_mock import Mocker


class TestInit(unittest.TestCase):
@classmethod
Expand All @@ -11,65 +13,74 @@ def setUp(cls):
"GENERAL": {"calls": 1, "period": 0.1},
"BULK_CREATE": {"calls": 1, "period": 0.2},
}
# Create one callhub object stored in cls.callhub (for most test cases)
# Create ten callhub objects stored in cls.callhubs (for bulk testing)
cls.callhubs = []

for i in range(11):
callhub = CallHub(api_key="123456789ABCDEF", rate_limit=cls.TESTING_API_LIMIT)

# Override all http methods with mocking so a poorly designed test can't mess with
callhub.session.get = MagicMock(returnvalue=None)
callhub.session.post = MagicMock(returnvalue=None)
callhub.session.put = MagicMock(returnvalue=None)
callhub.session.delete = MagicMock(returnvalue=None)
callhub.session.head = MagicMock(returnvalue=None)
callhub.session.options = MagicMock(returnvalue=None)

if i == 0:
cls.callhub = callhub
else:
cls.callhubs.append(callhub)

with Mocker() as mock:
mock.get("https://api.callhub.io/v1/agents/",
status_code=200,
json={'count': 1,
'next': None,
'previous': None,
'results': [{'email': 'user@example.com',
'id': 1111111111111111111,
'owner': [{'url': 'https://api.callhub.io/v1/users/0/',
'username': 'admin@example.com'}],
'teams': [],
'username': 'defaultuser'}]
},
)
# Create one callhub object stored in cls.callhub (for most test cases)
# Create ten callhub objects stored in cls.callhubs (for bulk testing)
cls.callhubs = []
for i in range(11):
callhub = CallHub(api_key="123456789ABCDEF", rate_limit=cls.TESTING_API_LIMIT)
# Override all http methods with mocking so a poorly designed test can't mess with
if i == 0:
cls.callhub = callhub
else:
cls.callhubs.append(callhub)

def test_repr(self):
self.assertEqual("<CallHub admin: admin@example.com>", self.callhub.__repr__())

def test_agent_leaderboard(self):
self.callhub.session.get.return_value.result.return_value.json.return_value = {
"plot_data": [
with Mocker() as mock:
mock.get("https://api.callhub.io/v1/analytics/agent-leaderboard/",
status_code=200,
json={
"plot_data": [
{
'connecttime': 3300,
'teams': ['Fundraising'],
'calls': 5,
'agent': 'jimmybru',
'talktime': 120
}
]
})
leaderboard = self.callhub.agent_leaderboard("2019-12-30", "2020-12-30")
expected_leaderboard = [
{
'connecttime': 3300,
'teams': ['Fundraising'],
'calls': 5,
'agent':'jimmybru',
'agent': 'jimmybru',
'talktime': 120
}
]
}

leaderboard = self.callhub.agent_leaderboard("2019-12-30", "2020-12-30")
expected_leaderboard = [
{
'connecttime': 3300,
'teams': ['Fundraising'],
'calls': 5,
'agent': 'jimmybru',
'talktime': 120
}
]
self.assertEqual(leaderboard,expected_leaderboard)
self.assertEqual(leaderboard, expected_leaderboard)

def test_bulk_create_success(self, test_specific_callhub_instance=None):
if test_specific_callhub_instance:
self.callhub = test_specific_callhub_instance

self.callhub.fields = MagicMock(return_value={"first name": 0, "phone number": 1})
self.callhub.session.post = MagicMock()
self.callhub.session.post.return_value.result.return_value.json.return_value = {
"message": "'Import in progress. You will get an email when import is complete'"}
result = self.callhub.bulk_create(
2325931969109558581,
[{"first name": "james", "phone number": "5555555555"}],
"CA")
self.assertEqual(result, True)
with Mocker() as mock:
mock.post("https://api.callhub.io/v1/contacts/bulk_create/",
status_code=200,
json={"message": "'Import in progress. You will get an email when import is complete'"})
if test_specific_callhub_instance:
self.callhub = test_specific_callhub_instance
self.callhub.fields = MagicMock(return_value={"first name": 0, "phone number": 1})
result = self.callhub.bulk_create(
2325931969109558581,
[{"first name": "james", "phone number": "5555555555"}],
"CA")
self.assertEqual(result, True)

def test_bulk_create_field_mismatch_failure(self):
self.callhub.fields = MagicMock(return_value={"foo": 0, "bar": 1})
Expand All @@ -81,36 +92,36 @@ def test_bulk_create_field_mismatch_failure(self):
)

def test_bulk_create_api_exceeded_or_other_failure(self):
self.callhub.fields = MagicMock(return_value={"first name": 0, "phone number": 1})
self.callhub.session.post = MagicMock()
self.callhub.session.post.return_value.result.return_value.json.return_value = {
"detail": "Request was throttled."}
self.assertRaises(RuntimeError,
self.callhub.bulk_create,
2325931969109558581,
[{"first name": "james", "phone number": "5555555555"}],
"CA"
)
self.callhub.session.post.return_value.result.return_value.json.return_value = {
"NON STANDARD KEY": "YOU MESSED UP FOR SOME REASON"}
self.assertRaises(RuntimeError,
self.callhub.bulk_create,
2325931969109558581,
[{"first name": "james", "phone number": "5555555555"}],
"CA"
)
with Mocker() as mock:
mock.post("https://api.callhub.io/v1/contacts/bulk_create/",
json={"detail": "Request was throttled."})
self.callhub.fields = MagicMock(return_value={"first name": 0, "phone number": 1})
self.assertRaises(RuntimeError,
self.callhub.bulk_create,
2325931969109558581,
[{"first name": "james", "phone number": "5555555555"}],
"CA"
)
mock.post("https://api.callhub.io/v1/contacts/bulk_create/",
json={"NON STANDARD KEY": "YOU MESSED UP FOR SOME REASON"})
self.assertRaises(RuntimeError,
self.callhub.bulk_create,
2325931969109558581,
[{"first name": "james", "phone number": "5555555555"}],
"CA"
)

def test_bulk_create_rate_limit(self):
start = time.perf_counter()
num_iterations=11
num_iterations = 11
for i in range(num_iterations):
self.test_bulk_create_success()
stop = time.perf_counter()

# Should run within 95% to 105% of ratelimit*num iterations -1
lower_bound = 0.95 * self.TESTING_API_LIMIT["BULK_CREATE"]["period"] * (num_iterations - 1)
upper_bound = 1.05 * self.TESTING_API_LIMIT["BULK_CREATE"]["period"] * (num_iterations - 1)
self.assertEqual(lower_bound <= stop-start <= upper_bound, True)
self.assertEqual(lower_bound <= stop - start <= upper_bound, True)

def test_bulk_create_many_objects_rate_limit(self):
start = time.perf_counter()
Expand All @@ -123,58 +134,60 @@ def test_bulk_create_many_objects_rate_limit(self):
# because the rate limiting should be on a per-object basis.
lower_bound = 0.95 * self.TESTING_API_LIMIT["BULK_CREATE"]["period"] * (num_iterations - 1)
upper_bound = 1.05 * self.TESTING_API_LIMIT["BULK_CREATE"]["period"] * (num_iterations - 1)
self.assertEqual(lower_bound <= stop-start <= upper_bound, True)

self.assertEqual(lower_bound <= stop - start <= upper_bound, True)

def test_fields(self):
self.callhub.session.get = MagicMock()
self.callhub.session.get.return_value.result.return_value.json.return_value = {'count': 4, 'results':
[{'id': 0, 'name': 'phone number'}, {'id': 1, 'name': 'mobile number'},
{'id': 2, 'name': 'last name'}, {'id': 3, 'name': 'first name'}]}
self.assertEqual(self.callhub.fields(),
{'phone number': 0, 'mobile number': 1, 'last name': 2, 'first name': 3})
with Mocker() as mock:
mock.get('https://api.callhub.io/v1/contacts/fields/',
json={'count': 4, 'results':
[{'id': 0, 'name': 'phone number'}, {'id': 1, 'name': 'mobile number'},
{'id': 2, 'name': 'last name'}, {'id': 3, 'name': 'first name'}]})
self.assertEqual(self.callhub.fields(),
{'phone number': 0, 'mobile number': 1, 'last name': 2, 'first name': 3})

def test_collect_fields(self):
contacts = [{"first name": "James", "contact": 5555555555}, {"last name": "Brunet", "contact": 1234567890}]
self.assertEqual(self.callhub._collect_fields(contacts), {"first name", "last name", "contact"})

def test_create_contact(self):
# Test if contact creation successful
self.callhub.fields = MagicMock(return_value={"first name": 0, "phone number": 1})
self.callhub.session.post = MagicMock()
expected_id = 123456
self.callhub.session.post.return_value.result.return_value.json.return_value = {"id": expected_id}
contact_id = self.callhub.create_contact({"first name": "Jimmy", "phone number": "5555555555"})
self.assertEqual(contact_id, expected_id)
with Mocker() as mock:
mock.post('https://api.callhub.io/v1/contacts/', json={"id": expected_id})

# Ensure contact creation fails on field mismatch
self.callhub.fields = MagicMock(return_value={"foo": 0, "bar": 1})
self.assertRaises(LookupError,
self.callhub.create_contact,
{"first name": "james", "phone number": "5555555555"},
)
# Test if contact creation successful
self.callhub.fields = MagicMock(return_value={"first name": 0, "phone number": 1})
contact_id = self.callhub.create_contact({"first name": "Jimmy", "phone number": "5555555555"})
self.assertEqual(contact_id, expected_id)

# Ensure contact creation fails on field mismatch
self.callhub.fields = MagicMock(return_value={"foo": 0, "bar": 1})
self.assertRaises(LookupError,
self.callhub.create_contact,
{"first name": "james", "phone number": "5555555555"},
)

def get_all_contacts(self, limit, count, status=200):
self.callhub.session.get = MagicMock()
self.callhub.session.get.return_value.result.return_value.status_code = status
page_json = {
"count": count,
"results": [
{"first name": "james"},
{"first name": "sumiya"}
]
}
self.callhub.session.get.return_value.result.return_value.json.return_value = page_json
expected_result = page_json["results"].copy()
# We expect get_contacts to fetch either the limit/page_size pages or the total/page_size pages, depending
# on which is smaller
expected_result *= min(math.ceil(limit/len(page_json["results"])), math.ceil(count/len(page_json["results"])))
expected_result *= min(math.ceil(limit / len(page_json["results"])),
math.ceil(count / len(page_json["results"])))
# We then expect get_contacts to trim the result to exactly the limit (because we fetch in batches equal to the
# page size but the limit is for the exact number of contacts)
expected_result = expected_result[:limit]
self.assertEqual(self.callhub.get_contacts(limit), expected_result)
# Test number of contacts matches size given
self.assertEqual(len(self.callhub.get_contacts(limit)), min(limit, count))
with Mocker() as mock:
mock.get('https://api.callhub.io/v1/contacts/', status_code=status, json=page_json)
# Test that the results of get_contacts match the expected results
self.assertEqual(self.callhub.get_contacts(limit), expected_result)
# Test number of contacts matches size given
self.assertEqual(len(self.callhub.get_contacts(limit)), min(limit, count))

def test_get_all_contacts(self):
# Test different variations of get_all_contacts with different numbers of contacts and different limits
Expand All @@ -191,5 +204,6 @@ def test_get_all_contacts(self):
# Test with 500 error
self.assertRaises(RuntimeError, self.get_all_contacts, limit=50, count=50, status=500)


if __name__ == '__main__':
unittest.main()