valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-07 08:58:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

rest of the unit tests for usage plans related functionality in boto_apigateway module.

Browse Source
This commit is contained in:
Winston Liu 2016-09-19 10:54:38 -07:00
parent 802fe3a76c
commit 962b0439a5
2 changed files with 117 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
salt/modules/boto_apigateway.py
View File

@ -1639,6 +1639,7 @@ def _update_usage_plan_apis(plan_id, apis, op, region=None, key=None, keyid=None
'path': '/apiStages',
'value': '{0}:{1}'.format(api['apiId'], api['stage'])
})
res = None
if patchOperations:
conn = _get_conn(region=region, key=key, keyid=keyid, profile=profile)
res = conn.update_usage_plan(usagePlanId=plan_id,
@ -1646,6 +1647,8 @@ def _update_usage_plan_apis(plan_id, apis, op, region=None, key=None, keyid=None
return {'success': True, 'result': res}
except ClientError as e:
return {'error': salt.utils.boto3.get_error(e)}
except Exception as e:
return {'error': e}
def attach_usage_plan_to_apis(plan_id, apis, region=None, key=None, keyid=None, profile=None):

115
tests/unit/modules/boto_apigateway_test.py
View File

@ -217,7 +217,7 @@ class BotoApiGatewayTestCaseMixin(object):
return False
#@skipIf(True, 'Skip these tests while investigating failures')
@skipIf(True, 'Skip these tests while investigating failures')
@skipIf(HAS_BOTO is False, 'The boto module must be installed.')
@skipIf(_has_required_boto() is False,
'The boto3 module must be greater than'
@ -1533,6 +1533,119 @@ class BotoApiGatewayTestCase(BotoApiGatewayTestCaseBase, BotoApiGatewayTestCaseM
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.conn.update_usage_plan.assert_called_once()
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.conn.delete_usage_plan.assert_not_called()
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.conn.delete_usage_plan.assert_called_once()
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.conn.delete_usage_plan.assert_not_called()
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.conn.delete_usage_plan.assert_called_once()
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.conn.update_usage_plan.assert_not_called()
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.conn.update_usage_plan.assert_not_called()
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.conn.update_usage_plan.assert_not_called()
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)
if __name__ == '__main__':
from integration import run_tests # pylint: disable=import-error
run_tests(BotoApiGatewayTestCase, needs_daemon=False)