Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_alias_on_non_existant_function_returns_error(self):
with self.app.test_request_context():
result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())
self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
result['message'])
iam_client = aws_stack.connect_to_service('iam')
lambda_client = aws_stack.connect_to_service('lambda')
# create lambda permission
action = 'lambda:InvokeFunction'
sid = 's3'
resp = lambda_client.add_permission(FunctionName=TEST_LAMBDA_NAME_PY, Action=action,
StatementId=sid, Principal='s3.amazonaws.com', SourceArn=aws_stack.s3_bucket_arn('test-bucket'))
self.assertIn('Statement', resp)
# fetch lambda policy
policy = lambda_client.get_policy(FunctionName=TEST_LAMBDA_NAME_PY)['Policy']
self.assertIsInstance(policy, six.string_types)
policy = json.loads(to_str(policy))
self.assertEqual(policy['Statement'][0]['Action'], action)
self.assertEqual(policy['Statement'][0]['Sid'], sid)
self.assertEqual(policy['Statement'][0]['Resource'], lambda_api.func_arn(TEST_LAMBDA_NAME_PY))
# fetch IAM policy
policies = iam_client.list_policies(Scope='Local', MaxItems=500)['Policies']
matching = [p for p in policies if p['PolicyName'] == 'lambda_policy_%s' % TEST_LAMBDA_NAME_PY]
self.assertEqual(len(matching), 1)
self.assertIn(':policy/', matching[0]['Arn'])
# remove permission that we just added
resp = lambda_client.remove_permission(FunctionName=TEST_LAMBDA_NAME_PY,
StatementId=resp['Statement'], Qualifier='qual1', RevisionId='r1')
self.assertEqual(resp['ResponseMetadata']['HTTPStatusCode'], 200)
def test_list_non_existant_function_aliases_returns_error(self):
with self.app.test_request_context():
result = json.loads(lambda_api.list_aliases(self.FUNCTION_NAME).get_data())
self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
result['message'])
def test_update_alias_on_non_existant_alias_returns_error(self):
with self.app.test_request_context():
self._create_function(self.FUNCTION_NAME)
result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())
alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME
self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])
self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])
def test_get_alias_on_non_existant_function_returns_error(self):
with self.app.test_request_context():
result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())
self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
result['message'])
def test_publish_function_version(self):
with self.app.test_request_context():
self._create_function(self.FUNCTION_NAME)
result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())
result2 = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())
result.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value
result2.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value
expected_result = dict()
expected_result['CodeSize'] = self.CODE_SIZE
expected_result['CodeSha256'] = self.CODE_SHA_256
expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'
expected_result['FunctionName'] = str(self.FUNCTION_NAME)
expected_result['Handler'] = str(self.HANDLER)
expected_result['Runtime'] = str(self.RUNTIME)
expected_result['Timeout'] = self.TIMEOUT
expected_result['Description'] = ''
expected_result['MemorySize'] = self.MEMORY_SIZE
expected_result['Role'] = self.ROLE
expected_result['LastModified'] = self.LAST_MODIFIED
expected_result['TracingConfig'] = self.TRACING_CONFIG
expected_result['Version'] = '1'
expected_result['State'] = 'Active'
expected_result2 = dict(expected_result)
expected_result2['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'
expected_result2['Version'] = '2'
self.assertDictEqual(expected_result, result)
self.assertDictEqual(expected_result2, result2)
def test_get_non_existent_function_returns_error(self):
with self.app.test_request_context():
result = json.loads(lambda_api.get_function('non_existent_function_name').get_data())
self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
self.assertEqual(
self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn('non_existent_function_name'),
result['message'])
def test_create_alias(self):
self._create_function(self.FUNCTION_NAME)
self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',
'Description': ''}))
result = json.loads(response.get_data())
result.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value
expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,
'FunctionVersion': '1', 'Description': '', 'Name': self.ALIAS_NAME}
self.assertDictEqual(expected_result, result)
def test_create_alias_returns_error_if_already_exists(self):
self._create_function(self.FUNCTION_NAME)
self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
data = json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''})
self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME), data=data)
response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
data=data)
result = json.loads(response.get_data())
alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME
self.assertEqual(self.ALIASEXISTS_EXCEPTION, result['__type'])
self.assertEqual(self.ALIASEXISTS_MESSAGE % alias_arn,
result['message'])
def test_tag_non_existent_function_returns_error(self):
with self.app.test_request_context():
arn = lambda_api.func_arn('non-existent-function')
response = self.client.post(
'{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn),
data=json.dumps({'Tags': self.TAGS}))
result = json.loads(response.get_data())
self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
self.assertEqual(
self.RESOURCENOTFOUND_MESSAGE % arn,
result['message'])