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

Merge remote-tracking branch 'original/master'

Browse source
This commit is contained in:
Stephan Huber 2018-06-06 09:57:12 +02:00
commit e118a678a6
57 changed files with 2158 additions and 93 deletions
Download patch file Download diff file Expand all files Collapse all files

47
tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
View file

@ -148,10 +148,41 @@ dummy_import_template = {
}
}
dummy_redrive_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"MainQueue": {
"Type": "AWS::SQS::Queue",
"Properties": {
"QueueName": "mainqueue.fifo",
"FifoQueue": True,
"ContentBasedDeduplication": False,
"RedrivePolicy": {
"deadLetterTargetArn": {
"Fn::GetAtt": [
"DeadLetterQueue",
"Arn"
]
},
"maxReceiveCount": 5
}
}
},
"DeadLetterQueue": {
"Type": "AWS::SQS::Queue",
"Properties": {
"FifoQueue": True
}
},
}
}
dummy_template_json = json.dumps(dummy_template)
dummy_update_template_json = json.dumps(dummy_update_template)
dummy_output_template_json = json.dumps(dummy_output_template)
dummy_import_template_json = json.dumps(dummy_import_template)
dummy_redrive_template_json = json.dumps(dummy_redrive_template)
@mock_cloudformation
@ -746,3 +777,19 @@ def test_stack_with_imports():
output = output_stack.outputs[0]['OutputValue']
queue = ec2_resource.get_queue_by_name(QueueName=output)
queue.should_not.be.none
@mock_sqs
@mock_cloudformation
def test_non_json_redrive_policy():
cf = boto3.resource('cloudformation', region_name='us-east-1')
stack = cf.create_stack(
StackName="test_stack1",
TemplateBody=dummy_redrive_template_json
)
stack.Resource('MainQueue').resource_status\
.should.equal("CREATE_COMPLETE")
stack.Resource('DeadLetterQueue').resource_status\
.should.equal("CREATE_COMPLETE")

87
tests/test_cloudformation/test_import_value.py Normal file
View file

@ -0,0 +1,87 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function, unicode_literals
# Standard library modules
import unittest
# Third-party modules
import boto3
from botocore.exceptions import ClientError
# Package modules
from moto import mock_cloudformation
AWS_REGION = 'us-west-1'
SG_STACK_NAME = 'simple-sg-stack'
SG_TEMPLATE = """
AWSTemplateFormatVersion: 2010-09-09
Description: Simple test CF template for moto_cloudformation
Resources:
SimpleSecurityGroup:
Type: AWS::EC2::SecurityGroup
Description: "A simple security group"
Properties:
GroupName: simple-security-group
GroupDescription: "A simple security group"
SecurityGroupEgress:
-
Description: "Egress to remote HTTPS servers"
CidrIp: 0.0.0.0/0
IpProtocol: tcp
FromPort: 443
ToPort: 443
Outputs:
SimpleSecurityGroupName:
Value: !GetAtt SimpleSecurityGroup.GroupId
Export:
Name: "SimpleSecurityGroup"
"""
EC2_STACK_NAME = 'simple-ec2-stack'
EC2_TEMPLATE = """
---
# The latest template format version is "2010-09-09" and as of 2018-04-09
# is currently the only valid value.
AWSTemplateFormatVersion: 2010-09-09
Description: Simple test CF template for moto_cloudformation
Resources:
SimpleInstance:
Type: AWS::EC2::Instance
Properties:
ImageId: ami-03cf127a
InstanceType: t2.micro
SecurityGroups: !Split [',', !ImportValue SimpleSecurityGroup]
"""
class TestSimpleInstance(unittest.TestCase):
def test_simple_instance(self):
"""Test that we can create a simple CloudFormation stack that imports values from an existing CloudFormation stack"""
with mock_cloudformation():
client = boto3.client('cloudformation', region_name=AWS_REGION)
client.create_stack(StackName=SG_STACK_NAME, TemplateBody=SG_TEMPLATE)
response = client.create_stack(StackName=EC2_STACK_NAME, TemplateBody=EC2_TEMPLATE)
self.assertIn('StackId', response)
response = client.describe_stacks(StackName=response['StackId'])
self.assertIn('Stacks', response)
stack_info = response['Stacks']
self.assertEqual(1, len(stack_info))
self.assertIn('StackName', stack_info[0])
self.assertEqual(EC2_STACK_NAME, stack_info[0]['StackName'])
def test_simple_instance_missing_export(self):
"""Test that we get an exception if a CloudFormation stack tries to imports a non-existent export value"""
with mock_cloudformation():
client = boto3.client('cloudformation', region_name=AWS_REGION)
with self.assertRaises(ClientError) as e:
client.create_stack(StackName=EC2_STACK_NAME, TemplateBody=EC2_TEMPLATE)
self.assertIn('Error', e.exception.response)
self.assertIn('Code', e.exception.response['Error'])
self.assertEqual('ExportNotFound', e.exception.response['Error']['Code'])

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

@ -162,6 +162,37 @@ def test_put_metric_data_no_dimensions():
metric['MetricName'].should.equal('metric')
@mock_cloudwatch
def test_put_metric_data_with_statistics():
conn = boto3.client('cloudwatch', region_name='us-east-1')
conn.put_metric_data(
Namespace='tester',
MetricData=[
dict(
MetricName='statmetric',
Timestamp=datetime(2015, 1, 1),
# 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
),
Unit='Milliseconds',
StorageResolution=123
)
]
)
metrics = conn.list_metrics()['Metrics']
metrics.should.have.length_of(1)
metric = metrics[0]
metric['Namespace'].should.equal('tester')
metric['MetricName'].should.equal('statmetric')
# TODO: test statistics - https://github.com/spulec/moto/issues/1615
@mock_cloudwatch
def test_get_metric_statistics():
conn = boto3.client('cloudwatch', region_name='us-east-1')
@ -173,6 +204,7 @@ def test_get_metric_statistics():
dict(
MetricName='metric',
Value=1.5,
Timestamp=utc_now
)
]
)
@ -180,7 +212,7 @@ def test_get_metric_statistics():
stats = conn.get_metric_statistics(
Namespace='tester',
MetricName='metric',
StartTime=utc_now,
StartTime=utc_now - timedelta(seconds=60),
EndTime=utc_now + timedelta(seconds=60),
Period=60,
Statistics=['SampleCount', 'Sum']

539
tests/test_cognitoidp/test_cognitoidp.py Normal file
View file

@ -0,0 +1,539 @@
from __future__ import unicode_literals
import boto3
import json
import os
import uuid
from jose import jws
from moto import mock_cognitoidp
import sure # noqa
@mock_cognitoidp
def test_create_user_pool():
conn = boto3.client("cognito-idp", "us-west-2")
name = str(uuid.uuid4())
value = str(uuid.uuid4())
result = conn.create_user_pool(
PoolName=name,
LambdaConfig={
"PreSignUp": value
}
)
result["UserPool"]["Id"].should_not.be.none
result["UserPool"]["Name"].should.equal(name)
result["UserPool"]["LambdaConfig"]["PreSignUp"].should.equal(value)
@mock_cognitoidp
def test_list_user_pools():
conn = boto3.client("cognito-idp", "us-west-2")
name = str(uuid.uuid4())
conn.create_user_pool(PoolName=name)
result = conn.list_user_pools(MaxResults=10)
result["UserPools"].should.have.length_of(1)
result["UserPools"][0]["Name"].should.equal(name)
@mock_cognitoidp
def test_describe_user_pool():
conn = boto3.client("cognito-idp", "us-west-2")
name = str(uuid.uuid4())
value = str(uuid.uuid4())
user_pool_details = conn.create_user_pool(
PoolName=name,
LambdaConfig={
"PreSignUp": value
}
)
result = conn.describe_user_pool(UserPoolId=user_pool_details["UserPool"]["Id"])
result["UserPool"]["Name"].should.equal(name)
result["UserPool"]["LambdaConfig"]["PreSignUp"].should.equal(value)
@mock_cognitoidp
def test_delete_user_pool():
conn = boto3.client("cognito-idp", "us-west-2")
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.list_user_pools(MaxResults=10)["UserPools"].should.have.length_of(1)
conn.delete_user_pool(UserPoolId=user_pool_id)
conn.list_user_pools(MaxResults=10)["UserPools"].should.have.length_of(0)
@mock_cognitoidp
def test_create_user_pool_domain():
conn = boto3.client("cognito-idp", "us-west-2")
domain = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
result = conn.create_user_pool_domain(UserPoolId=user_pool_id, Domain=domain)
result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@mock_cognitoidp
def test_describe_user_pool_domain():
conn = boto3.client("cognito-idp", "us-west-2")
domain = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_user_pool_domain(UserPoolId=user_pool_id, Domain=domain)
result = conn.describe_user_pool_domain(Domain=domain)
result["DomainDescription"]["Domain"].should.equal(domain)
result["DomainDescription"]["UserPoolId"].should.equal(user_pool_id)
result["DomainDescription"]["AWSAccountId"].should_not.be.none
@mock_cognitoidp
def test_delete_user_pool_domain():
conn = boto3.client("cognito-idp", "us-west-2")
domain = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_user_pool_domain(UserPoolId=user_pool_id, Domain=domain)
result = conn.delete_user_pool_domain(UserPoolId=user_pool_id, Domain=domain)
result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
result = conn.describe_user_pool_domain(Domain=domain)
# This is a surprising behavior of the real service: describing a missing domain comes
# back with status 200 and a DomainDescription of {}
result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
result["DomainDescription"].keys().should.have.length_of(0)
@mock_cognitoidp
def test_create_user_pool_client():
conn = boto3.client("cognito-idp", "us-west-2")
client_name = str(uuid.uuid4())
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
result = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=client_name,
CallbackURLs=[value],
)
result["UserPoolClient"]["UserPoolId"].should.equal(user_pool_id)
result["UserPoolClient"]["ClientId"].should_not.be.none
result["UserPoolClient"]["ClientName"].should.equal(client_name)
result["UserPoolClient"]["CallbackURLs"].should.have.length_of(1)
result["UserPoolClient"]["CallbackURLs"][0].should.equal(value)
@mock_cognitoidp
def test_list_user_pool_clients():
conn = boto3.client("cognito-idp", "us-west-2")
client_name = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_user_pool_client(UserPoolId=user_pool_id, ClientName=client_name)
result = conn.list_user_pool_clients(UserPoolId=user_pool_id, MaxResults=10)
result["UserPoolClients"].should.have.length_of(1)
result["UserPoolClients"][0]["ClientName"].should.equal(client_name)
@mock_cognitoidp
def test_describe_user_pool_client():
conn = boto3.client("cognito-idp", "us-west-2")
client_name = str(uuid.uuid4())
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_details = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=client_name,
CallbackURLs=[value],
)
result = conn.describe_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_details["UserPoolClient"]["ClientId"],
)
result["UserPoolClient"]["ClientName"].should.equal(client_name)
result["UserPoolClient"]["CallbackURLs"].should.have.length_of(1)
result["UserPoolClient"]["CallbackURLs"][0].should.equal(value)
@mock_cognitoidp
def test_update_user_pool_client():
conn = boto3.client("cognito-idp", "us-west-2")
old_client_name = str(uuid.uuid4())
new_client_name = str(uuid.uuid4())
old_value = str(uuid.uuid4())
new_value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_details = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=old_client_name,
CallbackURLs=[old_value],
)
result = conn.update_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_details["UserPoolClient"]["ClientId"],
ClientName=new_client_name,
CallbackURLs=[new_value],
)
result["UserPoolClient"]["ClientName"].should.equal(new_client_name)
result["UserPoolClient"]["CallbackURLs"].should.have.length_of(1)
result["UserPoolClient"]["CallbackURLs"][0].should.equal(new_value)
@mock_cognitoidp
def test_delete_user_pool_client():
conn = boto3.client("cognito-idp", "us-west-2")
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_details = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
)
conn.delete_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_details["UserPoolClient"]["ClientId"],
)
caught = False
try:
conn.describe_user_pool_client(
UserPoolId=user_pool_id,
ClientId=client_details["UserPoolClient"]["ClientId"],
)
except conn.exceptions.ResourceNotFoundException:
caught = True
caught.should.be.true
@mock_cognitoidp
def test_create_identity_provider():
conn = boto3.client("cognito-idp", "us-west-2")
provider_name = str(uuid.uuid4())
provider_type = "Facebook"
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
result = conn.create_identity_provider(
UserPoolId=user_pool_id,
ProviderName=provider_name,
ProviderType=provider_type,
ProviderDetails={
"thing": value
},
)
result["IdentityProvider"]["UserPoolId"].should.equal(user_pool_id)
result["IdentityProvider"]["ProviderName"].should.equal(provider_name)
result["IdentityProvider"]["ProviderType"].should.equal(provider_type)
result["IdentityProvider"]["ProviderDetails"]["thing"].should.equal(value)
@mock_cognitoidp
def test_list_identity_providers():
conn = boto3.client("cognito-idp", "us-west-2")
provider_name = str(uuid.uuid4())
provider_type = "Facebook"
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_identity_provider(
UserPoolId=user_pool_id,
ProviderName=provider_name,
ProviderType=provider_type,
ProviderDetails={},
)
result = conn.list_identity_providers(
UserPoolId=user_pool_id,
MaxResults=10,
)
result["Providers"].should.have.length_of(1)
result["Providers"][0]["ProviderName"].should.equal(provider_name)
result["Providers"][0]["ProviderType"].should.equal(provider_type)
@mock_cognitoidp
def test_describe_identity_providers():
conn = boto3.client("cognito-idp", "us-west-2")
provider_name = str(uuid.uuid4())
provider_type = "Facebook"
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_identity_provider(
UserPoolId=user_pool_id,
ProviderName=provider_name,
ProviderType=provider_type,
ProviderDetails={
"thing": value
},
)
result = conn.describe_identity_provider(
UserPoolId=user_pool_id,
ProviderName=provider_name,
)
result["IdentityProvider"]["UserPoolId"].should.equal(user_pool_id)
result["IdentityProvider"]["ProviderName"].should.equal(provider_name)
result["IdentityProvider"]["ProviderType"].should.equal(provider_type)
result["IdentityProvider"]["ProviderDetails"]["thing"].should.equal(value)
@mock_cognitoidp
def test_delete_identity_providers():
conn = boto3.client("cognito-idp", "us-west-2")
provider_name = str(uuid.uuid4())
provider_type = "Facebook"
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.create_identity_provider(
UserPoolId=user_pool_id,
ProviderName=provider_name,
ProviderType=provider_type,
ProviderDetails={
"thing": value
},
)
conn.delete_identity_provider(UserPoolId=user_pool_id, ProviderName=provider_name)
caught = False
try:
conn.describe_identity_provider(
UserPoolId=user_pool_id,
ProviderName=provider_name,
)
except conn.exceptions.ResourceNotFoundException:
caught = True
caught.should.be.true
@mock_cognitoidp
def test_admin_create_user():
conn = boto3.client("cognito-idp", "us-west-2")
username = str(uuid.uuid4())
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
result = conn.admin_create_user(
UserPoolId=user_pool_id,
Username=username,
UserAttributes=[
{"Name": "thing", "Value": value}
],
)
result["User"]["Username"].should.equal(username)
result["User"]["UserStatus"].should.equal("FORCE_CHANGE_PASSWORD")
result["User"]["Attributes"].should.have.length_of(1)
result["User"]["Attributes"][0]["Name"].should.equal("thing")
result["User"]["Attributes"][0]["Value"].should.equal(value)
@mock_cognitoidp
def test_admin_get_user():
conn = boto3.client("cognito-idp", "us-west-2")
username = str(uuid.uuid4())
value = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.admin_create_user(
UserPoolId=user_pool_id,
Username=username,
UserAttributes=[
{"Name": "thing", "Value": value}
],
)
result = conn.admin_get_user(UserPoolId=user_pool_id, Username=username)
result["Username"].should.equal(username)
result["UserAttributes"].should.have.length_of(1)
result["UserAttributes"][0]["Name"].should.equal("thing")
result["UserAttributes"][0]["Value"].should.equal(value)
@mock_cognitoidp
def test_list_users():
conn = boto3.client("cognito-idp", "us-west-2")
username = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.admin_create_user(UserPoolId=user_pool_id, Username=username)
result = conn.list_users(UserPoolId=user_pool_id)
result["Users"].should.have.length_of(1)
result["Users"][0]["Username"].should.equal(username)
@mock_cognitoidp
def test_admin_delete_user():
conn = boto3.client("cognito-idp", "us-west-2")
username = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
conn.admin_create_user(UserPoolId=user_pool_id, Username=username)
conn.admin_delete_user(UserPoolId=user_pool_id, Username=username)
caught = False
try:
conn.admin_get_user(UserPoolId=user_pool_id, Username=username)
except conn.exceptions.ResourceNotFoundException:
caught = True
caught.should.be.true
def authentication_flow(conn):
username = str(uuid.uuid4())
temporary_password = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
)["UserPoolClient"]["ClientId"]
conn.admin_create_user(
UserPoolId=user_pool_id,
Username=username,
TemporaryPassword=temporary_password,
)
result = conn.admin_initiate_auth(
UserPoolId=user_pool_id,
ClientId=client_id,
AuthFlow="ADMIN_NO_SRP_AUTH",
AuthParameters={
"USERNAME": username,
"PASSWORD": temporary_password
},
)
# A newly created user is forced to set a new password
result["ChallengeName"].should.equal("NEW_PASSWORD_REQUIRED")
result["Session"].should_not.be.none
# This sets a new password and logs the user in (creates tokens)
new_password = str(uuid.uuid4())
result = conn.respond_to_auth_challenge(
Session=result["Session"],
ClientId=client_id,
ChallengeName="NEW_PASSWORD_REQUIRED",
ChallengeResponses={
"USERNAME": username,
"NEW_PASSWORD": new_password
}
)
result["AuthenticationResult"]["IdToken"].should_not.be.none
result["AuthenticationResult"]["AccessToken"].should_not.be.none
return {
"user_pool_id": user_pool_id,
"client_id": client_id,
"id_token": result["AuthenticationResult"]["IdToken"],
"access_token": result["AuthenticationResult"]["AccessToken"],
"username": username,
"password": new_password,
}
@mock_cognitoidp
def test_authentication_flow():
conn = boto3.client("cognito-idp", "us-west-2")
authentication_flow(conn)
@mock_cognitoidp
def test_token_legitimacy():
conn = boto3.client("cognito-idp", "us-west-2")
path = "../../moto/cognitoidp/resources/jwks-public.json"
with open(os.path.join(os.path.dirname(__file__), path)) as f:
json_web_key = json.loads(f.read())["keys"][0]
outputs = authentication_flow(conn)
id_token = outputs["id_token"]
access_token = outputs["access_token"]
client_id = outputs["client_id"]
issuer = "https://cognito-idp.us-west-2.amazonaws.com/{}".format(outputs["user_pool_id"])
id_claims = json.loads(jws.verify(id_token, json_web_key, "RS256"))
id_claims["iss"].should.equal(issuer)
id_claims["aud"].should.equal(client_id)
access_claims = json.loads(jws.verify(access_token, json_web_key, "RS256"))
access_claims["iss"].should.equal(issuer)
access_claims["aud"].should.equal(client_id)
@mock_cognitoidp
def test_change_password():
conn = boto3.client("cognito-idp", "us-west-2")
outputs = authentication_flow(conn)
# Take this opportunity to test change_password, which requires an access token.
newer_password = str(uuid.uuid4())
conn.change_password(
AccessToken=outputs["access_token"],
PreviousPassword=outputs["password"],
ProposedPassword=newer_password,
)
# Log in again, which should succeed without a challenge because the user is no
# longer in the force-new-password state.
result = conn.admin_initiate_auth(
UserPoolId=outputs["user_pool_id"],
ClientId=outputs["client_id"],
AuthFlow="ADMIN_NO_SRP_AUTH",
AuthParameters={
"USERNAME": outputs["username"],
"PASSWORD": newer_password,
},
)
result["AuthenticationResult"].should_not.be.none
@mock_cognitoidp
def test_forgot_password():
conn = boto3.client("cognito-idp", "us-west-2")
result = conn.forgot_password(ClientId=str(uuid.uuid4()), Username=str(uuid.uuid4()))
result["CodeDeliveryDetails"].should_not.be.none
@mock_cognitoidp
def test_confirm_forgot_password():
conn = boto3.client("cognito-idp", "us-west-2")
username = str(uuid.uuid4())
user_pool_id = conn.create_user_pool(PoolName=str(uuid.uuid4()))["UserPool"]["Id"]
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
)["UserPoolClient"]["ClientId"]
conn.admin_create_user(
UserPoolId=user_pool_id,
Username=username,
TemporaryPassword=str(uuid.uuid4()),
)
conn.confirm_forgot_password(
ClientId=client_id,
Username=username,
ConfirmationCode=str(uuid.uuid4()),
Password=str(uuid.uuid4()),
)

31
tests/test_dynamodb2/test_dynamodb.py
View file

@ -6,6 +6,7 @@ import boto3
from boto3.dynamodb.conditions import Attr
import sure # noqa
import requests
from pytest import raises
from moto import mock_dynamodb2, mock_dynamodb2_deprecated
from moto.dynamodb2 import dynamodb_backend2
from boto.exception import JSONResponseError
@ -1052,6 +1053,7 @@ def test_query_missing_expr_names():
@mock_dynamodb2
def test_update_item_on_map():
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
client = boto3.client('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
dynamodb.create_table(
@ -1092,21 +1094,44 @@ def test_update_item_on_map():
resp = table.scan()
resp['Items'][0]['body'].should.equal({'nested': {'data': 'test'}})
# Nonexistent nested attributes are supported for existing top-level attributes.
table.update_item(Key={
'forum_name': 'the-key',
'subject': '123'
},
UpdateExpression='SET body.#nested.#data = :tb',
UpdateExpression='SET body.#nested.#data = :tb, body.nested.#nonexistentnested.#data = :tb2',
ExpressionAttributeNames={
'#nested': 'nested',
'#nonexistentnested': 'nonexistentnested',
'#data': 'data'
},
ExpressionAttributeValues={
':tb': 'new_value'
':tb': 'new_value',
':tb2': 'other_value'
})
resp = table.scan()
resp['Items'][0]['body'].should.equal({'nested': {'data': 'new_value'}})
resp['Items'][0]['body'].should.equal({
'nested': {
'data': 'new_value',
'nonexistentnested': {'data': 'other_value'}
}
})
# Test nested value for a nonexistent attribute.
with raises(client.exceptions.ConditionalCheckFailedException):
table.update_item(Key={
'forum_name': 'the-key',
'subject': '123'
},
UpdateExpression='SET nonexistent.#nested = :tb',
ExpressionAttributeNames={
'#nested': 'nested'
},
ExpressionAttributeValues={
':tb': 'new_value'
})
# https://github.com/spulec/moto/issues/1358

2
tests/test_ec2/test_elastic_block_store.py
View file

@ -8,6 +8,7 @@ import boto
import boto3
from botocore.exceptions import ClientError
from boto.exception import EC2ResponseError
from freezegun import freeze_time
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
@ -588,6 +589,7 @@ def test_volume_tag_escaping():
dict(snaps[0].tags).should.equal({'key': '</closed>'})
@freeze_time
@mock_ec2
def test_copy_snapshot():
ec2_client = boto3.client('ec2', region_name='eu-west-1')

11
tests/test_ec2/test_elastic_ip_addresses.py
View file

@ -62,6 +62,17 @@ def test_eip_allocate_vpc():
logging.debug("vpc alloc_id:".format(vpc.allocation_id))
vpc.release()
@mock_ec2
def test_specific_eip_allocate_vpc():
"""Allocate VPC EIP with specific address"""
service = boto3.resource('ec2', region_name='us-west-1')
client = boto3.client('ec2', region_name='us-west-1')
vpc = client.allocate_address(Domain="vpc", Address="127.38.43.222")
vpc['Domain'].should.be.equal("vpc")
vpc['PublicIp'].should.be.equal("127.38.43.222")
logging.debug("vpc alloc_id:".format(vpc['AllocationId']))
@mock_ec2_deprecated
def test_eip_allocate_invalid_domain():

13
tests/test_events/test_events.py
View file

@ -1,6 +1,7 @@
import random
import boto3
import json
from moto.events import mock_events
from botocore.exceptions import ClientError
@ -177,17 +178,19 @@ def test_remove_targets():
def test_permissions():
client = boto3.client('events', 'eu-central-1')
client.put_permission(Action='PutEvents', Principal='111111111111', StatementId='Account1')
client.put_permission(Action='PutEvents', Principal='222222222222', StatementId='Account2')
client.put_permission(Action='events:PutEvents', Principal='111111111111', StatementId='Account1')
client.put_permission(Action='events:PutEvents', Principal='222222222222', StatementId='Account2')
resp = client.describe_event_bus()
assert len(resp['Policy']['Statement']) == 2
resp_policy = json.loads(resp['Policy'])
assert len(resp_policy['Statement']) == 2
client.remove_permission(StatementId='Account2')
resp = client.describe_event_bus()
assert len(resp['Policy']['Statement']) == 1
assert resp['Policy']['Statement'][0]['Sid'] == 'Account1'
resp_policy = json.loads(resp['Policy'])
assert len(resp_policy['Statement']) == 1
assert resp_policy['Statement'][0]['Sid'] == 'Account1'
@mock_events

4
tests/test_logs/test_logs.py
View file

@ -13,6 +13,10 @@ def test_log_group_create():
conn = boto3.client('logs', 'us-west-2')
log_group_name = 'dummy'
response = conn.create_log_group(logGroupName=log_group_name)
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response['logGroups']) == 1
response = conn.delete_log_group(logGroupName=log_group_name)

7
tests/test_rds2/test_rds2.py
View file

@ -353,8 +353,6 @@ def test_describe_db_snapshots():
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
conn.describe_db_snapshots.when.called_with(
DBInstanceIdentifier="db-primary-1").should.throw(ClientError)
created = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-1').get('DBSnapshot')
@ -369,6 +367,11 @@ def test_describe_db_snapshots():
snapshot.should.equal(created)
snapshot.get('Engine').should.equal('postgres')
conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-2')
snapshots = conn.describe_db_snapshots(DBInstanceIdentifier='db-primary-1').get('DBSnapshots')
snapshots.should.have.length_of(2)
@mock_rds2
def test_delete_db_snapshot():

42
tests/test_redshift/test_redshift.py
View file

@ -818,6 +818,48 @@ def test_create_cluster_from_snapshot():
new_cluster['Endpoint']['Port'].should.equal(1234)
@mock_redshift
def test_create_cluster_from_snapshot_with_waiter():
client = boto3.client('redshift', region_name='us-east-1')
original_cluster_identifier = 'original-cluster'
original_snapshot_identifier = 'original-snapshot'
new_cluster_identifier = 'new-cluster'
client.create_cluster(
ClusterIdentifier=original_cluster_identifier,
ClusterType='single-node',
NodeType='ds2.xlarge',
MasterUsername='username',
MasterUserPassword='password',
)
client.create_cluster_snapshot(
SnapshotIdentifier=original_snapshot_identifier,
ClusterIdentifier=original_cluster_identifier
)
response = client.restore_from_cluster_snapshot(
ClusterIdentifier=new_cluster_identifier,
SnapshotIdentifier=original_snapshot_identifier,
Port=1234
)
response['Cluster']['ClusterStatus'].should.equal('creating')
client.get_waiter('cluster_restored').wait(
ClusterIdentifier=new_cluster_identifier,
WaiterConfig={
'Delay': 1,
'MaxAttempts': 2,
}
)
response = client.describe_clusters(
ClusterIdentifier=new_cluster_identifier
)
new_cluster = response['Clusters'][0]
new_cluster['NodeType'].should.equal('ds2.xlarge')
new_cluster['MasterUsername'].should.equal('username')
new_cluster['Endpoint']['Port'].should.equal(1234)
@mock_redshift
def test_create_cluster_from_non_existent_snapshot():
client = boto3.client('redshift', region_name='us-east-1')

13
tests/test_s3/test_s3.py
View file

@ -1405,6 +1405,19 @@ def test_boto3_deleted_versionings_list():
assert len(listed['Contents']) == 1
@mock_s3
def test_boto3_delete_versioned_bucket():
client = boto3.client('s3', region_name='us-east-1')
client.create_bucket(Bucket='blah')
client.put_bucket_versioning(Bucket='blah', VersioningConfiguration={'Status': 'Enabled'})
resp = client.put_object(Bucket='blah', Key='test1', Body=b'test1')
client.delete_object(Bucket='blah', Key='test1', VersionId=resp["VersionId"])
client.delete_bucket(Bucket='blah')
@mock_s3
def test_boto3_head_object_if_modified_since():
s3 = boto3.client('s3', region_name='us-east-1')

56
tests/test_ses/test_ses_boto3.py
View file

@ -136,3 +136,59 @@ def test_send_raw_email():
send_quota = conn.get_send_quota()
sent_count = int(send_quota['SentLast24Hours'])
sent_count.should.equal(2)
@mock_ses
def test_send_raw_email_without_source():
conn = boto3.client('ses', region_name='us-east-1')
message = MIMEMultipart()
message['Subject'] = 'Test'
message['From'] = 'test@example.com'
message['To'] = 'to@example.com, foo@example.com'
# Message body
part = MIMEText('test file attached')
message.attach(part)
# Attachment
part = MIMEText('contents of test file here')
part.add_header('Content-Disposition', 'attachment; filename=test.txt')
message.attach(part)
kwargs = dict(
RawMessage={'Data': message.as_string()},
)
conn.send_raw_email.when.called_with(**kwargs).should.throw(ClientError)
conn.verify_email_identity(EmailAddress="test@example.com")
conn.send_raw_email(**kwargs)
send_quota = conn.get_send_quota()
sent_count = int(send_quota['SentLast24Hours'])
sent_count.should.equal(2)
@mock_ses
def test_send_raw_email_without_source_or_from():
conn = boto3.client('ses', region_name='us-east-1')
message = MIMEMultipart()
message['Subject'] = 'Test'
message['To'] = 'to@example.com, foo@example.com'
# Message body
part = MIMEText('test file attached')
message.attach(part)
# Attachment
part = MIMEText('contents of test file here')
part.add_header('Content-Disposition', 'attachment; filename=test.txt')
message.attach(part)
kwargs = dict(
RawMessage={'Data': message.as_string()},
)
conn.send_raw_email.when.called_with(**kwargs).should.throw(ClientError)

25
tests/test_sns/test_publishing_boto3.py
View file

@ -42,6 +42,29 @@ def test_publish_to_sqs():
acquired_message.should.equal(expected)
@mock_sqs
@mock_sns
def test_publish_to_sqs_raw():
sns = boto3.resource('sns', region_name='us-east-1')
topic = sns.create_topic(Name='some-topic')
sqs = boto3.resource('sqs', region_name='us-east-1')
queue = sqs.create_queue(QueueName='test-queue')
subscription = topic.subscribe(
Protocol='sqs', Endpoint=queue.attributes['QueueArn'])
subscription.set_attributes(
AttributeName='RawMessageDelivery', AttributeValue='true')
message = 'my message'
with freeze_time("2015-01-01 12:00:00"):
topic.publish(Message=message)
messages = queue.receive_messages(MaxNumberOfMessages=1)
messages[0].body.should.equal(message)
@mock_sqs
@mock_sns
def test_publish_to_sqs_bad():
@ -230,7 +253,7 @@ def test_publish_to_sqs_in_different_region():
@mock_sns
def test_publish_to_http():
def callback(request):
request.headers["Content-Type"].should.equal("application/json")
request.headers["Content-Type"].should.equal("text/plain; charset=UTF-8")
json.loads.when.called_with(
request.body.decode()
).should_not.throw(Exception)

29
tests/test_sns/test_topics_boto3.py
View file

@ -13,23 +13,24 @@ from moto.sns.models import DEFAULT_TOPIC_POLICY, DEFAULT_EFFECTIVE_DELIVERY_POL
@mock_sns
def test_create_and_delete_topic():
conn = boto3.client("sns", region_name="us-east-1")
conn.create_topic(Name="some-topic")
for topic_name in ('some-topic', '-some-topic-', '_some-topic_', 'a' * 256):
conn.create_topic(Name=topic_name)
topics_json = conn.list_topics()
topics = topics_json["Topics"]
topics.should.have.length_of(1)
topics[0]['TopicArn'].should.equal(
"arn:aws:sns:{0}:123456789012:some-topic"
.format(conn._client_config.region_name)
)
topics_json = conn.list_topics()
topics = topics_json["Topics"]
topics.should.have.length_of(1)
topics[0]['TopicArn'].should.equal(
"arn:aws:sns:{0}:123456789012:{1}"
.format(conn._client_config.region_name, topic_name)
)
# Delete the topic
conn.delete_topic(TopicArn=topics[0]['TopicArn'])
# Delete the topic
conn.delete_topic(TopicArn=topics[0]['TopicArn'])
# And there should now be 0 topics
topics_json = conn.list_topics()
topics = topics_json["Topics"]
topics.should.have.length_of(0)
# And there should now be 0 topics
topics_json = conn.list_topics()
topics = topics_json["Topics"]
topics.should.have.length_of(0)
@mock_sns
def test_create_topic_should_be_indempodent():

132
tests/test_sqs/test_sqs.py
View file

@ -40,6 +40,29 @@ def test_create_fifo_queue_fail():
raise RuntimeError('Should of raised InvalidParameterValue Exception')
@mock_sqs
def test_create_queue_with_different_attributes_fail():
sqs = boto3.client('sqs', region_name='us-east-1')
sqs.create_queue(
QueueName='test-queue',
Attributes={
'VisibilityTimeout': '10',
}
)
try:
sqs.create_queue(
QueueName='test-queue',
Attributes={
'VisibilityTimeout': '60',
}
)
except botocore.exceptions.ClientError as err:
err.response['Error']['Code'].should.equal('QueueAlreadyExists')
else:
raise RuntimeError('Should of raised QueueAlreadyExists Exception')
@mock_sqs
def test_create_fifo_queue():
sqs = boto3.client('sqs', region_name='us-east-1')
@ -1063,3 +1086,112 @@ def test_redrive_policy_set_attributes():
assert 'RedrivePolicy' in copy.attributes
copy_policy = json.loads(copy.attributes['RedrivePolicy'])
assert copy_policy == redrive_policy
@mock_sqs
def test_receive_messages_with_message_group_id():
sqs = boto3.resource('sqs', region_name='us-east-1')
queue = sqs.create_queue(QueueName="test-queue.fifo",
Attributes={
'FifoQueue': 'true',
})
queue.set_attributes(Attributes={"VisibilityTimeout": "3600"})
queue.send_message(
MessageBody="message-1",
MessageGroupId="group"
)
queue.send_message(
MessageBody="message-2",
MessageGroupId="group"
)
messages = queue.receive_messages()
messages.should.have.length_of(1)
message = messages[0]
# received message is not deleted!
messages = queue.receive_messages(WaitTimeSeconds=0)
messages.should.have.length_of(0)
# message is now processed, next one should be available
message.delete()
messages = queue.receive_messages()
messages.should.have.length_of(1)
@mock_sqs
def test_receive_messages_with_message_group_id_on_requeue():
sqs = boto3.resource('sqs', region_name='us-east-1')
queue = sqs.create_queue(QueueName="test-queue.fifo",
Attributes={
'FifoQueue': 'true',
})
queue.set_attributes(Attributes={"VisibilityTimeout": "3600"})
queue.send_message(
MessageBody="message-1",
MessageGroupId="group"
)
queue.send_message(
MessageBody="message-2",
MessageGroupId="group"
)
messages = queue.receive_messages()
messages.should.have.length_of(1)
message = messages[0]
# received message is not deleted!
messages = queue.receive_messages(WaitTimeSeconds=0)
messages.should.have.length_of(0)
# message is now available again, next one should be available
message.change_visibility(VisibilityTimeout=0)
messages = queue.receive_messages()
messages.should.have.length_of(1)
messages[0].message_id.should.equal(message.message_id)
@mock_sqs
def test_receive_messages_with_message_group_id_on_visibility_timeout():
if os.environ.get('TEST_SERVER_MODE', 'false').lower() == 'true':
raise SkipTest('Cant manipulate time in server mode')
with freeze_time("2015-01-01 12:00:00"):
sqs = boto3.resource('sqs', region_name='us-east-1')
queue = sqs.create_queue(QueueName="test-queue.fifo",
Attributes={
'FifoQueue': 'true',
})
queue.set_attributes(Attributes={"VisibilityTimeout": "3600"})
queue.send_message(
MessageBody="message-1",
MessageGroupId="group"
)
queue.send_message(
MessageBody="message-2",
MessageGroupId="group"
)
messages = queue.receive_messages()
messages.should.have.length_of(1)
message = messages[0]
# received message is not deleted!
messages = queue.receive_messages(WaitTimeSeconds=0)
messages.should.have.length_of(0)
message.change_visibility(VisibilityTimeout=10)
with freeze_time("2015-01-01 12:00:05"):
# no timeout yet
messages = queue.receive_messages(WaitTimeSeconds=0)
messages.should.have.length_of(0)
with freeze_time("2015-01-01 12:00:15"):
# message is now available again, next one should be available
messages = queue.receive_messages()
messages.should.have.length_of(1)
messages[0].message_id.should.equal(message.message_id)