# -*- coding: utf-8 -*- # Import Python libs from __future__ import absolute_import from distutils.version import LooseVersion # pylint: disable=import-error,no-name-in-module import datetime from dateutil.tz import tzlocal from md5 import md5 # Import Salt Testing libs from salttesting.unit import skipIf, TestCase from salttesting.mock import NO_MOCK, NO_MOCK_REASON, MagicMock, patch from salttesting.helpers import ensure_in_syspath ensure_in_syspath('../../') # Import Salt libs import salt.config import salt.loader from salt.modules import boto_apigateway # Import 3rd-party libs import logging # pylint: disable=import-error,no-name-in-module try: import boto3 from botocore.exceptions import ClientError HAS_BOTO = True except ImportError: HAS_BOTO = False # pylint: enable=import-error,no-name-in-module # the boto_lambda module relies on the connect_to_region() method # which was added in boto 2.8.0 # https://github.com/boto/boto/commit/33ac26b416fbb48a60602542b4ce15dcc7029f12 required_boto3_version = '1.2.1' region = 'us-east-1' access_key = 'GKTADJGHEIQSXMKKRBJ08H' secret_key = 'askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs' conn_parameters = {'region': region, 'key': access_key, 'keyid': secret_key, 'profile': {}} error_message = 'An error occurred (101) when calling the {0} operation: Test-defined error' error_content = { 'Error': { 'Code': 101, 'Message': "Test-defined error" } } api_key_ret = { u'description': u'test-lambda-api-key', u'enabled': True, u'stageKeys': [u'123yd1l123/test'], u'lastUpdatedDate': datetime.datetime(2015, 11, 4, 19, 22, 18, tzinfo=tzlocal()), u'createdDate': datetime.datetime(2015, 11, 4, 19, 21, 7, tzinfo=tzlocal()), u'id': u'88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', u'name': u'test-salt-key', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} api_model_error_schema = u'{"properties":{"code":{"type":"integer","format":"int32"},"message":{"type":"string"},"fields":{"type":"string"}},"definitions":{}}' api_model_ret = { u'contentType': u'application/json', u'name': u'Error', u'description': u'Error Model', u'id': u'iltqcb', u'schema': api_model_error_schema, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} api_resources_ret = { u'items': [{u'id': u'hhg2t8f4h9', u'path': u'/'}, {u'id': u'isr8q2', u'parentId': u'hhg2t8f4h9', u'path': u'/api', u'pathPart': u'api'}, {u'id': u'5pvx7w', u'parentId': 'isr8q2', u'path': u'/api/users', u'pathPart': u'users', u'resourceMethods': {u'OPTIONS': {}, u'POST': {}}}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} api_create_resource_ret = { u'id': u'123abc', u'parentId': u'hhg2t8f4h9', u'path': u'/api3', u'pathPart': u'api3', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} log = logging.getLogger(__name__) opts = salt.config.DEFAULT_MINION_OPTS context = {} utils = salt.loader.utils(opts, whitelist=['boto3'], context=context) boto_apigateway.__utils__ = utils boto_apigateway.__init__(opts) boto_apigateway.__salt__ = {} def _has_required_boto(): ''' Returns True/False boolean depending on if Boto is installed and correct version. ''' if not HAS_BOTO: return False elif LooseVersion(boto3.__version__) < LooseVersion(required_boto3_version): return False else: return True class BotoApiGatewayTestCaseBase(TestCase): conn = None # Set up MagicMock to replace the boto3 session def setUp(self): global context context.clear() self.patcher = patch('boto3.session.Session') self.addCleanup(self.patcher.stop) mock_session = self.patcher.start() session_instance = mock_session.return_value self.conn = MagicMock() session_instance.client.return_value = self.conn class BotoApiGatewayTestCaseMixin(object): def _diff_list_dicts(self, listdict1, listdict2, sortkey): ''' Compares the two list of dictionaries to ensure they have same content. Returns True if there is difference, else False ''' if len(listdict1) != len(listdict2): return True listdict1_sorted = sorted(listdict1, key=lambda x: x[sortkey]) listdict2_sorted = sorted(listdict2, key=lambda x: x[sortkey]) for item1, item2 in zip(listdict1_sorted, listdict2_sorted): if len(set(item1) & set(item2)) != len(set(item2)): return True return False @skipIf(HAS_BOTO is False, 'The boto module must be installed.') @skipIf(_has_required_boto() is False, 'The boto3 module must be greater than' ' or equal to version {0}' .format(required_boto3_version)) @skipIf(NO_MOCK, NO_MOCK_REASON) class BotoApiGatewayTestCase(BotoApiGatewayTestCaseBase, BotoApiGatewayTestCaseMixin): ''' TestCase for salt.modules.boto_apigateway module ''' def test_that_when_checking_if_a_rest_api_exists_and_a_rest_api_exists_the_api_exists_method_returns_true(self): ''' Tests checking an apigateway rest api existence when api's name exists ''' self.conn.get_rest_apis.return_value = {'items': [{'name': 'myapi', 'id': '1234def'}]} api_exists_result = boto_apigateway.api_exists(name='myapi', **conn_parameters) self.assertTrue(api_exists_result['exists']) def test_that_when_checking_if_a_rest_api_exists_and_multiple_rest_api_exist_the_api_exists_method_returns_true(self): ''' Tests checking an apigateway rest api existence when multiple api's with same name exists ''' self.conn.get_rest_apis.return_value = {'items': [{'name': 'myapi', 'id': '1234abc'}, {'name': 'myapi', 'id': '1234def'}]} api_exists_result = boto_apigateway.api_exists(name='myapi', **conn_parameters) self.assertTrue(api_exists_result['exists']) def test_that_when_checking_if_a_rest_api_exists_and_no_rest_api_exists_the_api_exists_method_returns_false(self): ''' Tests checking an apigateway rest api existence when no matching rest api name exists ''' self.conn.get_rest_apis.return_value = {'items': [{'name': 'myapi', 'id': '1234abc'}, {'name': 'myapi', 'id': '1234def'}]} api_exists_result = boto_apigateway.api_exists(name='myapi123', **conn_parameters) self.assertFalse(api_exists_result['exists']) def test_that_when_describing_rest_apis_and_no_name_given_the_describe_apis_method_returns_list_of_all_rest_apis(self): ''' Tests that all rest apis defined for a region is returned ''' self.conn.get_rest_apis.return_value = {u'items': [{u'description': u'A sample API that uses a petstore as an example to demonstrate features in the swagger-2.0 specification', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'id': u'2ut6i4vyle', u'name': u'Swagger Petstore'}, {u'description': u'testingabcd', u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58, tzinfo=tzlocal()), u'id': u'g41ls77hz0', u'name': u'testingabc'}, {u'description': u'a simple food delivery service test', u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28, tzinfo=tzlocal()), u'id': u'h7pbwydho9', u'name': u'Food Delivery Service'}, {u'description': u'Created by AWS Lambda', u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41, tzinfo=tzlocal()), u'id': u'i2yyd1ldvj', u'name': u'LambdaMicroservice'}, {u'description': u'cloud tap service with combination of API GW and Lambda', u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18, tzinfo=tzlocal()), u'id': u'rm06h9oac4', u'name': u'API Gateway Cloudtap Service'}, {u'description': u'testing1234', u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44, tzinfo=tzlocal()), u'id': u'vtir6ssxvd', u'name': u'testing123'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} items = self.conn.get_rest_apis.return_value['items'] get_apis_result = boto_apigateway.describe_apis(**conn_parameters) items_dt = map(boto_apigateway._convert_datetime_str, items) apis = get_apis_result.get('restapi') diff = self._diff_list_dicts(apis, items_dt, 'id') self.assertTrue(apis) self.assertEqual(len(apis), len(items)) self.assertFalse(diff) def test_that_when_describing_rest_apis_and_name_is_testing123_the_describe_apis_method_returns_list_of_two_rest_apis(self): ''' Tests that exactly 2 apis are returned matching 'testing123' ''' self.conn.get_rest_apis.return_value = {u'items': [{u'description': u'A sample API that uses a petstore as an example to demonstrate features in the swagger-2.0 specification', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'id': u'2ut6i4vyle', u'name': u'Swagger Petstore'}, {u'description': u'testingabcd', u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58, tzinfo=tzlocal()), u'id': u'g41ls77hz0', u'name': u'testing123'}, {u'description': u'a simple food delivery service test', u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28, tzinfo=tzlocal()), u'id': u'h7pbwydho9', u'name': u'Food Delivery Service'}, {u'description': u'Created by AWS Lambda', u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41, tzinfo=tzlocal()), u'id': u'i2yyd1ldvj', u'name': u'LambdaMicroservice'}, {u'description': u'cloud tap service with combination of API GW and Lambda', u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18, tzinfo=tzlocal()), u'id': u'rm06h9oac4', u'name': u'API Gateway Cloudtap Service'}, {u'description': u'testing1234', u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44, tzinfo=tzlocal()), u'id': u'vtir6ssxvd', u'name': u'testing123'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} expected_items = [{u'description': u'testingabcd', u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58, tzinfo=tzlocal()), u'id': u'g41ls77hz0', u'name': u'testing123'}, {u'description': u'testing1234', u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44, tzinfo=tzlocal()), u'id': u'vtir6ssxvd', u'name': u'testing123'}] get_apis_result = boto_apigateway.describe_apis(name='testing123', **conn_parameters) expected_items_dt = map(boto_apigateway._convert_datetime_str, expected_items) apis = get_apis_result.get('restapi') diff = self._diff_list_dicts(apis, expected_items_dt, 'id') self.assertTrue(apis and not diff) def test_that_when_describing_rest_apis_and_name_is_testing123_the_describe_apis_method_returns_no_matching_items(self): ''' Tests that no apis are returned matching 'testing123' ''' self.conn.get_rest_apis.return_value = {u'items': [{u'description': u'A sample API that uses a petstore as an example to demonstrate features in the swagger-2.0 specification', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'id': u'2ut6i4vyle', u'name': u'Swagger Petstore'}, {u'description': u'a simple food delivery service test', u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28, tzinfo=tzlocal()), u'id': u'h7pbwydho9', u'name': u'Food Delivery Service'}, {u'description': u'Created by AWS Lambda', u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41, tzinfo=tzlocal()), u'id': u'i2yyd1ldvj', u'name': u'LambdaMicroservice'}, {u'description': u'cloud tap service with combination of API GW and Lambda', u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18, tzinfo=tzlocal()), u'id': u'rm06h9oac4', u'name': u'API Gateway Cloudtap Service'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} get_apis_result = boto_apigateway.describe_apis(name='testing123', **conn_parameters) apis = get_apis_result.get('restapi') self.assertTrue(not apis) def test_that_when_creating_a_rest_api_succeeds_the_create_api_method_returns_true(self): ''' test True if rest api is created ''' created_date = datetime.datetime.now() assigned_api_id = unicode(md5('{0}'.format(created_date)).hexdigest()[:10]) self.conn.create_rest_api.return_value = {u'description': u'unit-testing1234', u'createdDate': created_date, u'id': assigned_api_id, u'name': u'unit-testing123', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} create_api_result = boto_apigateway.create_api(name='unit-testing123', description='unit-testing1234', **conn_parameters) api = create_api_result.get('restapi') self.assertTrue(create_api_result.get('created') and api and api['id'] == assigned_api_id and api['createdDate'] == '{0}'.format(created_date) and api['name'] == 'unit-testing123' and api['description'] == 'unit-testing1234') def test_that_when_creating_a_rest_api_fails_the_create_api_method_returns_error(self): ''' test True for rest api creation error. ''' self.conn.create_rest_api.side_effect = ClientError(error_content, 'create_rest_api') create_api_result = boto_apigateway.create_api(name='unit-testing123', description='unit-testing1234', **conn_parameters) api = create_api_result.get('restapi') self.assertEqual(create_api_result.get('error').get('message'), error_message.format('create_rest_api')) def test_that_when_deleting_rest_apis_and_name_is_testing123_matching_two_apis_the_delete_api_method_returns_delete_count_of_two(self): ''' test True if the deleted count for "testing123" api is 2. ''' self.conn.get_rest_apis.return_value = {u'items': [{u'description': u'A sample API that uses a petstore as an example to demonstrate features in the swagger-2.0 specification', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'id': u'2ut6i4vyle', u'name': u'Swagger Petstore'}, {u'description': u'testingabcd', u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58, tzinfo=tzlocal()), u'id': u'g41ls77hz0', u'name': u'testing123'}, {u'description': u'a simple food delivery service test', u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28, tzinfo=tzlocal()), u'id': u'h7pbwydho9', u'name': u'Food Delivery Service'}, {u'description': u'Created by AWS Lambda', u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41, tzinfo=tzlocal()), u'id': u'i2yyd1ldvj', u'name': u'LambdaMicroservice'}, {u'description': u'cloud tap service with combination of API GW and Lambda', u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18, tzinfo=tzlocal()), u'id': u'rm06h9oac4', u'name': u'API Gateway Cloudtap Service'}, {u'description': u'testing1234', u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44, tzinfo=tzlocal()), u'id': u'vtir6ssxvd', u'name': u'testing123'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} self.conn.delete_rest_api.return_value = None delete_api_result = boto_apigateway.delete_api(name='testing123', **conn_parameters) self.assertTrue(delete_api_result.get('deleted') and delete_api_result.get('count') == 2) def test_that_when_deleting_rest_apis_and_name_given_provides_no_match_the_delete_api_method_returns_false(self): ''' Test that the given api name doesn't exists, and delete_api should return deleted status of False ''' self.conn.get_rest_apis.return_value = {u'items': [{u'description': u'testing1234', u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44, tzinfo=tzlocal()), u'id': u'vtir6ssxvd', u'name': u'testing1234'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} self.conn.delete_rest_api.return_value = None delete_api_result = boto_apigateway.delete_api(name='testing123', **conn_parameters) self.assertFalse(delete_api_result.get('deleted')) def test_that_describing_api_keys_the_describe_api_keys_method_returns_all_api_keys(self): ''' tests True if all api_keys are returned. ''' self.conn.get_api_keys.return_value = { u'items': [{u'description': u'test-lambda-api-key', u'enabled': True, u'stageKeys': [u'123yd1l123/test'], u'lastUpdatedDate': datetime.datetime(2015, 11, 4, 19, 22, 18, tzinfo=tzlocal()), u'createdDate': datetime.datetime(2015, 11, 4, 19, 21, 7, tzinfo=tzlocal()), u'id': u'88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', u'name': u'test-salt-key'}, {u'description': u'testing_salt_123', u'enabled': True, u'stageKeys': [], u'lastUpdatedDate': datetime.datetime(2015, 12, 5, 0, 14, 49, tzinfo=tzlocal()), u'createdDate': datetime.datetime(2015, 12, 4, 22, 29, 33, tzinfo=tzlocal()), u'id': u'999999989b8cNSp4505pL6OgDe3oW7oY29Z3eIZ4', u'name': u'testing_salt'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} items = self.conn.get_api_keys.return_value['items'] get_api_keys_result = boto_apigateway.describe_api_keys(**conn_parameters) items_dt = map(boto_apigateway._convert_datetime_str, items) api_keys = get_api_keys_result.get('apiKeys') diff = False if len(api_keys) != len(items): diff = True else: # compare individual items. diff = self._diff_list_dicts(api_keys, items_dt, 'id') self.assertTrue(api_keys and not diff) def test_that_describing_api_keys_fails_the_desribe_api_keys_method_returns_error(self): ''' test True for describe api keys error. ''' self.conn.get_api_keys.side_effect = ClientError(error_content, 'get_api_keys') result = boto_apigateway.describe_api_keys(**conn_parameters) self.assertEqual(result.get('error', {}).get('message'), error_message.format('get_api_keys')) def test_that_describing_an_api_key_the_describe_api_key_method_returns_matching_api_key(self): ''' tests True if the key is found. ''' self.conn.get_api_key.return_value = api_key_ret result = boto_apigateway.describe_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertEqual(result.get('apiKey', {}).get('id'), self.conn.get_api_key.return_value.get('id')) def test_that_describing_an_api_key_that_does_not_exists_the_desribe_api_key_method_returns_error(self): ''' test True for error being thrown. ''' self.conn.get_api_key.side_effect = ClientError(error_content, 'get_api_keys') result = boto_apigateway.describe_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertEqual(result.get('error', {}).get('message'), error_message.format('get_api_keys')) def test_that_when_creating_an_api_key_succeeds_the_create_api_key_method_returns_true(self): ''' tests that we can successfully create an api key and the createDat and lastUpdateDate are converted to string ''' now = datetime.datetime.now() self.conn.create_api_key.return_value = { u'description': u'test-lambda-api-key', u'enabled': True, u'stageKeys': [u'123yd1l123/test'], u'lastUpdatedDate': now, u'createdDate': now, u'id': u'88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', u'name': u'test-salt-key', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} create_api_key_result = boto_apigateway.create_api_key('test-salt-key', 'test-lambda-api-key', **conn_parameters) api_key = create_api_key_result.get('apiKey') now_str = '{0}'.format(now) self.assertTrue(create_api_key_result.get('created')) self.assertEqual(api_key.get('lastUpdatedDate'), now_str) self.assertEqual(api_key.get('createdDate'), now_str) def test_that_when_creating_an_api_key_fails_the_create_api_key_method_returns_error(self): ''' tests that we properly handle errors when create an api key fails. ''' self.conn.create_api_key.side_effect = ClientError(error_content, 'create_api_key') create_api_key_result = boto_apigateway.create_api_key('test-salt-key', 'unit-testing1234') api_key = create_api_key_result.get('apiKey') self.assertFalse(api_key) self.assertIs(create_api_key_result.get('created'), False) self.assertEqual(create_api_key_result.get('error').get('message'), error_message.format('create_api_key')) def test_that_when_deleting_an_api_key_that_exists_the_delete_api_key_method_returns_true(self): ''' test True if the api key is successfully deleted. ''' self.conn.delete_api_key.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.delete_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_an_api_key_that_does_not_exist_the_delete_api_key_method_returns_false(self): ''' Test that the given api key doesn't exists, and delete_api_key should return deleted status of False ''' self.conn.delete_api_key.side_effect = ClientError(error_content, 'delete_api_key') result = boto_apigateway.delete_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_updating_an_api_key_description_successfully_the_update_api_key_description_method_returns_true(self): ''' Test True if api key descriptipn update is successful ''' self.conn.update_api_key.return_value = api_key_ret result = boto_apigateway.update_api_key_description(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', description='test-lambda-api-key', **conn_parameters) self.assertTrue(result.get('updated')) def test_that_when_updating_an_api_key_description_for_a_key_that_does_not_exist_the_update_api_key_description_method_returns_false(self): ''' Test False if api key doesn't exists for the update request ''' self.conn.update_api_key.side_effect = ClientError(error_content, 'update_api_key') result = boto_apigateway.update_api_key_description(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', description='test-lambda-api-key', **conn_parameters) self.assertFalse(result.get('updated')) def test_that_when_enabling_an_api_key_that_exists_the_enable_api_key_method_returns_api_key(self): ''' Test True for the status of the enabled flag of the returned api key ''' self.conn.update_api_key.return_value = api_key_ret result = boto_apigateway.enable_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertTrue(result.get('apiKey', {}).get('enabled')) def test_that_when_enabling_an_api_key_that_does_not_exist_the_enable_api_key_method_returns_error(self): ''' Test Equality of the returned value of 'erorr' ''' self.conn.update_api_key.side_effect = ClientError(error_content, 'update_api_key') result = boto_apigateway.enable_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('update_api_key')) def test_that_when_disabling_an_api_key_that_exists_the_disable_api_key_method_returns_api_key(self): ''' Test False for the status of the enabled flag of the returned api key ''' self.conn.update_api_key.return_value = api_key_ret.copy() self.conn.update_api_key.return_value['enabled'] = False result = boto_apigateway.disable_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertFalse(result.get('apiKey', {}).get('enabled')) def test_that_when_disabling_an_api_key_that_does_not_exist_the_disable_api_key_method_returns_error(self): ''' Test Equality of the returned value of 'erorr' ''' self.conn.update_api_key.side_effect = ClientError(error_content, 'update_api_key') result = boto_apigateway.disable_api_key(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('update_api_key')) def test_that_when_associating_stages_to_an_api_key_that_exists_the_associate_api_key_stagekeys_method_returns_true(self): ''' Test True for returned value of 'associated' ''' self.conn.update_api_key.retuen_value = api_key_ret result = boto_apigateway.associate_api_key_stagekeys(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', stagekeyslist=[u'123yd1l123/test'], **conn_parameters) self.assertTrue(result.get('associated')) def test_that_when_associating_stages_to_an_api_key_that_does_not_exist_the_associate_api_key_stagekeys_method_returns_false(self): ''' Test False returned value of 'associated' ''' self.conn.update_api_key.side_effect = ClientError(error_content, 'update_api_key') result = boto_apigateway.associate_api_key_stagekeys(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', stagekeyslist=[u'123yd1l123/test'], **conn_parameters) self.assertFalse(result.get('associated')) def test_that_when_disassociating_stages_to_an_api_key_that_exists_the_disassociate_api_key_stagekeys_method_returns_true(self): ''' Test True for returned value of 'associated' ''' self.conn.update_api_key.retuen_value = None result = boto_apigateway.disassociate_api_key_stagekeys(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', stagekeyslist=[u'123yd1l123/test'], **conn_parameters) self.assertTrue(result.get('disassociated')) def test_that_when_disassociating_stages_to_an_api_key_that_does_not_exist_the_disassociate_api_key_stagekeys_method_returns_false(self): ''' Test False returned value of 'associated' ''' self.conn.update_api_key.side_effect = ClientError(error_content, 'update_api_key') result = boto_apigateway.disassociate_api_key_stagekeys(apiKey='88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2', stagekeyslist=[u'123yd1l123/test'], **conn_parameters) self.assertFalse(result.get('disassociated')) def test_that_when_describing_api_deployments_the_describe_api_deployments_method_returns_list_of_deployments(self): ''' Test Equality for number of deployments is 2 ''' self.conn.get_deployments.return_value = {u'items': [{u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'id': u'n05smo'}, {u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44, tzinfo=tzlocal()), u'id': u'n05sm1'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.describe_api_deployments(restApiId='rm06h9oac4', **conn_parameters) self.assertEqual(len(result.get('deployments', {})), 2) def test_that_when_describing_api_deployments_and_an_error_occurred_the_describe_api_deployments_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_deployments.side_effect = ClientError(error_content, 'get_deployments') result = boto_apigateway.describe_api_deployments(restApiId='rm06h9oac4', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_deployments')) def test_that_when_describing_an_api_deployment_the_describe_api_deployment_method_returns_the_deployment(self): ''' Test True for the returned deployment ''' self.conn.get_deployment.return_value = {u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'id': u'n05smo', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.describe_api_deployment(restApiId='rm06h9oac4', deploymentId='n05smo', **conn_parameters) self.assertTrue(result.get('deployment')) def test_that_when_describing_api_deployment_that_does_not_exist_the_describe_api_deployment_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_deployment.side_effect = ClientError(error_content, 'get_deployment') result = boto_apigateway.describe_api_deployment(restApiId='rm06h9oac4', deploymentId='n05smo', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_deployment')) def test_that_when_activating_api_deployment_for_stage_and_deployment_that_exist_the_activate_api_deployment_method_returns_true(self): ''' Test True for value of 'set' ''' self.conn.update_stage.return_value = {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAAILABLE', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': 'n05smo', u'description': 'test', u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': 'test', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.activate_api_deployment(restApiId='rm06h9oac4', stageName='test', deploymentId='n05smo', **conn_parameters) self.assertTrue(result.get('set')) def test_that_when_activating_api_deployment_for_stage_that_does_not_exist_the_activate_api_deployment_method_returns_false(self): ''' Test False for value of 'set' ''' self.conn.update_stage.side_effect = ClientError(error_content, 'update_stage') result = boto_apigateway.activate_api_deployment(restApiId='rm06h9oac4', stageName='test', deploymentId='n05smo', **conn_parameters) self.assertFalse(result.get('set')) def test_that_when_creating_an_api_deployment_succeeds_the_create_api_deployment_method_returns_true(self): ''' tests that we can successfully create an api deployment and the createDate is converted to string ''' now = datetime.datetime.now() self.conn.create_deployment.return_value = { u'description': u'test-lambda-api-key', u'id': 'n05smo', u'createdDate': now, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.create_api_deployment(restApiId='rm06h9oac4', stageName='test', **conn_parameters) deployment = result.get('deployment') now_str = '{0}'.format(now) self.assertTrue(result.get('created')) self.assertEqual(deployment.get('createdDate'), now_str) def test_that_when_creating_an_deployment_fails_the_create_api_deployment_method_returns_error(self): ''' tests that we properly handle errors when create an api deployment fails. ''' self.conn.create_deployment.side_effect = ClientError(error_content, 'create_deployment') result = boto_apigateway.create_api_deployment(restApiId='rm06h9oac4', stageName='test', **conn_parameters) self.assertIs(result.get('created'), False) self.assertEqual(result.get('error').get('message'), error_message.format('create_deployment')) def test_that_when_deleting_an_api_deployment_that_exists_the_delete_api_deployment_method_returns_true(self): ''' test True if the api deployment is successfully deleted. ''' self.conn.delete_deployment.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.delete_api_deployment(restApiId='rm06h9oac4', deploymentId='n05smo', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_an_api_deployment_that_does_not_exist_the_delete_api_deployment_method_returns_false(self): ''' Test that the given api deployment doesn't exists, and delete_api_deployment should return deleted status of False ''' self.conn.delete_deployment.side_effect = ClientError(error_content, 'delete_deployment') result = boto_apigateway.delete_api_deployment(restApiId='rm06h9oac4', deploymentId='n05smo1', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_describing_api_stages_the_describe_api_stages_method_returns_list_of_stages(self): ''' Test Equality for number of stages for the given deployment is 2 ''' self.conn.get_stages.return_value = {u'item': [{u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': u'n05smo', u'description': u'test', u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': u'test'}, {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': datetime.datetime(2015, 12, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': u'n05smo', u'description': u'dev', u'lastUpdatedDate': datetime.datetime(2015, 12, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': u'dev'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.describe_api_stages(restApiId='rm06h9oac4', deploymentId='n05smo', **conn_parameters) self.assertEqual(len(result.get('stages', {})), 2) def test_that_when_describing_api_stages_and_that_the_deployment_does_not_exist_the_describe_api_stages_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_stages.side_effect = ClientError(error_content, 'get_stages') result = boto_apigateway.describe_api_stages(restApiId='rm06h9oac4', deploymentId='n05smo', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_stages')) def test_that_when_describing_an_api_stage_the_describe_api_stage_method_returns_the_stage(self): ''' Test True for the returned stage ''' self.conn.get_stage.return_value = {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': u'n05smo', u'description': u'test', u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': u'test', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.describe_api_stage(restApiId='rm06h9oac4', stageName='test', **conn_parameters) self.assertTrue(result.get('stage')) def test_that_when_describing_api_stage_that_does_not_exist_the_describe_api_stage_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_stage.side_effect = ClientError(error_content, 'get_stage') result = boto_apigateway.describe_api_stage(restApiId='rm06h9oac4', stageName='no_such_stage', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_stage')) def test_that_when_overwriting_stage_variables_to_an_existing_stage_the_overwrite_api_stage_variables_method_returns_the_updated_stage(self): ''' Test True for the returned stage ''' self.conn.get_stage.return_value = {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': u'n05smo', u'description': u'test', u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': u'test', u'variables': {'key1': 'val1'}, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} self.conn.update_stage.return_value = {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': u'n05smo', u'description': u'test', u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': u'test', u'variables': {'key1': 'val2'}, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.overwrite_api_stage_variables(restApiId='rm06h9oac4', stageName='test', variables=dict(key1='val2'), **conn_parameters) self.assertTrue(result.get('stage').get('variables').get('key1') == 'val2') def test_that_when_overwriting_stage_variables_to_a_nonexisting_stage_the_overwrite_api_stage_variables_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_stage.side_effect = ClientError(error_content, 'get_stage') result = boto_apigateway.overwrite_api_stage_variables(restApiId='rm06h9oac4', stageName='no_such_stage', variables=dict(key1="val1", key2="val2"), **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_stage')) def test_that_when_overwriting_stage_variables_to_an_existing_stage_the_overwrite_api_stage_variables_method_returns_error(self): ''' Test Equality of error returned due to update_stage ''' self.conn.get_stage.return_value = {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'deploymentId': u'n05smo', u'description': u'test', u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50, tzinfo=tzlocal()), u'stageName': u'test', u'variables': {'key1': 'val1'}, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} self.conn.update_stage.side_effect = ClientError(error_content, 'update_stage') result = boto_apigateway.overwrite_api_stage_variables(restApiId='rm06h9oac4', stageName='test', variables=dict(key1='val2'), **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('update_stage')) def test_that_when_creating_an_api_stage_succeeds_the_create_api_stage_method_returns_true(self): ''' tests that we can successfully create an api stage and the createDate is converted to string ''' now = datetime.datetime.now() self.conn.create_stage.return_value = {u'cacheClusterEnabled': False, u'cacheClusterStatus': 'NOT_AVAILABLE', u'createdDate': now, u'deploymentId': u'n05smo', u'description': u'test', u'lastUpdatedDate': now, u'stageName': u'test', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.create_api_stage(restApiId='rm06h9oac4', stageName='test', deploymentId='n05smo', **conn_parameters) stage = result.get('stage') now_str = '{0}'.format(now) self.assertIs(result.get('created'), True) self.assertEqual(stage.get('createdDate'), now_str) self.assertEqual(stage.get('lastUpdatedDate'), now_str) def test_that_when_creating_an_api_stage_fails_the_create_api_stage_method_returns_error(self): ''' tests that we properly handle errors when create an api stage fails. ''' self.conn.create_stage.side_effect = ClientError(error_content, 'create_stage') result = boto_apigateway.create_api_stage(restApiId='rm06h9oac4', stageName='test', deploymentId='n05smo', **conn_parameters) self.assertIs(result.get('created'), False) self.assertEqual(result.get('error').get('message'), error_message.format('create_stage')) def test_that_when_deleting_an_api_stage_that_exists_the_delete_api_stage_method_returns_true(self): ''' test True if the api stage is successfully deleted. ''' self.conn.delete_stage.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.delete_api_stage(restApiId='rm06h9oac4', stageName='test', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_an_api_stage_that_does_not_exist_the_delete_api_stage_method_returns_false(self): ''' Test that the given api stage doesn't exists, and delete_api_stage should return deleted status of False ''' self.conn.delete_stage.side_effect = ClientError(error_content, 'delete_stage') result = boto_apigateway.delete_api_stage(restApiId='rm06h9oac4', stageName='no_such_stage', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_flushing_api_stage_cache_for_an_existing_stage_the_flush_api_stage_cache_method_returns_true(self): ''' Test True for 'flushed' ''' self.conn.flush_stage_cache.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.flush_api_stage_cache(restApiId='rm06h9oac4', stageName='no_such_stage', **conn_parameters) self.assertTrue(result.get('flushed')) def test_that_when_flushing_api_stage_cache_and_the_stage_does_not_exist_the_flush_api_stage_cache_method_returns_false(self): ''' Test False for 'flushed' ''' self.conn.flush_stage_cache.side_effect = ClientError(error_content, 'flush_stage_cache') result = boto_apigateway.flush_api_stage_cache(restApiId='rm06h9oac4', stageName='no_such_stage', **conn_parameters) self.assertFalse(result.get('flushed')) def test_that_when_describing_api_models_the_describe_api_models_method_returns_list_of_models(self): ''' Test Equality for number of models for the given api is 2 ''' self.conn.get_models.return_value = {u'items': [{u'contentType': u'application/json', u'name': u'Error', u'description': u'Error Model', u'id': u'iltqcb', u'schema': u'{"properties":{"code":{"type":"integer","format":"int32"},"message":{"type":"string"},"fields":{"type":"string"}},"definitions":{}}'}, {u'contentType': u'application/json', u'name': u'User', u'description': u'User Model', u'id': u'iltqcc', u'schema': u'{"properties":{"username":{"type":"string","description":"A unique username for the user"},"password":{"type":"string","description":"A password for the new user"}},"definitions":{}}'}], 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.describe_api_models(restApiId='rm06h9oac4', **conn_parameters) self.assertEqual(len(result.get('models', {})), 2) def test_that_when_describing_api_models_and_that_the_api_does_not_exist_the_describe_api_models_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_models.side_effect = ClientError(error_content, 'get_models') result = boto_apigateway.describe_api_models(restApiId='rm06h9oac4', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_models')) def test_that_when_describing_api_model_the_describe_api_model_method_returns_the_model(self): ''' Test True for the returned stage ''' self.conn.get_model.return_value = api_model_ret result = boto_apigateway.describe_api_model(restApiId='rm06h9oac4', modelName='Error', **conn_parameters) self.assertTrue(result.get('model')) def test_that_when_describing_api_model_and_that_the_model_does_not_exist_the_describe_api_model_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_model.side_effect = ClientError(error_content, 'get_model') result = boto_apigateway.describe_api_model(restApiId='rm06h9oac4', modelName='Error', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_model')) def test_that_model_exists_the_api_model_exists_method_returns_true(self): ''' Tests True when model exists ''' self.conn.get_model.return_value = api_model_ret result = boto_apigateway.api_model_exists(restApiId='rm06h9oac4', modelName='Error', **conn_parameters) self.assertTrue(result.get('exists')) def test_that_model_does_not_exists_the_api_model_exists_method_returns_false(self): ''' Tests False when model does not exist ''' self.conn.get_model.side_effect = ClientError(error_content, 'get_model') result = boto_apigateway.api_model_exists(restApiId='rm06h9oac4', modelName='Error', **conn_parameters) self.assertFalse(result.get('exists')) def test_that_updating_model_schema_the_update_api_model_schema_method_returns_true(self): ''' Tests True when model schema is updated. ''' self.conn.update_model.return_value = api_model_ret result = boto_apigateway.update_api_model_schema(restApiId='rm06h9oac4', modelName='Error', schema=api_model_error_schema, **conn_parameters) self.assertTrue(result.get('updated')) def test_that_updating_model_schema_when_model_does_not_exist_the_update_api_model_schema_emthod_returns_false(self): ''' Tests False when model schema is not upated. ''' self.conn.update_model.side_effect = ClientError(error_content, 'update_model') result = boto_apigateway.update_api_model_schema(restApiId='rm06h9oac4', modelName='no_such_model', schema=api_model_error_schema, **conn_parameters) self.assertFalse(result.get('updated')) def test_that_when_creating_an_api_model_succeeds_the_create_api_model_method_returns_true(self): ''' tests that we can successfully create an api model ''' self.conn.create_model.return_value = api_model_ret result = boto_apigateway.create_api_model(restApiId='rm06h9oac4', modelName='Error', modelDescription='Error Model', schema=api_model_error_schema, **conn_parameters) self.assertTrue(result.get('created')) def test_that_when_creating_an_api_model_fails_the_create_api_model_method_returns_error(self): ''' tests that we properly handle errors when create an api model fails. ''' self.conn.create_model.side_effect = ClientError(error_content, 'create_model') result = boto_apigateway.create_api_model(restApiId='rm06h9oac4', modelName='Error', modelDescription='Error Model', schema=api_model_error_schema, **conn_parameters) self.assertFalse(result.get('created')) def test_that_when_deleting_an_api_model_that_exists_the_delete_api_model_method_returns_true(self): ''' test True if the api model is successfully deleted. ''' self.conn.delete_model.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.delete_api_model(restApiId='rm06h9oac4', modelName='Error', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_an_api_model_that_does_not_exist_the_delete_api_model_method_returns_false(self): ''' Test that the given api model doesn't exists, and delete_api_model should return deleted status of False ''' self.conn.delete_model.side_effect = ClientError(error_content, 'delete_model') result = boto_apigateway.delete_api_model(restApiId='rm06h9oac4', modelName='no_such_model', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_describing_api_resources_the_describe_api_resources_method_returns_list_of_3_resources(self): ''' Test Equality for number of resources for the given api is 3 ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_resources(restApiId='rm06h9oac4', **conn_parameters) self.assertEqual(len(result.get('resources')), len(api_resources_ret.get('items'))) def test_that_when_describing_api_resources_and_that_the_api_does_not_exist_the_describe_api_resources_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_resources.side_effect = ClientError(error_content, 'get_resources') result = boto_apigateway.describe_api_resources(restApiId='rm06h9oac4', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_resources')) def test_that_when_describing_an_api_resource_that_exists_the_describe_api_resource_method_returns_the_resource(self): ''' Test Equality of the resource path returned is /api ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_resource(restApiId='rm06h9oac4', path="/api", **conn_parameters) self.assertEqual(result.get('resource', {}).get('path'), '/api') def test_that_when_describing_an_api_resource_that_does_not_exist_the_describe_api_resource_method_returns_the_resource_as_none(self): ''' Test Equality of the 'resource' is None ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_resource(restApiId='rm06h9oac4', path='/path/does/not/exist', **conn_parameters) self.assertEqual(result.get('resource'), None) def test_that_when_describing_an_api_resource_and_that_the_api_does_not_exist_the_describe_api_resource_method_returns_error(self): ''' Test Equality of error returned ''' self.conn.get_resources.side_effect = ClientError(error_content, 'get_resources') result = boto_apigateway.describe_api_resource(restApiId='bad_id', path="/api", **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_resources')) def test_that_when_creating_api_resources_for_a_path_that_creates_one_new_resource_the_create_resources_api_method_returns_all_resources(self): ''' Tests that a path of '/api3' returns 2 resources, named '/' and '/api'. ''' self.conn.get_resources.return_value = api_resources_ret self.conn.create_resource.return_value = api_create_resource_ret result = boto_apigateway.create_api_resources(restApiId='rm06h9oac4', path='/api3') resources = result.get('resources') self.assertIs(result.get('created'), True) self.assertEqual(len(resources), 2) self.assertEqual(resources[0].get('path'), '/') self.assertEqual(resources[1].get('path'), '/api3') def test_that_when_creating_api_resources_for_a_path_whose_resources_exist_the_create_resources_api_method_returns_all_resources(self): ''' Tests that a path of '/api/users' as defined in api_resources_ret return resources named '/', '/api', and '/api/users' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.create_api_resources(restApiId='rm06h9oac4', path='/api/users', **conn_parameters) resources = result.get('resources') self.assertIs(result.get('created'), True) self.assertEqual(len(resources), len(api_resources_ret.get('items'))) self.assertEqual(resources[0].get('path'), '/') self.assertEqual(resources[1].get('path'), '/api') self.assertEqual(resources[2].get('path'), '/api/users') def test_that_when_creating_api_resource_fails_the_create_resources_api_method_returns_false(self): ''' Tests False if we failed to create a resource ''' self.conn.get_resources.return_value = api_resources_ret self.conn.create_resource.side_effect = ClientError(error_content, 'create_resource') result = boto_apigateway.create_api_resources(restApiId='rm06h9oac4', path='/api4', **conn_parameters) self.assertFalse(result.get('created')) def test_that_when_deleting_api_resources_for_a_resource_that_exists_the_delete_api_resources_method_returns_true(self): ''' Tests True for '/api' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.delete_api_resources(restApiId='rm06h9oac4', path='/api', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_api_resources_for_a_resource_that_does_not_exist_the_delete_api_resources_method_returns_false(self): ''' Tests False for '/api5' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.delete_api_resources(restApiId='rm06h9oac4', path='/api5', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_deleting_the_root_api_resource_the_delete_api_resources_method_returns_false(self): ''' Tests False for '/' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.delete_api_resources(restApiId='rm06h9oac4', path='/', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_deleting_api_resources_and_delete_resource_throws_error_the_delete_api_resources_method_returns_false(self): ''' Tests False delete_resource side side_effect ''' self.conn.get_resources.return_value = api_resources_ret self.conn.delete_resource.side_effect = ClientError(error_content, 'delete_resource') result = boto_apigateway.delete_api_resources(restApiId='rm06h9oac4', path='/api', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_describing_an_api_resource_method_that_exists_the_describe_api_resource_method_returns_the_method(self): ''' Tests True for '/api/users' and POST ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_method.return_value = {u'httpMethod': 'POST', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.describe_api_resource_method(restApiId='rm06h9oac4', resourcepath='/api/users', httpMethod='POST', **conn_parameters) self.assertTrue(result.get('method')) def test_that_when_describing_an_api_resource_method_whose_method_does_not_exist_the_describe_api_resource_method_returns_error(self): ''' Tests Equality of returned error for '/api/users' and PUT ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_method.side_effect = ClientError(error_content, 'get_method') result = boto_apigateway.describe_api_resource_method(restApiId='rm06h9oac4', resourcepath='/api/users', httpMethod='PUT', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_method')) def test_that_when_describing_an_api_resource_method_whose_resource_does_not_exist_the_describe_api_resrouce_method_returns_error(self): ''' Tests True for resource not found error for '/does/not/exist' and POST ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_resource_method(restApiId='rm06h9oac4', resourcepath='/does/not/exist', httpMethod='POST', **conn_parameters) self.assertTrue(result.get('error')) def test_that_when_creating_an_api_method_the_create_api_method_method_returns_true(self): ''' Tests True on 'created' for '/api/users' and 'GET' ''' self.conn.get_resources.return_value = api_resources_ret self.conn.put_method.return_value = {u'httpMethod': 'GET', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.create_api_method(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='GET', authorizationType='NONE', **conn_parameters) self.assertTrue(result.get('created')) def test_that_when_creating_an_api_method_and_resource_does_not_exist_the_create_api_method_method_returns_false(self): ''' Tests False on 'created' for '/api5', and 'GET' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.create_api_method(restApiId='rm06h9oac4', resourcePath='/api5', httpMethod='GET', authorizationType='NONE', **conn_parameters) self.assertFalse(result.get('created')) def test_that_when_creating_an_api_method_and_error_thrown_on_put_method_the_create_api_method_method_returns_false(self): ''' Tests False on 'created' for '/api/users' and 'GET' ''' self.conn.get_resources.return_value = api_resources_ret self.conn.put_method.side_effect = ClientError(error_content, 'put_method') result = boto_apigateway.create_api_method(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='GET', authorizationType='NONE', **conn_parameters) self.assertFalse(result.get('created')) def test_that_when_deleting_an_api_method_for_a_method_that_exist_the_delete_api_method_method_returns_true(self): ''' Tests True for '/api/users' and 'POST' ''' self.conn.get_resources.return_value = api_resources_ret self.conn.delete_method.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.delete_api_method(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_an_api_method_for_a_method_that_does_not_exist_the_delete_api_method_method_returns_false(self): ''' Tests False for '/api/users' and 'GET' ''' self.conn.get_resources.return_value = api_resources_ret self.conn.delete_method.side_effect = ClientError(error_content, 'delete_method') result = boto_apigateway.delete_api_method(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='GET', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_deleting_an_api_method_for_a_resource_that_does_not_exist_the_delete_api_method_method_returns_false(self): ''' Tests False for '/api/users5' and 'POST' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.delete_api_method(restApiId='rm06h9oac4', resourcePath='/api/users5', httpMethod='POST', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_describing_an_api_method_response_that_exists_the_describe_api_method_respond_method_returns_the_response(self): ''' Tests True for 'response' for '/api/users', 'POST', and 200 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_method_response.return_value = {u'statusCode': 200, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.describe_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode=200, **conn_parameters) self.assertTrue(result.get('response')) def test_that_when_describing_an_api_method_response_and_response_code_does_not_exist_the_describe_api_method_response_method_returns_error(self): ''' Tests Equality of error msg thrown from get_method_response for '/api/users', 'POST', and 250 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_method_response.side_effect = ClientError(error_content, 'get_method_response') result = boto_apigateway.describe_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode=250, **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_method_response')) def test_that_when_describing_an_api_method_response_and_resource_does_not_exist_the_describe_api_method_response_method_returns_error(self): ''' Tests True for existence of 'error' for '/api5/users', 'POST', and 200 ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_method_response(restApiId='rm06h9oac4', resourcePath='/api5/users', httpMethod='POST', statusCode=200, **conn_parameters) self.assertTrue(result.get('error')) def test_that_when_creating_an_api_method_response_the_create_api_method_response_method_returns_true(self): ''' Tests True on 'created' for '/api/users', 'POST', 201 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.put_method_response.return_value = {u'statusCode': '201', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.create_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode='201', **conn_parameters) self.assertTrue(result.get('created')) def test_that_when_creating_an_api_method_response_and_resource_does_not_exist_the_create_api_method_response_method_returns_false(self): ''' Tests False on 'created' for '/api5', 'POST', 200 ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.create_api_method_response(restApiId='rm06h9oac4', resourcePath='/api5', httpMethod='POST', statusCode='200', **conn_parameters) self.assertFalse(result.get('created')) def test_that_when_creating_an_api_method_response_and_error_thrown_on_put_method_response_the_create_api_method_response_method_returns_false(self): ''' Tests False on 'created' for '/api/users', 'POST', 200 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.put_method_response.side_effect = ClientError(error_content, 'put_method_response') result = boto_apigateway.create_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode='200', **conn_parameters) self.assertFalse(result.get('created')) def test_that_when_deleting_an_api_method_response_for_a_response_that_exist_the_delete_api_method_response_method_returns_true(self): ''' Tests True for '/api/users', 'POST', 200 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.delete_method_response.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}} result = boto_apigateway.delete_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode='200', **conn_parameters) self.assertTrue(result.get('deleted')) def test_that_when_deleting_an_api_method_response_for_a_response_that_does_not_exist_the_delete_api_method_response_method_returns_false(self): ''' Tests False for '/api/users', 'POST', 201 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.delete_method_response.side_effect = ClientError(error_content, 'delete_method_response') result = boto_apigateway.delete_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='GET', statusCode='201', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_deleting_an_api_method_response_for_a_resource_that_does_not_exist_the_delete_api_method_response_method_returns_false(self): ''' Tests False for '/api/users5', 'POST', 200 ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.delete_api_method_response(restApiId='rm06h9oac4', resourcePath='/api/users5', httpMethod='POST', statusCode='200', **conn_parameters) self.assertFalse(result.get('deleted')) def test_that_when_describing_an_api_integration_that_exists_the_describe_api_integration_method_returns_the_intgration(self): ''' Tests True for 'integration' for '/api/users', 'POST' ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_integration.return_value = {u'type': 'AWS', u'uri': 'arn:aws:apigateway:us-west-2:lambda:path/2015-03-31/functions/arn:aws:lambda:us-west-2:1234568992820:function:echo_event/invocations', u'credentials': 'testing', u'httpMethod': 'POST', u'intgrationResponses': {'200': {}}, u'requestTemplates': {'application/json': {}}, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.describe_api_integration(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', **conn_parameters) self.assertTrue(result.get('integration')) def test_that_when_describing_an_api_integration_and_method_does_not_have_integration_defined_the_describe_api_integration_method_returns_error(self): ''' Tests Equality of error msg thrown from get_method_response for '/api/users', 'GET' ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_integration.side_effect = ClientError(error_content, 'get_integration') result = boto_apigateway.describe_api_integration(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='GET', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_integration')) def test_that_when_describing_an_api_integration_and_resource_does_not_exist_the_describe_api_integration_method_returns_error(self): ''' Tests True for existence of 'error' for '/api5/users', 'POST' ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_integration(restApiId='rm06h9oac4', resourcePath='/api5/users', httpMethod='POST', **conn_parameters) self.assertTrue(result.get('error')) def test_that_when_describing_an_api_integration_response_that_exists_the_describe_api_integration_response_method_returns_the_intgration(self): ''' Tests True for 'response' for '/api/users', 'POST', 200 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_integration_response.return_value = {u'responseParameters': {}, u'statusCode': 200, 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '7cc233dd-9dc8-11e5-ba47-1b7350cc2757'}} result = boto_apigateway.describe_api_integration_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode='200', **conn_parameters) self.assertTrue(result.get('response')) def test_that_when_describing_an_api_integration_response_and_status_code_does_not_exist_the_describe_api_integration_response_method_returns_error(self): ''' Tests Equality of error msg thrown from get_method_response for '/api/users', 'POST', 201 ''' self.conn.get_resources.return_value = api_resources_ret self.conn.get_integration_response.side_effect = ClientError(error_content, 'get_integration_response') result = boto_apigateway.describe_api_integration_response(restApiId='rm06h9oac4', resourcePath='/api/users', httpMethod='POST', statusCode='201', **conn_parameters) self.assertEqual(result.get('error').get('message'), error_message.format('get_integration_response')) def test_that_when_describing_an_api_integration_response_and_resource_does_not_exist_the_describe_api_integration_response_method_returns_error(self): ''' Tests True for existence of 'error' for '/api5/users', 'POST', 200 ''' self.conn.get_resources.return_value = api_resources_ret result = boto_apigateway.describe_api_integration_response(restApiId='rm06h9oac4', resourcePath='/api5/users', httpMethod='POST', statusCode='200', **conn_parameters) self.assertTrue(result.get('error')) if __name__ == '__main__': from integration import run_tests # pylint: disable=import-error run_tests(BotoApiGatewayTestCase, needs_daemon=False)