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

Merge branch 'master' of https://github.com/spulec/moto into spulec-master

Browse source
This commit is contained in:
Stephan Huber 2018-10-16 15:29:56 +02:00
commit 0ba213ffcc
76 changed files with 7929 additions and 4971 deletions
Download patch file Download diff file Expand all files Collapse all files

14
tests/test_cognitoidentity/test_cognitoidentity.py
View file

@ -31,6 +31,7 @@ def test_create_identity_pool():
# testing a helper function
def test_get_random_identity_id():
assert len(get_random_identity_id('us-west-2')) > 0
assert len(get_random_identity_id('us-west-2').split(':')[1]) == 19
@mock_cognitoidentity
@ -69,3 +70,16 @@ def test_get_open_id_token_for_developer_identity():
)
assert len(result['Token'])
assert result['IdentityId'] == '12345'
@mock_cognitoidentity
def test_get_open_id_token_for_developer_identity_when_no_explicit_identity_id():
conn = boto3.client('cognito-identity', 'us-west-2')
result = conn.get_open_id_token_for_developer_identity(
IdentityPoolId='us-west-2:12345',
Logins={
'someurl': '12345'
},
TokenDuration=123
)
assert len(result['Token']) > 0
assert len(result['IdentityId']) > 0

14
tests/test_cognitoidp/test_cognitoidp.py
View file

@ -6,6 +6,7 @@ import os
import uuid
from jose import jws
from moto import mock_cognitoidp
import sure # noqa
@ -24,6 +25,7 @@ def test_create_user_pool():
)
result["UserPool"]["Id"].should_not.be.none
result["UserPool"]["Id"].should.match(r'[\w-]+_[0-9a-zA-Z]+')
result["UserPool"]["Name"].should.equal(name)
result["UserPool"]["LambdaConfig"]["PreSignUp"].should.equal(value)
@ -399,15 +401,22 @@ 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"]
user_attribute_name = str(uuid.uuid4())
user_attribute_value = str(uuid.uuid4())
client_id = conn.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName=str(uuid.uuid4()),
ReadAttributes=[user_attribute_name]
)["UserPoolClient"]["ClientId"]
conn.admin_create_user(
UserPoolId=user_pool_id,
Username=username,
TemporaryPassword=temporary_password,
UserAttributes=[{
'Name': user_attribute_name,
'Value': user_attribute_value
}]
)
result = conn.admin_initiate_auth(
@ -446,6 +455,9 @@ def authentication_flow(conn):
"access_token": result["AuthenticationResult"]["AccessToken"],
"username": username,
"password": new_password,
"additional_fields": {
user_attribute_name: user_attribute_value
}
}
@ -475,6 +487,8 @@ def test_token_legitimacy():
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)
for k, v in outputs["additional_fields"].items():
access_claims[k].should.equal(v)
@mock_cognitoidp

11
tests/test_core/test_decorator_calls.py
View file

@ -85,3 +85,14 @@ class TesterWithSetup(unittest.TestCase):
def test_still_the_same(self):
bucket = self.conn.get_bucket('mybucket')
bucket.name.should.equal("mybucket")
@mock_s3_deprecated
class TesterWithStaticmethod(object):
@staticmethod
def static(*args):
assert not args or not isinstance(args[0], TesterWithStaticmethod)
def test_no_instance_sent_to_staticmethod(self):
self.static()

52
tests/test_dynamodb2/test_dynamodb.py
View file

@ -201,6 +201,48 @@ def test_item_add_empty_string_exception():
)
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_update_item_with_empty_string_exception():
name = 'TestTable'
conn = boto3.client('dynamodb',
region_name='us-west-2',
aws_access_key_id="ak",
aws_secret_access_key="sk")
conn.create_table(TableName=name,
KeySchema=[{'AttributeName':'forum_name','KeyType':'HASH'}],
AttributeDefinitions=[{'AttributeName':'forum_name','AttributeType':'S'}],
ProvisionedThroughput={'ReadCapacityUnits':5,'WriteCapacityUnits':5})
conn.put_item(
TableName=name,
Item={
'forum_name': { 'S': 'LOLCat Forum' },
'subject': { 'S': 'Check this out!' },
'Body': { 'S': 'http://url_to_lolcat.gif'},
'SentBy': { 'S': "test" },
'ReceivedTime': { 'S': '12/9/2011 11:36:03 PM'},
}
)
with assert_raises(ClientError) as ex:
conn.update_item(
TableName=name,
Key={
'forum_name': { 'S': 'LOLCat Forum'},
},
UpdateExpression='set Body=:Body',
ExpressionAttributeValues={
':Body': {'S': ''}
})
ex.exception.response['Error']['Code'].should.equal('ValidationException')
ex.exception.response['ResponseMetadata']['HTTPStatusCode'].should.equal(400)
ex.exception.response['Error']['Message'].should.equal(
'One or more parameter values were invalid: An AttributeValue may not contain an empty string'
)
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_query_invalid_table():
@ -658,8 +700,8 @@ def test_filter_expression():
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('Id IN :v0', {}, {':v0': {'NS': [7, 8, 9]}})
filter_expr.expr(row1).should.be(True)
# attribute function tests
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('attribute_exists(Id) AND attribute_not_exists(User)', {}, {})
# attribute function tests (with extra spaces)
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('attribute_exists(Id) AND attribute_not_exists (User)', {}, {})
filter_expr.expr(row1).should.be(True)
filter_expr = moto.dynamodb2.comparisons.get_filter_expression('attribute_type(Id, N)', {}, {})
@ -1178,7 +1220,8 @@ def test_update_if_not_exists():
'forum_name': 'the-key',
'subject': '123'
},
UpdateExpression='SET created_at = if_not_exists(created_at, :created_at)',
# if_not_exists without space
UpdateExpression='SET created_at=if_not_exists(created_at,:created_at)',
ExpressionAttributeValues={
':created_at': 123
}
@ -1191,7 +1234,8 @@ def test_update_if_not_exists():
'forum_name': 'the-key',
'subject': '123'
},
UpdateExpression='SET created_at = if_not_exists(created_at, :created_at)',
# if_not_exists with space
UpdateExpression='SET created_at = if_not_exists (created_at, :created_at)',
ExpressionAttributeValues={
':created_at': 456
}

4
tests/test_ec2/test_elastic_block_store.py
View file

@ -615,8 +615,8 @@ def test_copy_snapshot():
dest = dest_ec2.Snapshot(copy_snapshot_response['SnapshotId'])
attribs = ['data_encryption_key_id', 'encrypted',
'kms_key_id', 'owner_alias', 'owner_id', 'progress',
'start_time', 'state', 'state_message',
'kms_key_id', 'owner_alias', 'owner_id',
'progress', 'state', 'state_message',
'tags', 'volume_id', 'volume_size']
for attrib in attribs:

39
tests/test_ec2/test_vpc_peering.py
View file

@ -2,12 +2,15 @@ from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
from moto.ec2.exceptions import EC2ClientError
from botocore.exceptions import ClientError
import boto3
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from moto import mock_ec2_deprecated
from moto import mock_ec2, mock_ec2_deprecated
from tests.helpers import requires_boto_gte
@ -93,3 +96,37 @@ def test_vpc_peering_connections_delete():
cm.exception.code.should.equal('InvalidVpcPeeringConnectionId.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_vpc_peering_connections_cross_region():
# create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource('ec2', region_name='us-west-1')
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock='10.90.0.0/16')
ec2_apn1 = boto3.resource('ec2', region_name='ap-northeast-1')
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock='10.20.0.0/16')
# create peering
vpc_pcx = ec2_usw1.create_vpc_peering_connection(
VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion='ap-northeast-1',
)
vpc_pcx.status['Code'].should.equal('initiating-request')
vpc_pcx.requester_vpc.id.should.equal(vpc_usw1.id)
vpc_pcx.accepter_vpc.id.should.equal(vpc_apn1.id)
@mock_ec2
def test_vpc_peering_connections_cross_region_fail():
# create vpc in us-west-1 and ap-northeast-1
ec2_usw1 = boto3.resource('ec2', region_name='us-west-1')
vpc_usw1 = ec2_usw1.create_vpc(CidrBlock='10.90.0.0/16')
ec2_apn1 = boto3.resource('ec2', region_name='ap-northeast-1')
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock='10.20.0.0/16')
# create peering wrong region with no vpc
with assert_raises(ClientError) as cm:
ec2_usw1.create_vpc_peering_connection(
VpcId=vpc_usw1.id,
PeerVpcId=vpc_apn1.id,
PeerRegion='ap-northeast-2')
cm.exception.response['Error']['Code'].should.equal('InvalidVpcID.NotFound')

115
tests/test_ecs/test_ecs_boto3.py
View file

@ -304,6 +304,52 @@ def test_create_service():
response['service']['status'].should.equal('ACTIVE')
response['service']['taskDefinition'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:task-definition/test_ecs_task:1')
response['service']['schedulingStrategy'].should.equal('REPLICA')
@mock_ecs
def test_create_service_scheduling_strategy():
client = boto3.client('ecs', region_name='us-east-1')
_ = client.create_cluster(
clusterName='test_ecs_cluster'
)
_ = client.register_task_definition(
family='test_ecs_task',
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'}
}
]
)
response = client.create_service(
cluster='test_ecs_cluster',
serviceName='test_ecs_service',
taskDefinition='test_ecs_task',
desiredCount=2,
schedulingStrategy='DAEMON',
)
response['service']['clusterArn'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:cluster/test_ecs_cluster')
response['service']['desiredCount'].should.equal(2)
len(response['service']['events']).should.equal(0)
len(response['service']['loadBalancers']).should.equal(0)
response['service']['pendingCount'].should.equal(0)
response['service']['runningCount'].should.equal(0)
response['service']['serviceArn'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:service/test_ecs_service')
response['service']['serviceName'].should.equal('test_ecs_service')
response['service']['status'].should.equal('ACTIVE')
response['service']['taskDefinition'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:task-definition/test_ecs_task:1')
response['service']['schedulingStrategy'].should.equal('DAEMON')
@mock_ecs
@ -411,6 +457,72 @@ def test_describe_services():
response['services'][0]['deployments'][0]['status'].should.equal('PRIMARY')
@mock_ecs
def test_describe_services_scheduling_strategy():
client = boto3.client('ecs', region_name='us-east-1')
_ = client.create_cluster(
clusterName='test_ecs_cluster'
)
_ = client.register_task_definition(
family='test_ecs_task',
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.create_service(
cluster='test_ecs_cluster',
serviceName='test_ecs_service1',
taskDefinition='test_ecs_task',
desiredCount=2
)
_ = client.create_service(
cluster='test_ecs_cluster',
serviceName='test_ecs_service2',
taskDefinition='test_ecs_task',
desiredCount=2,
schedulingStrategy='DAEMON'
)
_ = client.create_service(
cluster='test_ecs_cluster',
serviceName='test_ecs_service3',
taskDefinition='test_ecs_task',
desiredCount=2
)
response = client.describe_services(
cluster='test_ecs_cluster',
services=['test_ecs_service1',
'arn:aws:ecs:us-east-1:012345678910:service/test_ecs_service2',
'test_ecs_service3']
)
len(response['services']).should.equal(3)
response['services'][0]['serviceArn'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:service/test_ecs_service1')
response['services'][0]['serviceName'].should.equal('test_ecs_service1')
response['services'][1]['serviceArn'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:service/test_ecs_service2')
response['services'][1]['serviceName'].should.equal('test_ecs_service2')
response['services'][0]['deployments'][0]['desiredCount'].should.equal(2)
response['services'][0]['deployments'][0]['pendingCount'].should.equal(2)
response['services'][0]['deployments'][0]['runningCount'].should.equal(0)
response['services'][0]['deployments'][0]['status'].should.equal('PRIMARY')
response['services'][0]['schedulingStrategy'].should.equal('REPLICA')
response['services'][1]['schedulingStrategy'].should.equal('DAEMON')
response['services'][2]['schedulingStrategy'].should.equal('REPLICA')
@mock_ecs
def test_update_service():
client = boto3.client('ecs', region_name='us-east-1')
@ -449,6 +561,7 @@ def test_update_service():
desiredCount=0
)
response['service']['desiredCount'].should.equal(0)
response['service']['schedulingStrategy'].should.equal('REPLICA')
@mock_ecs
@ -515,8 +628,10 @@ def test_delete_service():
'arn:aws:ecs:us-east-1:012345678910:service/test_ecs_service')
response['service']['serviceName'].should.equal('test_ecs_service')
response['service']['status'].should.equal('ACTIVE')
response['service']['schedulingStrategy'].should.equal('REPLICA')
response['service']['taskDefinition'].should.equal(
'arn:aws:ecs:us-east-1:012345678910:task-definition/test_ecs_task:1')
@mock_ec2

34
tests/test_elb/test_elb.py
View file

@ -723,6 +723,40 @@ def test_describe_instance_health():
instances_health[0].state.should.equal('InService')
@mock_ec2
@mock_elb
def test_describe_instance_health_boto3():
elb = boto3.client('elb', region_name="us-east-1")
ec2 = boto3.client('ec2', region_name="us-east-1")
instances = ec2.run_instances(MinCount=2, MaxCount=2)['Instances']
lb_name = "my_load_balancer"
elb.create_load_balancer(
Listeners=[{
'InstancePort': 80,
'LoadBalancerPort': 8080,
'Protocol': 'HTTP'
}],
LoadBalancerName=lb_name,
)
elb.register_instances_with_load_balancer(
LoadBalancerName=lb_name,
Instances=[{'InstanceId': instances[0]['InstanceId']}]
)
instances_health = elb.describe_instance_health(
LoadBalancerName=lb_name,
Instances=[{'InstanceId': instance['InstanceId']} for instance in instances]
)
instances_health['InstanceStates'].should.have.length_of(2)
instances_health['InstanceStates'][0]['InstanceId'].\
should.equal(instances[0]['InstanceId'])
instances_health['InstanceStates'][0]['State'].\
should.equal('InService')
instances_health['InstanceStates'][1]['InstanceId'].\
should.equal(instances[1]['InstanceId'])
instances_health['InstanceStates'][1]['State'].\
should.equal('Unknown')
@mock_elb
def test_add_remove_tags():
client = boto3.client('elb', region_name='us-east-1')

25
tests/test_glue/fixtures/datacatalog.py
View file

@ -29,3 +29,28 @@ TABLE_INPUT = {
},
'TableType': 'EXTERNAL_TABLE',
}
PARTITION_INPUT = {
# 'DatabaseName': 'dbname',
'StorageDescriptor': {
'BucketColumns': [],
'Columns': [],
'Compressed': False,
'InputFormat': 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat',
'Location': 's3://.../partition=value',
'NumberOfBuckets': -1,
'OutputFormat': 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat',
'Parameters': {},
'SerdeInfo': {
'Parameters': {'path': 's3://...', 'serialization.format': '1'},
'SerializationLibrary': 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'},
'SkewedInfo': {'SkewedColumnNames': [],
'SkewedColumnValueLocationMaps': {},
'SkewedColumnValues': []},
'SortColumns': [],
'StoredAsSubDirectories': False,
},
# 'TableName': 'source_table',
# 'Values': ['2018-06-26'],
}

81
tests/test_glue/helpers.py
View file

@ -2,7 +2,7 @@ from __future__ import unicode_literals
import copy
from .fixtures.datacatalog import TABLE_INPUT
from .fixtures.datacatalog import TABLE_INPUT, PARTITION_INPUT
def create_database(client, database_name):
@ -17,22 +17,38 @@ def get_database(client, database_name):
return client.get_database(Name=database_name)
def create_table_input(table_name, s3_location, columns=[], partition_keys=[]):
def create_table_input(database_name, table_name, columns=[], partition_keys=[]):
table_input = copy.deepcopy(TABLE_INPUT)
table_input['Name'] = table_name
table_input['PartitionKeys'] = partition_keys
table_input['StorageDescriptor']['Columns'] = columns
table_input['StorageDescriptor']['Location'] = s3_location
table_input['StorageDescriptor']['Location'] = 's3://my-bucket/{database_name}/{table_name}'.format(
database_name=database_name,
table_name=table_name
)
return table_input
def create_table(client, database_name, table_name, table_input):
def create_table(client, database_name, table_name, table_input=None, **kwargs):
if table_input is None:
table_input = create_table_input(database_name, table_name, **kwargs)
return client.create_table(
DatabaseName=database_name,
TableInput=table_input
)
def update_table(client, database_name, table_name, table_input=None, **kwargs):
if table_input is None:
table_input = create_table_input(database_name, table_name, **kwargs)
return client.update_table(
DatabaseName=database_name,
TableInput=table_input,
)
def get_table(client, database_name, table_name):
return client.get_table(
DatabaseName=database_name,
@ -44,3 +60,60 @@ def get_tables(client, database_name):
return client.get_tables(
DatabaseName=database_name
)
def get_table_versions(client, database_name, table_name):
return client.get_table_versions(
DatabaseName=database_name,
TableName=table_name
)
def get_table_version(client, database_name, table_name, version_id):
return client.get_table_version(
DatabaseName=database_name,
TableName=table_name,
VersionId=version_id,
)
def create_partition_input(database_name, table_name, values=[], columns=[]):
root_path = 's3://my-bucket/{database_name}/{table_name}'.format(
database_name=database_name,
table_name=table_name
)
part_input = copy.deepcopy(PARTITION_INPUT)
part_input['Values'] = values
part_input['StorageDescriptor']['Columns'] = columns
part_input['StorageDescriptor']['SerdeInfo']['Parameters']['path'] = root_path
return part_input
def create_partition(client, database_name, table_name, partiton_input=None, **kwargs):
if partiton_input is None:
partiton_input = create_partition_input(database_name, table_name, **kwargs)
return client.create_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionInput=partiton_input
)
def update_partition(client, database_name, table_name, old_values=[], partiton_input=None, **kwargs):
if partiton_input is None:
partiton_input = create_partition_input(database_name, table_name, **kwargs)
return client.update_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionInput=partiton_input,
PartitionValueList=old_values,
)
def get_partition(client, database_name, table_name, values):
return client.get_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionValues=values,
)

362
tests/test_glue/test_datacatalog.py
View file

@ -1,10 +1,15 @@
from __future__ import unicode_literals
import sure # noqa
import re
from nose.tools import assert_raises
import boto3
from botocore.client import ClientError
from datetime import datetime
import pytz
from moto import mock_glue
from . import helpers
@ -30,7 +35,19 @@ def test_create_database_already_exists():
with assert_raises(ClientError) as exc:
helpers.create_database(client, database_name)
exc.exception.response['Error']['Code'].should.equal('DatabaseAlreadyExistsException')
exc.exception.response['Error']['Code'].should.equal('AlreadyExistsException')
@mock_glue
def test_get_database_not_exits():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'nosuchdatabase'
with assert_raises(ClientError) as exc:
helpers.get_database(client, database_name)
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('Database nosuchdatabase not found')
@mock_glue
@ -40,12 +57,7 @@ def test_create_table():
helpers.create_database(client, database_name)
table_name = 'myspecialtable'
s3_location = 's3://my-bucket/{database_name}/{table_name}'.format(
database_name=database_name,
table_name=table_name
)
table_input = helpers.create_table_input(table_name, s3_location)
table_input = helpers.create_table_input(database_name, table_name)
helpers.create_table(client, database_name, table_name, table_input)
response = helpers.get_table(client, database_name, table_name)
@ -63,18 +75,12 @@ def test_create_table_already_exists():
helpers.create_database(client, database_name)
table_name = 'cantcreatethistabletwice'
s3_location = 's3://my-bucket/{database_name}/{table_name}'.format(
database_name=database_name,
table_name=table_name
)
table_input = helpers.create_table_input(table_name, s3_location)
helpers.create_table(client, database_name, table_name, table_input)
helpers.create_table(client, database_name, table_name)
with assert_raises(ClientError) as exc:
helpers.create_table(client, database_name, table_name, table_input)
helpers.create_table(client, database_name, table_name)
exc.exception.response['Error']['Code'].should.equal('TableAlreadyExistsException')
exc.exception.response['Error']['Code'].should.equal('AlreadyExistsException')
@mock_glue
@ -87,11 +93,7 @@ def test_get_tables():
table_inputs = {}
for table_name in table_names:
s3_location = 's3://my-bucket/{database_name}/{table_name}'.format(
database_name=database_name,
table_name=table_name
)
table_input = helpers.create_table_input(table_name, s3_location)
table_input = helpers.create_table_input(database_name, table_name)
table_inputs[table_name] = table_input
helpers.create_table(client, database_name, table_name, table_input)
@ -99,10 +101,326 @@ def test_get_tables():
tables = response['TableList']
assert len(tables) == 3
tables.should.have.length_of(3)
for table in tables:
table_name = table['Name']
table_name.should.equal(table_inputs[table_name]['Name'])
table['StorageDescriptor'].should.equal(table_inputs[table_name]['StorageDescriptor'])
table['PartitionKeys'].should.equal(table_inputs[table_name]['PartitionKeys'])
@mock_glue
def test_get_table_versions():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
helpers.create_database(client, database_name)
table_name = 'myfirsttable'
version_inputs = {}
table_input = helpers.create_table_input(database_name, table_name)
helpers.create_table(client, database_name, table_name, table_input)
version_inputs["1"] = table_input
columns = [{'Name': 'country', 'Type': 'string'}]
table_input = helpers.create_table_input(database_name, table_name, columns=columns)
helpers.update_table(client, database_name, table_name, table_input)
version_inputs["2"] = table_input
# Updateing with an indentical input should still create a new version
helpers.update_table(client, database_name, table_name, table_input)
version_inputs["3"] = table_input
response = helpers.get_table_versions(client, database_name, table_name)
vers = response['TableVersions']
vers.should.have.length_of(3)
vers[0]['Table']['StorageDescriptor']['Columns'].should.equal([])
vers[-1]['Table']['StorageDescriptor']['Columns'].should.equal(columns)
for n, ver in enumerate(vers):
n = str(n + 1)
ver['VersionId'].should.equal(n)
ver['Table']['Name'].should.equal(table_name)
ver['Table']['StorageDescriptor'].should.equal(version_inputs[n]['StorageDescriptor'])
ver['Table']['PartitionKeys'].should.equal(version_inputs[n]['PartitionKeys'])
response = helpers.get_table_version(client, database_name, table_name, "3")
ver = response['TableVersion']
ver['VersionId'].should.equal("3")
ver['Table']['Name'].should.equal(table_name)
ver['Table']['StorageDescriptor']['Columns'].should.equal(columns)
@mock_glue
def test_get_table_version_not_found():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
with assert_raises(ClientError) as exc:
helpers.get_table_version(client, database_name, 'myfirsttable', "20")
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('version', re.I)
@mock_glue
def test_get_table_version_invalid_input():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
with assert_raises(ClientError) as exc:
helpers.get_table_version(client, database_name, 'myfirsttable', "10not-an-int")
exc.exception.response['Error']['Code'].should.equal('InvalidInputException')
@mock_glue
def test_get_table_not_exits():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
helpers.create_database(client, database_name)
with assert_raises(ClientError) as exc:
helpers.get_table(client, database_name, 'myfirsttable')
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('Table myfirsttable not found')
@mock_glue
def test_get_table_when_database_not_exits():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'nosuchdatabase'
with assert_raises(ClientError) as exc:
helpers.get_table(client, database_name, 'myfirsttable')
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('Database nosuchdatabase not found')
@mock_glue
def test_get_partitions_empty():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
response = client.get_partitions(DatabaseName=database_name, TableName=table_name)
response['Partitions'].should.have.length_of(0)
@mock_glue
def test_create_partition():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
values = ['2018-10-01']
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
before = datetime.now(pytz.utc)
part_input = helpers.create_partition_input(database_name, table_name, values=values)
helpers.create_partition(client, database_name, table_name, part_input)
after = datetime.now(pytz.utc)
response = client.get_partitions(DatabaseName=database_name, TableName=table_name)
partitions = response['Partitions']
partitions.should.have.length_of(1)
partition = partitions[0]
partition['TableName'].should.equal(table_name)
partition['StorageDescriptor'].should.equal(part_input['StorageDescriptor'])
partition['Values'].should.equal(values)
partition['CreationTime'].should.be.greater_than(before)
partition['CreationTime'].should.be.lower_than(after)
@mock_glue
def test_create_partition_already_exist():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
values = ['2018-10-01']
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
helpers.create_partition(client, database_name, table_name, values=values)
with assert_raises(ClientError) as exc:
helpers.create_partition(client, database_name, table_name, values=values)
exc.exception.response['Error']['Code'].should.equal('AlreadyExistsException')
@mock_glue
def test_get_partition_not_found():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
values = ['2018-10-01']
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
with assert_raises(ClientError) as exc:
helpers.get_partition(client, database_name, table_name, values)
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('partition')
@mock_glue
def test_get_partition():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
values = [['2018-10-01'], ['2018-09-01']]
helpers.create_partition(client, database_name, table_name, values=values[0])
helpers.create_partition(client, database_name, table_name, values=values[1])
response = client.get_partition(DatabaseName=database_name, TableName=table_name, PartitionValues=values[1])
partition = response['Partition']
partition['TableName'].should.equal(table_name)
partition['Values'].should.equal(values[1])
@mock_glue
def test_update_partition_not_found_moving():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
with assert_raises(ClientError) as exc:
helpers.update_partition(client, database_name, table_name, old_values=['0000-00-00'], values=['2018-10-02'])
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('partition')
@mock_glue
def test_update_partition_not_found_change_in_place():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
values = ['2018-10-01']
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
with assert_raises(ClientError) as exc:
helpers.update_partition(client, database_name, table_name, old_values=values, values=values)
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('partition')
@mock_glue
def test_update_partition_cannot_overwrite():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
values = [['2018-10-01'], ['2018-09-01']]
helpers.create_partition(client, database_name, table_name, values=values[0])
helpers.create_partition(client, database_name, table_name, values=values[1])
with assert_raises(ClientError) as exc:
helpers.update_partition(client, database_name, table_name, old_values=values[0], values=values[1])
exc.exception.response['Error']['Code'].should.equal('AlreadyExistsException')
@mock_glue
def test_update_partition():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
values = ['2018-10-01']
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
helpers.create_partition(client, database_name, table_name, values=values)
response = helpers.update_partition(
client,
database_name,
table_name,
old_values=values,
values=values,
columns=[{'Name': 'country', 'Type': 'string'}],
)
response = client.get_partition(DatabaseName=database_name, TableName=table_name, PartitionValues=values)
partition = response['Partition']
partition['TableName'].should.equal(table_name)
partition['StorageDescriptor']['Columns'].should.equal([{'Name': 'country', 'Type': 'string'}])
@mock_glue
def test_update_partition_move():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
table_name = 'myfirsttable'
values = ['2018-10-01']
new_values = ['2018-09-01']
helpers.create_database(client, database_name)
helpers.create_table(client, database_name, table_name)
helpers.create_partition(client, database_name, table_name, values=values)
response = helpers.update_partition(
client,
database_name,
table_name,
old_values=values,
values=new_values,
columns=[{'Name': 'country', 'Type': 'string'}],
)
with assert_raises(ClientError) as exc:
helpers.get_partition(client, database_name, table_name, values)
# Old partition shouldn't exist anymore
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
response = client.get_partition(DatabaseName=database_name, TableName=table_name, PartitionValues=new_values)
partition = response['Partition']
partition['TableName'].should.equal(table_name)
partition['StorageDescriptor']['Columns'].should.equal([{'Name': 'country', 'Type': 'string'}])

31
tests/test_iam/test_iam.py
View file

@ -286,6 +286,16 @@ def test_create_policy_versions():
PolicyDocument='{"some":"policy"}')
version.get('PolicyVersion').get('Document').should.equal({'some': 'policy'})
@mock_iam
def test_get_policy():
conn = boto3.client('iam', region_name='us-east-1')
response = conn.create_policy(
PolicyName="TestGetPolicy",
PolicyDocument='{"some":"policy"}')
policy = conn.get_policy(
PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicy")
response['Policy']['Arn'].should.equal("arn:aws:iam::123456789012:policy/TestGetPolicy")
@mock_iam
def test_get_policy_version():
@ -314,17 +324,22 @@ def test_list_policy_versions():
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions")
conn.create_policy(
PolicyName="TestListPolicyVersions",
PolicyDocument='{"some":"policy"}')
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions",
PolicyDocument='{"first":"policy"}')
versions = conn.list_policy_versions(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions")
versions.get('Versions')[0].get('VersionId').should.equal('v1')
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions",
PolicyDocument='{"second":"policy"}')
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions",
PolicyDocument='{"third":"policy"}')
versions = conn.list_policy_versions(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions")
versions.get('Versions')[0].get('Document').should.equal({'first': 'policy'})
print(versions.get('Versions'))
versions.get('Versions')[1].get('Document').should.equal({'second': 'policy'})
versions.get('Versions')[2].get('Document').should.equal({'third': 'policy'})
@mock_iam
@ -332,20 +347,20 @@ def test_delete_policy_version():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_policy(
PolicyName="TestDeletePolicyVersion",
PolicyDocument='{"some":"policy"}')
PolicyDocument='{"first":"policy"}')
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
PolicyDocument='{"first":"policy"}')
PolicyDocument='{"second":"policy"}')
with assert_raises(ClientError):
conn.delete_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
VersionId='v2-nope-this-does-not-exist')
conn.delete_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
VersionId='v1')
VersionId='v2')
versions = conn.list_policy_versions(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion")
len(versions.get('Versions')).should.equal(0)
len(versions.get('Versions')).should.equal(1)
@mock_iam_deprecated()

102
tests/test_kms/test_kms.py
View file

@ -1,5 +1,5 @@
from __future__ import unicode_literals
import re
import os, re
import boto3
import boto.kms
@ -8,6 +8,9 @@ from boto.kms.exceptions import AlreadyExistsException, NotFoundException
import sure # noqa
from moto import mock_kms, mock_kms_deprecated
from nose.tools import assert_raises
from freezegun import freeze_time
from datetime import datetime, timedelta
from dateutil.tz import tzlocal
@mock_kms_deprecated
@ -617,3 +620,100 @@ def test_kms_encrypt_boto3():
response = client.decrypt(CiphertextBlob=response['CiphertextBlob'])
response['Plaintext'].should.equal(b'bar')
@mock_kms
def test_disable_key():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='disable-key')
client.disable_key(
KeyId=key['KeyMetadata']['KeyId']
)
result = client.describe_key(KeyId=key['KeyMetadata']['KeyId'])
assert result["KeyMetadata"]["Enabled"] == False
assert result["KeyMetadata"]["KeyState"] == 'Disabled'
@mock_kms
def test_enable_key():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='enable-key')
client.disable_key(
KeyId=key['KeyMetadata']['KeyId']
)
client.enable_key(
KeyId=key['KeyMetadata']['KeyId']
)
result = client.describe_key(KeyId=key['KeyMetadata']['KeyId'])
assert result["KeyMetadata"]["Enabled"] == True
assert result["KeyMetadata"]["KeyState"] == 'Enabled'
@mock_kms
def test_schedule_key_deletion():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='schedule-key-deletion')
if os.environ.get('TEST_SERVER_MODE', 'false').lower() == 'false':
with freeze_time("2015-01-01 12:00:00"):
response = client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId']
)
assert response['KeyId'] == key['KeyMetadata']['KeyId']
assert response['DeletionDate'] == datetime(2015, 1, 31, 12, 0, tzinfo=tzlocal())
else:
# Can't manipulate time in server mode
response = client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId']
)
assert response['KeyId'] == key['KeyMetadata']['KeyId']
result = client.describe_key(KeyId=key['KeyMetadata']['KeyId'])
assert result["KeyMetadata"]["Enabled"] == False
assert result["KeyMetadata"]["KeyState"] == 'PendingDeletion'
assert 'DeletionDate' in result["KeyMetadata"]
@mock_kms
def test_schedule_key_deletion_custom():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='schedule-key-deletion')
if os.environ.get('TEST_SERVER_MODE', 'false').lower() == 'false':
with freeze_time("2015-01-01 12:00:00"):
response = client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId'],
PendingWindowInDays=7
)
assert response['KeyId'] == key['KeyMetadata']['KeyId']
assert response['DeletionDate'] == datetime(2015, 1, 8, 12, 0, tzinfo=tzlocal())
else:
# Can't manipulate time in server mode
response = client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId'],
PendingWindowInDays=7
)
assert response['KeyId'] == key['KeyMetadata']['KeyId']
result = client.describe_key(KeyId=key['KeyMetadata']['KeyId'])
assert result["KeyMetadata"]["Enabled"] == False
assert result["KeyMetadata"]["KeyState"] == 'PendingDeletion'
assert 'DeletionDate' in result["KeyMetadata"]
@mock_kms
def test_cancel_key_deletion():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='cancel-key-deletion')
client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId']
)
response = client.cancel_key_deletion(
KeyId=key['KeyMetadata']['KeyId']
)
assert response['KeyId'] == key['KeyMetadata']['KeyId']
result = client.describe_key(KeyId=key['KeyMetadata']['KeyId'])
assert result["KeyMetadata"]["Enabled"] == False
assert result["KeyMetadata"]["KeyState"] == 'Disabled'
assert 'DeletionDate' not in result["KeyMetadata"]

14
tests/test_logs/test_logs.py
View file

@ -1,5 +1,6 @@
import boto3
import sure # noqa
import six
from botocore.exceptions import ClientError
from moto import mock_logs, settings
@ -47,7 +48,7 @@ def test_exceptions():
logEvents=[
{
'timestamp': 0,
'message': 'line'
'message': 'line'
},
],
)
@ -79,7 +80,7 @@ def test_put_logs():
{'timestamp': 0, 'message': 'hello'},
{'timestamp': 0, 'message': 'world'}
]
conn.put_log_events(
putRes = conn.put_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
logEvents=messages
@ -89,6 +90,9 @@ def test_put_logs():
logStreamName=log_stream_name
)
events = res['events']
nextSequenceToken = putRes['nextSequenceToken']
assert isinstance(nextSequenceToken, six.string_types) == True
assert len(nextSequenceToken) == 56
events.should.have.length_of(2)
@ -117,4 +121,8 @@ def test_filter_logs_interleaved():
interleaved=True,
)
events = res['events']
events.should.have.length_of(2)
for original_message, resulting_event in zip(messages, events):
resulting_event['eventId'].should.equal(str(resulting_event['eventId']))
resulting_event['timestamp'].should.equal(original_message['timestamp'])
resulting_event['message'].should.equal(original_message['message'])

0
tests/test_organizations/__init__.py Normal file
View file

136
tests/test_organizations/organizations_test_utils.py Normal file
View file

@ -0,0 +1,136 @@
from __future__ import unicode_literals
import six
import sure # noqa
import datetime
from moto.organizations import utils
EMAIL_REGEX = "^.+@[a-zA-Z0-9-.]+.[a-zA-Z]{2,3}|[0-9]{1,3}$"
ORG_ID_REGEX = r'o-[a-z0-9]{%s}' % utils.ORG_ID_SIZE
ROOT_ID_REGEX = r'r-[a-z0-9]{%s}' % utils.ROOT_ID_SIZE
OU_ID_REGEX = r'ou-[a-z0-9]{%s}-[a-z0-9]{%s}' % (utils.ROOT_ID_SIZE, utils.OU_ID_SUFFIX_SIZE)
ACCOUNT_ID_REGEX = r'[0-9]{%s}' % utils.ACCOUNT_ID_SIZE
CREATE_ACCOUNT_STATUS_ID_REGEX = r'car-[a-z0-9]{%s}' % utils.CREATE_ACCOUNT_STATUS_ID_SIZE
def test_make_random_org_id():
org_id = utils.make_random_org_id()
org_id.should.match(ORG_ID_REGEX)
def test_make_random_root_id():
root_id = utils.make_random_root_id()
root_id.should.match(ROOT_ID_REGEX)
def test_make_random_ou_id():
root_id = utils.make_random_root_id()
ou_id = utils.make_random_ou_id(root_id)
ou_id.should.match(OU_ID_REGEX)
def test_make_random_account_id():
account_id = utils.make_random_account_id()
account_id.should.match(ACCOUNT_ID_REGEX)
def test_make_random_create_account_status_id():
create_account_status_id = utils.make_random_create_account_status_id()
create_account_status_id.should.match(CREATE_ACCOUNT_STATUS_ID_REGEX)
def validate_organization(response):
org = response['Organization']
sorted(org.keys()).should.equal([
'Arn',
'AvailablePolicyTypes',
'FeatureSet',
'Id',
'MasterAccountArn',
'MasterAccountEmail',
'MasterAccountId',
])
org['Id'].should.match(ORG_ID_REGEX)
org['MasterAccountId'].should.equal(utils.MASTER_ACCOUNT_ID)
org['MasterAccountArn'].should.equal(utils.MASTER_ACCOUNT_ARN_FORMAT.format(
org['MasterAccountId'],
org['Id'],
))
org['Arn'].should.equal(utils.ORGANIZATION_ARN_FORMAT.format(
org['MasterAccountId'],
org['Id'],
))
org['MasterAccountEmail'].should.equal(utils.MASTER_ACCOUNT_EMAIL)
org['FeatureSet'].should.be.within(['ALL', 'CONSOLIDATED_BILLING'])
org['AvailablePolicyTypes'].should.equal([{
'Type': 'SERVICE_CONTROL_POLICY',
'Status': 'ENABLED'
}])
def validate_roots(org, response):
response.should.have.key('Roots').should.be.a(list)
response['Roots'].should_not.be.empty
root = response['Roots'][0]
root.should.have.key('Id').should.match(ROOT_ID_REGEX)
root.should.have.key('Arn').should.equal(utils.ROOT_ARN_FORMAT.format(
org['MasterAccountId'],
org['Id'],
root['Id'],
))
root.should.have.key('Name').should.be.a(six.string_types)
root.should.have.key('PolicyTypes').should.be.a(list)
root['PolicyTypes'][0].should.have.key('Type').should.equal('SERVICE_CONTROL_POLICY')
root['PolicyTypes'][0].should.have.key('Status').should.equal('ENABLED')
def validate_organizational_unit(org, response):
response.should.have.key('OrganizationalUnit').should.be.a(dict)
ou = response['OrganizationalUnit']
ou.should.have.key('Id').should.match(OU_ID_REGEX)
ou.should.have.key('Arn').should.equal(utils.OU_ARN_FORMAT.format(
org['MasterAccountId'],
org['Id'],
ou['Id'],
))
ou.should.have.key('Name').should.be.a(six.string_types)
def validate_account(org, account):
sorted(account.keys()).should.equal([
'Arn',
'Email',
'Id',
'JoinedMethod',
'JoinedTimestamp',
'Name',
'Status',
])
account['Id'].should.match(ACCOUNT_ID_REGEX)
account['Arn'].should.equal(utils.ACCOUNT_ARN_FORMAT.format(
org['MasterAccountId'],
org['Id'],
account['Id'],
))
account['Email'].should.match(EMAIL_REGEX)
account['JoinedMethod'].should.be.within(['INVITED', 'CREATED'])
account['Status'].should.be.within(['ACTIVE', 'SUSPENDED'])
account['Name'].should.be.a(six.string_types)
account['JoinedTimestamp'].should.be.a(datetime.datetime)
def validate_create_account_status(create_status):
sorted(create_status.keys()).should.equal([
'AccountId',
'AccountName',
'CompletedTimestamp',
'Id',
'RequestedTimestamp',
'State',
])
create_status['Id'].should.match(CREATE_ACCOUNT_STATUS_ID_REGEX)
create_status['AccountId'].should.match(ACCOUNT_ID_REGEX)
create_status['AccountName'].should.be.a(six.string_types)
create_status['State'].should.equal('SUCCEEDED')
create_status['RequestedTimestamp'].should.be.a(datetime.datetime)
create_status['CompletedTimestamp'].should.be.a(datetime.datetime)

322
tests/test_organizations/test_organizations_boto3.py Normal file
View file

@ -0,0 +1,322 @@
from __future__ import unicode_literals
import boto3
import sure # noqa
from botocore.exceptions import ClientError
from nose.tools import assert_raises
from moto import mock_organizations
from moto.organizations import utils
from .organizations_test_utils import (
validate_organization,
validate_roots,
validate_organizational_unit,
validate_account,
validate_create_account_status,
)
@mock_organizations
def test_create_organization():
client = boto3.client('organizations', region_name='us-east-1')
response = client.create_organization(FeatureSet='ALL')
validate_organization(response)
response['Organization']['FeatureSet'].should.equal('ALL')
@mock_organizations
def test_describe_organization():
client = boto3.client('organizations', region_name='us-east-1')
client.create_organization(FeatureSet='ALL')
response = client.describe_organization()
validate_organization(response)
@mock_organizations
def test_describe_organization_exception():
client = boto3.client('organizations', region_name='us-east-1')
with assert_raises(ClientError) as e:
response = client.describe_organization()
ex = e.exception
ex.operation_name.should.equal('DescribeOrganization')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('AWSOrganizationsNotInUseException')
# Organizational Units
@mock_organizations
def test_list_roots():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
response = client.list_roots()
validate_roots(org, response)
@mock_organizations
def test_create_organizational_unit():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
ou_name = 'ou01'
response = client.create_organizational_unit(
ParentId=root_id,
Name=ou_name,
)
validate_organizational_unit(org, response)
response['OrganizationalUnit']['Name'].should.equal(ou_name)
@mock_organizations
def test_describe_organizational_unit():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
ou_id = client.create_organizational_unit(
ParentId=root_id,
Name='ou01',
)['OrganizationalUnit']['Id']
response = client.describe_organizational_unit(OrganizationalUnitId=ou_id)
validate_organizational_unit(org, response)
@mock_organizations
def test_describe_organizational_unit_exception():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
with assert_raises(ClientError) as e:
response = client.describe_organizational_unit(
OrganizationalUnitId=utils.make_random_root_id()
)
ex = e.exception
ex.operation_name.should.equal('DescribeOrganizationalUnit')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('OrganizationalUnitNotFoundException')
@mock_organizations
def test_list_organizational_units_for_parent():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
client.create_organizational_unit(ParentId=root_id, Name='ou01')
client.create_organizational_unit(ParentId=root_id, Name='ou02')
client.create_organizational_unit(ParentId=root_id, Name='ou03')
response = client.list_organizational_units_for_parent(ParentId=root_id)
response.should.have.key('OrganizationalUnits').should.be.a(list)
for ou in response['OrganizationalUnits']:
validate_organizational_unit(org, dict(OrganizationalUnit=ou))
@mock_organizations
def test_list_organizational_units_for_parent_exception():
client = boto3.client('organizations', region_name='us-east-1')
with assert_raises(ClientError) as e:
response = client.list_organizational_units_for_parent(
ParentId=utils.make_random_root_id()
)
ex = e.exception
ex.operation_name.should.equal('ListOrganizationalUnitsForParent')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('ParentNotFoundException')
# Accounts
mockname = 'mock-account'
mockdomain = 'moto-example.org'
mockemail = '@'.join([mockname, mockdomain])
@mock_organizations
def test_create_account():
client = boto3.client('organizations', region_name='us-east-1')
client.create_organization(FeatureSet='ALL')
create_status = client.create_account(
AccountName=mockname, Email=mockemail
)['CreateAccountStatus']
validate_create_account_status(create_status)
create_status['AccountName'].should.equal(mockname)
@mock_organizations
def test_describe_account():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
account_id = client.create_account(
AccountName=mockname, Email=mockemail
)['CreateAccountStatus']['AccountId']
response = client.describe_account(AccountId=account_id)
validate_account(org, response['Account'])
response['Account']['Name'].should.equal(mockname)
response['Account']['Email'].should.equal(mockemail)
@mock_organizations
def test_describe_account_exception():
client = boto3.client('organizations', region_name='us-east-1')
with assert_raises(ClientError) as e:
response = client.describe_account(AccountId=utils.make_random_account_id())
ex = e.exception
ex.operation_name.should.equal('DescribeAccount')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('AccountNotFoundException')
@mock_organizations
def test_list_accounts():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
for i in range(5):
name = mockname + str(i)
email = name + '@' + mockdomain
client.create_account(AccountName=name, Email=email)
response = client.list_accounts()
response.should.have.key('Accounts')
accounts = response['Accounts']
len(accounts).should.equal(5)
for account in accounts:
validate_account(org, account)
accounts[3]['Name'].should.equal(mockname + '3')
accounts[2]['Email'].should.equal(mockname + '2' + '@' + mockdomain)
@mock_organizations
def test_list_accounts_for_parent():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
account_id = client.create_account(
AccountName=mockname,
Email=mockemail,
)['CreateAccountStatus']['AccountId']
response = client.list_accounts_for_parent(ParentId=root_id)
account_id.should.be.within([account['Id'] for account in response['Accounts']])
@mock_organizations
def test_move_account():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
account_id = client.create_account(
AccountName=mockname, Email=mockemail
)['CreateAccountStatus']['AccountId']
ou01 = client.create_organizational_unit(ParentId=root_id, Name='ou01')
ou01_id = ou01['OrganizationalUnit']['Id']
client.move_account(
AccountId=account_id,
SourceParentId=root_id,
DestinationParentId=ou01_id,
)
response = client.list_accounts_for_parent(ParentId=ou01_id)
account_id.should.be.within([account['Id'] for account in response['Accounts']])
@mock_organizations
def test_list_parents_for_ou():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
ou01 = client.create_organizational_unit(ParentId=root_id, Name='ou01')
ou01_id = ou01['OrganizationalUnit']['Id']
response01 = client.list_parents(ChildId=ou01_id)
response01.should.have.key('Parents').should.be.a(list)
response01['Parents'][0].should.have.key('Id').should.equal(root_id)
response01['Parents'][0].should.have.key('Type').should.equal('ROOT')
ou02 = client.create_organizational_unit(ParentId=ou01_id, Name='ou02')
ou02_id = ou02['OrganizationalUnit']['Id']
response02 = client.list_parents(ChildId=ou02_id)
response02.should.have.key('Parents').should.be.a(list)
response02['Parents'][0].should.have.key('Id').should.equal(ou01_id)
response02['Parents'][0].should.have.key('Type').should.equal('ORGANIZATIONAL_UNIT')
@mock_organizations
def test_list_parents_for_accounts():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
ou01 = client.create_organizational_unit(ParentId=root_id, Name='ou01')
ou01_id = ou01['OrganizationalUnit']['Id']
account01_id = client.create_account(
AccountName='account01',
Email='account01@moto-example.org'
)['CreateAccountStatus']['AccountId']
account02_id = client.create_account(
AccountName='account02',
Email='account02@moto-example.org'
)['CreateAccountStatus']['AccountId']
client.move_account(
AccountId=account02_id,
SourceParentId=root_id,
DestinationParentId=ou01_id,
)
response01 = client.list_parents(ChildId=account01_id)
response01.should.have.key('Parents').should.be.a(list)
response01['Parents'][0].should.have.key('Id').should.equal(root_id)
response01['Parents'][0].should.have.key('Type').should.equal('ROOT')
response02 = client.list_parents(ChildId=account02_id)
response02.should.have.key('Parents').should.be.a(list)
response02['Parents'][0].should.have.key('Id').should.equal(ou01_id)
response02['Parents'][0].should.have.key('Type').should.equal('ORGANIZATIONAL_UNIT')
@mock_organizations
def test_list_children():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
ou01 = client.create_organizational_unit(ParentId=root_id, Name='ou01')
ou01_id = ou01['OrganizationalUnit']['Id']
ou02 = client.create_organizational_unit(ParentId=ou01_id, Name='ou02')
ou02_id = ou02['OrganizationalUnit']['Id']
account01_id = client.create_account(
AccountName='account01',
Email='account01@moto-example.org'
)['CreateAccountStatus']['AccountId']
account02_id = client.create_account(
AccountName='account02',
Email='account02@moto-example.org'
)['CreateAccountStatus']['AccountId']
client.move_account(
AccountId=account02_id,
SourceParentId=root_id,
DestinationParentId=ou01_id,
)
response01 = client.list_children(ParentId=root_id, ChildType='ACCOUNT')
response02 = client.list_children(ParentId=root_id, ChildType='ORGANIZATIONAL_UNIT')
response03 = client.list_children(ParentId=ou01_id, ChildType='ACCOUNT')
response04 = client.list_children(ParentId=ou01_id, ChildType='ORGANIZATIONAL_UNIT')
response01['Children'][0]['Id'].should.equal(account01_id)
response01['Children'][0]['Type'].should.equal('ACCOUNT')
response02['Children'][0]['Id'].should.equal(ou01_id)
response02['Children'][0]['Type'].should.equal('ORGANIZATIONAL_UNIT')
response03['Children'][0]['Id'].should.equal(account02_id)
response03['Children'][0]['Type'].should.equal('ACCOUNT')
response04['Children'][0]['Id'].should.equal(ou02_id)
response04['Children'][0]['Type'].should.equal('ORGANIZATIONAL_UNIT')
@mock_organizations
def test_list_children_exception():
client = boto3.client('organizations', region_name='us-east-1')
org = client.create_organization(FeatureSet='ALL')['Organization']
root_id = client.list_roots()['Roots'][0]['Id']
with assert_raises(ClientError) as e:
response = client.list_children(
ParentId=utils.make_random_root_id(),
ChildType='ACCOUNT'
)
ex = e.exception
ex.operation_name.should.equal('ListChildren')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('ParentNotFoundException')
with assert_raises(ClientError) as e:
response = client.list_children(
ParentId=root_id,
ChildType='BLEE'
)
ex = e.exception
ex.operation_name.should.equal('ListChildren')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('InvalidInputException')

155
tests/test_rds2/test_rds2.py
View file

@ -33,6 +33,7 @@ def test_create_database():
db_instance['DBInstanceIdentifier'].should.equal("db-master-1")
db_instance['IAMDatabaseAuthenticationEnabled'].should.equal(False)
db_instance['DbiResourceId'].should.contain("db-")
db_instance['CopyTagsToSnapshot'].should.equal(False)
@mock_rds2
@ -339,6 +340,49 @@ def test_create_db_snapshots():
snapshot.get('Engine').should.equal('postgres')
snapshot.get('DBInstanceIdentifier').should.equal('db-primary-1')
snapshot.get('DBSnapshotIdentifier').should.equal('g-1')
result = conn.list_tags_for_resource(ResourceName=snapshot['DBSnapshotArn'])
result['TagList'].should.equal([])
@mock_rds2
def test_create_db_snapshots_copy_tags():
conn = boto3.client('rds', region_name='us-west-2')
conn.create_db_snapshot.when.called_with(
DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-1').should.throw(ClientError)
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"],
CopyTagsToSnapshot=True,
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='g-1').get('DBSnapshot')
snapshot.get('Engine').should.equal('postgres')
snapshot.get('DBInstanceIdentifier').should.equal('db-primary-1')
snapshot.get('DBSnapshotIdentifier').should.equal('g-1')
result = conn.list_tags_for_resource(ResourceName=snapshot['DBSnapshotArn'])
result['TagList'].should.equal([{'Value': 'bar',
'Key': 'foo'},
{'Value': 'bar1',
'Key': 'foo1'}])
@mock_rds2
@ -656,6 +700,117 @@ def test_remove_tags_db():
len(result['TagList']).should.equal(1)
@mock_rds2
def test_list_tags_snapshot():
conn = boto3.client('rds', region_name='us-west-2')
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:foo')
result['TagList'].should.equal([])
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-with-tags',
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
result = conn.list_tags_for_resource(ResourceName=snapshot['DBSnapshot']['DBSnapshotArn'])
result['TagList'].should.equal([{'Value': 'bar',
'Key': 'foo'},
{'Value': 'bar1',
'Key': 'foo1'}])
@mock_rds2
def test_add_tags_snapshot():
conn = boto3.client('rds', region_name='us-west-2')
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-without-tags',
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-without-tags')
list(result['TagList']).should.have.length_of(2)
conn.add_tags_to_resource(ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-without-tags',
Tags=[
{
'Key': 'foo',
'Value': 'fish',
},
{
'Key': 'foo2',
'Value': 'bar2',
},
])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-without-tags')
list(result['TagList']).should.have.length_of(3)
@mock_rds2
def test_remove_tags_snapshot():
conn = boto3.client('rds', region_name='us-west-2')
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-with-tags',
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-with-tags')
list(result['TagList']).should.have.length_of(2)
conn.remove_tags_from_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-with-tags', TagKeys=['foo'])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-with-tags')
len(result['TagList']).should.equal(1)
@mock_rds2
def test_add_tags_option_group():
conn = boto3.client('rds', region_name='us-west-2')

4
tests/test_redshift/test_redshift.py
View file

@ -1,5 +1,7 @@
from __future__ import unicode_literals
import datetime
import boto
import boto3
from boto.redshift.exceptions import (
@ -32,6 +34,8 @@ def test_create_cluster_boto3():
MasterUserPassword='password',
)
response['Cluster']['NodeType'].should.equal('ds2.xlarge')
create_time = response['Cluster']['ClusterCreateTime']
create_time.should.be.lower_than(datetime.datetime.now(create_time.tzinfo))
@mock_redshift

66
tests/test_s3/test_s3.py
View file

@ -2471,6 +2471,72 @@ def test_boto3_delete_markers():
oldest['Key'].should.equal('key-with-versions-and-unicode-ó')
@mock_s3
def test_boto3_multiple_delete_markers():
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'mybucket'
key = u'key-with-versions-and-unicode-ó'
s3.create_bucket(Bucket=bucket_name)
s3.put_bucket_versioning(
Bucket=bucket_name,
VersioningConfiguration={
'Status': 'Enabled'
}
)
items = (six.b('v1'), six.b('v2'))
for body in items:
s3.put_object(
Bucket=bucket_name,
Key=key,
Body=body
)
# Delete the object twice to add multiple delete markers
s3.delete_object(Bucket=bucket_name, Key=key)
s3.delete_object(Bucket=bucket_name, Key=key)
response = s3.list_object_versions(Bucket=bucket_name)
response['DeleteMarkers'].should.have.length_of(2)
with assert_raises(ClientError) as e:
s3.get_object(
Bucket=bucket_name,
Key=key
)
e.response['Error']['Code'].should.equal('404')
# Remove both delete markers to restore the object
s3.delete_object(
Bucket=bucket_name,
Key=key,
VersionId='2'
)
s3.delete_object(
Bucket=bucket_name,
Key=key,
VersionId='3'
)
response = s3.get_object(
Bucket=bucket_name,
Key=key
)
response['Body'].read().should.equal(items[-1])
response = s3.list_object_versions(Bucket=bucket_name)
response['Versions'].should.have.length_of(2)
# We've asserted there is only 2 records so one is newest, one is oldest
latest = list(filter(lambda item: item['IsLatest'], response['Versions']))[0]
oldest = list(filter(lambda item: not item['IsLatest'], response['Versions']))[0]
# Double check ordering of version ID's
latest['VersionId'].should.equal('1')
oldest['VersionId'].should.equal('0')
# Double check the name is still unicode
latest['Key'].should.equal('key-with-versions-and-unicode-ó')
oldest['Key'].should.equal('key-with-versions-and-unicode-ó')
@mock_s3
def test_get_stream_gzipped():
payload = b"this is some stuff here"

121
tests/test_s3/test_s3_lifecycle.py
View file

@ -191,6 +191,127 @@ def test_lifecycle_with_eodm():
assert err.exception.response["Error"]["Code"] == "MalformedXML"
@mock_s3
def test_lifecycle_with_nve():
client = boto3.client("s3")
client.create_bucket(Bucket="bucket")
lfc = {
"Rules": [
{
"NoncurrentVersionExpiration": {
"NoncurrentDays": 30
},
"ID": "wholebucket",
"Filter": {
"Prefix": ""
},
"Status": "Enabled"
}
]
}
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["NoncurrentVersionExpiration"]["NoncurrentDays"] == 30
# Change NoncurrentDays:
lfc["Rules"][0]["NoncurrentVersionExpiration"]["NoncurrentDays"] = 10
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["NoncurrentVersionExpiration"]["NoncurrentDays"] == 10
# TODO: Add test for failures due to missing children
@mock_s3
def test_lifecycle_with_nvt():
client = boto3.client("s3")
client.create_bucket(Bucket="bucket")
lfc = {
"Rules": [
{
"NoncurrentVersionTransitions": [{
"NoncurrentDays": 30,
"StorageClass": "ONEZONE_IA"
}],
"ID": "wholebucket",
"Filter": {
"Prefix": ""
},
"Status": "Enabled"
}
]
}
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["NoncurrentVersionTransitions"][0]["NoncurrentDays"] == 30
assert result["Rules"][0]["NoncurrentVersionTransitions"][0]["StorageClass"] == "ONEZONE_IA"
# Change NoncurrentDays:
lfc["Rules"][0]["NoncurrentVersionTransitions"][0]["NoncurrentDays"] = 10
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["NoncurrentVersionTransitions"][0]["NoncurrentDays"] == 10
# Change StorageClass:
lfc["Rules"][0]["NoncurrentVersionTransitions"][0]["StorageClass"] = "GLACIER"
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["NoncurrentVersionTransitions"][0]["StorageClass"] == "GLACIER"
# With failures for missing children:
del lfc["Rules"][0]["NoncurrentVersionTransitions"][0]["NoncurrentDays"]
with assert_raises(ClientError) as err:
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
assert err.exception.response["Error"]["Code"] == "MalformedXML"
lfc["Rules"][0]["NoncurrentVersionTransitions"][0]["NoncurrentDays"] = 30
del lfc["Rules"][0]["NoncurrentVersionTransitions"][0]["StorageClass"]
with assert_raises(ClientError) as err:
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
assert err.exception.response["Error"]["Code"] == "MalformedXML"
@mock_s3
def test_lifecycle_with_aimu():
client = boto3.client("s3")
client.create_bucket(Bucket="bucket")
lfc = {
"Rules": [
{
"AbortIncompleteMultipartUpload": {
"DaysAfterInitiation": 7
},
"ID": "wholebucket",
"Filter": {
"Prefix": ""
},
"Status": "Enabled"
}
]
}
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["AbortIncompleteMultipartUpload"]["DaysAfterInitiation"] == 7
# Change DaysAfterInitiation:
lfc["Rules"][0]["AbortIncompleteMultipartUpload"]["DaysAfterInitiation"] = 30
client.put_bucket_lifecycle_configuration(Bucket="bucket", LifecycleConfiguration=lfc)
result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
assert len(result["Rules"]) == 1
assert result["Rules"][0]["AbortIncompleteMultipartUpload"]["DaysAfterInitiation"] == 30
# TODO: Add test for failures due to missing children
@mock_s3_deprecated
def test_lifecycle_with_glacier_transition():
conn = boto.s3.connect_to_region("us-west-1")

109
tests/test_secretsmanager/test_secretsmanager.py
View file

@ -26,13 +26,13 @@ def test_get_secret_that_does_not_exist():
result = conn.get_secret_value(SecretId='i-dont-exist')
@mock_secretsmanager
def test_get_secret_with_mismatched_id():
def test_get_secret_that_does_not_match():
conn = boto3.client('secretsmanager', region_name='us-west-2')
create_secret = conn.create_secret(Name='java-util-test-password',
SecretString="foosecret")
with assert_raises(ClientError):
result = conn.get_secret_value(SecretId='i-dont-exist')
result = conn.get_secret_value(SecretId='i-dont-match')
@mock_secretsmanager
def test_create_secret():
@ -179,3 +179,108 @@ def test_describe_secret_that_does_not_match():
with assert_raises(ClientError):
result = conn.get_secret_value(SecretId='i-dont-match')
@mock_secretsmanager
def test_rotate_secret():
secret_name = 'test-secret'
conn = boto3.client('secretsmanager', region_name='us-west-2')
conn.create_secret(Name=secret_name,
SecretString='foosecret')
rotated_secret = conn.rotate_secret(SecretId=secret_name)
assert rotated_secret
assert rotated_secret['ARN'] == (
'arn:aws:secretsmanager:us-west-2:1234567890:secret:test-secret-rIjad'
)
assert rotated_secret['Name'] == secret_name
assert rotated_secret['VersionId'] != ''
@mock_secretsmanager
def test_rotate_secret_enable_rotation():
secret_name = 'test-secret'
conn = boto3.client('secretsmanager', region_name='us-west-2')
conn.create_secret(Name=secret_name,
SecretString='foosecret')
initial_description = conn.describe_secret(SecretId=secret_name)
assert initial_description
assert initial_description['RotationEnabled'] is False
assert initial_description['RotationRules']['AutomaticallyAfterDays'] == 0
conn.rotate_secret(SecretId=secret_name,
RotationRules={'AutomaticallyAfterDays': 42})
rotated_description = conn.describe_secret(SecretId=secret_name)
assert rotated_description
assert rotated_description['RotationEnabled'] is True
assert rotated_description['RotationRules']['AutomaticallyAfterDays'] == 42
@mock_secretsmanager
def test_rotate_secret_that_does_not_exist():
conn = boto3.client('secretsmanager', 'us-west-2')
with assert_raises(ClientError):
result = conn.rotate_secret(SecretId='i-dont-exist')
@mock_secretsmanager
def test_rotate_secret_that_does_not_match():
conn = boto3.client('secretsmanager', region_name='us-west-2')
conn.create_secret(Name='test-secret',
SecretString='foosecret')
with assert_raises(ClientError):
result = conn.rotate_secret(SecretId='i-dont-match')
@mock_secretsmanager
def test_rotate_secret_client_request_token_too_short():
# Test is intentionally empty. Boto3 catches too short ClientRequestToken
# and raises ParamValidationError before Moto can see it.
# test_server actually handles this error.
assert True
@mock_secretsmanager
def test_rotate_secret_client_request_token_too_long():
secret_name = 'test-secret'
conn = boto3.client('secretsmanager', region_name='us-west-2')
conn.create_secret(Name=secret_name,
SecretString='foosecret')
client_request_token = (
'ED9F8B6C-85B7-446A-B7E4-38F2A3BEB13C-'
'ED9F8B6C-85B7-446A-B7E4-38F2A3BEB13C'
)
with assert_raises(ClientError):
result = conn.rotate_secret(SecretId=secret_name,
ClientRequestToken=client_request_token)
@mock_secretsmanager
def test_rotate_secret_rotation_lambda_arn_too_long():
secret_name = 'test-secret'
conn = boto3.client('secretsmanager', region_name='us-west-2')
conn.create_secret(Name=secret_name,
SecretString='foosecret')
rotation_lambda_arn = '85B7-446A-B7E4' * 147 # == 2058 characters
with assert_raises(ClientError):
result = conn.rotate_secret(SecretId=secret_name,
RotationLambdaARN=rotation_lambda_arn)
@mock_secretsmanager
def test_rotate_secret_rotation_period_zero():
# Test is intentionally empty. Boto3 catches zero day rotation period
# and raises ParamValidationError before Moto can see it.
# test_server actually handles this error.
assert True
@mock_secretsmanager
def test_rotate_secret_rotation_period_too_long():
secret_name = 'test-secret'
conn = boto3.client('secretsmanager', region_name='us-west-2')
conn.create_secret(Name=secret_name,
SecretString='foosecret')
rotation_rules = {'AutomaticallyAfterDays': 1001}
with assert_raises(ClientError):
result = conn.rotate_secret(SecretId=secret_name,
RotationRules=rotation_rules)

286
tests/test_secretsmanager/test_server.py
View file

@ -49,6 +49,27 @@ def test_get_secret_that_does_not_exist():
assert json_data['message'] == "Secrets Manager can't find the specified secret"
assert json_data['__type'] == 'ResourceNotFoundException'
@mock_secretsmanager
def test_get_secret_that_does_not_match():
backend = server.create_backend_app("secretsmanager")
test_client = backend.test_client()
create_secret = test_client.post('/',
data={"Name": "test-secret",
"SecretString": "foo-secret"},
headers={
"X-Amz-Target": "secretsmanager.CreateSecret"},
)
get_secret = test_client.post('/',
data={"SecretId": "i-dont-match",
"VersionStage": "AWSCURRENT"},
headers={
"X-Amz-Target": "secretsmanager.GetSecretValue"},
)
json_data = json.loads(get_secret.data.decode("utf-8"))
assert json_data['message'] == "Secrets Manager can't find the specified secret"
assert json_data['__type'] == 'ResourceNotFoundException'
@mock_secretsmanager
def test_create_secret():
@ -133,3 +154,268 @@ def test_describe_secret_that_does_not_match():
json_data = json.loads(describe_secret.data.decode("utf-8"))
assert json_data['message'] == "Secrets Manager can't find the specified secret"
assert json_data['__type'] == 'ResourceNotFoundException'
@mock_secretsmanager
def test_rotate_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"
},
)
client_request_token = "EXAMPLE2-90ab-cdef-fedc-ba987SECRET2"
rotate_secret = test_client.post('/',
data={"SecretId": "test-secret",
"ClientRequestToken": client_request_token},
headers={
"X-Amz-Target": "secretsmanager.RotateSecret"
},
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert json_data # Returned dict is not empty
assert json_data['ARN'] == (
'arn:aws:secretsmanager:us-east-1:1234567890:secret:test-secret-rIjad'
)
assert json_data['Name'] == 'test-secret'
assert json_data['VersionId'] == client_request_token
# @mock_secretsmanager
# def test_rotate_secret_enable_rotation():
# 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"
# },
# )
# initial_description = test_client.post(
# '/',
# data={
# "SecretId": "test-secret"
# },
# headers={
# "X-Amz-Target": "secretsmanager.DescribeSecret"
# },
# )
# json_data = json.loads(initial_description.data.decode("utf-8"))
# assert json_data # Returned dict is not empty
# assert json_data['RotationEnabled'] is False
# assert json_data['RotationRules']['AutomaticallyAfterDays'] == 0
# rotate_secret = test_client.post(
# '/',
# data={
# "SecretId": "test-secret",
# "RotationRules": {"AutomaticallyAfterDays": 42}
# },
# headers={
# "X-Amz-Target": "secretsmanager.RotateSecret"
# },
# )
# rotated_description = test_client.post(
# '/',
# data={
# "SecretId": "test-secret"
# },
# headers={
# "X-Amz-Target": "secretsmanager.DescribeSecret"
# },
# )
# json_data = json.loads(rotated_description.data.decode("utf-8"))
# assert json_data # Returned dict is not empty
# assert json_data['RotationEnabled'] is True
# assert json_data['RotationRules']['AutomaticallyAfterDays'] == 42
@mock_secretsmanager
def test_rotate_secret_that_does_not_exist():
backend = server.create_backend_app('secretsmanager')
test_client = backend.test_client()
rotate_secret = test_client.post('/',
data={"SecretId": "i-dont-exist"},
headers={
"X-Amz-Target": "secretsmanager.RotateSecret"
},
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert json_data['message'] == "Secrets Manager can't find the specified secret"
assert json_data['__type'] == 'ResourceNotFoundException'
@mock_secretsmanager
def test_rotate_secret_that_does_not_match():
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"
},
)
rotate_secret = test_client.post('/',
data={"SecretId": "i-dont-match"},
headers={
"X-Amz-Target": "secretsmanager.RotateSecret"
},
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert json_data['message'] == "Secrets Manager can't find the specified secret"
assert json_data['__type'] == 'ResourceNotFoundException'
@mock_secretsmanager
def test_rotate_secret_client_request_token_too_short():
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"
},
)
client_request_token = "ED9F8B6C-85B7-B7E4-38F2A3BEB13C"
rotate_secret = test_client.post('/',
data={"SecretId": "test-secret",
"ClientRequestToken": client_request_token},
headers={
"X-Amz-Target": "secretsmanager.RotateSecret"
},
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert json_data['message'] == "ClientRequestToken must be 32-64 characters long."
assert json_data['__type'] == 'InvalidParameterException'
@mock_secretsmanager
def test_rotate_secret_client_request_token_too_long():
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"
},
)
client_request_token = (
'ED9F8B6C-85B7-446A-B7E4-38F2A3BEB13C-'
'ED9F8B6C-85B7-446A-B7E4-38F2A3BEB13C'
)
rotate_secret = test_client.post('/',
data={"SecretId": "test-secret",
"ClientRequestToken": client_request_token},
headers={
"X-Amz-Target": "secretsmanager.RotateSecret"
},
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert json_data['message'] == "ClientRequestToken must be 32-64 characters long."
assert json_data['__type'] == 'InvalidParameterException'
@mock_secretsmanager
def test_rotate_secret_rotation_lambda_arn_too_long():
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"
},
)
rotation_lambda_arn = '85B7-446A-B7E4' * 147 # == 2058 characters
rotate_secret = test_client.post('/',
data={"SecretId": "test-secret",
"RotationLambdaARN": rotation_lambda_arn},
headers={
"X-Amz-Target": "secretsmanager.RotateSecret"
},
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert json_data['message'] == "RotationLambdaARN must <= 2048 characters long."
assert json_data['__type'] == 'InvalidParameterException'
#
# The following tests should work, but fail on the embedded dict in
# RotationRules. The error message suggests a problem deeper in the code, which
# needs further investigation.
#
# @mock_secretsmanager
# def test_rotate_secret_rotation_period_zero():
# 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"
# },
# )
# rotate_secret = test_client.post('/',
# data={"SecretId": "test-secret",
# "RotationRules": {"AutomaticallyAfterDays": 0}},
# headers={
# "X-Amz-Target": "secretsmanager.RotateSecret"
# },
# )
# json_data = json.loads(rotate_secret.data.decode("utf-8"))
# assert json_data['message'] == "RotationRules.AutomaticallyAfterDays must be within 1-1000."
# assert json_data['__type'] == 'InvalidParameterException'
# @mock_secretsmanager
# def test_rotate_secret_rotation_period_too_long():
# 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"
# },
# )
# rotate_secret = test_client.post('/',
# data={"SecretId": "test-secret",
# "RotationRules": {"AutomaticallyAfterDays": 1001}},
# headers={
# "X-Amz-Target": "secretsmanager.RotateSecret"
# },
# )
# json_data = json.loads(rotate_secret.data.decode("utf-8"))
# assert json_data['message'] == "RotationRules.AutomaticallyAfterDays must be within 1-1000."
# assert json_data['__type'] == 'InvalidParameterException'

40
tests/test_sqs/test_sqs.py
View file

@ -40,6 +40,33 @@ def test_create_fifo_queue_fail():
raise RuntimeError('Should of raised InvalidParameterValue Exception')
@mock_sqs
def test_create_queue_with_same_attributes():
sqs = boto3.client('sqs', region_name='us-east-1')
dlq_url = sqs.create_queue(QueueName='test-queue-dlq')['QueueUrl']
dlq_arn = sqs.get_queue_attributes(QueueUrl=dlq_url)['Attributes']['QueueArn']
attributes = {
'DelaySeconds': '900',
'MaximumMessageSize': '262144',
'MessageRetentionPeriod': '1209600',
'ReceiveMessageWaitTimeSeconds': '20',
'RedrivePolicy': '{"deadLetterTargetArn": "%s", "maxReceiveCount": 100}' % (dlq_arn),
'VisibilityTimeout': '43200'
}
sqs.create_queue(
QueueName='test-queue',
Attributes=attributes
)
sqs.create_queue(
QueueName='test-queue',
Attributes=attributes
)
@mock_sqs
def test_create_queue_with_different_attributes_fail():
sqs = boto3.client('sqs', region_name='us-east-1')
@ -1195,3 +1222,16 @@ def test_receive_messages_with_message_group_id_on_visibility_timeout():
messages = queue.receive_messages()
messages.should.have.length_of(1)
messages[0].message_id.should.equal(message.message_id)
@mock_sqs
def test_receive_message_for_queue_with_receive_message_wait_time_seconds_set():
sqs = boto3.resource('sqs', region_name='us-east-1')
queue = sqs.create_queue(
QueueName='test-queue',
Attributes={
'ReceiveMessageWaitTimeSeconds': '2',
}
)
queue.receive_messages()

118
tests/test_ssm/test_ssm_boto3.py
View file

@ -5,11 +5,12 @@ import botocore.exceptions
import sure # noqa
import datetime
import uuid
import json
from botocore.exceptions import ClientError
from nose.tools import assert_raises
from moto import mock_ssm
from moto import mock_ssm, mock_cloudformation
@mock_ssm
@ -668,3 +669,118 @@ def test_list_commands():
with assert_raises(ClientError):
response = client.list_commands(
CommandId=str(uuid.uuid4()))
@mock_ssm
def test_get_command_invocation():
client = boto3.client('ssm', region_name='us-east-1')
ssm_document = 'AWS-RunShellScript'
params = {'commands': ['#!/bin/bash\necho \'hello world\'']}
response = client.send_command(
InstanceIds=['i-123456', 'i-234567', 'i-345678'],
DocumentName=ssm_document,
Parameters=params,
OutputS3Region='us-east-2',
OutputS3BucketName='the-bucket',
OutputS3KeyPrefix='pref')
cmd = response['Command']
cmd_id = cmd['CommandId']
instance_id = 'i-345678'
invocation_response = client.get_command_invocation(
CommandId=cmd_id,
InstanceId=instance_id,
PluginName='aws:runShellScript')
invocation_response['CommandId'].should.equal(cmd_id)
invocation_response['InstanceId'].should.equal(instance_id)
# test the error case for an invalid instance id
with assert_raises(ClientError):
invocation_response = client.get_command_invocation(
CommandId=cmd_id,
InstanceId='i-FAKE')
# test the error case for an invalid plugin name
with assert_raises(ClientError):
invocation_response = client.get_command_invocation(
CommandId=cmd_id,
InstanceId=instance_id,
PluginName='FAKE')
@mock_ssm
@mock_cloudformation
def test_get_command_invocations_from_stack():
stack_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "Test Stack",
"Resources": {
"EC2Instance1": {
"Type": "AWS::EC2::Instance",
"Properties": {
"ImageId": "ami-test-image-id",
"KeyName": "test",
"InstanceType": "t2.micro",
"Tags": [
{
"Key": "Test Description",
"Value": "Test tag"
},
{
"Key": "Test Name",
"Value": "Name tag for tests"
}
]
}
}
},
"Outputs": {
"test": {
"Description": "Test Output",
"Value": "Test output value",
"Export": {
"Name": "Test value to export"
}
},
"PublicIP": {
"Value": "Test public ip"
}
}
}
cloudformation_client = boto3.client(
'cloudformation',
region_name='us-east-1')
stack_template_str = json.dumps(stack_template)
response = cloudformation_client.create_stack(
StackName='test_stack',
TemplateBody=stack_template_str,
Capabilities=('CAPABILITY_IAM', ))
client = boto3.client('ssm', region_name='us-east-1')
ssm_document = 'AWS-RunShellScript'
params = {'commands': ['#!/bin/bash\necho \'hello world\'']}
response = client.send_command(
Targets=[{
'Key': 'tag:aws:cloudformation:stack-name',
'Values': ('test_stack', )}],
DocumentName=ssm_document,
Parameters=params,
OutputS3Region='us-east-2',
OutputS3BucketName='the-bucket',
OutputS3KeyPrefix='pref')
cmd = response['Command']
cmd_id = cmd['CommandId']
instance_ids = cmd['InstanceIds']
invocation_response = client.get_command_invocation(
CommandId=cmd_id,
InstanceId=instance_ids[0],
PluginName='aws:runShellScript')