Skip to content

Instantly share code, notes, and snippets.

@leoybkim

leoybkim/moto_test.py

Created October 10, 2017 20:18
Show Gist options
  • Save leoybkim/dacea850d94e89807214a64021d6a89f to your computer and use it in GitHub Desktop.
Save leoybkim/dacea850d94e89807214a64021d6a89f to your computer and use it in GitHub Desktop.
Download ZIP
moto issue #1249
Raw
moto_test.py
from __future__ import unicode_literals
import base64
import botocore.client
import boto3
import hashlib
import io
import json
import zipfile
import sure # noqa
from freezegun import freeze_time
from moto import mock_lambda, mock_s3, mock_ec2, settings
_lambda_region = 'us-east-1' if settings.TEST_SERVER_MODE else 'us-west-2'
def _process_lambda(func_str):
zip_output = io.BytesIO()
zip_file = zipfile.ZipFile(zip_output, 'w', zipfile.ZIP_DEFLATED)
zip_file.writestr('lambda_function.py', func_str)
zip_file.close()
zip_output.seek(0)
return zip_output.read()
def get_test_zip_file1():
pfunc = """
def lambda_handler(event, context):
return event
"""
return _process_lambda(pfunc)
def get_test_zip_file2():
func_str = """
import boto3
def lambda_handler(event, context):
ec2 = boto3.resource('ec2', region_name='us-west-2', endpoint_url='http://{base_url}')
volume_id = event.get('volume_id')
vol = ec2.Volume(volume_id)
print('get volume details for %s\\nVolume - %s state=%s, size=%s' % (volume_id, volume_id, vol.state, vol.size))
return event
""".format(base_url="motoserver:5000" if settings.TEST_SERVER_MODE else "ec2.us-west-2.amazonaws.com")
return _process_lambda(func_str)
@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
def test_invoke_requestresponse_function():
conn = boto3.client('lambda', 'us-west-2')
conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file1(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
in_data = {'msg': 'So long and thanks for all the fish'}
success_result = conn.invoke(FunctionName='testFunction', InvocationType='RequestResponse',
Payload=json.dumps(in_data))
success_result["StatusCode"].should.equal(202)
result_obj = json.loads(
base64.b64decode(success_result["LogResult"]).decode('utf-8'))
result_obj.should.equal(in_data)
payload = success_result["Payload"].read().decode('utf-8')
json.loads(payload).should.equal(in_data)
@mock_lambda
def test_invoke_event_function():
conn = boto3.client('lambda', 'us-west-2')
conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file1(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
conn.invoke.when.called_with(
FunctionName='notAFunction',
InvocationType='Event',
Payload='{}'
).should.throw(botocore.client.ClientError)
in_data = {'msg': 'So long and thanks for all the fish'}
success_result = conn.invoke(
FunctionName='testFunction', InvocationType='Event', Payload=json.dumps(in_data))
success_result["StatusCode"].should.equal(202)
json.loads(success_result['Payload'].read().decode(
'utf-8')).should.equal({})
if settings.TEST_SERVER_MODE:
@mock_ec2
@mock_lambda
def test_invoke_function_get_ec2_volume():
conn = boto3.resource("ec2", "us-west-2")
vol = conn.create_volume(Size=99, AvailabilityZone='us-west-2')
vol = conn.Volume(vol.id)
conn = boto3.client('lambda', 'us-west-2')
conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file2(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
in_data = {'volume_id': vol.id}
result = conn.invoke(FunctionName='testFunction',
InvocationType='RequestResponse', Payload=json.dumps(in_data))
result["StatusCode"].should.equal(202)
msg = 'get volume details for %s\nVolume - %s state=%s, size=%s\n%s' % (
vol.id, vol.id, vol.state, vol.size, json.dumps(in_data))
log_result = base64.b64decode(result["LogResult"]).decode('utf-8')
# fix for running under travis (TODO: investigate why it has an extra newline)
log_result = log_result.replace('\n\n', '\n')
log_result.should.equal(msg)
payload = result['Payload'].read().decode('utf-8')
# fix for running under travis (TODO: investigate why it has an extra newline)
payload = payload.replace('\n\n', '\n')
payload.should.equal(msg)
@mock_lambda
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.lambda_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_file2()
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.lambda_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"],
},
)
# this is hard to match against, so remove it
result['ResponseMetadata'].pop('HTTPHeaders', None)
# Botocore inserts retry attempts not seen in Python27
result['ResponseMetadata'].pop('RetryAttempts', None)
result.pop('LastModified')
result.should.equal({
'FunctionName': 'testFunction',
'FunctionArn': 'arn:aws:lambda:{}:123456789012:function:testFunction'.format(_lambda_region),
'Runtime': 'python2.7',
'Role': 'test-iam-role',
'Handler': 'lambda_function.lambda_handler',
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
"CodeSize": len(zip_content),
'Description': 'test lambda function',
'Timeout': 3,
'MemorySize': 128,
'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_file1()
result = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': zip_content,
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
# this is hard to match against, so remove it
result['ResponseMetadata'].pop('HTTPHeaders', None)
# Botocore inserts retry attempts not seen in Python27
result['ResponseMetadata'].pop('RetryAttempts', None)
result.pop('LastModified')
result.should.equal({
'FunctionName': 'testFunction',
'FunctionArn': 'arn:aws:lambda:{}:123456789012:function:testFunction'.format(_lambda_region),
'Runtime': 'python2.7',
'Role': 'test-iam-role',
'Handler': 'lambda_function.lambda_handler',
'CodeSize': len(zip_content),
'Description': 'test lambda function',
'Timeout': 3,
'MemorySize': 128,
'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_file1()
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.lambda_handler',
Code={
'S3Bucket': 'test-bucket',
'S3Key': 'test.zip',
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
result = conn.get_function(FunctionName='testFunction')
# this is hard to match against, so remove it
result['ResponseMetadata'].pop('HTTPHeaders', None)
# Botocore inserts retry attempts not seen in Python27
result['ResponseMetadata'].pop('RetryAttempts', None)
result['Configuration'].pop('LastModified')
result.should.equal({
"Code": {
"Location": "s3://awslambda-{0}-tasks.s3-{0}.amazonaws.com/test.zip".format(_lambda_region),
"RepositoryType": "S3"
},
"Configuration": {
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
"CodeSize": len(zip_content),
"Description": "test lambda function",
"FunctionArn": 'arn:aws:lambda:{}:123456789012:function:testFunction'.format(_lambda_region),
"FunctionName": "testFunction",
"Handler": "lambda_function.lambda_handler",
"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_file2()
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.lambda_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')
# this is hard to match against, so remove it
success_result['ResponseMetadata'].pop('HTTPHeaders', None)
# Botocore inserts retry attempts not seen in Python27
success_result['ResponseMetadata'].pop('RetryAttempts', None)
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_file2()
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.lambda_handler',
Code={
'S3Bucket': 'test-bucket',
'S3Key': 'test.zip',
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
expected_function_result = {
"Code": {
"Location": "s3://awslambda-{0}-tasks.s3-{0}.amazonaws.com/test.zip".format(_lambda_region),
"RepositoryType": "S3"
},
"Configuration": {
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
"CodeSize": len(zip_content),
"Description": "test lambda function",
"FunctionArn": 'arn:aws:lambda:{}:123456789012:function:testFunction'.format(_lambda_region),
"FunctionName": "testFunction",
"Handler": "lambda_function.lambda_handler",
"MemorySize": 128,
"Role": "test-iam-role",
"Runtime": "python2.7",
"Timeout": 3,
"Version": '$LATEST',
"VpcConfig": {
"SecurityGroupIds": [],
"SubnetIds": [],
}
},
'ResponseMetadata': {'HTTPStatusCode': 200},
}
func = conn.list_functions()['Functions'][0]
func.pop('LastModified')
func.should.equal(expected_function_result['Configuration'])
func = conn.get_function(FunctionName='testFunction')
# this is hard to match against, so remove it
func['ResponseMetadata'].pop('HTTPHeaders', None)
# Botocore inserts retry attempts not seen in Python27
func['ResponseMetadata'].pop('RetryAttempts', None)
func['Configuration'].pop('LastModified')
func.should.equal(expected_function_result)
conn.delete_function(FunctionName='testFunction')
conn.list_functions()['Functions'].should.have.length_of(0)
@mock_lambda
def test_invoke_lambda_error():
lambda_fx = """
def lambda_handler(event, context):
raise Exception('failsauce')
"""
zip_output = io.BytesIO()
zip_file = zipfile.ZipFile(zip_output, 'w', zipfile.ZIP_DEFLATED)
zip_file.writestr('lambda_function.py', lambda_fx)
zip_file.close()
zip_output.seek(0)
client = boto3.client('lambda', region_name='us-east-1')
client.create_function(
FunctionName='test-lambda-fx',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
Code={
'ZipFile': zip_output.read()
},
)
result = client.invoke(
FunctionName='test-lambda-fx',
InvocationType='RequestResponse',
LogType='Tail'
)
assert 'FunctionError' in result
assert result['FunctionError'] == 'Handled'
@mock_lambda
@mock_s3
def test_tags():
"""
test list_tags -> tag_resource -> list_tags -> tag_resource -> list_tags -> untag_resource -> list_tags integration
"""
s3_conn = boto3.client('s3', 'us-west-2')
s3_conn.create_bucket(Bucket='test-bucket')
zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
conn = boto3.client('lambda', 'us-west-2')
function = 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,
)
# List tags when there are none
conn.list_tags(
Resource=function['FunctionArn']
)['Tags'].should.equal(dict())
# List tags when there is one
conn.tag_resource(
Resource=function['FunctionArn'],
Tags=dict(spam='eggs')
)['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
conn.list_tags(
Resource=function['FunctionArn']
)['Tags'].should.equal(dict(spam='eggs'))
# List tags when another has been added
conn.tag_resource(
Resource=function['FunctionArn'],
Tags=dict(foo='bar')
)['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
conn.list_tags(
Resource=function['FunctionArn']
)['Tags'].should.equal(dict(spam='eggs', foo='bar'))
# Untag resource
conn.untag_resource(
Resource=function['FunctionArn'],
TagKeys=['spam', 'trolls']
)['ResponseMetadata']['HTTPStatusCode'].should.equal(204)
conn.list_tags(
Resource=function['FunctionArn']
)['Tags'].should.equal(dict(foo='bar'))
# Untag a tag that does not exist (no error and no change)
conn.untag_resource(
Resource=function['FunctionArn'],
TagKeys=['spam']
)['ResponseMetadata']['HTTPStatusCode'].should.equal(204)
@mock_lambda
def test_tags_not_found():
"""
Test list_tags and tag_resource when the lambda with the given arn does not exist
"""
conn = boto3.client('lambda', 'us-west-2')
conn.list_tags.when.called_with(
Resource='arn:aws:lambda:123456789012:function:not-found'
).should.throw(botocore.client.ClientError)
conn.tag_resource.when.called_with(
Resource='arn:aws:lambda:123456789012:function:not-found',
Tags=dict(spam='eggs')
).should.throw(botocore.client.ClientError)
conn.untag_resource.when.called_with(
Resource='arn:aws:lambda:123456789012:function:not-found',
TagKeys=['spam']
).should.throw(botocore.client.ClientError)
@mock_lambda
def test_invoke_async_function():
conn = boto3.client('lambda', 'us-west-2')
conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.handler',
Code={
'ZipFile': get_test_zip_file1(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
success_result = conn.invoke_async(
FunctionName='testFunction',
InvokeArgs=json.dumps({ 'test': 'event' })
)
success_result['Status'].should.equal(202)
@mock_lambda
@freeze_time('2015-01-01 00:00:00')
def test_get_function_created_with_zipfile():
conn = boto3.client('lambda', 'us-west-2')
zip_content = get_test_zip_file1()
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,
)
response = conn.get_function(
FunctionName='testFunction'
)
response['Configuration'].pop('LastModified')
response['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
assert len(response['Code']) == 2
assert response['Code']['RepositoryType'] == 'S3'
assert response['Code']['Location'].startswith('s3://awslambda-{0}-tasks.s3-{0}.amazonaws.com'.format(_lambda_region))
response['Configuration'].should.equal(
{
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
"CodeSize": len(zip_content),
"Description": "test lambda function",
"FunctionArn":'arn:aws:lambda:{}:123456789012:function:testFunction'.format(_lambda_region),
"FunctionName": "testFunction",
"Handler": "lambda_function.handler",
"MemorySize": 128,
"Role": "test-iam-role",
"Runtime": "python2.7",
"Timeout": 3,
"Version": '$LATEST',
"VpcConfig": {
"SecurityGroupIds": [],
"SubnetIds": [],
}
},
)
@mock_lambda
def add_function_permission():
conn = boto3.client('lambda', 'us-west-2')
zip_content = get_test_zip_file1()
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,
)
response = conn.add_permission(
FunctionName='testFunction',
StatementId='1',
Action="lambda:InvokeFunction",
Principal='432143214321',
SourceArn="arn:aws:lambda:us-west-2:account-id:function:helloworld",
SourceAccount='123412341234',
EventSourceToken='blah',
Qualifier='2'
)
assert 'Statement' in response
res = json.loads(response['Statement'])
assert res['Action'] == "lambda:InvokeFunction"
@mock_lambda
def get_function_policy():
conn = boto3.client('lambda', 'us-west-2')
zip_content = get_test_zip_file1()
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,
)
response = conn.add_permission(
FunctionName='testFunction',
StatementId='1',
Action="lambda:InvokeFunction",
Principal='432143214321',
SourceArn="arn:aws:lambda:us-west-2:account-id:function:helloworld",
SourceAccount='123412341234',
EventSourceToken='blah',
Qualifier='2'
)
response = conn.get_policy(
FunctionName='testFunction'
)
assert 'Policy' in response
assert isinstance(response['Policy'], str)
res = json.loads(response['Policy'])
assert res['Statement'][0]['Action'] == 'lambda:InvokeFunction'
if __name__ == '__main__':
get_function_policy()
Raw
requirements.txt
asn1crypto==0.23.0
backports.ssl-match-hostname==3.5.0.1
backports.tempfile==1.0rc1
backports.weakref==1.0.post1
boto==2.48.0
boto3==1.4.7
botocore==1.7.25
certifi==2017.7.27.1
cffi==1.11.1
chardet==3.0.4
cookies==2.2.1
cryptography==2.0.3
docker==2.5.1
docker-pycreds==0.2.1
docutils==0.14
enum34==1.1.6
freezegun==0.3.9
funcsigs==1.0.2
futures==3.1.1
idna==2.6
ipaddress==1.0.18
Jinja2==2.9.6
jmespath==0.9.3
MarkupSafe==1.0
mock==2.0.0
moto==1.1.21
pbr==3.1.1
pyaml==17.10.0
pycparser==2.18
python-dateutil==2.6.1
pytz==2017.2
PyYAML==3.12
requests==2.18.4
s3transfer==0.1.11
six==1.11.0
sure==1.4.7
urllib3==1.22
websocket-client==0.44.0
Werkzeug==0.12.2
xmltodict==0.11.0
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment