catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge remote master

Browse source
This commit is contained in:
Ilya Shmygol 2019-12-11 16:17:21 +01:00
commit 3a42079ec7
45 changed files with 1642 additions and 192 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/test_apigateway/test_apigateway.py
View file

@ -563,7 +563,7 @@ def test_create_stage():
api_id = response["id"]
create_method_integration(client, api_id)
response = client.create_deployment(restApiId=api_id, stageName=stage_name,)
response = client.create_deployment(restApiId=api_id, stageName=stage_name)
deployment_id = response["id"]
response = client.get_deployment(restApiId=api_id, deploymentId=deployment_id)

143
tests/test_awslambda/test_lambda.py
View file

@ -15,6 +15,7 @@ from freezegun import freeze_time
from moto import (
mock_dynamodb2,
mock_lambda,
mock_iam,
mock_s3,
mock_ec2,
mock_sns,
@ -22,6 +23,7 @@ from moto import (
settings,
mock_sqs,
)
from moto.sts.models import ACCOUNT_ID
from nose.tools import assert_raises
from botocore.exceptions import ClientError
@ -96,7 +98,7 @@ def test_invoke_requestresponse_function():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file1()},
Description="test lambda function",
@ -129,7 +131,7 @@ def test_invoke_event_function():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file1()},
Description="test lambda function",
@ -163,7 +165,7 @@ if settings.TEST_SERVER_MODE:
conn.create_function(
FunctionName="testFunction",
Runtime="python3.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file2()},
Description="test lambda function",
@ -218,7 +220,7 @@ def test_invoke_function_from_sns():
result = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -262,7 +264,7 @@ def test_create_based_on_s3_with_missing_bucket():
conn.create_function.when.called_with(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "this-bucket-does-not-exist", "S3Key": "test.zip"},
Description="test lambda function",
@ -287,7 +289,7 @@ def test_create_function_from_aws_bucket():
result = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -308,7 +310,7 @@ def test_create_function_from_aws_bucket():
_lambda_region
),
"Runtime": "python2.7",
"Role": "test-iam-role",
"Role": result["Role"],
"Handler": "lambda_function.lambda_handler",
"CodeSha256": hashlib.sha256(zip_content).hexdigest(),
"CodeSize": len(zip_content),
@ -334,7 +336,7 @@ def test_create_function_from_zipfile():
result = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
@ -355,7 +357,7 @@ def test_create_function_from_zipfile():
_lambda_region
),
"Runtime": "python2.7",
"Role": "test-iam-role",
"Role": result["Role"],
"Handler": "lambda_function.lambda_handler",
"CodeSize": len(zip_content),
"Description": "test lambda function",
@ -383,7 +385,7 @@ def test_get_function():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -414,7 +416,7 @@ def test_get_function():
result["Configuration"]["FunctionName"].should.equal("testFunction")
result["Configuration"]["Handler"].should.equal("lambda_function.lambda_handler")
result["Configuration"]["MemorySize"].should.equal(128)
result["Configuration"]["Role"].should.equal("test-iam-role")
result["Configuration"]["Role"].should.equal(get_role_name())
result["Configuration"]["Runtime"].should.equal("python2.7")
result["Configuration"]["Timeout"].should.equal(3)
result["Configuration"]["Version"].should.equal("$LATEST")
@ -451,7 +453,7 @@ def test_get_function_by_arn():
fnc = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": bucket_name, "S3Key": "test.zip"},
Description="test lambda function",
@ -477,7 +479,7 @@ def test_delete_function():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -512,7 +514,7 @@ def test_delete_function_by_arn():
fnc = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": bucket_name, "S3Key": "test.zip"},
Description="test lambda function",
@ -547,7 +549,7 @@ def test_publish():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -599,7 +601,7 @@ def test_list_create_list_get_delete_list():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -624,7 +626,7 @@ def test_list_create_list_get_delete_list():
"FunctionName": "testFunction",
"Handler": "lambda_function.lambda_handler",
"MemorySize": 128,
"Role": "test-iam-role",
"Role": get_role_name(),
"Runtime": "python2.7",
"Timeout": 3,
"Version": "$LATEST",
@ -665,7 +667,7 @@ def lambda_handler(event, context):
client.create_function(
FunctionName="test-lambda-fx",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Description="test lambda function",
Timeout=3,
@ -698,7 +700,7 @@ def test_tags():
function = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -766,7 +768,7 @@ def test_invoke_async_function():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file1()},
Description="test lambda function",
@ -790,7 +792,7 @@ def test_get_function_created_with_zipfile():
result = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
@ -819,7 +821,7 @@ def test_get_function_created_with_zipfile():
"FunctionName": "testFunction",
"Handler": "lambda_function.handler",
"MemorySize": 128,
"Role": "test-iam-role",
"Role": get_role_name(),
"Runtime": "python2.7",
"Timeout": 3,
"Version": "$LATEST",
@ -835,7 +837,7 @@ def test_add_function_permission():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=(get_role_name()),
Handler="lambda_function.handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
@ -866,7 +868,7 @@ def test_get_function_policy():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
@ -906,7 +908,7 @@ def test_list_versions_by_function():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="arn:aws:iam::123456789012:role/test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -935,7 +937,7 @@ def test_list_versions_by_function():
conn.create_function(
FunctionName="testFunction_2",
Runtime="python2.7",
Role="arn:aws:iam::123456789012:role/test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -964,7 +966,7 @@ def test_create_function_with_already_exists():
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -976,7 +978,7 @@ def test_create_function_with_already_exists():
response = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -1008,7 +1010,7 @@ def test_create_event_source_mapping():
func = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1038,7 +1040,7 @@ def test_invoke_function_from_sqs():
func = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1098,7 +1100,7 @@ def test_invoke_function_from_dynamodb():
func = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function executed after a DynamoDB table is updated",
@ -1149,7 +1151,7 @@ def test_invoke_function_from_sqs_exception():
func = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file4()},
Description="test lambda function",
@ -1208,7 +1210,7 @@ def test_list_event_source_mappings():
func = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1240,7 +1242,7 @@ def test_get_event_source_mapping():
func = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1270,7 +1272,7 @@ def test_update_event_source_mapping():
func1 = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1281,7 +1283,7 @@ def test_update_event_source_mapping():
func2 = conn.create_function(
FunctionName="testFunction2",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1314,7 +1316,7 @@ def test_delete_event_source_mapping():
func1 = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file3()},
Description="test lambda function",
@ -1350,7 +1352,7 @@ def test_update_configuration():
fxn = conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -1395,7 +1397,7 @@ def test_update_function_zip():
fxn = conn.create_function(
FunctionName="testFunctionZip",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": zip_content_one},
Description="test lambda function",
@ -1430,7 +1432,7 @@ def test_update_function_zip():
"FunctionName": "testFunctionZip",
"Handler": "lambda_function.lambda_handler",
"MemorySize": 128,
"Role": "test-iam-role",
"Role": fxn["Role"],
"Runtime": "python2.7",
"Timeout": 3,
"Version": "2",
@ -1453,7 +1455,7 @@ def test_update_function_s3():
fxn = conn.create_function(
FunctionName="testFunctionS3",
Runtime="python2.7",
Role="test-iam-role",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"S3Bucket": "test-bucket", "S3Key": "test.zip"},
Description="test lambda function",
@ -1492,10 +1494,67 @@ def test_update_function_s3():
"FunctionName": "testFunctionS3",
"Handler": "lambda_function.lambda_handler",
"MemorySize": 128,
"Role": "test-iam-role",
"Role": fxn["Role"],
"Runtime": "python2.7",
"Timeout": 3,
"Version": "2",
"VpcConfig": {"SecurityGroupIds": [], "SubnetIds": []},
}
)
@mock_lambda
def test_create_function_with_invalid_arn():
err = create_invalid_lambda("test-iam-role")
err.exception.response["Error"]["Message"].should.equal(
"1 validation error detected: Value 'test-iam-role' at 'role' failed to satisfy constraint: Member must satisfy regular expression pattern: arn:(aws[a-zA-Z-]*)?:iam::(\d{12}):role/?[a-zA-Z_0-9+=,.@\-_/]+"
)
@mock_lambda
def test_create_function_with_arn_from_different_account():
err = create_invalid_lambda("arn:aws:iam::000000000000:role/example_role")
err.exception.response["Error"]["Message"].should.equal(
"Cross-account pass role is not allowed."
)
@mock_lambda
def test_create_function_with_unknown_arn():
err = create_invalid_lambda(
"arn:aws:iam::" + str(ACCOUNT_ID) + ":role/service-role/unknown_role"
)
err.exception.response["Error"]["Message"].should.equal(
"The role defined for the function cannot be assumed by Lambda."
)
def create_invalid_lambda(role):
conn = boto3.client("lambda", "us-west-2")
zip_content = get_test_zip_file1()
with assert_raises(ClientError) as err:
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role=role,
Handler="lambda_function.handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)
return err
def get_role_name():
with mock_iam():
iam = boto3.client("iam", region_name="us-west-2")
try:
return iam.get_role(RoleName="my-role")["Role"]["Arn"]
except ClientError:
return iam.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
)["Role"]["Arn"]

138
tests/test_awslambda/test_lambda_cloudformation.py Normal file
View file

@ -0,0 +1,138 @@
import boto3
import io
import sure # noqa
import zipfile
from botocore.exceptions import ClientError
from moto import mock_cloudformation, mock_iam, mock_lambda, mock_s3
from nose.tools import assert_raises
from string import Template
from uuid import uuid4
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_zip_file():
pfunc = """
def lambda_handler1(event, context):
return event
def lambda_handler2(event, context):
return event
"""
return _process_lambda(pfunc)
template = Template(
"""{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"LF3ABOV": {
"Type": "AWS::Lambda::Function",
"Properties": {
"Handler": "$handler",
"Role": "$role_arn",
"Runtime": "$runtime",
"Code": {
"S3Bucket": "$bucket_name",
"S3Key": "$key"
},
}
}
}
}"""
)
@mock_cloudformation
@mock_lambda
@mock_s3
def test_lambda_can_be_updated_by_cloudformation():
s3 = boto3.client("s3", "us-east-1")
cf = boto3.client("cloudformation", region_name="us-east-1")
lmbda = boto3.client("lambda", region_name="us-east-1")
body2, stack = create_stack(cf, s3)
created_fn_name = get_created_function_name(cf, stack)
# Verify function has been created
created_fn = lmbda.get_function(FunctionName=created_fn_name)
created_fn["Configuration"]["Handler"].should.equal(
"lambda_function.lambda_handler1"
)
created_fn["Configuration"]["Runtime"].should.equal("python3.7")
created_fn["Code"]["Location"].should.match("/test1.zip")
# Update CF stack
cf.update_stack(StackName="teststack", TemplateBody=body2)
updated_fn_name = get_created_function_name(cf, stack)
# Verify function has been updated
updated_fn = lmbda.get_function(FunctionName=updated_fn_name)
updated_fn["Configuration"]["FunctionArn"].should.equal(
created_fn["Configuration"]["FunctionArn"]
)
updated_fn["Configuration"]["Handler"].should.equal(
"lambda_function.lambda_handler2"
)
updated_fn["Configuration"]["Runtime"].should.equal("python3.8")
updated_fn["Code"]["Location"].should.match("/test2.zip")
@mock_cloudformation
@mock_lambda
@mock_s3
def test_lambda_can_be_deleted_by_cloudformation():
s3 = boto3.client("s3", "us-east-1")
cf = boto3.client("cloudformation", region_name="us-east-1")
lmbda = boto3.client("lambda", region_name="us-east-1")
_, stack = create_stack(cf, s3)
created_fn_name = get_created_function_name(cf, stack)
# Delete Stack
cf.delete_stack(StackName=stack["StackId"])
# Verify function was deleted
with assert_raises(ClientError) as e:
lmbda.get_function(FunctionName=created_fn_name)
e.exception.response["Error"]["Code"].should.equal("404")
def create_stack(cf, s3):
bucket_name = str(uuid4())
s3.create_bucket(Bucket=bucket_name)
s3.put_object(Bucket=bucket_name, Key="test1.zip", Body=get_zip_file())
s3.put_object(Bucket=bucket_name, Key="test2.zip", Body=get_zip_file())
body1 = get_template(bucket_name, "1", "python3.7")
body2 = get_template(bucket_name, "2", "python3.8")
stack = cf.create_stack(StackName="teststack", TemplateBody=body1)
return body2, stack
def get_created_function_name(cf, stack):
res = cf.list_stack_resources(StackName=stack["StackId"])
return res["StackResourceSummaries"][0]["PhysicalResourceId"]
def get_template(bucket_name, version, runtime):
key = "test" + version + ".zip"
handler = "lambda_function.lambda_handler" + version
return template.substitute(
bucket_name=bucket_name,
key=key,
handler=handler,
role_arn=get_role_arn(),
runtime=runtime,
)
def get_role_arn():
with mock_iam():
iam = boto3.client("iam", region_name="us-west-2")
try:
return iam.get_role(RoleName="my-role")["Role"]["Arn"]
except ClientError:
return iam.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
)["Role"]["Arn"]

0
tests/test_awslambda/test_lambda_eventsourcemappings.py
View file

13
tests/test_cloudformation/test_cloudformation_stack_crud.py
View file

@ -4,6 +4,7 @@ import os
import json
import boto
import boto.iam
import boto.s3
import boto.s3.key
import boto.cloudformation
@ -18,6 +19,7 @@ from moto import (
mock_cloudformation_deprecated,
mock_s3_deprecated,
mock_route53_deprecated,
mock_iam_deprecated,
)
from moto.cloudformation import cloudformation_backends
@ -516,7 +518,7 @@ def test_create_stack_lambda_and_dynamodb():
"Code": {"S3Bucket": "bucket_123", "S3Key": "key_123"},
"FunctionName": "func1",
"Handler": "handler.handler",
"Role": "role1",
"Role": get_role_name(),
"Runtime": "python2.7",
"Description": "descr",
"MemorySize": 12345,
@ -591,3 +593,12 @@ def test_create_stack_kinesis():
stack = conn.describe_stacks()[0]
resources = stack.list_resources()
assert len(resources) == 1
def get_role_name():
with mock_iam_deprecated():
iam = boto.connect_iam()
role = iam.create_role("my-role")["create_role_response"]["create_role_result"][
"role"
]["arn"]
return role

19
tests/test_cloudformation/test_cloudformation_stack_integration.py
View file

@ -1773,11 +1773,25 @@ def lambda_handler(event, context):
"Handler": "lambda_function.handler",
"Description": "Test function",
"MemorySize": 128,
"Role": "test-role",
"Role": {"Fn::GetAtt": ["MyRole", "Arn"]},
"Runtime": "python2.7",
"Environment": {"Variables": {"TEST_ENV_KEY": "test-env-val"}},
},
}
},
"MyRole": {
"Type": "AWS::IAM::Role",
"Properties": {
"AssumeRolePolicyDocument": {
"Statement": [
{
"Action": ["sts:AssumeRole"],
"Effect": "Allow",
"Principal": {"Service": ["ec2.amazonaws.com"]},
}
]
}
},
},
},
}
@ -1791,7 +1805,6 @@ def lambda_handler(event, context):
result["Functions"][0]["Description"].should.equal("Test function")
result["Functions"][0]["Handler"].should.equal("lambda_function.handler")
result["Functions"][0]["MemorySize"].should.equal(128)
result["Functions"][0]["Role"].should.equal("test-role")
result["Functions"][0]["Runtime"].should.equal("python2.7")
result["Functions"][0]["Environment"].should.equal(
{"Variables": {"TEST_ENV_KEY": "test-env-val"}}

3
tests/test_cloudwatch/test_cloudwatch.py
View file

@ -1,8 +1,5 @@
import boto
from boto.ec2.cloudwatch.alarm import MetricAlarm
import boto3
from datetime import datetime, timedelta
import pytz
import sure # noqa
from moto import mock_cloudwatch_deprecated

90
tests/test_cloudwatch/test_cloudwatch_boto3.py Executable file → Normal file
View file

@ -1,8 +1,10 @@
from __future__ import unicode_literals
# from __future__ import unicode_literals
import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
from nose.tools import assert_raises
from uuid import uuid4
import pytz
import sure # noqa
@ -155,13 +157,14 @@ def test_put_metric_data_no_dimensions():
@mock_cloudwatch
def test_put_metric_data_with_statistics():
conn = boto3.client("cloudwatch", region_name="us-east-1")
utc_now = datetime.now(tz=pytz.utc)
conn.put_metric_data(
Namespace="tester",
MetricData=[
dict(
MetricName="statmetric",
Timestamp=datetime(2015, 1, 1),
Timestamp=utc_now,
# no Value to test https://github.com/spulec/moto/issues/1615
StatisticValues=dict(
SampleCount=123.0, Sum=123.0, Minimum=123.0, Maximum=123.0
@ -203,3 +206,86 @@ def test_get_metric_statistics():
datapoint = stats["Datapoints"][0]
datapoint["SampleCount"].should.equal(1.0)
datapoint["Sum"].should.equal(1.5)
@mock_cloudwatch
def test_list_metrics():
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
# Verify namespace has to exist
res = cloudwatch.list_metrics(Namespace="unknown/")["Metrics"]
res.should.be.empty
# Create some metrics to filter on
create_metrics(cloudwatch, namespace="list_test_1/", metrics=4, data_points=2)
create_metrics(cloudwatch, namespace="list_test_2/", metrics=4, data_points=2)
# Verify we can retrieve everything
res = cloudwatch.list_metrics()["Metrics"]
len(res).should.equal(16) # 2 namespaces * 4 metrics * 2 data points
# Verify we can filter by namespace/metric name
res = cloudwatch.list_metrics(Namespace="list_test_1/")["Metrics"]
len(res).should.equal(8) # 1 namespace * 4 metrics * 2 data points
res = cloudwatch.list_metrics(Namespace="list_test_1/", MetricName="metric1")[
"Metrics"
]
len(res).should.equal(2) # 1 namespace * 1 metrics * 2 data points
# Verify format
res.should.equal(
[
{u"Namespace": "list_test_1/", u"Dimensions": [], u"MetricName": "metric1"},
{u"Namespace": "list_test_1/", u"Dimensions": [], u"MetricName": "metric1"},
]
)
# Verify unknown namespace still has no results
res = cloudwatch.list_metrics(Namespace="unknown/")["Metrics"]
res.should.be.empty
@mock_cloudwatch
def test_list_metrics_paginated():
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
# Verify that only a single page of metrics is returned
cloudwatch.list_metrics()["Metrics"].should.be.empty
# Verify we can't pass a random NextToken
with assert_raises(ClientError) as e:
cloudwatch.list_metrics(NextToken=str(uuid4()))
e.exception.response["Error"]["Message"].should.equal(
"Request parameter NextToken is invalid"
)
# Add a boatload of metrics
create_metrics(cloudwatch, namespace="test", metrics=100, data_points=1)
# Verify that a single page is returned until we've reached 500
first_page = cloudwatch.list_metrics()
first_page["Metrics"].shouldnt.be.empty
len(first_page["Metrics"]).should.equal(100)
create_metrics(cloudwatch, namespace="test", metrics=200, data_points=2)
first_page = cloudwatch.list_metrics()
len(first_page["Metrics"]).should.equal(500)
first_page.shouldnt.contain("NextToken")
# Verify that adding more data points results in pagination
create_metrics(cloudwatch, namespace="test", metrics=60, data_points=10)
first_page = cloudwatch.list_metrics()
len(first_page["Metrics"]).should.equal(500)
first_page["NextToken"].shouldnt.be.empty
# Retrieve second page - and verify there's more where that came from
second_page = cloudwatch.list_metrics(NextToken=first_page["NextToken"])
len(second_page["Metrics"]).should.equal(500)
second_page.should.contain("NextToken")
# Last page should only have the last 100 results, and no NextToken (indicating that pagination is finished)
third_page = cloudwatch.list_metrics(NextToken=second_page["NextToken"])
len(third_page["Metrics"]).should.equal(100)
third_page.shouldnt.contain("NextToken")
# Verify that we can't reuse an existing token
with assert_raises(ClientError) as e:
cloudwatch.list_metrics(NextToken=first_page["NextToken"])
e.exception.response["Error"]["Message"].should.equal(
"Request parameter NextToken is invalid"
)
def create_metrics(cloudwatch, namespace, metrics=5, data_points=5):
for i in range(0, metrics):
metric_name = "metric" + str(i)
for j in range(0, data_points):
cloudwatch.put_metric_data(
Namespace=namespace,
MetricData=[{"MetricName": metric_name, "Value": j, "Unit": "Seconds"}],
)

131
tests/test_dynamodb2/test_dynamodb.py
View file

@ -3274,6 +3274,7 @@ def test_update_supports_complex_expression_attribute_values():
@mock_dynamodb2
def test_update_supports_list_append():
# Verify whether the list_append operation works as expected
client = boto3.client("dynamodb", region_name="us-east-1")
client.create_table(
@ -3307,6 +3308,132 @@ def test_update_supports_list_append():
)
@mock_dynamodb2
def test_update_supports_nested_list_append():
# Verify whether we can append a list that's inside a map
client = boto3.client("dynamodb", region_name="us-east-1")
client.create_table(
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
TableName="TestTable",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
client.put_item(
TableName="TestTable",
Item={
"id": {"S": "nested_list_append"},
"a": {"M": {"b": {"L": [{"S": "bar1"}]}}},
},
)
# Update item using list_append expression
client.update_item(
TableName="TestTable",
Key={"id": {"S": "nested_list_append"}},
UpdateExpression="SET a.#b = list_append(a.#b, :i)",
ExpressionAttributeValues={":i": {"L": [{"S": "bar2"}]}},
ExpressionAttributeNames={"#b": "b"},
)
# Verify item is appended to the existing list
result = client.get_item(
TableName="TestTable", Key={"id": {"S": "nested_list_append"}}
)["Item"]
result.should.equal(
{
"id": {"S": "nested_list_append"},
"a": {"M": {"b": {"L": [{"S": "bar1"}, {"S": "bar2"}]}}},
}
)
@mock_dynamodb2
def test_update_supports_multiple_levels_nested_list_append():
# Verify whether we can append a list that's inside a map that's inside a map (Inception!)
client = boto3.client("dynamodb", region_name="us-east-1")
client.create_table(
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
TableName="TestTable",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
client.put_item(
TableName="TestTable",
Item={
"id": {"S": "nested_list_append"},
"a": {"M": {"b": {"M": {"c": {"L": [{"S": "bar1"}]}}}}},
},
)
# Update item using list_append expression
client.update_item(
TableName="TestTable",
Key={"id": {"S": "nested_list_append"}},
UpdateExpression="SET a.#b.c = list_append(a.#b.#c, :i)",
ExpressionAttributeValues={":i": {"L": [{"S": "bar2"}]}},
ExpressionAttributeNames={"#b": "b", "#c": "c"},
)
# Verify item is appended to the existing list
result = client.get_item(
TableName="TestTable", Key={"id": {"S": "nested_list_append"}}
)["Item"]
result.should.equal(
{
"id": {"S": "nested_list_append"},
"a": {"M": {"b": {"M": {"c": {"L": [{"S": "bar1"}, {"S": "bar2"}]}}}}},
}
)
@mock_dynamodb2
def test_update_supports_nested_list_append_onto_another_list():
# Verify whether we can take the contents of one list, and use that to fill another list
# Note that the contents of the other list is completely overwritten
client = boto3.client("dynamodb", region_name="us-east-1")
client.create_table(
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
TableName="TestTable",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
client.put_item(
TableName="TestTable",
Item={
"id": {"S": "list_append_another"},
"a": {"M": {"b": {"L": [{"S": "bar1"}]}, "c": {"L": [{"S": "car1"}]}}},
},
)
# Update item using list_append expression
client.update_item(
TableName="TestTable",
Key={"id": {"S": "list_append_another"}},
UpdateExpression="SET a.#c = list_append(a.#b, :i)",
ExpressionAttributeValues={":i": {"L": [{"S": "bar2"}]}},
ExpressionAttributeNames={"#b": "b", "#c": "c"},
)
# Verify item is appended to the existing list
result = client.get_item(
TableName="TestTable", Key={"id": {"S": "list_append_another"}}
)["Item"]
result.should.equal(
{
"id": {"S": "list_append_another"},
"a": {
"M": {
"b": {"L": [{"S": "bar1"}]},
"c": {"L": [{"S": "bar1"}, {"S": "bar2"}]},
}
},
}
)
@mock_dynamodb2
def test_update_catches_invalid_list_append_operation():
client = boto3.client("dynamodb", region_name="us-east-1")
@ -3372,7 +3499,7 @@ def test_update_item_if_original_value_is_none():
table.update_item(
Key={"job_id": "a"},
UpdateExpression="SET job_name = :output",
ExpressionAttributeValues={":output": "updated",},
ExpressionAttributeValues={":output": "updated"},
)
table.scan()["Items"][0]["job_name"].should.equal("updated")
@ -3391,7 +3518,7 @@ def test_update_nested_item_if_original_value_is_none():
table.update_item(
Key={"job_id": "a"},
UpdateExpression="SET job_details.job_name = :output",
ExpressionAttributeValues={":output": "updated",},
ExpressionAttributeValues={":output": "updated"},
)
table.scan()["Items"][0]["job_details"]["job_name"].should.equal("updated")

73
tests/test_dynamodb2/test_dynamodb_table_with_range_key.py
View file

@ -1289,6 +1289,16 @@ def test_update_item_add_with_expression():
current_item["str_set"] = current_item["str_set"].union({"item4"})
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
# Update item to add a string value to a non-existing set
# Should just create the set in the background
table.update_item(
Key=item_key,
UpdateExpression="ADD non_existing_str_set :v",
ExpressionAttributeValues={":v": {"item4"}},
)
current_item["non_existing_str_set"] = {"item4"}
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
# Update item to add a num value to a num set
table.update_item(
Key=item_key,
@ -1336,6 +1346,69 @@ def test_update_item_add_with_expression():
).should.have.raised(ClientError)
@mock_dynamodb2
def test_update_item_add_with_nested_sets():
table = _create_table_with_range_key()
item_key = {"forum_name": "the-key", "subject": "123"}
current_item = {
"forum_name": "the-key",
"subject": "123",
"nested": {"str_set": {"item1", "item2", "item3"}},
}
# Put an entry in the DB to play with
table.put_item(Item=current_item)
# Update item to add a string value to a nested string set
table.update_item(
Key=item_key,
UpdateExpression="ADD nested.str_set :v",
ExpressionAttributeValues={":v": {"item4"}},
)
current_item["nested"]["str_set"] = current_item["nested"]["str_set"].union(
{"item4"}
)
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
# Update item to add a string value to a non-existing set
# Should just create the set in the background
table.update_item(
Key=item_key,
UpdateExpression="ADD #ns.#ne :v",
ExpressionAttributeNames={"#ns": "nested", "#ne": "non_existing_str_set"},
ExpressionAttributeValues={":v": {"new_item"}},
)
current_item["nested"]["non_existing_str_set"] = {"new_item"}
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
@mock_dynamodb2
def test_update_item_delete_with_nested_sets():
table = _create_table_with_range_key()
item_key = {"forum_name": "the-key", "subject": "123"}
current_item = {
"forum_name": "the-key",
"subject": "123",
"nested": {"str_set": {"item1", "item2", "item3"}},
}
# Put an entry in the DB to play with
table.put_item(Item=current_item)
# Update item to add a string value to a nested string set
table.update_item(
Key=item_key,
UpdateExpression="DELETE nested.str_set :v",
ExpressionAttributeValues={":v": {"item3"}},
)
current_item["nested"]["str_set"] = current_item["nested"]["str_set"].difference(
{"item3"}
)
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
@mock_dynamodb2
def test_update_item_delete_with_expression():
table = _create_table_with_range_key()

9
tests/test_dynamodbstreams/test_dynamodbstreams.py
View file

@ -213,7 +213,7 @@ class TestEdges:
resp = conn.update_table(
TableName="test-streams",
StreamSpecification={"StreamViewType": "KEYS_ONLY"},
StreamSpecification={"StreamViewType": "KEYS_ONLY", "StreamEnabled": True},
)
assert "StreamSpecification" in resp["TableDescription"]
assert resp["TableDescription"]["StreamSpecification"] == {
@ -226,7 +226,10 @@ class TestEdges:
with assert_raises(conn.exceptions.ResourceInUseException):
resp = conn.update_table(
TableName="test-streams",
StreamSpecification={"StreamViewType": "OLD_IMAGES"},
StreamSpecification={
"StreamViewType": "OLD_IMAGES",
"StreamEnabled": True,
},
)
def test_stream_with_range_key(self):
@ -243,7 +246,7 @@ class TestEdges:
{"AttributeName": "color", "AttributeType": "S"},
],
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
StreamSpecification={"StreamViewType": "NEW_IMAGES"},
StreamSpecification={"StreamViewType": "NEW_IMAGES", "StreamEnabled": True},
)
stream_arn = resp["TableDescription"]["LatestStreamArn"]

30
tests/test_ec2/test_security_groups.py
View file

@ -833,3 +833,33 @@ def test_get_all_security_groups_filter_with_same_vpc_id():
cm.exception.code.should.equal("InvalidGroup.NotFound")
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_revoke_security_group_egress():
ec2 = boto3.resource("ec2", "us-east-1")
sg = ec2.create_security_group(Description="Test SG", GroupName="test-sg")
sg.ip_permissions_egress.should.equal(
[
{
"IpProtocol": "-1",
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
"UserIdGroupPairs": [],
}
]
)
sg.revoke_egress(
IpPermissions=[
{
"FromPort": 0,
"IpProtocol": "-1",
"IpRanges": [{"CidrIp": "0.0.0.0/0"},],
"ToPort": 123,
},
]
)
sg.reload()
sg.ip_permissions_egress.should.have.length_of(0)

65
tests/test_ecs/test_ecs_boto3.py
View file

@ -171,6 +171,69 @@ def test_list_task_definitions():
)
@mock_ecs
def test_list_task_definitions_with_family_prefix():
client = boto3.client("ecs", region_name="us-east-1")
_ = client.register_task_definition(
family="test_ecs_task_a",
containerDefinitions=[
{
"name": "hello_world",
"image": "docker/hello-world:latest",
"cpu": 1024,
"memory": 400,
"essential": True,
"environment": [
{"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY"}
],
"logConfiguration": {"logDriver": "json-file"},
}
],
)
_ = client.register_task_definition(
family="test_ecs_task_a",
containerDefinitions=[
{
"name": "hello_world",
"image": "docker/hello-world:latest",
"cpu": 1024,
"memory": 400,
"essential": True,
"environment": [
{"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY"}
],
"logConfiguration": {"logDriver": "json-file"},
}
],
)
_ = client.register_task_definition(
family="test_ecs_task_b",
containerDefinitions=[
{
"name": "hello_world2",
"image": "docker/hello-world2:latest",
"cpu": 1024,
"memory": 400,
"essential": True,
"environment": [
{"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY2"}
],
"logConfiguration": {"logDriver": "json-file"},
}
],
)
empty_response = client.list_task_definitions(familyPrefix="test_ecs_task")
len(empty_response["taskDefinitionArns"]).should.equal(0)
filtered_response = client.list_task_definitions(familyPrefix="test_ecs_task_a")
len(filtered_response["taskDefinitionArns"]).should.equal(2)
filtered_response["taskDefinitionArns"][0].should.equal(
"arn:aws:ecs:us-east-1:012345678910:task-definition/test_ecs_task_a:1"
)
filtered_response["taskDefinitionArns"][1].should.equal(
"arn:aws:ecs:us-east-1:012345678910:task-definition/test_ecs_task_a:2"
)
@mock_ecs
def test_describe_task_definition():
client = boto3.client("ecs", region_name="us-east-1")
@ -1756,7 +1819,7 @@ def test_update_task_definition_family_through_cloudformation_should_trigger_a_r
cfn_conn.update_stack(StackName="test_stack", TemplateBody=template2_json)
ecs_conn = boto3.client("ecs", region_name="us-west-1")
resp = ecs_conn.list_task_definitions(familyPrefix="testTaskDefinition")
resp = ecs_conn.list_task_definitions(familyPrefix="testTaskDefinition2")
len(resp["taskDefinitionArns"]).should.equal(1)
resp["taskDefinitionArns"][0].endswith("testTaskDefinition2:1").should.be.true

4
tests/test_iam/test_iam.py
View file

@ -2526,9 +2526,7 @@ def test_get_account_summary():
)
client.create_instance_profile(InstanceProfileName="test-profile")
client.create_open_id_connect_provider(
Url="https://example.com", ThumbprintList=[],
)
client.create_open_id_connect_provider(Url="https://example.com", ThumbprintList=[])
response_policy = client.create_policy(
PolicyName="test-policy", PolicyDocument=MOCK_POLICY
)

272
tests/test_iot/test_iot.py
View file

@ -581,6 +581,278 @@ def test_delete_principal_thing():
client.delete_certificate(certificateId=cert_id)
@mock_iot
def test_describe_thing_group_metadata_hierarchy():
client = boto3.client("iot", region_name="ap-northeast-1")
group_name_1a = "my-group-name-1a"
group_name_1b = "my-group-name-1b"
group_name_2a = "my-group-name-2a"
group_name_2b = "my-group-name-2b"
group_name_3a = "my-group-name-3a"
group_name_3b = "my-group-name-3b"
group_name_3c = "my-group-name-3c"
group_name_3d = "my-group-name-3d"
# --1a
# |--2a
# | |--3a
# | |--3b
# |
# |--2b
# |--3c
# |--3d
# --1b
# create thing groups tree
# 1
thing_group1a = client.create_thing_group(thingGroupName=group_name_1a)
thing_group1a.should.have.key("thingGroupName").which.should.equal(group_name_1a)
thing_group1a.should.have.key("thingGroupArn")
thing_group1b = client.create_thing_group(thingGroupName=group_name_1b)
thing_group1b.should.have.key("thingGroupName").which.should.equal(group_name_1b)
thing_group1b.should.have.key("thingGroupArn")
# 2
thing_group2a = client.create_thing_group(
thingGroupName=group_name_2a, parentGroupName=group_name_1a
)
thing_group2a.should.have.key("thingGroupName").which.should.equal(group_name_2a)
thing_group2a.should.have.key("thingGroupArn")
thing_group2b = client.create_thing_group(
thingGroupName=group_name_2b, parentGroupName=group_name_1a
)
thing_group2b.should.have.key("thingGroupName").which.should.equal(group_name_2b)
thing_group2b.should.have.key("thingGroupArn")
# 3
thing_group3a = client.create_thing_group(
thingGroupName=group_name_3a, parentGroupName=group_name_2a
)
thing_group3a.should.have.key("thingGroupName").which.should.equal(group_name_3a)
thing_group3a.should.have.key("thingGroupArn")
thing_group3b = client.create_thing_group(
thingGroupName=group_name_3b, parentGroupName=group_name_2a
)
thing_group3b.should.have.key("thingGroupName").which.should.equal(group_name_3b)
thing_group3b.should.have.key("thingGroupArn")
thing_group3c = client.create_thing_group(
thingGroupName=group_name_3c, parentGroupName=group_name_2b
)
thing_group3c.should.have.key("thingGroupName").which.should.equal(group_name_3c)
thing_group3c.should.have.key("thingGroupArn")
thing_group3d = client.create_thing_group(
thingGroupName=group_name_3d, parentGroupName=group_name_2b
)
thing_group3d.should.have.key("thingGroupName").which.should.equal(group_name_3d)
thing_group3d.should.have.key("thingGroupArn")
# describe groups
# groups level 1
# 1a
thing_group_description1a = client.describe_thing_group(
thingGroupName=group_name_1a
)
thing_group_description1a.should.have.key("thingGroupName").which.should.equal(
group_name_1a
)
thing_group_description1a.should.have.key("thingGroupProperties")
thing_group_description1a.should.have.key("thingGroupMetadata")
thing_group_description1a["thingGroupMetadata"].should.have.key("creationDate")
thing_group_description1a.should.have.key("version")
# 1b
thing_group_description1b = client.describe_thing_group(
thingGroupName=group_name_1b
)
thing_group_description1b.should.have.key("thingGroupName").which.should.equal(
group_name_1b
)
thing_group_description1b.should.have.key("thingGroupProperties")
thing_group_description1b.should.have.key("thingGroupMetadata")
thing_group_description1b["thingGroupMetadata"].should.have.length_of(1)
thing_group_description1b["thingGroupMetadata"].should.have.key("creationDate")
thing_group_description1b.should.have.key("version")
# groups level 2
# 2a
thing_group_description2a = client.describe_thing_group(
thingGroupName=group_name_2a
)
thing_group_description2a.should.have.key("thingGroupName").which.should.equal(
group_name_2a
)
thing_group_description2a.should.have.key("thingGroupProperties")
thing_group_description2a.should.have.key("thingGroupMetadata")
thing_group_description2a["thingGroupMetadata"].should.have.length_of(3)
thing_group_description2a["thingGroupMetadata"].should.have.key(
"parentGroupName"
).being.equal(group_name_1a)
thing_group_description2a["thingGroupMetadata"].should.have.key(
"rootToParentThingGroups"
)
thing_group_description2a["thingGroupMetadata"][
"rootToParentThingGroups"
].should.have.length_of(1)
thing_group_description2a["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupName"
].should.match(group_name_1a)
thing_group_description2a["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
thing_group_description2a.should.have.key("version")
# 2b
thing_group_description2b = client.describe_thing_group(
thingGroupName=group_name_2b
)
thing_group_description2b.should.have.key("thingGroupName").which.should.equal(
group_name_2b
)
thing_group_description2b.should.have.key("thingGroupProperties")
thing_group_description2b.should.have.key("thingGroupMetadata")
thing_group_description2b["thingGroupMetadata"].should.have.length_of(3)
thing_group_description2b["thingGroupMetadata"].should.have.key(
"parentGroupName"
).being.equal(group_name_1a)
thing_group_description2b["thingGroupMetadata"].should.have.key(
"rootToParentThingGroups"
)
thing_group_description2b["thingGroupMetadata"][
"rootToParentThingGroups"
].should.have.length_of(1)
thing_group_description2b["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupName"
].should.match(group_name_1a)
thing_group_description2b["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
thing_group_description2b.should.have.key("version")
# groups level 3
# 3a
thing_group_description3a = client.describe_thing_group(
thingGroupName=group_name_3a
)
thing_group_description3a.should.have.key("thingGroupName").which.should.equal(
group_name_3a
)
thing_group_description3a.should.have.key("thingGroupProperties")
thing_group_description3a.should.have.key("thingGroupMetadata")
thing_group_description3a["thingGroupMetadata"].should.have.length_of(3)
thing_group_description3a["thingGroupMetadata"].should.have.key(
"parentGroupName"
).being.equal(group_name_2a)
thing_group_description3a["thingGroupMetadata"].should.have.key(
"rootToParentThingGroups"
)
thing_group_description3a["thingGroupMetadata"][
"rootToParentThingGroups"
].should.have.length_of(2)
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupName"
].should.match(group_name_1a)
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2a)
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2a["thingGroupArn"])
thing_group_description3a.should.have.key("version")
# 3b
thing_group_description3b = client.describe_thing_group(
thingGroupName=group_name_3b
)
thing_group_description3b.should.have.key("thingGroupName").which.should.equal(
group_name_3b
)
thing_group_description3b.should.have.key("thingGroupProperties")
thing_group_description3b.should.have.key("thingGroupMetadata")
thing_group_description3b["thingGroupMetadata"].should.have.length_of(3)
thing_group_description3b["thingGroupMetadata"].should.have.key(
"parentGroupName"
).being.equal(group_name_2a)
thing_group_description3b["thingGroupMetadata"].should.have.key(
"rootToParentThingGroups"
)
thing_group_description3b["thingGroupMetadata"][
"rootToParentThingGroups"
].should.have.length_of(2)
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupName"
].should.match(group_name_1a)
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2a)
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2a["thingGroupArn"])
thing_group_description3b.should.have.key("version")
# 3c
thing_group_description3c = client.describe_thing_group(
thingGroupName=group_name_3c
)
thing_group_description3c.should.have.key("thingGroupName").which.should.equal(
group_name_3c
)
thing_group_description3c.should.have.key("thingGroupProperties")
thing_group_description3c.should.have.key("thingGroupMetadata")
thing_group_description3c["thingGroupMetadata"].should.have.length_of(3)
thing_group_description3c["thingGroupMetadata"].should.have.key(
"parentGroupName"
).being.equal(group_name_2b)
thing_group_description3c["thingGroupMetadata"].should.have.key(
"rootToParentThingGroups"
)
thing_group_description3c["thingGroupMetadata"][
"rootToParentThingGroups"
].should.have.length_of(2)
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupName"
].should.match(group_name_1a)
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2b)
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2b["thingGroupArn"])
thing_group_description3c.should.have.key("version")
# 3d
thing_group_description3d = client.describe_thing_group(
thingGroupName=group_name_3d
)
thing_group_description3d.should.have.key("thingGroupName").which.should.equal(
group_name_3d
)
thing_group_description3d.should.have.key("thingGroupProperties")
thing_group_description3d.should.have.key("thingGroupMetadata")
thing_group_description3d["thingGroupMetadata"].should.have.length_of(3)
thing_group_description3d["thingGroupMetadata"].should.have.key(
"parentGroupName"
).being.equal(group_name_2b)
thing_group_description3d["thingGroupMetadata"].should.have.key(
"rootToParentThingGroups"
)
thing_group_description3d["thingGroupMetadata"][
"rootToParentThingGroups"
].should.have.length_of(2)
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupName"
].should.match(group_name_1a)
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2b)
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2b["thingGroupArn"])
thing_group_description3d.should.have.key("version")
@mock_iot
def test_thing_groups():
client = boto3.client("iot", region_name="ap-northeast-1")

208
tests/test_logs/test_logs.py
View file

@ -166,70 +166,202 @@ def test_delete_retention_policy():
@mock_logs
def test_get_log_events():
conn = boto3.client("logs", "us-west-2")
client = boto3.client("logs", "us-west-2")
log_group_name = "test"
log_stream_name = "stream"
conn.create_log_group(logGroupName=log_group_name)
conn.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
client.create_log_group(logGroupName=log_group_name)
client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
events = [{"timestamp": x, "message": str(x)} for x in range(20)]
conn.put_log_events(
client.put_log_events(
logGroupName=log_group_name, logStreamName=log_stream_name, logEvents=events
)
resp = conn.get_log_events(
resp = client.get_log_events(
logGroupName=log_group_name, logStreamName=log_stream_name, limit=10
)
resp["events"].should.have.length_of(10)
resp.should.have.key("nextForwardToken")
resp.should.have.key("nextBackwardToken")
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000010"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000000"
)
for i in range(10):
resp["events"][i]["timestamp"].should.equal(i)
resp["events"][i]["message"].should.equal(str(i))
next_token = resp["nextForwardToken"]
resp = conn.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=next_token,
limit=10,
)
resp["events"].should.have.length_of(10)
resp.should.have.key("nextForwardToken")
resp.should.have.key("nextBackwardToken")
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000020"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000000"
)
for i in range(10):
resp["events"][i]["timestamp"].should.equal(i + 10)
resp["events"][i]["message"].should.equal(str(i + 10))
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000019"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000010"
)
resp = conn.get_log_events(
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp["nextBackwardToken"],
limit=20,
)
resp["events"].should.have.length_of(10)
for i in range(10):
resp["events"][i]["timestamp"].should.equal(i)
resp["events"][i]["message"].should.equal(str(i))
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000009"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000000"
)
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp["nextBackwardToken"],
limit=10,
)
resp["events"].should.have.length_of(0)
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000000"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000000"
)
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp["nextForwardToken"],
limit=1,
)
resp["events"].should.have.length_of(1)
resp["events"][0]["timestamp"].should.equal(1)
resp["events"][0]["message"].should.equal(str(1))
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000001"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000001"
)
@mock_logs
def test_get_log_events_with_start_from_head():
client = boto3.client("logs", "us-west-2")
log_group_name = "test"
log_stream_name = "stream"
client.create_log_group(logGroupName=log_group_name)
client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
events = [{"timestamp": x, "message": str(x)} for x in range(20)]
client.put_log_events(
logGroupName=log_group_name, logStreamName=log_stream_name, logEvents=events
)
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
limit=10,
startFromHead=True, # this parameter is only relevant without the usage of nextToken
)
resp["events"].should.have.length_of(10)
resp.should.have.key("nextForwardToken")
resp.should.have.key("nextBackwardToken")
for i in range(10):
resp["events"][i]["timestamp"].should.equal(i)
resp["events"][i]["message"].should.equal(str(i))
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000009"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000000"
)
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp["nextForwardToken"],
limit=20,
)
resp["events"].should.have.length_of(10)
for i in range(10):
resp["events"][i]["timestamp"].should.equal(i + 10)
resp["events"][i]["message"].should.equal(str(i + 10))
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000019"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000010"
)
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp["nextForwardToken"],
limit=10,
)
resp["events"].should.have.length_of(0)
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000019"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000019"
)
resp = client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp["nextBackwardToken"],
limit=1,
)
resp["events"].should.have.length_of(1)
resp["events"][0]["timestamp"].should.equal(18)
resp["events"][0]["message"].should.equal(str(18))
resp["nextForwardToken"].should.equal(
"f/00000000000000000000000000000000000000000000000000000018"
)
resp["nextBackwardToken"].should.equal(
"b/00000000000000000000000000000000000000000000000000000018"
)
@mock_logs
def test_get_log_events_errors():
client = boto3.client("logs", "us-west-2")
log_group_name = "test"
log_stream_name = "stream"
client.create_log_group(logGroupName=log_group_name)
client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
with assert_raises(ClientError) as e:
client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken="n/00000000000000000000000000000000000000000000000000000000",
)
ex = e.exception
ex.operation_name.should.equal("GetLogEvents")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.equal("InvalidParameterException")
ex.response["Error"]["Message"].should.contain(
"The specified nextToken is invalid."
)
with assert_raises(ClientError) as e:
client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken="not-existing-token",
)
ex = e.exception
ex.operation_name.should.equal("GetLogEvents")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.equal("InvalidParameterException")
ex.response["Error"]["Message"].should.contain(
"The specified nextToken is invalid."
)
@mock_logs

108
tests/test_organizations/test_organizations_boto3.py
View file

@ -3,6 +3,7 @@ from __future__ import unicode_literals
import boto3
import json
import six
import sure # noqa
from botocore.exceptions import ClientError
from nose.tools import assert_raises
@ -605,3 +606,110 @@ def test_list_targets_for_policy_exception():
ex.operation_name.should.equal("ListTargetsForPolicy")
ex.response["Error"]["Code"].should.equal("400")
ex.response["Error"]["Message"].should.contain("InvalidInputException")
@mock_organizations
def test_tag_resource():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
account_id = client.create_account(AccountName=mockname, Email=mockemail)[
"CreateAccountStatus"
]["AccountId"]
client.tag_resource(ResourceId=account_id, Tags=[{"Key": "key", "Value": "value"}])
response = client.list_tags_for_resource(ResourceId=account_id)
response["Tags"].should.equal([{"Key": "key", "Value": "value"}])
# adding a tag with an existing key, will update the value
client.tag_resource(
ResourceId=account_id, Tags=[{"Key": "key", "Value": "new-value"}]
)
response = client.list_tags_for_resource(ResourceId=account_id)
response["Tags"].should.equal([{"Key": "key", "Value": "new-value"}])
@mock_organizations
def test_tag_resource_errors():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
with assert_raises(ClientError) as e:
client.tag_resource(
ResourceId="000000000000", Tags=[{"Key": "key", "Value": "value"},]
)
ex = e.exception
ex.operation_name.should.equal("TagResource")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidInputException")
ex.response["Error"]["Message"].should.equal(
"You provided a value that does not match the required pattern."
)
@mock_organizations
def test_list_tags_for_resource():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
account_id = client.create_account(AccountName=mockname, Email=mockemail)[
"CreateAccountStatus"
]["AccountId"]
client.tag_resource(ResourceId=account_id, Tags=[{"Key": "key", "Value": "value"}])
response = client.list_tags_for_resource(ResourceId=account_id)
response["Tags"].should.equal([{"Key": "key", "Value": "value"}])
@mock_organizations
def test_list_tags_for_resource_errors():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
with assert_raises(ClientError) as e:
client.list_tags_for_resource(ResourceId="000000000000")
ex = e.exception
ex.operation_name.should.equal("ListTagsForResource")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidInputException")
ex.response["Error"]["Message"].should.equal(
"You provided a value that does not match the required pattern."
)
@mock_organizations
def test_untag_resource():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
account_id = client.create_account(AccountName=mockname, Email=mockemail)[
"CreateAccountStatus"
]["AccountId"]
client.tag_resource(ResourceId=account_id, Tags=[{"Key": "key", "Value": "value"}])
response = client.list_tags_for_resource(ResourceId=account_id)
response["Tags"].should.equal([{"Key": "key", "Value": "value"}])
# removing a non existing tag should not raise any error
client.untag_resource(ResourceId=account_id, TagKeys=["not-existing"])
response = client.list_tags_for_resource(ResourceId=account_id)
response["Tags"].should.equal([{"Key": "key", "Value": "value"}])
client.untag_resource(ResourceId=account_id, TagKeys=["key"])
response = client.list_tags_for_resource(ResourceId=account_id)
response["Tags"].should.have.length_of(0)
@mock_organizations
def test_untag_resource_errors():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
with assert_raises(ClientError) as e:
client.untag_resource(ResourceId="000000000000", TagKeys=["key"])
ex = e.exception
ex.operation_name.should.equal("UntagResource")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidInputException")
ex.response["Error"]["Message"].should.equal(
"You provided a value that does not match the required pattern."
)

24
tests/test_secretsmanager/test_secretsmanager.py
View file

@ -26,6 +26,18 @@ def test_get_secret_value():
assert result["SecretString"] == "foosecret"
@mock_secretsmanager
def test_get_secret_value_by_arn():
conn = boto3.client("secretsmanager", region_name="us-west-2")
secret_value = "test_get_secret_value_by_arn"
result = conn.create_secret(
Name="java-util-test-password", SecretString=secret_value
)
result = conn.get_secret_value(SecretId=result["ARN"])
assert result["SecretString"] == secret_value
@mock_secretsmanager
def test_get_secret_value_binary():
conn = boto3.client("secretsmanager", region_name="us-west-2")
@ -361,6 +373,18 @@ def test_describe_secret():
assert secret_description_2["ARN"] != "" # Test arn not empty
@mock_secretsmanager
def test_describe_secret_with_arn():
conn = boto3.client("secretsmanager", region_name="us-west-2")
results = conn.create_secret(Name="test-secret", SecretString="foosecret")
secret_description = conn.describe_secret(SecretId=results["ARN"])
assert secret_description # Returned dict is not empty
assert secret_description["Name"] == ("test-secret")
assert secret_description["ARN"] != results["ARN"]
@mock_secretsmanager
def test_describe_secret_that_does_not_exist():
conn = boto3.client("secretsmanager", region_name="us-west-2")

23
tests/test_secretsmanager/test_server.py
View file

@ -586,6 +586,29 @@ def test_can_list_secret_version_ids():
].sort() == returned_version_ids.sort()
@mock_secretsmanager
def test_get_resource_policy_secret():
backend = server.create_backend_app("secretsmanager")
test_client = backend.test_client()
create_secret = test_client.post(
"/",
data={"Name": "test-secret", "SecretString": "foosecret"},
headers={"X-Amz-Target": "secretsmanager.CreateSecret"},
)
describe_secret = test_client.post(
"/",
data={"SecretId": "test-secret"},
headers={"X-Amz-Target": "secretsmanager.GetResourcePolicy"},
)
json_data = json.loads(describe_secret.data.decode("utf-8"))
assert json_data # Returned dict is not empty
assert json_data["ARN"] != ""
assert json_data["Name"] == "test-secret"
#
# The following tests should work, but fail on the embedded dict in
# RotationRules. The error message suggests a problem deeper in the code, which