Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add S3 LocationConstraint to Lambda tests

Browse Source
This commit is contained in:
Bert Blommers 2020-02-02 11:48:32 +00:00
parent f74f08581a
commit a86cba79de
3 changed files with 95 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
tests/test_awslambda/test_lambda.py
View File

@ -86,14 +86,14 @@ def lambda_handler(event, context):
@mock_lambda @mock_lambda
def test_list_functions(): def test_list_functions():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
result = conn.list_functions() result = conn.list_functions()
result["Functions"].should.have.length_of(0) result["Functions"].should.have.length_of(0)
@mock_lambda @mock_lambda
def test_invoke_requestresponse_function(): def test_invoke_requestresponse_function():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
Runtime="python2.7", Runtime="python2.7",
@ -126,7 +126,7 @@ def test_invoke_requestresponse_function():
@mock_lambda @mock_lambda
def test_invoke_event_function(): def test_invoke_event_function():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
Runtime="python2.7", Runtime="python2.7",
@ -156,11 +156,11 @@ if settings.TEST_SERVER_MODE:
@mock_ec2 @mock_ec2
@mock_lambda @mock_lambda
def test_invoke_function_get_ec2_volume(): def test_invoke_function_get_ec2_volume():
conn = boto3.resource("ec2", "us-west-2") conn = boto3.resource("ec2", _lambda_region)
vol = conn.create_volume(Size=99, AvailabilityZone="us-west-2") vol = conn.create_volume(Size=99, AvailabilityZone=_lambda_region)
vol = conn.Volume(vol.id) vol = conn.Volume(vol.id)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
Runtime="python3.7", Runtime="python3.7",
@ -190,14 +190,14 @@ if settings.TEST_SERVER_MODE:
@mock_ec2 @mock_ec2
@mock_lambda @mock_lambda
def test_invoke_function_from_sns(): def test_invoke_function_from_sns():
logs_conn = boto3.client("logs", region_name="us-west-2") logs_conn = boto3.client("logs", region_name=_lambda_region)
sns_conn = boto3.client("sns", region_name="us-west-2") sns_conn = boto3.client("sns", region_name=_lambda_region)
sns_conn.create_topic(Name="some-topic") sns_conn.create_topic(Name="some-topic")
topics_json = sns_conn.list_topics() topics_json = sns_conn.list_topics()
topics = topics_json["Topics"] topics = topics_json["Topics"]
topic_arn = topics[0]["TopicArn"] topic_arn = topics[0]["TopicArn"]
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
result = conn.create_function( result = conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
Runtime="python2.7", Runtime="python2.7",
@ -240,7 +240,7 @@ def test_invoke_function_from_sns():
@mock_lambda @mock_lambda
def test_create_based_on_s3_with_missing_bucket(): def test_create_based_on_s3_with_missing_bucket():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function.when.called_with( conn.create_function.when.called_with(
FunctionName="testFunction", FunctionName="testFunction",
@ -260,12 +260,15 @@ def test_create_based_on_s3_with_missing_bucket():
@mock_s3 @mock_s3
@freeze_time("2015-01-01 00:00:00") @freeze_time("2015-01-01 00:00:00")
def test_create_function_from_aws_bucket(): def test_create_function_from_aws_bucket():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
result = conn.create_function( result = conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -313,7 +316,7 @@ def test_create_function_from_aws_bucket():
@mock_lambda @mock_lambda
@freeze_time("2015-01-01 00:00:00") @freeze_time("2015-01-01 00:00:00")
def test_create_function_from_zipfile(): def test_create_function_from_zipfile():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
result = conn.create_function( result = conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -358,12 +361,15 @@ def test_create_function_from_zipfile():
@mock_s3 @mock_s3
@freeze_time("2015-01-01 00:00:00") @freeze_time("2015-01-01 00:00:00")
def test_get_function(): def test_get_function():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -427,7 +433,10 @@ def test_get_function():
def test_get_function_by_arn(): def test_get_function_by_arn():
bucket_name = "test-bucket" bucket_name = "test-bucket"
s3_conn = boto3.client("s3", "us-east-1") s3_conn = boto3.client("s3", "us-east-1")
s3_conn.create_bucket(Bucket=bucket_name) s3_conn.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket=bucket_name, Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket=bucket_name, Key="test.zip", Body=zip_content)
@ -452,12 +461,15 @@ def test_get_function_by_arn():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_delete_function(): def test_delete_function():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -488,7 +500,10 @@ def test_delete_function():
def test_delete_function_by_arn(): def test_delete_function_by_arn():
bucket_name = "test-bucket" bucket_name = "test-bucket"
s3_conn = boto3.client("s3", "us-east-1") s3_conn = boto3.client("s3", "us-east-1")
s3_conn.create_bucket(Bucket=bucket_name) s3_conn.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket=bucket_name, Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket=bucket_name, Key="test.zip", Body=zip_content)
@ -513,7 +528,7 @@ def test_delete_function_by_arn():
@mock_lambda @mock_lambda
def test_delete_unknown_function(): def test_delete_unknown_function():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.delete_function.when.called_with( conn.delete_function.when.called_with(
FunctionName="testFunctionThatDoesntExist" FunctionName="testFunctionThatDoesntExist"
).should.throw(botocore.client.ClientError) ).should.throw(botocore.client.ClientError)
@ -522,12 +537,15 @@ def test_delete_unknown_function():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_publish(): def test_publish():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -572,12 +590,15 @@ def test_list_create_list_get_delete_list():
test `list -> create -> list -> get -> delete -> list` integration test `list -> create -> list -> get -> delete -> list` integration
""" """
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.list_functions()["Functions"].should.have.length_of(0) conn.list_functions()["Functions"].should.have.length_of(0)
@ -674,12 +695,15 @@ def test_tags():
""" """
test list_tags -> tag_resource -> list_tags -> tag_resource -> list_tags -> untag_resource -> list_tags integration 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 = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
function = conn.create_function( function = conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -731,7 +755,7 @@ def test_tags_not_found():
""" """
Test list_tags and tag_resource when the lambda with the given arn does not exist Test list_tags and tag_resource when the lambda with the given arn does not exist
""" """
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.list_tags.when.called_with( conn.list_tags.when.called_with(
Resource="arn:aws:lambda:{}:function:not-found".format(ACCOUNT_ID) Resource="arn:aws:lambda:{}:function:not-found".format(ACCOUNT_ID)
).should.throw(botocore.client.ClientError) ).should.throw(botocore.client.ClientError)
@ -749,7 +773,7 @@ def test_tags_not_found():
@mock_lambda @mock_lambda
def test_invoke_async_function(): def test_invoke_async_function():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
Runtime="python2.7", Runtime="python2.7",
@ -772,7 +796,7 @@ def test_invoke_async_function():
@mock_lambda @mock_lambda
@freeze_time("2015-01-01 00:00:00") @freeze_time("2015-01-01 00:00:00")
def test_get_function_created_with_zipfile(): def test_get_function_created_with_zipfile():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
result = conn.create_function( result = conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -818,7 +842,7 @@ def test_get_function_created_with_zipfile():
@mock_lambda @mock_lambda
def test_add_function_permission(): def test_add_function_permission():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -849,7 +873,7 @@ def test_add_function_permission():
@mock_lambda @mock_lambda
def test_get_function_policy(): def test_get_function_policy():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -884,12 +908,15 @@ def test_get_function_policy():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_list_versions_by_function(): def test_list_versions_by_function():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -940,12 +967,15 @@ def test_list_versions_by_function():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_create_function_with_already_exists(): def test_create_function_with_already_exists():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
conn.create_function( conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -977,7 +1007,7 @@ def test_create_function_with_already_exists():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_list_versions_by_function_for_nonexistent_function(): def test_list_versions_by_function_for_nonexistent_function():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
versions = conn.list_versions_by_function(FunctionName="testFunction") versions = conn.list_versions_by_function(FunctionName="testFunction")
assert len(versions["Versions"]) == 0 assert len(versions["Versions"]) == 0
@ -1326,12 +1356,15 @@ def test_delete_event_source_mapping():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_update_configuration(): def test_update_configuration():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file2() zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
fxn = conn.create_function( fxn = conn.create_function(
FunctionName="testFunction", FunctionName="testFunction",
@ -1374,7 +1407,7 @@ def test_update_configuration():
@mock_lambda @mock_lambda
def test_update_function_zip(): def test_update_function_zip():
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
zip_content_one = get_test_zip_file1() zip_content_one = get_test_zip_file1()
@ -1429,13 +1462,16 @@ def test_update_function_zip():
@mock_lambda @mock_lambda
@mock_s3 @mock_s3
def test_update_function_s3(): def test_update_function_s3():
s3_conn = boto3.client("s3", "us-west-2") s3_conn = boto3.client("s3", _lambda_region)
s3_conn.create_bucket(Bucket="test-bucket") s3_conn.create_bucket(
Bucket="test-bucket",
CreateBucketConfiguration={"LocationConstraint": _lambda_region},
)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content) s3_conn.put_object(Bucket="test-bucket", Key="test.zip", Body=zip_content)
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
fxn = conn.create_function( fxn = conn.create_function(
FunctionName="testFunctionS3", FunctionName="testFunctionS3",
@ -1516,7 +1552,7 @@ def test_create_function_with_unknown_arn():
def create_invalid_lambda(role): def create_invalid_lambda(role):
conn = boto3.client("lambda", "us-west-2") conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1() zip_content = get_test_zip_file1()
with assert_raises(ClientError) as err: with assert_raises(ClientError) as err:
conn.create_function( conn.create_function(
@ -1535,7 +1571,7 @@ def create_invalid_lambda(role):
def get_role_name(): def get_role_name():
with mock_iam(): with mock_iam():
iam = boto3.client("iam", region_name="us-west-2") iam = boto3.client("iam", region_name=_lambda_region)
try: try:
return iam.get_role(RoleName="my-role")["Role"]["Arn"] return iam.get_role(RoleName="my-role")["Role"]["Arn"]
except ClientError: except ClientError:

5
tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
View File

@ -21,7 +21,10 @@ def test_get_resources_s3():
# Create 4 buckets # Create 4 buckets
for i in range(1, 5): for i in range(1, 5):
i_str = str(i) i_str = str(i)
s3_client.create_bucket(Bucket="test_bucket" + i_str) s3_client.create_bucket(
Bucket="test_bucket" + i_str,
CreateBucketConfiguration={"LocationConstraint": "eu-central-1"},
)
s3_client.put_bucket_tagging( s3_client.put_bucket_tagging(
Bucket="test_bucket" + i_str, Bucket="test_bucket" + i_str,
Tagging={"TagSet": [{"Key": "key" + i_str, "Value": "value" + i_str}]}, Tagging={"TagSet": [{"Key": "key" + i_str, "Value": "value" + i_str}]},

3
tests/test_s3/test_s3.py
View File

@ -1546,7 +1546,8 @@ def test_bucket_create_force_us_east_1():
s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
with assert_raises(ClientError) as exc: with assert_raises(ClientError) as exc:
s3.create_bucket( s3.create_bucket(
Bucket="blah", CreateBucketConfiguration={"LocationConstraint": DEFAULT_REGION_NAME} Bucket="blah",
CreateBucketConfiguration={"LocationConstraint": DEFAULT_REGION_NAME},
) )
exc.exception.response["Error"]["Code"].should.equal("InvalidLocationConstraint") exc.exception.response["Error"]["Code"].should.equal("InvalidLocationConstraint")