salt/tests/unit/modules/test_boto_apigateway.py

1653 lines
96 KiB
Python
Raw Normal View History

# -*- coding: utf-8 -*-
# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals
2015-12-07 21:33:00 +00:00
import datetime
2016-02-12 01:56:28 +00:00
import logging
import random
import string
# Import Salt Testing libs
2017-02-19 16:43:29 +00:00
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.unit import skipIf, TestCase
from tests.support.mock import NO_MOCK, NO_MOCK_REASON, MagicMock, patch
# Import Salt libs
import salt.loader
import salt.modules.boto_apigateway as boto_apigateway
from salt.utils.versions import LooseVersion
# Import 3rd-party libs
# pylint: disable=import-error,no-name-in-module
try:
import boto3
import botocore
from botocore.exceptions import ClientError
HAS_BOTO = True
except ImportError:
HAS_BOTO = False
from salt.ext.six.moves import range, zip
Remove repr formatting flag in places where it is used solely for quoting (#34183) * salt/cloud/__init__.py: remove repr formatting * salt/cloud/clouds/azurearm.py: remove repr formatting * salt/cloud/clouds/ec2.py: remove repr formatting * salt/cloud/clouds/profitbricks.py: remove repr formatting * salt/loader.py: remove repr formatting * salt/modules/win_file.py: remove repr formatting * salt/modules/zypper.py: remove repr formatting * salt/pillar/consul_pillar.py: remove repr formatting * salt/renderers/pyobjects.py: remove repr formatting * salt/returners/sentry_return.py: remove repr formatting * salt/states/bower.py: remove repr formatting * salt/states/cabal.py: remove repr formatting * salt/states/cmd.py: remove repr formatting * salt/states/composer.py: remove repr formatting * salt/states/win_network.py: remove repr formatting * salt/states/eselect.py: remove repr formatting * salt/states/file.py: remove repr formatting * salt/states/htpasswd.py: remove repr formatting * salt/states/memcached.py: remove repr formatting * salt/states/npm.py: remove repr formatting * salt/states/pip_state.py: remove repr formatting * salt/states/pkg.py: remove repr formatting * salt/states/pkgrepo.py: remove repr formatting * salt/states/supervisord.py: remove repr formatting * salt/states/timezone.py: remove repr formatting * salt/states/virtualenv_mod.py: remove repr formatting * salt/states/dockerio.py: remove repr formatting * salt/states/win_system.py: remove repr formatting * salt/utils/nb_popen.py: remove repr formatting * salt/utils/cloud.py: remove repr formatting * Add pylint disable due to legit usage of repr flag See https://github.com/saltstack/salt-pylint/pull/6 * Fix composer tests These tests needed to be updated because quoting was changed in the state module in 9dc9146. There was an unnecessary !r used for the exception class there, which means that instead of the exception class being passed through the formatter and coming out with the equivalent value of err.__str__(), we get a repr'ed instance of the exception class (i.e. SaltException('',)) in the state output. The unit test was asserting that we have that repr'ed instance of SaltException in the output, a case of writing the test to confirm the badly-conceived output in the state. This has also been corrected. * salt/cloud/clouds/azurearm.py: lint fixes * salt/modules/boto_s3_bucket.py: lint fixes * salt/modules/minion.py: lint fixes * salt/modules/reg.py: lint fixes * salt/modules/testinframod.py: lint fixes * salt/modules/win_iis.py: lint fixes * salt/pillar/csvpillar.py: lint fixes * salt/utils/win_functions.py: lint fixes * salt/states/nxos.py: lint fixes * salt/returners/mongo_future_return.py: lint fixes * tests/integration/__init__.py: lint fixes * tests/unit/context_test.py: lint fixes * tests/integration/states/file.py: lint fixes * tests/integration/utils/test_reactor.py: lint fixes * tests/integration/utils/testprogram.py: lint fixes * tests/unit/__init__.py: lint fixes * tests/integration/shell/minion.py: lint fixes * tests/unit/modules/boto_apigateway_test.py: lint fixes * tests/unit/modules/boto_cognitoidentity_test.py: lint fixes * tests/unit/modules/boto_elasticsearch_domain_test.py: lint fixes * tests/unit/modules/k8s_test.py: lint fixes * tests/unit/modules/reg_win_test.py: lint fixes * tests/unit/states/boto_apigateway_test.py: lint fixes * tests/unit/states/boto_cognitoidentity_test.py: lint fixes * tests/unit/states/boto_elasticsearch_domain_test.py: lint fixes
2016-06-29 20:30:18 +00:00
# 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'
required_botocore_version = '1.4.49'
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'],
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 4, 19, 22, 18),
u'createdDate': datetime.datetime(2015, 11, 4, 19, 21, 7),
u'id': u'88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2',
u'name': u'test-salt-key',
'ResponseMetadata': {'HTTPStatusCode': 200,
2016-02-10 23:05:53 +00:00
'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'}}
usage_plan1 = dict(
id='plan1_id',
name='plan1_name',
description='plan1_desc',
apiStages=[],
throttle=dict(
burstLimit=123,
rateLimit=123.0
),
quota=dict(
limit=123,
offset=123,
period='DAY'
)
)
usage_plan2 = dict(
id='plan2_id',
name='plan2_name',
description='plan2_desc',
apiStages=[],
throttle=dict(
burstLimit=123,
rateLimit=123.0
),
quota=dict(
limit=123,
offset=123,
period='DAY'
)
)
usage_plan1b = dict(
id='another_plan1_id',
name='plan1_name',
description='another_plan1_desc',
apiStages=[],
throttle=dict(
burstLimit=123,
rateLimit=123.0
),
quota=dict(
limit=123,
offset=123,
period='DAY'
)
)
usage_plans_ret = dict(
items=[
usage_plan1, usage_plan2, usage_plan1b
]
)
log = logging.getLogger(__name__)
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
2016-02-10 23:05:53 +00:00
def _has_required_botocore():
'''
Returns True/False boolean depending on if botocore supports usage plan
'''
if not HAS_BOTO:
return False
elif LooseVersion(botocore.__version__) < LooseVersion(required_botocore_version):
return False
else:
return True
2017-02-19 16:43:29 +00:00
class BotoApiGatewayTestCaseBase(TestCase, LoaderModuleMockMixin):
conn = None
def setup_loader_modules(self):
2017-02-19 16:43:29 +00:00
self.opts = opts = salt.config.DEFAULT_MINION_OPTS
utils = salt.loader.utils(opts, whitelist=['boto3'])
return {
boto_apigateway: {
'__opts__': opts,
'__utils__': utils,
}
2017-02-19 16:43:29 +00:00
}
def setUp(self):
2017-02-19 16:43:29 +00:00
TestCase.setUp(self)
# __virtual__ must be caller in order for _get_conn to be injected
boto_apigateway.__init__(self.opts)
delattr(self, 'opts')
# Set up MagicMock to replace the boto3 session
# connections keep getting cached from prior tests, can't find the
# correct context object to clear it. So randomize the cache key, to prevent any
# cache hits
conn_parameters['key'] = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(50))
self.patcher = patch('boto3.session.Session')
self.addCleanup(self.patcher.stop)
2017-02-19 16:43:29 +00:00
self.addCleanup(delattr, self, 'patcher')
mock_session = self.patcher.start()
session_instance = mock_session.return_value
self.conn = MagicMock()
session_instance.client.return_value = self.conn
2017-02-19 16:43:29 +00:00
self.addCleanup(delattr, self, 'conn')
2016-02-10 23:05:53 +00:00
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
2017-02-19 16:43:29 +00:00
#@skipIf(True, 'Skip these tests while investigating failures')
@skipIf(HAS_BOTO is False, 'The boto3 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(_has_required_botocore() is False,
'The botocore module must be greater than'
' or equal to version {0}'.format(required_botocore_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):
2015-12-07 21:33:00 +00:00
'''
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'id': u'2ut6i4vyle',
u'name': u'Swagger Petstore'},
{u'description': u'testingabcd',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58),
u'id': u'g41ls77hz0',
u'name': u'testingabc'},
{u'description': u'a simple food delivery service test',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28),
u'id': u'h7pbwydho9',
u'name': u'Food Delivery Service'},
{u'description': u'Created by AWS Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41),
u'id': u'i2yyd1ldvj',
u'name': u'LambdaMicroservice'},
{u'description': u'cloud tap service with combination of API GW and Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18),
u'id': u'rm06h9oac4',
u'name': u'API Gateway Cloudtap Service'},
{u'description': u'testing1234',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44),
u'id': u'vtir6ssxvd',
u'name': u'testing123'}],
'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}}
2015-12-07 21:33:00 +00:00
items = self.conn.get_rest_apis.return_value['items']
get_apis_result = boto_apigateway.describe_apis(**conn_parameters)
2016-02-11 16:38:29 +00:00
items_dt = [boto_apigateway._convert_datetime_str(item) for item in items]
2015-12-07 21:33:00 +00:00
apis = get_apis_result.get('restapi')
2016-02-10 23:05:53 +00:00
diff = self._diff_list_dicts(apis, items_dt, 'id')
2016-02-10 23:05:53 +00:00
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'id': u'2ut6i4vyle',
u'name': u'Swagger Petstore'},
{u'description': u'testingabcd',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58),
u'id': u'g41ls77hz0',
u'name': u'testing123'},
{u'description': u'a simple food delivery service test',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28),
u'id': u'h7pbwydho9',
u'name': u'Food Delivery Service'},
{u'description': u'Created by AWS Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41),
u'id': u'i2yyd1ldvj',
u'name': u'LambdaMicroservice'},
{u'description': u'cloud tap service with combination of API GW and Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18),
u'id': u'rm06h9oac4',
u'name': u'API Gateway Cloudtap Service'},
{u'description': u'testing1234',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44),
u'id': u'vtir6ssxvd',
u'name': u'testing123'}],
'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': '2d31072c-9d15-11e5-9977-6d9fcfda9c0a'}}
2016-02-11 22:24:35 +00:00
expected_items = [{u'description': u'testingabcd', u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58),
u'id': u'g41ls77hz0', u'name': u'testing123'},
2016-02-11 22:24:35 +00:00
{u'description': u'testing1234', u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44),
u'id': u'vtir6ssxvd', u'name': u'testing123'}]
get_apis_result = boto_apigateway.describe_apis(name='testing123', **conn_parameters)
2016-02-11 16:38:29 +00:00
expected_items_dt = [boto_apigateway._convert_datetime_str(item) for item in expected_items]
apis = get_apis_result.get('restapi')
diff = self._diff_list_dicts(apis, expected_items_dt, 'id')
2015-12-07 21:33:00 +00:00
2016-02-11 22:24:35 +00:00
self.assertTrue(apis)
self.assertIs(diff, False)
2015-12-07 21:33:00 +00:00
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'id': u'2ut6i4vyle',
u'name': u'Swagger Petstore'},
{u'description': u'a simple food delivery service test',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28),
u'id': u'h7pbwydho9',
u'name': u'Food Delivery Service'},
{u'description': u'Created by AWS Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41),
u'id': u'i2yyd1ldvj',
u'name': u'LambdaMicroservice'},
{u'description': u'cloud tap service with combination of API GW and Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18),
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')
2016-02-11 22:24:35 +00:00
self.assertFalse(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()
2016-02-11 22:24:35 +00:00
assigned_api_id = 'created_api_id'
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')
2016-02-11 22:24:35 +00:00
self.assertTrue(create_api_result.get('created'))
self.assertTrue(api)
self.assertEqual(api['id'], assigned_api_id)
self.assertEqual(api['createdDate'], '{0}'.format(created_date))
self.assertEqual(api['name'], 'unit-testing123')
self.assertEqual(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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'id': u'2ut6i4vyle',
u'name': u'Swagger Petstore'},
{u'description': u'testingabcd',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 3, 21, 57, 58),
u'id': u'g41ls77hz0',
u'name': u'testing123'},
{u'description': u'a simple food delivery service test',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 23, 57, 28),
u'id': u'h7pbwydho9',
u'name': u'Food Delivery Service'},
{u'description': u'Created by AWS Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 4, 17, 55, 41),
u'id': u'i2yyd1ldvj',
u'name': u'LambdaMicroservice'},
{u'description': u'cloud tap service with combination of API GW and Lambda',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 22, 3, 18),
u'id': u'rm06h9oac4',
u'name': u'API Gateway Cloudtap Service'},
{u'description': u'testing1234',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44),
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)
2016-02-11 22:24:35 +00:00
self.assertTrue(delete_api_result.get('deleted'))
self.assertEqual(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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44),
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'],
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 4, 19, 22, 18),
u'createdDate': datetime.datetime(2015, 11, 4, 19, 21, 7),
u'id': u'88883333amaa1ZMVGCoLeaTrQk8kzOC36vCgRcT2',
u'name': u'test-salt-key'},
{u'description': u'testing_salt_123', u'enabled': True,
u'stageKeys': [],
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 12, 5, 0, 14, 49),
u'createdDate': datetime.datetime(2015, 12, 4, 22, 29, 33),
u'id': u'999999989b8cNSp4505pL6OgDe3oW7oY29Z3eIZ4',
u'name': u'testing_salt'}],
'ResponseMetadata': {'HTTPStatusCode': 200,
2016-02-10 23:05:53 +00:00
'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)
2016-02-11 16:38:29 +00:00
items_dt = [boto_apigateway._convert_datetime_str(item) for item in 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')
2016-02-11 22:24:35 +00:00
self.assertTrue(api_keys)
self.assertIs(diff, False)
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,
2016-02-10 23:05:53 +00:00
'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)
2016-02-10 23:05:53 +00:00
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')
2016-02-10 23:05:53 +00:00
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
'''
2016-02-11 22:24:35 +00:00
self.conn.get_deployments.return_value = {u'items': [{u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'id': u'n05smo'},
2016-02-11 22:24:35 +00:00
{u'createdDate': datetime.datetime(2015, 12, 2, 19, 51, 44),
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
'''
2016-02-11 22:24:35 +00:00
self.conn.get_deployment.return_value = {u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'deploymentId': 'n05smo',
u'description': 'test',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
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,
2016-02-10 23:05:53 +00:00
'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)
2016-02-10 23:05:53 +00:00
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)
2016-02-10 23:05:53 +00:00
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'deploymentId': u'n05smo',
u'description': u'test',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'stageName': u'test'},
{u'cacheClusterEnabled': False,
u'cacheClusterStatus': 'NOT_AVAILABLE',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 12, 17, 16, 33, 50),
u'deploymentId': u'n05smo',
u'description': u'dev',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 12, 17, 16, 33, 50),
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'deploymentId': u'n05smo',
u'description': u'test',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'deploymentId': u'n05smo',
u'description': u'test',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'deploymentId': u'n05smo',
u'description': u'test',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
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)
2016-02-11 22:24:35 +00:00
self.assertEqual(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',
2016-02-11 22:24:35 +00:00
u'createdDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
u'deploymentId': u'n05smo',
u'description': u'test',
2016-02-11 22:24:35 +00:00
u'lastUpdatedDate': datetime.datetime(2015, 11, 17, 16, 33, 50),
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)
2016-02-10 23:05:53 +00:00
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)
2016-02-10 23:05:53 +00:00
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',
2016-02-10 23:05:53 +00:00
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',
2016-02-10 23:05:53 +00:00
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', **conn_parameters)
resources = result.get('resources')
2016-02-10 23:05:53 +00:00
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')
2016-02-10 23:05:53 +00:00
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',
2016-02-10 23:05:53 +00:00
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',
2016-02-10 23:05:53 +00:00
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',
2016-02-10 23:05:53 +00:00
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'))
def test_that_when_describing_usage_plans_and_an_exception_is_thrown_in_get_usage_plans(self):
'''
Tests True for existence of 'error'
'''
self.conn.get_usage_plans.side_effect = ClientError(error_content, 'get_usage_plans_exception')
result = boto_apigateway.describe_usage_plans(name='some plan', **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format('get_usage_plans_exception'))
def test_that_when_describing_usage_plans_and_plan_name_or_id_does_not_exist_that_results_have_empty_plans_list(self):
'''
Tests for plans equaling empty list
'''
self.conn.get_usage_plans.return_value = usage_plans_ret
result = boto_apigateway.describe_usage_plans(name='does not exist', **conn_parameters)
self.assertEqual(result.get('plans'), [])
result = boto_apigateway.describe_usage_plans(plan_id='does not exist', **conn_parameters)
self.assertEqual(result.get('plans'), [])
result = boto_apigateway.describe_usage_plans(name='does not exist', plan_id='does not exist', **conn_parameters)
self.assertEqual(result.get('plans'), [])
result = boto_apigateway.describe_usage_plans(name='plan1_name', plan_id='does not exist', **conn_parameters)
self.assertEqual(result.get('plans'), [])
result = boto_apigateway.describe_usage_plans(name='does not exist', plan_id='plan1_id', **conn_parameters)
self.assertEqual(result.get('plans'), [])
def test_that_when_describing_usage_plans_for_plans_that_exist_that_the_function_returns_all_matching_plans(self):
'''
Tests for plans filtering properly if they exist
'''
self.conn.get_usage_plans.return_value = usage_plans_ret
result = boto_apigateway.describe_usage_plans(name=usage_plan1['name'], **conn_parameters)
self.assertEqual(len(result.get('plans')), 2)
for plan in result['plans']:
self.assertTrue(plan in [usage_plan1, usage_plan1b])
def test_that_when_creating_or_updating_a_usage_plan_and_throttle_or_quota_failed_to_validate_that_an_error_is_returned(self):
'''
Tests for TypeError and ValueError in throttle and quota
'''
for throttle, quota in (([], None), (None, []), ('abc', None), (None, 'def')):
res = boto_apigateway.create_usage_plan('plan1_name', description=None, throttle=throttle, quota=quota, **conn_parameters)
self.assertNotEqual(None, res.get('error'))
res = boto_apigateway.update_usage_plan('plan1_id', throttle=throttle, quota=quota, **conn_parameters)
self.assertNotEqual(None, res.get('error'))
for quota in ({'limit': 123}, {'period': 123}, {'period': 'DAY'}):
res = boto_apigateway.create_usage_plan('plan1_name', description=None, throttle=None, quota=quota, **conn_parameters)
self.assertNotEqual(None, res.get('error'))
res = boto_apigateway.update_usage_plan('plan1_id', quota=quota, **conn_parameters)
self.assertNotEqual(None, res.get('error'))
self.assertTrue(self.conn.get_usage_plans.call_count == 0)
self.assertTrue(self.conn.create_usage_plan.call_count == 0)
self.assertTrue(self.conn.update_usage_plan.call_count == 0)
def test_that_when_creating_a_usage_plan_and_create_usage_plan_throws_an_exception_that_an_error_is_returned(self):
'''
tests for ClientError
'''
self.conn.create_usage_plan.side_effect = ClientError(error_content, 'create_usage_plan_exception')
result = boto_apigateway.create_usage_plan(name='some plan', **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format('create_usage_plan_exception'))
def test_that_create_usage_plan_succeeds(self):
'''
tests for success user plan creation
'''
res = 'unit test create_usage_plan succeeded'
self.conn.create_usage_plan.return_value = res
result = boto_apigateway.create_usage_plan(name='some plan', **conn_parameters)
self.assertEqual(result.get('created'), True)
self.assertEqual(result.get('result'), res)
def test_that_when_udpating_a_usage_plan_and_update_usage_plan_throws_an_exception_that_an_error_is_returned(self):
'''
tests for ClientError
'''
self.conn.update_usage_plan.side_effect = ClientError(error_content, 'update_usage_plan_exception')
result = boto_apigateway.update_usage_plan(plan_id='plan1_id', **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format('update_usage_plan_exception'))
def test_that_when_updating_a_usage_plan_and_if_throttle_and_quota_parameters_are_none_update_usage_plan_removes_throttle_and_quota(self):
'''
tests for throttle and quota removal
'''
ret = 'some success status'
self.conn.update_usage_plan.return_value = ret
result = boto_apigateway.update_usage_plan(plan_id='plan1_id', throttle=None, quota=None, **conn_parameters)
self.assertEqual(result.get('updated'), True)
self.assertEqual(result.get('result'), ret)
self.assertTrue(self.conn.update_usage_plan.call_count >= 1)
def test_that_when_deleting_usage_plan_and_describe_usage_plans_had_error_that_the_same_error_is_returned(self):
'''
tests for error in describe_usage_plans returns error
'''
ret = 'get_usage_plans_exception'
self.conn.get_usage_plans.side_effect = ClientError(error_content, ret)
result = boto_apigateway.delete_usage_plan(plan_id='some plan id', **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format(ret))
self.assertTrue(self.conn.delete_usage_plan.call_count == 0)
def test_that_when_deleting_usage_plan_and_plan_exists_that_the_functions_returns_deleted_true(self):
self.conn.get_usage_plans.return_value = usage_plans_ret
ret = 'delete_usage_plan_retval'
self.conn.delete_usage_plan.return_value = ret
result = boto_apigateway.delete_usage_plan(plan_id='plan1_id', **conn_parameters)
self.assertEqual(result.get('deleted'), True)
self.assertEqual(result.get('usagePlanId'), 'plan1_id')
self.assertTrue(self.conn.delete_usage_plan.call_count >= 1)
def test_that_when_deleting_usage_plan_and_plan_does_not_exist_that_the_functions_returns_deleted_true(self):
'''
tests for ClientError
'''
self.conn.get_usage_plans.return_value = dict(
items=[]
)
ret = 'delete_usage_plan_retval'
self.conn.delete_usage_plan.return_value = ret
result = boto_apigateway.delete_usage_plan(plan_id='plan1_id', **conn_parameters)
self.assertEqual(result.get('deleted'), True)
self.assertEqual(result.get('usagePlanId'), 'plan1_id')
self.assertTrue(self.conn.delete_usage_plan.call_count == 0)
def test_that_when_deleting_usage_plan_and_delete_usage_plan_throws_exception_that_an_error_is_returned(self):
'''
tests for ClientError
'''
self.conn.get_usage_plans.return_value = usage_plans_ret
error_msg = 'delete_usage_plan_exception'
self.conn.delete_usage_plan.side_effect = ClientError(error_content, error_msg)
result = boto_apigateway.delete_usage_plan(plan_id='plan1_id', **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format(error_msg))
self.assertTrue(self.conn.delete_usage_plan.call_count >= 1)
def test_that_attach_or_detach_usage_plan_when_apis_is_empty_that_success_is_returned(self):
'''
tests for border cases when apis is empty list
'''
result = boto_apigateway.attach_usage_plan_to_apis(plan_id='plan1_id', apis=[], **conn_parameters)
self.assertEqual(result.get('success'), True)
self.assertEqual(result.get('result', 'no result?'), None)
self.assertTrue(self.conn.update_usage_plan.call_count == 0)
result = boto_apigateway.detach_usage_plan_from_apis(plan_id='plan1_id', apis=[], **conn_parameters)
self.assertEqual(result.get('success'), True)
self.assertEqual(result.get('result', 'no result?'), None)
self.assertTrue(self.conn.update_usage_plan.call_count == 0)
def test_that_attach_or_detach_usage_plan_when_api_does_not_contain_apiId_or_stage_that_an_error_is_returned(self):
'''
tests for invalid key in api object
'''
for api in ({'apiId': 'some Id'}, {'stage': 'some stage'}, {}):
result = boto_apigateway.attach_usage_plan_to_apis(plan_id='plan1_id', apis=[api], **conn_parameters)
self.assertNotEqual(result.get('error'), None)
result = boto_apigateway.detach_usage_plan_from_apis(plan_id='plan1_id', apis=[api], **conn_parameters)
self.assertNotEqual(result.get('error'), None)
self.assertTrue(self.conn.update_usage_plan.call_count == 0)
def test_that_attach_or_detach_usage_plan_and_update_usage_plan_throws_exception_that_an_error_is_returned(self):
'''
tests for ClientError
'''
api = {'apiId': 'some_id', 'stage': 'some_stage'}
error_msg = 'update_usage_plan_exception'
self.conn.update_usage_plan.side_effect = ClientError(error_content, error_msg)
result = boto_apigateway.attach_usage_plan_to_apis(plan_id='plan1_id', apis=[api], **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format(error_msg))
result = boto_apigateway.detach_usage_plan_from_apis(plan_id='plan1_id', apis=[api], **conn_parameters)
self.assertEqual(result.get('error').get('message'), error_message.format(error_msg))
def test_that_attach_or_detach_usage_plan_updated_successfully(self):
'''
tests for update_usage_plan called
'''
api = {'apiId': 'some_id', 'stage': 'some_stage'}
attach_ret = 'update_usage_plan_add_op_succeeded'
detach_ret = 'update_usage_plan_remove_op_succeeded'
self.conn.update_usage_plan.side_effect = [attach_ret, detach_ret]
result = boto_apigateway.attach_usage_plan_to_apis(plan_id='plan1_id', apis=[api], **conn_parameters)
self.assertEqual(result.get('success'), True)
self.assertEqual(result.get('result'), attach_ret)
result = boto_apigateway.detach_usage_plan_from_apis(plan_id='plan1_id', apis=[api], **conn_parameters)
self.assertEqual(result.get('success'), True)
self.assertEqual(result.get('result'), detach_ret)