Merge pull request #542 from spulec/feature/lambda
AWS Lambda basic support: ListFunctions, CreateFunction, GetFunction, DeleteFunction
This commit is contained in:
Steve Pulec
commit
dba038683b
@ -83,6 +83,8 @@ It gets even better! Moto isn't just S3. Here's the status of the other AWS serv
|
||||
|------------------------------------------------------------------------------|
|
||||
| IAM | @mock_iam | core endpoints done |
|
||||
|------------------------------------------------------------------------------|
|
||||
| Lambda | @mock_lambda | basic endpoints done |
|
||||
|------------------------------------------------------------------------------|
|
||||
| Kinesis | @mock_kinesis | core endpoints done |
|
||||
|------------------------------------------------------------------------------|
|
||||
| RDS | @mock_rds | core endpoints done |
|
||||
|
||||
@ -6,6 +6,7 @@ __title__ = 'moto'
|
||||
__version__ = '0.4.21'
|
||||
|
||||
from .autoscaling import mock_autoscaling # flake8: noqa
|
||||
from .awslambda import mock_lambda # flake8: noqa
|
||||
from .cloudformation import mock_cloudformation # flake8: noqa
|
||||
from .cloudwatch import mock_cloudwatch # flake8: noqa
|
||||
from .datapipeline import mock_datapipeline # flake8: noqa
|
||||
|
||||
13
moto/awslambda/__init__.py
Normal file
13
moto/awslambda/__init__.py
Normal file
@ -0,0 +1,13 @@
|
||||
from __future__ import unicode_literals
|
||||
from .models import lambda_backends
|
||||
from ..core.models import MockAWS
|
||||
|
||||
|
||||
lambda_backend = lambda_backends['us-east-1']
|
||||
|
||||
|
||||
def mock_lambda(func=None):
|
||||
if func:
|
||||
return MockAWS(lambda_backends)(func)
|
||||
else:
|
||||
return MockAWS(lambda_backends)
|
||||
143
moto/awslambda/models.py
Normal file
143
moto/awslambda/models.py
Normal file
@ -0,0 +1,143 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import base64
|
||||
import datetime
|
||||
import hashlib
|
||||
|
||||
import boto.awslambda
|
||||
from moto.core import BaseBackend
|
||||
from moto.s3.models import s3_backend
|
||||
from moto.s3.exceptions import MissingBucket
|
||||
|
||||
|
||||
class LambdaFunction(object):
|
||||
|
||||
def __init__(self, spec):
|
||||
# required
|
||||
self.code = spec['Code']
|
||||
self.function_name = spec['FunctionName']
|
||||
self.handler = spec['Handler']
|
||||
self.role = spec['Role']
|
||||
self.run_time = spec['Runtime']
|
||||
|
||||
# optional
|
||||
self.description = spec.get('Description', '')
|
||||
self.memory_size = spec.get('MemorySize', 128)
|
||||
self.publish = spec.get('Publish', False) # this is ignored currently
|
||||
self.timeout = spec.get('Timeout', 3)
|
||||
|
||||
# this isn't finished yet. it needs to find out the VpcId value
|
||||
self._vpc_config = spec.get('VpcConfig', {'SubnetIds': [], 'SecurityGroupIds': []})
|
||||
|
||||
# auto-generated
|
||||
self.version = '$LATEST'
|
||||
self.last_modified = datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
|
||||
if 'ZipFile' in self.code:
|
||||
code = base64.b64decode(self.code['ZipFile'])
|
||||
self.code_size = len(code)
|
||||
self.code_sha_256 = hashlib.sha256(code).hexdigest()
|
||||
else:
|
||||
# validate s3 bucket
|
||||
try:
|
||||
# FIXME: does not validate bucket region
|
||||
key = s3_backend.get_key(self.code['S3Bucket'], self.code['S3Key'])
|
||||
except MissingBucket:
|
||||
raise ValueError(
|
||||
"InvalidParameterValueException",
|
||||
"Error occurred while GetObject. S3 Error Code: NoSuchBucket. S3 Error Message: The specified bucket does not exist")
|
||||
else:
|
||||
# validate s3 key
|
||||
if key is None:
|
||||
raise ValueError(
|
||||
"InvalidParameterValueException",
|
||||
"Error occurred while GetObject. S3 Error Code: NoSuchKey. S3 Error Message: The specified key does not exist.")
|
||||
else:
|
||||
self.code_size = key.size
|
||||
self.code_sha_256 = hashlib.sha256(key.value).hexdigest()
|
||||
self.function_arn = 'arn:aws:lambda:123456789012:function:{0}'.format(self.function_name)
|
||||
|
||||
@property
|
||||
def vpc_config(self):
|
||||
config = self._vpc_config.copy()
|
||||
if config['SecurityGroupIds']:
|
||||
config.update({"VpcId": "vpc-123abc"})
|
||||
return config
|
||||
|
||||
def __repr__(self):
|
||||
return json.dumps(self.get_configuration())
|
||||
|
||||
def get_configuration(self):
|
||||
return {
|
||||
"CodeSha256": self.code_sha_256,
|
||||
"CodeSize": self.code_size,
|
||||
"Description": self.description,
|
||||
"FunctionArn": self.function_arn,
|
||||
"FunctionName": self.function_name,
|
||||
"Handler": self.handler,
|
||||
"LastModified": self.last_modified,
|
||||
"MemorySize": self.memory_size,
|
||||
"Role": self.role,
|
||||
"Runtime": self.run_time,
|
||||
"Timeout": self.timeout,
|
||||
"Version": self.version,
|
||||
"VpcConfig": self.vpc_config,
|
||||
}
|
||||
|
||||
def get_code(self):
|
||||
return {
|
||||
"Code": {
|
||||
"Location": "s3://lambda-functions.aws.amazon.com/{0}".format(self.code['S3Key']),
|
||||
"RepositoryType": "S3"
|
||||
},
|
||||
"Configuration": self.get_configuration(),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
|
||||
properties = cloudformation_json['Properties']
|
||||
|
||||
# required
|
||||
spec = {
|
||||
'Code': properties['Code'],
|
||||
'FunctionName': resource_name,
|
||||
'Handler': properties['Handler'],
|
||||
'Role': properties['Role'],
|
||||
'Runtime': properties['Runtime'],
|
||||
}
|
||||
optional_properties = 'Description MemorySize Publish Timeout VpcConfig'.split()
|
||||
# NOTE: Not doing `properties.get(k, DEFAULT)` to avoid duplicating the default logic
|
||||
for prop in optional_properties:
|
||||
if prop in properties:
|
||||
spec[prop] = properties[prop]
|
||||
|
||||
backend = lambda_backends[region_name]
|
||||
fn = backend.create_function(spec)
|
||||
return fn
|
||||
|
||||
|
||||
class LambdaBackend(BaseBackend):
|
||||
|
||||
def __init__(self):
|
||||
self._functions = {}
|
||||
|
||||
def has_function(self, function_name):
|
||||
return function_name in self._functions
|
||||
|
||||
def create_function(self, spec):
|
||||
fn = LambdaFunction(spec)
|
||||
self._functions[fn.function_name] = fn
|
||||
return fn
|
||||
|
||||
def get_function(self, function_name):
|
||||
return self._functions[function_name]
|
||||
|
||||
def delete_function(self, function_name):
|
||||
del self._functions[function_name]
|
||||
|
||||
def list_functions(self):
|
||||
return self._functions.values()
|
||||
|
||||
|
||||
lambda_backends = {}
|
||||
for region in boto.awslambda.regions():
|
||||
lambda_backends[region.name] = LambdaBackend()
|
||||
82
moto/awslambda/responses.py
Normal file
82
moto/awslambda/responses.py
Normal file
@ -0,0 +1,82 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import json
|
||||
import re
|
||||
import uuid
|
||||
|
||||
from moto.core.responses import BaseResponse
|
||||
from .models import lambda_backends
|
||||
|
||||
|
||||
class LambdaResponse(BaseResponse):
|
||||
|
||||
@classmethod
|
||||
def root(cls, request, full_url, headers):
|
||||
if request.method == 'GET':
|
||||
return cls()._list_functions(request, full_url, headers)
|
||||
elif request.method == 'POST':
|
||||
return cls()._create_function(request, full_url, headers)
|
||||
else:
|
||||
raise ValueError("Cannot handle request")
|
||||
|
||||
@classmethod
|
||||
def function(cls, request, full_url, headers):
|
||||
if request.method == 'GET':
|
||||
return cls()._get_function(request, full_url, headers)
|
||||
elif request.method == 'DELETE':
|
||||
return cls()._delete_function(request, full_url, headers)
|
||||
else:
|
||||
raise ValueError("Cannot handle request")
|
||||
|
||||
def _list_functions(self, request, full_url, headers):
|
||||
lambda_backend = self.get_lambda_backend(full_url)
|
||||
return 200, headers, json.dumps({
|
||||
"Functions": [fn.get_configuration() for fn in lambda_backend.list_functions()],
|
||||
"NextMarker": str(uuid.uuid4()),
|
||||
})
|
||||
|
||||
def _create_function(self, request, full_url, headers):
|
||||
lambda_backend = self.get_lambda_backend(full_url)
|
||||
spec = json.loads(request.body.decode('utf-8'))
|
||||
try:
|
||||
fn = lambda_backend.create_function(spec)
|
||||
except ValueError as e:
|
||||
return 400, headers, json.dumps({"Error": {"Code": e.args[0], "Message": e.args[1]}})
|
||||
else:
|
||||
config = fn.get_configuration()
|
||||
return 201, headers, json.dumps(config)
|
||||
|
||||
def _delete_function(self, request, full_url, headers):
|
||||
lambda_backend = self.get_lambda_backend(full_url)
|
||||
|
||||
function_name = request.path.split('/')[-1]
|
||||
|
||||
if lambda_backend.has_function(function_name):
|
||||
lambda_backend.delete_function(function_name)
|
||||
return 204, headers, ""
|
||||
else:
|
||||
return 404, headers, "{}"
|
||||
|
||||
def _get_function(self, request, full_url, headers):
|
||||
lambda_backend = self.get_lambda_backend(full_url)
|
||||
|
||||
function_name = request.path.split('/')[-1]
|
||||
|
||||
if lambda_backend.has_function(function_name):
|
||||
fn = lambda_backend.get_function(function_name)
|
||||
code = fn.get_code()
|
||||
return 200, headers, json.dumps(code)
|
||||
else:
|
||||
return 404, headers, "{}"
|
||||
|
||||
def get_lambda_backend(self, full_url):
|
||||
from moto.awslambda.models import lambda_backends
|
||||
region = self._get_aws_region(full_url)
|
||||
return lambda_backends[region]
|
||||
|
||||
def _get_aws_region(self, full_url):
|
||||
region = re.search(self.region_regex, full_url)
|
||||
if region:
|
||||
return region.group(1)
|
||||
else:
|
||||
return self.default_region
|
||||
12
moto/awslambda/urls.py
Normal file
12
moto/awslambda/urls.py
Normal file
@ -0,0 +1,12 @@
|
||||
from __future__ import unicode_literals
|
||||
from .responses import LambdaResponse
|
||||
|
||||
url_bases = [
|
||||
"https?://lambda.(.+).amazonaws.com",
|
||||
]
|
||||
|
||||
url_paths = {
|
||||
# double curly braces because the `format()` method is called on the strings
|
||||
'{0}/\d{{4}}-\d{{2}}-\d{{2}}/functions/?$': LambdaResponse.root,
|
||||
'{0}/\d{{4}}-\d{{2}}-\d{{2}}/functions/(?P<function_name>[\w_-]+)/?$': LambdaResponse.function,
|
||||
}
|
||||
@ -1,5 +1,6 @@
|
||||
from __future__ import unicode_literals
|
||||
from moto.autoscaling import autoscaling_backend
|
||||
from moto.awslambda import lambda_backend
|
||||
from moto.cloudwatch import cloudwatch_backend
|
||||
from moto.cloudformation import cloudformation_backend
|
||||
from moto.datapipeline import datapipeline_backend
|
||||
@ -43,7 +44,8 @@ BACKENDS = {
|
||||
'sns': sns_backend,
|
||||
'sqs': sqs_backend,
|
||||
'sts': sts_backend,
|
||||
'route53': route53_backend
|
||||
'route53': route53_backend,
|
||||
'lambda': lambda_backend,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -4,6 +4,7 @@ import functools
|
||||
import logging
|
||||
|
||||
from moto.autoscaling import models as autoscaling_models
|
||||
from moto.awslambda import models as lambda_models
|
||||
from moto.datapipeline import models as datapipeline_models
|
||||
from moto.ec2 import models as ec2_models
|
||||
from moto.elb import models as elb_models
|
||||
@ -21,6 +22,7 @@ from boto.exception import BotoServerError
|
||||
MODEL_MAP = {
|
||||
"AWS::AutoScaling::AutoScalingGroup": autoscaling_models.FakeAutoScalingGroup,
|
||||
"AWS::AutoScaling::LaunchConfiguration": autoscaling_models.FakeLaunchConfiguration,
|
||||
"AWS::Lambda::Function": lambda_models.LambdaFunction,
|
||||
"AWS::EC2::EIP": ec2_models.ElasticAddress,
|
||||
"AWS::EC2::Instance": ec2_models.Instance,
|
||||
"AWS::EC2::InternetGateway": ec2_models.InternetGateway,
|
||||
|
||||
@ -5,5 +5,6 @@ sure>=1.2.24
|
||||
coverage
|
||||
freezegun
|
||||
flask
|
||||
boto3
|
||||
boto3>=1.2.3
|
||||
botocore>=1.3.26
|
||||
six
|
||||
301
tests/test_awslambda/test_lambda.py
Normal file
301
tests/test_awslambda/test_lambda.py
Normal file
@ -0,0 +1,301 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import botocore.client
|
||||
import boto3
|
||||
import hashlib
|
||||
import io
|
||||
import zipfile
|
||||
import sure # noqa
|
||||
|
||||
from freezegun import freeze_time
|
||||
from moto import mock_lambda, mock_s3
|
||||
|
||||
|
||||
def get_test_zip_file():
|
||||
zip_output = io.BytesIO()
|
||||
zip_file = zipfile.ZipFile(zip_output, 'w')
|
||||
zip_file.writestr('lambda_function.py', b'''\
|
||||
def handler(event, context):
|
||||
return "hello world"
|
||||
''')
|
||||
zip_file.close()
|
||||
zip_output.seek(0)
|
||||
return zip_output.read()
|
||||
|
||||
|
||||
@mock_lambda
|
||||
def test_list_functions():
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
result = conn.list_functions()
|
||||
|
||||
result['Functions'].should.have.length_of(0)
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@freeze_time('2015-01-01 00:00:00')
|
||||
def test_create_based_on_s3_with_missing_bucket():
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
conn.create_function.when.called_with(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.handler',
|
||||
Code={
|
||||
'S3Bucket': 'this-bucket-does-not-exist',
|
||||
'S3Key': 'test.zip',
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
VpcConfig={
|
||||
"SecurityGroupIds": ["sg-123abc"],
|
||||
"SubnetIds": ["subnet-123abc"],
|
||||
},
|
||||
).should.throw(botocore.client.ClientError)
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
@freeze_time('2015-01-01 00:00:00')
|
||||
def test_create_function_from_aws_bucket():
|
||||
s3_conn = boto3.client('s3', 'us-west-2')
|
||||
s3_conn.create_bucket(Bucket='test-bucket')
|
||||
|
||||
zip_content = get_test_zip_file()
|
||||
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
result = conn.create_function(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.handler',
|
||||
Code={
|
||||
'S3Bucket': 'test-bucket',
|
||||
'S3Key': 'test.zip',
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
VpcConfig={
|
||||
"SecurityGroupIds": ["sg-123abc"],
|
||||
"SubnetIds": ["subnet-123abc"],
|
||||
},
|
||||
)
|
||||
result.should.equal({
|
||||
'FunctionName': 'testFunction',
|
||||
'FunctionArn': 'arn:aws:lambda:123456789012:function:testFunction',
|
||||
'Runtime': 'python2.7',
|
||||
'Role': 'test-iam-role',
|
||||
'Handler': 'lambda_function.handler',
|
||||
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
|
||||
"CodeSize": len(zip_content),
|
||||
'Description': 'test lambda function',
|
||||
'Timeout': 3,
|
||||
'MemorySize': 128,
|
||||
'LastModified': '2015-01-01 00:00:00',
|
||||
'Version': '$LATEST',
|
||||
'VpcConfig': {
|
||||
"SecurityGroupIds": ["sg-123abc"],
|
||||
"SubnetIds": ["subnet-123abc"],
|
||||
"VpcId": "vpc-123abc"
|
||||
},
|
||||
|
||||
'ResponseMetadata': {'HTTPStatusCode': 201},
|
||||
})
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@freeze_time('2015-01-01 00:00:00')
|
||||
def test_create_function_from_zipfile():
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
zip_content = get_test_zip_file()
|
||||
result = conn.create_function(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.handler',
|
||||
Code={
|
||||
'ZipFile': zip_content,
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
)
|
||||
result.should.equal({
|
||||
'FunctionName': 'testFunction',
|
||||
'FunctionArn': 'arn:aws:lambda:123456789012:function:testFunction',
|
||||
'Runtime': 'python2.7',
|
||||
'Role': 'test-iam-role',
|
||||
'Handler': 'lambda_function.handler',
|
||||
'CodeSize': len(zip_content),
|
||||
'Description': 'test lambda function',
|
||||
'Timeout': 3,
|
||||
'MemorySize': 128,
|
||||
'LastModified': '2015-01-01 00:00:00',
|
||||
'CodeSha256': hashlib.sha256(zip_content).hexdigest(),
|
||||
'Version': '$LATEST',
|
||||
'VpcConfig': {
|
||||
"SecurityGroupIds": [],
|
||||
"SubnetIds": [],
|
||||
},
|
||||
|
||||
'ResponseMetadata': {'HTTPStatusCode': 201},
|
||||
})
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
@freeze_time('2015-01-01 00:00:00')
|
||||
def test_get_function():
|
||||
s3_conn = boto3.client('s3', 'us-west-2')
|
||||
s3_conn.create_bucket(Bucket='test-bucket')
|
||||
|
||||
zip_content = get_test_zip_file()
|
||||
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
conn.create_function(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.handler',
|
||||
Code={
|
||||
'S3Bucket': 'test-bucket',
|
||||
'S3Key': 'test.zip',
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
)
|
||||
|
||||
result = conn.get_function(FunctionName='testFunction')
|
||||
|
||||
result.should.equal({
|
||||
"Code": {
|
||||
"Location": "s3://lambda-functions.aws.amazon.com/test.zip",
|
||||
"RepositoryType": "S3"
|
||||
},
|
||||
"Configuration": {
|
||||
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
|
||||
"CodeSize": len(zip_content),
|
||||
"Description": "test lambda function",
|
||||
"FunctionArn": "arn:aws:lambda:123456789012:function:testFunction",
|
||||
"FunctionName": "testFunction",
|
||||
"Handler": "lambda_function.handler",
|
||||
"LastModified": "2015-01-01 00:00:00",
|
||||
"MemorySize": 128,
|
||||
"Role": "test-iam-role",
|
||||
"Runtime": "python2.7",
|
||||
"Timeout": 3,
|
||||
"Version": '$LATEST',
|
||||
"VpcConfig": {
|
||||
"SecurityGroupIds": [],
|
||||
"SubnetIds": [],
|
||||
}
|
||||
},
|
||||
'ResponseMetadata': {'HTTPStatusCode': 200},
|
||||
})
|
||||
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
def test_delete_function():
|
||||
s3_conn = boto3.client('s3', 'us-west-2')
|
||||
s3_conn.create_bucket(Bucket='test-bucket')
|
||||
|
||||
zip_content = get_test_zip_file()
|
||||
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
conn.create_function(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.handler',
|
||||
Code={
|
||||
'S3Bucket': 'test-bucket',
|
||||
'S3Key': 'test.zip',
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
)
|
||||
|
||||
success_result = conn.delete_function(FunctionName='testFunction')
|
||||
success_result.should.equal({'ResponseMetadata': {'HTTPStatusCode': 204}})
|
||||
|
||||
conn.delete_function.when.called_with(FunctionName='testFunctionThatDoesntExist').should.throw(botocore.client.ClientError)
|
||||
|
||||
|
||||
@mock_lambda
|
||||
@mock_s3
|
||||
@freeze_time('2015-01-01 00:00:00')
|
||||
def test_list_create_list_get_delete_list():
|
||||
"""
|
||||
test `list -> create -> list -> get -> delete -> list` integration
|
||||
|
||||
"""
|
||||
s3_conn = boto3.client('s3', 'us-west-2')
|
||||
s3_conn.create_bucket(Bucket='test-bucket')
|
||||
|
||||
zip_content = get_test_zip_file()
|
||||
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
|
||||
conn = boto3.client('lambda', 'us-west-2')
|
||||
|
||||
conn.list_functions()['Functions'].should.have.length_of(0)
|
||||
|
||||
conn.create_function(
|
||||
FunctionName='testFunction',
|
||||
Runtime='python2.7',
|
||||
Role='test-iam-role',
|
||||
Handler='lambda_function.handler',
|
||||
Code={
|
||||
'S3Bucket': 'test-bucket',
|
||||
'S3Key': 'test.zip',
|
||||
},
|
||||
Description='test lambda function',
|
||||
Timeout=3,
|
||||
MemorySize=128,
|
||||
Publish=True,
|
||||
)
|
||||
expected_function_result = {
|
||||
"Code": {
|
||||
"Location": "s3://lambda-functions.aws.amazon.com/test.zip",
|
||||
"RepositoryType": "S3"
|
||||
},
|
||||
"Configuration": {
|
||||
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
|
||||
"CodeSize": len(zip_content),
|
||||
"Description": "test lambda function",
|
||||
"FunctionArn": "arn:aws:lambda:123456789012:function:testFunction",
|
||||
"FunctionName": "testFunction",
|
||||
"Handler": "lambda_function.handler",
|
||||
"LastModified": "2015-01-01 00:00:00",
|
||||
"MemorySize": 128,
|
||||
"Role": "test-iam-role",
|
||||
"Runtime": "python2.7",
|
||||
"Timeout": 3,
|
||||
"Version": '$LATEST',
|
||||
"VpcConfig": {
|
||||
"SecurityGroupIds": [],
|
||||
"SubnetIds": [],
|
||||
}
|
||||
},
|
||||
'ResponseMetadata': {'HTTPStatusCode': 200},
|
||||
}
|
||||
conn.list_functions()['Functions'].should.equal([expected_function_result['Configuration']])
|
||||
|
||||
conn.get_function(FunctionName='testFunction').should.equal(expected_function_result)
|
||||
conn.delete_function(FunctionName='testFunction')
|
||||
|
||||
conn.list_functions()['Functions'].should.have.length_of(0)
|
||||
@ -13,6 +13,7 @@ import boto.redshift
|
||||
import boto.sns
|
||||
import boto.sqs
|
||||
import boto.vpc
|
||||
import boto3
|
||||
import sure # noqa
|
||||
|
||||
from moto import (
|
||||
@ -22,6 +23,7 @@ from moto import (
|
||||
mock_ec2,
|
||||
mock_elb,
|
||||
mock_iam,
|
||||
mock_lambda,
|
||||
mock_rds,
|
||||
mock_redshift,
|
||||
mock_route53,
|
||||
@ -1478,3 +1480,49 @@ def test_datapipeline():
|
||||
stack_resources = cf_conn.list_stack_resources(stack_id)
|
||||
stack_resources.should.have.length_of(1)
|
||||
stack_resources[0].physical_resource_id.should.equal(data_pipelines['pipelineIdList'][0]['id'])
|
||||
|
||||
|
||||
@mock_cloudformation
|
||||
@mock_lambda
|
||||
def test_lambda_function():
|
||||
conn = boto3.client('lambda', 'us-east-1')
|
||||
template = {
|
||||
"AWSTemplateFormatVersion": "2010-09-09",
|
||||
"Resources": {
|
||||
"lambdaTest": {
|
||||
"Type": "AWS::Lambda::Function",
|
||||
"Properties": {
|
||||
"Code": {
|
||||
"ZipFile": {"Fn::Join": [
|
||||
"\n",
|
||||
"""
|
||||
exports.handler = function(event, context) {
|
||||
context.succeed();
|
||||
}
|
||||
""".splitlines()
|
||||
]}
|
||||
},
|
||||
"Handler": "index.handler",
|
||||
"Description": "Test function",
|
||||
"MemorySize": 128,
|
||||
"Role": "test-role",
|
||||
"Runtime": "nodejs",
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
template_json = json.dumps(template)
|
||||
cf_conn = boto.cloudformation.connect_to_region("us-east-1")
|
||||
cf_conn.create_stack(
|
||||
"test_stack",
|
||||
template_body=template_json,
|
||||
)
|
||||
|
||||
result = conn.list_functions()
|
||||
result['Functions'].should.have.length_of(1)
|
||||
result['Functions'][0]['Description'].should.equal('Test function')
|
||||
result['Functions'][0]['Handler'].should.equal('index.handler')
|
||||
result['Functions'][0]['MemorySize'].should.equal(128)
|
||||
result['Functions'][0]['Role'].should.equal('test-role')
|
||||
result['Functions'][0]['Runtime'].should.equal('nodejs')
|
||||
|
||||
Loading…
Reference in New Issue
Block a user