Skip to content

DX-2751 Refactor Integration Tests to Use Hamcrest #105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ckoegel merged 7 commits into from
Aug 29, 2022
Merged

DX-2751 Refactor Integration Tests to Use Hamcrest #105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9e122a7
DX-2751 Refactor Integration Tests to Use Hamcrest
ckoegel Aug 26, 2022
2fb7fc2
fix messages tests
ckoegel Aug 26, 2022
f9b5152
fix recordings tests
ckoegel Aug 26, 2022
5b33440
media tests
ckoegel Aug 26, 2022
a4ba56c
empty array
ckoegel Aug 26, 2022
3cbd274
update tests to use env var helper
ckoegel Aug 29, 2022
e7962c3
_ :upside_down_face
ckoegel Aug 29, 2022
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
29 changes: 12 additions & 17 deletions test/integration/test_media_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import logging

import bandwidth
from hamcrest import *
from bandwidth.api import media_api
from bandwidth.model.media import Media
from bandwidth.exceptions import ApiException, NotFoundException
Expand Down Expand Up @@ -43,13 +44,13 @@ def uploadMedia(self) -> None:
account_id=self.account_id,
media_id=media_id,
body=self.original_file,
content_type=content_type,
_content_type=content_type,
cache_control=cache_control,
_return_http_data_only=False
)

logging.debug(api_response_with_http_info)
self.assertEqual(api_response_with_http_info[1], 204)
assert_that(api_response_with_http_info[1], equal_to(204))

# reopen the media file
# the client automatically closes any files passed into request bodies
Expand All @@ -60,7 +61,7 @@ def uploadMedia(self) -> None:
account_id=self.account_id,
media_id=media_id,
body=reopened_file,
content_type=content_type,
_content_type=content_type,
cache_control=cache_control,
_return_http_data_only=False
)
Expand All @@ -71,12 +72,12 @@ def listMedia(self) -> None:
api_response_with_http_info = self.api_instance.list_media(
self.account_id, _return_http_data_only=False)

self.assertEqual(api_response_with_http_info[1], 200)
assert_that(api_response_with_http_info[1], equal_to(200))

api_response = self.api_instance.list_media(self.account_id)
logging.debug("List Media" + str(api_response))

self.assertIs(type(api_response[0]), Media)
assert_that(api_response[0], instance_of(Media))
pass

def getMedia(self) -> None:
Expand All @@ -86,16 +87,16 @@ def getMedia(self) -> None:
self.account_id, self.media_id, _return_http_data_only=False)

logging.debug(api_response_with_http_info)
self.assertEqual(api_response_with_http_info[1], 200)
assert_that(api_response_with_http_info[1], equal_to(200))

api_response = self.api_instance.get_media(
self.account_id, self.media_id, _preload_content=False)

with open(self.media_path + self.download_file_path, "wb") as download_file:
download_file.write(api_response.data)

self.assertTrue(filecmp.cmp(self.media_path + self.media_file,
self.media_path + self.download_file_path))
assert_that(filecmp.cmp(self.media_path + self.media_file,
self.media_path + self.download_file_path), equal_to(True))
download_file.close()

def deleteMedia(self) -> None:
Expand All @@ -105,7 +106,7 @@ def deleteMedia(self) -> None:
self.account_id, self.media_id, _return_http_data_only=False)

logging.debug(api_response_with_http_info)
self.assertEqual(api_response_with_http_info[1], 204)
assert_that(api_response_with_http_info[1], equal_to(204))

# returns void
self.api_instance.delete_media(self.account_id, self.media_id)
Expand All @@ -127,11 +128,5 @@ def testGetMediaWithBandwidthId(self) -> None:
# use a nonexistent mediaId - results in a 404
media_id = "abcd1234-e5f6-1111-2222-3456ghi7890/image123456.jpg"

with self.assertRaises(NotFoundException) as context:
api_response = self.api_instance.get_media(
self.account_id,
media_id,
_preload_content=False
)

self.assertEqual(context.exception.status, 404)
assert_that(calling(self.api_instance.get_media).with_args(
self.account_id, media_id, _preload_content=False)), raises(NotFoundException)
136 changes: 58 additions & 78 deletions test/integration/test_messages_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from datetime import datetime

import bandwidth
from hamcrest import *
from bandwidth.api import messages_api
from bandwidth.model.list_message_direction_enum import ListMessageDirectionEnum
from bandwidth.model.list_message_item import ListMessageItem
Expand All @@ -26,6 +27,7 @@
from bandwidth.model.priority_enum import PriorityEnum
from bandwidth.model.message import Message
from bandwidth.exceptions import ApiException, UnauthorizedException
from test.utils.env_variables import *


class TestMessagesApi(unittest.TestCase):
Expand All @@ -34,21 +36,21 @@ class TestMessagesApi(unittest.TestCase):
def setUp(self):
# API Client
configuration = bandwidth.Configuration(
username = os.environ.get('BW_USERNAME'),
password = os.environ.get('BW_PASSWORD')
username = BW_USERNAME,
password = BW_PASSWORD
)
api_client = bandwidth.ApiClient(configuration)
self.api_instance = messages_api.MessagesApi(api_client)
self.account_id = os.environ.get('BW_ACCOUNT_ID')
self.account_id = BW_ACCOUNT_ID

# Unauthorized API Client
self.unauthorized_api_client = bandwidth.ApiClient()
self.unauthorized_api_instance = messages_api.MessagesApi(self.unauthorized_api_client)

# Message Properties
self.application_id = os.environ.get('BW_MESSAGING_APPLICATION_ID')
self.to_number = [os.environ.get('USER_NUMBER')]
self.from_number = os.environ.get('BW_NUMBER')
self.application_id = BW_MESSAGING_APPLICATION_ID
self.to_number = [USER_NUMBER]
self.from_number = BW_NUMBER
self.text = 'python integration'
self.media = ['https://cdn2.thecatapi.com/images/MTY3ODIyMQ.jpg']
self.tag = 'python integration tag'
Expand Down Expand Up @@ -77,104 +79,82 @@ def setUp(self):
def test_create_message(self):
response = self.api_instance.create_message(self.account_id, self.message_request, _return_http_data_only=False)

self.assertEqual(response[1], 202)
assert_that(response[1], equal_to(202))

api_response = response[0]
self.assertIsInstance(api_response, Message)
self.assertEqual(api_response.application_id,self.application_id)
self.assertEqual(api_response.to, self.to_number)
self.assertEqual(api_response._from, self.from_number)
self.assertEqual(api_response.owner, self.from_number)
self.assertEqual(api_response.text, self.text)
self.assertEqual(api_response.media, self.media)
self.assertEqual(api_response.tag, self.tag)
self.assertIsInstance(api_response.priority, PriorityEnum)
self.assertEqual(api_response.priority, self.priority)
self.assertEqual(api_response.segment_count, 1)
self.assertTrue(datetime.fromisoformat(api_response.time[:-1]))


def test_create_message_bad_request(self):
with self.assertRaises(ApiException) as context:
self.api_instance.create_message(self.account_id, self.invalid_message_request)

self.assertEqual(context.exception.status, 400)

e = json.loads(context.exception.body)
self.assertEqual(e['type'], 'request-validation')
self.assertIsInstance(e['description'], str)
self.assertIsInstance(e['fieldErrors'], list)

field_error = e['fieldErrors'][0]
self.assertEqual(field_error['fieldName'], 'to')
self.assertEqual(field_error['description'], "'+invalid' must be replaced with a valid E164 formatted telephone number")

assert_that(api_response, instance_of(Message))
assert_that(api_response, has_properties(
'application_id', self.application_id,
'to', self.to_number,
'_from', self.from_number,
'owner', self.from_number,
'text', self.text,
'media', self.media,
'tag', self.tag,
'priority', instance_of(PriorityEnum),
'priority', self.priority,
'segment_count', 1,
)
)
assert_that(datetime.fromisoformat(api_response.time[:-1]), instance_of(datetime))

def test_create_message_unauthorized(self):
with self.assertRaises(UnauthorizedException) as context:
self.unauthorized_api_instance.create_message(self.account_id, self.invalid_message_request)

self.assertEqual(context.exception.status, 401)

e = json.loads(context.exception.body)
self.assertEqual(e['type'], 'unauthorized')
self.assertEqual(e['description'], 'Authentication Failed')
def test_create_message_bad_request(self):
assert_that(calling(self.api_instance.create_message).with_args(
self.account_id, self.invalid_message_request)), raises(ApiException)


def test_create_message_unauthorized(self):
assert_that(calling(self.unauthorized_api_instance.create_message).with_args(
self.account_id, self.invalid_message_request)), raises(UnauthorizedException)


@unittest.skip('The SDK catches incorrect content-type before making the request and attempts to create an ApiException,\
but the creation of the exception fails since there is no response body. This should probably create some\
kind of Client Exception instead, since this is not an actual API Exception.')
def test_create_message_invalid_media(self):
with self.assertRaises(ApiException) as context:
self.api_instance.create_message(self.account_id, self.message_request, _content_type='application/xml')
assert_that(calling(self.api_instance.create_message).with_args(
self.account_id, self.message_request, _content_type='application/xml')), raises(ApiException)


def test_list_messages(self):
message_direction = ListMessageDirectionEnum("OUTBOUND")

response = self.api_instance.list_messages(self.account_id, message_direction=message_direction, _return_http_data_only=False)

self.assertEqual(response[1], 200)
assert_that(response[1], equal_to(200))

api_response = response[0]
self.assertIsInstance(api_response, MessagesList)
self.assertGreater(api_response.total_count, 0)
self.assertTrue(api_response.messages[0], ListMessageItem)
assert_that(api_response, instance_of(MessagesList))
assert_that(api_response, has_properties(
'total_count', greater_than(0),
'messages', instance_of(list)
))

assert_that(api_response.messages[0], instance_of(ListMessageItem))

message = api_response.messages[0]
self.assertEqual(message.account_id, self.account_id)
self.assertRegex(message.destination_tn, '^\\+[1-9]\\d{1,14}$')
self.assertEqual(message.message_direction, ListMessageDirectionEnum("OUTBOUND"))
self.assertTrue(message.message_id)
self.assertIsInstance(message.message_status, MessageStatusEnum)
self.assertIsInstance(message.message_type, MessageTypeEnum)
self.assertTrue(datetime.fromisoformat(message.receive_time[:-1]))
self.assertTrue(message.segment_count)
self.assertRegex(message.source_tn, '^\\+[1-9]\\d{1,14}$')
assert_that(message, has_properties(
'account_id', self.account_id,
'destination_tn', matches_regexp('^\\+[1-9]\\d{1,14}$'),
'message_direction', ListMessageDirectionEnum("OUTBOUND"),
'message_id', matches_regexp('^.+$'),
'message_status', instance_of(MessageStatusEnum),
'message_type', instance_of(MessageTypeEnum),
'segment_count', greater_than(0),
'source_tn', matches_regexp('^\\+[1-9]\\d{1,14}$')
))
assert_that(datetime.fromisoformat(message.receive_time[:-1]), instance_of(datetime))


def test_list_messages_bad_request(self):

with self.assertRaises(ApiException) as context:
self.api_instance.list_messages(self.account_id)

self.assertEqual(context.exception.status, 400)

e = json.loads(context.exception.body)
self.assertEqual(e['type'], 'bad-request')
self.assertIsInstance(e['description'], str)
def test_list_messages_bad_request(self):
assert_that(calling(self.api_instance.list_messages).with_args(
self.account_id), raises(ApiException))


def test_list_messages_unauthorized(self):

with self.assertRaises(UnauthorizedException) as context:
self.unauthorized_api_instance.list_messages(self.account_id)

self.assertEqual(context.exception.status, 401)

e = json.loads(context.exception.body)
self.assertEqual(e['type'], 'unauthorized')
self.assertEqual(e['description'], 'Your request could not be authenticated')
assert_that(calling(self.unauthorized_api_instance.list_messages).with_args(
self.account_id), raises(UnauthorizedException))


if __name__ == '__main__':
Expand Down
28 changes: 14 additions & 14 deletions test/integration/test_multi_factor_authentication.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,8 @@
from bandwidth.model.verify_code_response import VerifyCodeResponse
from bandwidth.model.voice_code_response import VoiceCodeResponse
from bandwidth.exceptions import ApiException, UnauthorizedException, ForbiddenException
from test.utils.env_variables import *

import hamcrest
from hamcrest.core import *
from hamcrest.library import *

Expand All @@ -27,31 +27,31 @@ class TestMultiFactorAuthentication(unittest.TestCase):

def setUp(self) -> None:
configuration = bandwidth.Configuration(
username=os.environ['BW_USERNAME'],
password=os.environ['BW_PASSWORD']
username=BW_USERNAME,
password=BW_PASSWORD
)
api_client = bandwidth.ApiClient(configuration)
self.api_instance = mfa_api.MFAApi(api_client)
self.account_id = os.environ['BW_ACCOUNT_ID']
self.account_id = BW_ACCOUNT_ID
self.messaging_code_request = CodeRequest(
to=os.environ['USER_NUMBER'],
_from=os.environ['BW_NUMBER'],
application_id=os.environ['BW_MESSAGING_APPLICATION_ID'],
to=USER_NUMBER,
_from=BW_NUMBER,
application_id=BW_MESSAGING_APPLICATION_ID,
scope="scope",
message="Your temporary {NAME} {SCOPE} code is {CODE}",
digits=6,
)
self.voice_code_request = CodeRequest(
to=os.environ['USER_NUMBER'],
_from=os.environ['BW_NUMBER'],
application_id=os.environ['BW_VOICE_APPLICATION_ID'],
to=USER_NUMBER,
_from=BW_NUMBER,
application_id=BW_VOICE_APPLICATION_ID,
scope="scope",
message="Your temporary {NAME} {SCOPE} code is {CODE}",
digits=6,
)
self.bad_code_request = CodeRequest(
to=os.environ['USER_NUMBER'],
_from=os.environ['BW_NUMBER'],
to=USER_NUMBER,
_from=BW_NUMBER,
application_id='not_an_application_id',
scope="scope",
message="Your temporary {NAME} {SCOPE} code is {CODE}",
Expand Down Expand Up @@ -154,8 +154,8 @@ def testForbiddenRequest(self) -> None:
"""Validate a forbidden (403) request
"""
configuration = bandwidth.Configuration(
username=os.environ['BW_USERNAME_FORBIDDEN'],
# password=os.environ['BW_PASSWORD_FORBIDDEN'],
username=FORBIDDEN_USERNAME,
# password=FORBIDDEN_PASSWORD,
password='bad_password'
)
forbidden_api_client = bandwidth.ApiClient(configuration)
Expand Down
Loading