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

Merge branch 'master' into mark-alias-target

Browse source
This commit is contained in:
Steve Pulec 2019-07-07 23:11:53 -05:00 • committed by GitHub
commit 414ff930ed
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
43 changed files with 26142 additions and 7134 deletions
Download patch file Download diff file Expand all files Collapse all files

48
tests/test_core/test_socket.py Normal file
View file

@ -0,0 +1,48 @@
import unittest
from moto import mock_dynamodb2_deprecated, mock_dynamodb2
import socket
from six import PY3
class TestSocketPair(unittest.TestCase):
@mock_dynamodb2_deprecated
def test_asyncio_deprecated(self):
if PY3:
self.assertIn(
'moto.packages.httpretty.core.fakesock.socket',
str(socket.socket),
'Our mock should be present'
)
import asyncio
self.assertIsNotNone(asyncio.get_event_loop())
@mock_dynamodb2_deprecated
def test_socket_pair_deprecated(self):
# In Python2, the fakesocket is not set, for some reason.
if PY3:
self.assertIn(
'moto.packages.httpretty.core.fakesock.socket',
str(socket.socket),
'Our mock should be present'
)
a, b = socket.socketpair()
self.assertIsNotNone(a)
self.assertIsNotNone(b)
if a:
a.close()
if b:
b.close()
@mock_dynamodb2
def test_socket_pair(self):
a, b = socket.socketpair()
self.assertIsNotNone(a)
self.assertIsNotNone(b)
if a:
a.close()
if b:
b.close()

162
tests/test_dynamodb2/test_dynamodb.py
View file

@ -452,6 +452,90 @@ def test_basic_projection_expressions():
assert 'body' in results['Items'][1]
assert 'forum_name' in results['Items'][1]
@mock_dynamodb2
def test_basic_projection_expressions_using_scan():
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
table = dynamodb.create_table(
TableName='users',
KeySchema=[
{
'AttributeName': 'forum_name',
'KeyType': 'HASH'
},
{
'AttributeName': 'subject',
'KeyType': 'RANGE'
},
],
AttributeDefinitions=[
{
'AttributeName': 'forum_name',
'AttributeType': 'S'
},
{
'AttributeName': 'subject',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
table = dynamodb.Table('users')
table.put_item(Item={
'forum_name': 'the-key',
'subject': '123',
'body': 'some test message'
})
table.put_item(Item={
'forum_name': 'not-the-key',
'subject': '123',
'body': 'some other test message'
})
# Test a scan returning all items
results = table.scan(
FilterExpression=Key('forum_name').eq(
'the-key'),
ProjectionExpression='body, subject'
)
assert 'body' in results['Items'][0]
assert results['Items'][0]['body'] == 'some test message'
assert 'subject' in results['Items'][0]
table.put_item(Item={
'forum_name': 'the-key',
'subject': '1234',
'body': 'yet another test message'
})
results = table.scan(
FilterExpression=Key('forum_name').eq(
'the-key'),
ProjectionExpression='body'
)
assert 'body' in results['Items'][0]
assert 'subject' not in results['Items'][0]
assert 'forum_name' not in results['Items'][0]
assert 'body' in results['Items'][1]
assert 'subject' not in results['Items'][1]
assert 'forum_name' not in results['Items'][1]
# The projection expression should not remove data from storage
results = table.query(
KeyConditionExpression=Key('forum_name').eq(
'the-key'),
)
assert 'subject' in results['Items'][0]
assert 'body' in results['Items'][1]
assert 'forum_name' in results['Items'][1]
@mock_dynamodb2
def test_basic_projection_expressions_with_attr_expression_names():
@ -519,6 +603,84 @@ def test_basic_projection_expressions_with_attr_expression_names():
assert 'attachment' in results['Items'][0]
assert results['Items'][0]['attachment'] == 'something'
@mock_dynamodb2
def test_basic_projection_expressions_using_scan_with_attr_expression_names():
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
table = dynamodb.create_table(
TableName='users',
KeySchema=[
{
'AttributeName': 'forum_name',
'KeyType': 'HASH'
},
{
'AttributeName': 'subject',
'KeyType': 'RANGE'
},
],
AttributeDefinitions=[
{
'AttributeName': 'forum_name',
'AttributeType': 'S'
},
{
'AttributeName': 'subject',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
table = dynamodb.Table('users')
table.put_item(Item={
'forum_name': 'the-key',
'subject': '123',
'body': 'some test message',
'attachment': 'something'
})
table.put_item(Item={
'forum_name': 'not-the-key',
'subject': '123',
'body': 'some other test message',
'attachment': 'something'
})
# Test a scan returning all items
results = table.scan(
FilterExpression=Key('forum_name').eq(
'the-key'),
ProjectionExpression='#rl, #rt, subject',
ExpressionAttributeNames={
'#rl': 'body',
'#rt': 'attachment'
},
)
assert 'body' in results['Items'][0]
assert 'attachment' in results['Items'][0]
assert 'subject' in results['Items'][0]
assert 'form_name' not in results['Items'][0]
# Test without a FilterExpression
results = table.scan(
ProjectionExpression='#rl, #rt, subject',
ExpressionAttributeNames={
'#rl': 'body',
'#rt': 'attachment'
},
)
assert 'body' in results['Items'][0]
assert 'attachment' in results['Items'][0]
assert 'subject' in results['Items'][0]
assert 'form_name' not in results['Items'][0]
@mock_dynamodb2
def test_put_item_returns_consumed_capacity():

41
tests/test_ec2/test_instances.py
View file

@ -1,5 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
from botocore.exceptions import ClientError
import tests.backport_assert_raises
from nose.tools import assert_raises
@ -679,8 +681,8 @@ def test_modify_instance_attribute_security_groups():
reservation = conn.run_instances('ami-1234abcd')
instance = reservation.instances[0]
sg_id = 'sg-1234abcd'
sg_id2 = 'sg-abcd4321'
sg_id = conn.create_security_group('test security group', 'this is a test security group').id
sg_id2 = conn.create_security_group('test security group 2', 'this is a test security group 2').id
with assert_raises(EC2ResponseError) as ex:
instance.modify_attribute("groupSet", [sg_id, sg_id2], dry_run=True)
@ -1255,6 +1257,7 @@ def test_create_instance_ebs_optimized():
instance.load()
instance.ebs_optimized.should.be(False)
@mock_ec2
def test_run_multiple_instances_in_same_command():
instance_count = 4
@ -1269,3 +1272,37 @@ def test_run_multiple_instances_in_same_command():
instances = reservations[0]['Instances']
for i in range(0, instance_count):
instances[i]['AmiLaunchIndex'].should.be(i)
@mock_ec2
def test_describe_instance_attribute():
client = boto3.client('ec2', region_name='us-east-1')
security_group_id = client.create_security_group(
GroupName='test security group', Description='this is a test security group')['GroupId']
client.run_instances(ImageId='ami-1234abcd',
MinCount=1,
MaxCount=1,
SecurityGroupIds=[security_group_id])
instance_id = client.describe_instances()['Reservations'][0]['Instances'][0]['InstanceId']
valid_instance_attributes = ['instanceType', 'kernel', 'ramdisk', 'userData', 'disableApiTermination', 'instanceInitiatedShutdownBehavior', 'rootDeviceName', 'blockDeviceMapping', 'productCodes', 'sourceDestCheck', 'groupSet', 'ebsOptimized', 'sriovNetSupport']
for valid_instance_attribute in valid_instance_attributes:
response = client.describe_instance_attribute(InstanceId=instance_id, Attribute=valid_instance_attribute)
if valid_instance_attribute == "groupSet":
response.should.have.key("Groups")
response["Groups"].should.have.length_of(1)
response["Groups"][0]["GroupId"].should.equal(security_group_id)
elif valid_instance_attribute == "userData":
response.should.have.key("UserData")
response["UserData"].should.be.empty
invalid_instance_attributes = ['abc', 'Kernel', 'RamDisk', 'userdata', 'iNsTaNcEtYpE']
for invalid_instance_attribute in invalid_instance_attributes:
with assert_raises(ClientError) as ex:
client.describe_instance_attribute(InstanceId=instance_id, Attribute=invalid_instance_attribute)
ex.exception.response['Error']['Code'].should.equal('InvalidParameterValue')
ex.exception.response['ResponseMetadata']['HTTPStatusCode'].should.equal(400)
message = 'Value ({invalid_instance_attribute}) for parameter attribute is invalid. Unknown attribute.'.format(invalid_instance_attribute=invalid_instance_attribute)
ex.exception.response['Error']['Message'].should.equal(message)

111
tests/test_ecr/test_ecr_boto3.py
View file

@ -3,6 +3,8 @@ from __future__ import unicode_literals
import hashlib
import json
from datetime import datetime
from freezegun import freeze_time
import os
from random import random
import re
@ -13,6 +15,7 @@ from botocore.exceptions import ClientError, ParamValidationError
from dateutil.tz import tzlocal
from moto import mock_ecr
from nose import SkipTest
def _create_image_digest(contents=None):
@ -198,6 +201,42 @@ def test_put_image():
response['image']['repositoryName'].should.equal('test_repository')
response['image']['registryId'].should.equal('012345678910')
@mock_ecr
def test_put_image_with_push_date():
if os.environ.get('TEST_SERVER_MODE', 'false').lower() == 'true':
raise SkipTest('Cant manipulate time in server mode')
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
repositoryName='test_repository'
)
with freeze_time('2018-08-28 00:00:00'):
image1_date = datetime.now()
_ = client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(_create_image_manifest()),
imageTag='latest'
)
with freeze_time('2019-05-31 00:00:00'):
image2_date = datetime.now()
_ = client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(_create_image_manifest()),
imageTag='latest'
)
describe_response = client.describe_images(repositoryName='test_repository')
type(describe_response['imageDetails']).should.be(list)
len(describe_response['imageDetails']).should.be(2)
set([describe_response['imageDetails'][0]['imagePushedAt'],
describe_response['imageDetails'][1]['imagePushedAt']]).should.equal(set([image1_date, image2_date]))
@mock_ecr
def test_put_image_with_multiple_tags():
client = boto3.client('ecr', region_name='us-east-1')
@ -240,6 +279,7 @@ def test_put_image_with_multiple_tags():
len(response2['imageDetails'][0]['imageTags']).should.be(2)
response2['imageDetails'][0]['imageTags'].should.be.equal(['v1', 'latest'])
@mock_ecr
def test_list_images():
client = boto3.client('ecr', region_name='us-east-1')
@ -700,7 +740,7 @@ def test_batch_get_image_no_tags():
@mock_ecr
def test_batch_delete_image_by_tag():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -708,14 +748,13 @@ def test_batch_delete_image_by_tag():
tags = ['v1', 'v1.0', 'latest']
for tag in tags:
put_response = client.put_image(
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(manifest),
imageTag=tag,
)
describe_response1 = client.describe_images(repositoryName='test_repository')
image_digest = describe_response1['imageDetails'][0]['imageDigest']
batch_delete_response = client.batch_delete_image(
registryId='012345678910',
@ -744,10 +783,52 @@ def test_batch_delete_image_by_tag():
len(batch_delete_response['failures']).should.be(0)
@mock_ecr
def test_batch_delete_image_delete_last_tag():
client = boto3.client('ecr', region_name='us-east-1')
client.create_repository(
repositoryName='test_repository'
)
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(_create_image_manifest()),
imageTag='v1',
)
describe_response1 = client.describe_images(repositoryName='test_repository')
batch_delete_response = client.batch_delete_image(
registryId='012345678910',
repositoryName='test_repository',
imageIds=[
{
'imageTag': 'v1'
},
],
)
describe_response2 = client.describe_images(repositoryName='test_repository')
type(describe_response1['imageDetails'][0]['imageTags']).should.be(list)
len(describe_response1['imageDetails'][0]['imageTags']).should.be(1)
type(describe_response2['imageDetails']).should.be(list)
len(describe_response2['imageDetails']).should.be(0)
type(batch_delete_response['imageIds']).should.be(list)
len(batch_delete_response['imageIds']).should.be(1)
batch_delete_response['imageIds'][0]['imageTag'].should.equal("v1")
type(batch_delete_response['failures']).should.be(list)
len(batch_delete_response['failures']).should.be(0)
@mock_ecr
def test_batch_delete_image_with_nonexistent_tag():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -755,14 +836,13 @@ def test_batch_delete_image_with_nonexistent_tag():
tags = ['v1', 'v1.0', 'latest']
for tag in tags:
put_response = client.put_image(
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(manifest),
imageTag=tag,
)
describe_response = client.describe_images(repositoryName='test_repository')
image_digest = describe_response['imageDetails'][0]['imageDigest']
missing_tag = "missing-tag"
batch_delete_response = client.batch_delete_image(
@ -792,7 +872,7 @@ def test_batch_delete_image_with_nonexistent_tag():
@mock_ecr
def test_batch_delete_image_by_digest():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -800,7 +880,7 @@ def test_batch_delete_image_by_digest():
tags = ['v1', 'v2', 'latest']
for tag in tags:
put_response = client.put_image(
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(manifest),
imageTag=tag
@ -843,7 +923,7 @@ def test_batch_delete_image_by_digest():
@mock_ecr
def test_batch_delete_image_with_invalid_digest():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -851,13 +931,12 @@ def test_batch_delete_image_with_invalid_digest():
tags = ['v1', 'v2', 'latest']
for tag in tags:
put_response = client.put_image(
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(manifest),
imageTag=tag
)
describe_response = client.describe_images(repositoryName='test_repository')
invalid_image_digest = 'sha256:invalid-digest'
batch_delete_response = client.batch_delete_image(
@ -884,7 +963,7 @@ def test_batch_delete_image_with_invalid_digest():
@mock_ecr
def test_batch_delete_image_with_missing_parameters():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -910,7 +989,7 @@ def test_batch_delete_image_with_missing_parameters():
@mock_ecr
def test_batch_delete_image_with_matching_digest_and_tag():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -918,7 +997,7 @@ def test_batch_delete_image_with_matching_digest_and_tag():
tags = ['v1', 'v1.0', 'latest']
for tag in tags:
put_response = client.put_image(
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(manifest),
imageTag=tag
@ -962,7 +1041,7 @@ def test_batch_delete_image_with_matching_digest_and_tag():
@mock_ecr
def test_batch_delete_image_with_mismatched_digest_and_tag():
client = boto3.client('ecr', region_name='us-east-1')
_ = client.create_repository(
client.create_repository(
repositoryName='test_repository'
)
@ -970,7 +1049,7 @@ def test_batch_delete_image_with_mismatched_digest_and_tag():
tags = ['v1', 'latest']
for tag in tags:
put_response = client.put_image(
client.put_image(
repositoryName='test_repository',
imageManifest=json.dumps(manifest),
imageTag=tag

195
tests/test_glue/test_datacatalog.py
View file

@ -229,6 +229,26 @@ def test_delete_table():
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('Table myspecialtable not found')
@mock_glue
def test_batch_delete_table():
client = boto3.client('glue', region_name='us-east-1')
database_name = 'myspecialdatabase'
helpers.create_database(client, database_name)
table_name = 'myspecialtable'
table_input = helpers.create_table_input(database_name, table_name)
helpers.create_table(client, database_name, table_name, table_input)
result = client.batch_delete_table(DatabaseName=database_name, TablesToDelete=[table_name])
result['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
# confirm table is deleted
with assert_raises(ClientError) as exc:
helpers.get_table(client, database_name, table_name)
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('Table myspecialtable not found')
@mock_glue
def test_get_partitions_empty():
@ -310,6 +330,72 @@ def test_get_partition_not_found():
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
exc.exception.response['Error']['Message'].should.match('partition')
@mock_glue
def test_batch_create_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)
before = datetime.now(pytz.utc)
partition_inputs = []
for i in range(0, 20):
values = ["2018-10-{:2}".format(i)]
part_input = helpers.create_partition_input(database_name, table_name, values=values)
partition_inputs.append(part_input)
client.batch_create_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionInputList=partition_inputs
)
after = datetime.now(pytz.utc)
response = client.get_partitions(DatabaseName=database_name, TableName=table_name)
partitions = response['Partitions']
partitions.should.have.length_of(20)
for idx, partition in enumerate(partitions):
partition_input = partition_inputs[idx]
partition['TableName'].should.equal(table_name)
partition['StorageDescriptor'].should.equal(partition_input['StorageDescriptor'])
partition['Values'].should.equal(partition_input['Values'])
partition['CreationTime'].should.be.greater_than(before)
partition['CreationTime'].should.be.lower_than(after)
@mock_glue
def test_batch_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)
partition_input = helpers.create_partition_input(database_name, table_name, values=values)
response = client.batch_create_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionInputList=[partition_input]
)
response.should.have.key('Errors')
response['Errors'].should.have.length_of(1)
response['Errors'][0]['PartitionValues'].should.equal(values)
response['Errors'][0]['ErrorDetail']['ErrorCode'].should.equal('AlreadyExistsException')
@mock_glue
def test_get_partition():
@ -445,3 +531,112 @@ def test_update_partition_move():
partition['TableName'].should.equal(table_name)
partition['StorageDescriptor']['Columns'].should.equal([{'Name': 'country', 'Type': 'string'}])
@mock_glue
def test_delete_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)
part_input = helpers.create_partition_input(database_name, table_name, values=values)
helpers.create_partition(client, database_name, table_name, part_input)
client.delete_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionValues=values,
)
response = client.get_partitions(DatabaseName=database_name, TableName=table_name)
partitions = response['Partitions']
partitions.should.be.empty
@mock_glue
def test_delete_partition_bad_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)
with assert_raises(ClientError) as exc:
client.delete_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionValues=values,
)
exc.exception.response['Error']['Code'].should.equal('EntityNotFoundException')
@mock_glue
def test_batch_delete_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)
partition_inputs = []
for i in range(0, 20):
values = ["2018-10-{:2}".format(i)]
part_input = helpers.create_partition_input(database_name, table_name, values=values)
partition_inputs.append(part_input)
client.batch_create_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionInputList=partition_inputs
)
partition_values = [{"Values": p["Values"]} for p in partition_inputs]
response = client.batch_delete_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionsToDelete=partition_values,
)
response.should_not.have.key('Errors')
@mock_glue
def test_batch_delete_partition_with_bad_partitions():
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)
partition_inputs = []
for i in range(0, 20):
values = ["2018-10-{:2}".format(i)]
part_input = helpers.create_partition_input(database_name, table_name, values=values)
partition_inputs.append(part_input)
client.batch_create_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionInputList=partition_inputs
)
partition_values = [{"Values": p["Values"]} for p in partition_inputs]
partition_values.insert(5, {"Values": ["2018-11-01"]})
partition_values.insert(10, {"Values": ["2018-11-02"]})
partition_values.insert(15, {"Values": ["2018-11-03"]})
response = client.batch_delete_partition(
DatabaseName=database_name,
TableName=table_name,
PartitionsToDelete=partition_values,
)
response.should.have.key('Errors')
response['Errors'].should.have.length_of(3)
error_partitions = map(lambda x: x['PartitionValues'], response['Errors'])
['2018-11-01'].should.be.within(error_partitions)
['2018-11-02'].should.be.within(error_partitions)
['2018-11-03'].should.be.within(error_partitions)

221
tests/test_iam/test_iam.py
View file

@ -1,5 +1,6 @@
from __future__ import unicode_literals
import base64
import json
import boto
import boto3
@ -29,6 +30,44 @@ FyDHrtlrS80dPUQWNYHw++oACDpWO01LGLPPrGmuO/7cOdojPEd852q5gd+7W9xt
8vUH+pBa6IBLbvBp+szli51V3TLSWcoyy4ceJNQU2vCkTLoFdS0RLd/7tQ==
-----END CERTIFICATE-----"""
MOCK_POLICY = """
{
"Version": "2012-10-17",
"Statement":
{
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": "arn:aws:s3:::example_bucket"
}
}
"""
MOCK_POLICY_2 = """
{
"Version": "2012-10-17",
"Id": "2",
"Statement":
{
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": "arn:aws:s3:::example_bucket"
}
}
"""
MOCK_POLICY_3 = """
{
"Version": "2012-10-17",
"Id": "3",
"Statement":
{
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": "arn:aws:s3:::example_bucket"
}
}
"""
@mock_iam_deprecated()
def test_get_all_server_certs():
@ -243,12 +282,12 @@ def test_list_instance_profiles_for_role():
def test_list_role_policies():
conn = boto.connect_iam()
conn.create_role("my-role")
conn.put_role_policy("my-role", "test policy", "my policy")
conn.put_role_policy("my-role", "test policy", MOCK_POLICY)
role = conn.list_role_policies("my-role")
role.policy_names.should.have.length_of(1)
role.policy_names[0].should.equal("test policy")
conn.put_role_policy("my-role", "test policy 2", "another policy")
conn.put_role_policy("my-role", "test policy 2", MOCK_POLICY)
role = conn.list_role_policies("my-role")
role.policy_names.should.have.length_of(2)
@ -266,12 +305,21 @@ def test_put_role_policy():
conn = boto.connect_iam()
conn.create_role(
"my-role", assume_role_policy_document="some policy", path="my-path")
conn.put_role_policy("my-role", "test policy", "my policy")
conn.put_role_policy("my-role", "test policy", MOCK_POLICY)
policy = conn.get_role_policy(
"my-role", "test policy")['get_role_policy_response']['get_role_policy_result']['policy_name']
policy.should.equal("test policy")
@mock_iam
def test_get_role_policy():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_role(
RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="my-path")
with assert_raises(conn.exceptions.NoSuchEntityException):
conn.get_role_policy(RoleName="my-role", PolicyName="does-not-exist")
@mock_iam_deprecated()
def test_update_assume_role_policy():
conn = boto.connect_iam()
@ -286,7 +334,7 @@ def test_create_policy():
conn = boto3.client('iam', region_name='us-east-1')
response = conn.create_policy(
PolicyName="TestCreatePolicy",
PolicyDocument='{"some":"policy"}')
PolicyDocument=MOCK_POLICY)
response['Policy']['Arn'].should.equal("arn:aws:iam::123456789012:policy/TestCreatePolicy")
@ -299,20 +347,62 @@ def test_create_policy_versions():
PolicyDocument='{"some":"policy"}')
conn.create_policy(
PolicyName="TestCreatePolicyVersion",
PolicyDocument='{"some":"policy"}')
PolicyDocument=MOCK_POLICY)
version = conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion",
PolicyDocument='{"some":"policy"}',
PolicyDocument=MOCK_POLICY,
SetAsDefault=True)
version.get('PolicyVersion').get('Document').should.equal({'some': 'policy'})
version.get('PolicyVersion').get('Document').should.equal(json.loads(MOCK_POLICY))
version.get('PolicyVersion').get('VersionId').should.equal("v2")
version.get('PolicyVersion').get('IsDefaultVersion').should.be.ok
conn.delete_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion",
VersionId="v1")
version = conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion",
PolicyDocument='{"some":"policy"}')
PolicyDocument=MOCK_POLICY)
version.get('PolicyVersion').get('VersionId').should.equal("v3")
version.get('PolicyVersion').get('IsDefaultVersion').shouldnt.be.ok
@mock_iam
def test_create_many_policy_versions():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_policy(
PolicyName="TestCreateManyPolicyVersions",
PolicyDocument=MOCK_POLICY)
for _ in range(0, 4):
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestCreateManyPolicyVersions",
PolicyDocument=MOCK_POLICY)
with assert_raises(ClientError):
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestCreateManyPolicyVersions",
PolicyDocument=MOCK_POLICY)
@mock_iam
def test_set_default_policy_version():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_policy(
PolicyName="TestSetDefaultPolicyVersion",
PolicyDocument=MOCK_POLICY)
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestSetDefaultPolicyVersion",
PolicyDocument=MOCK_POLICY_2,
SetAsDefault=True)
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestSetDefaultPolicyVersion",
PolicyDocument=MOCK_POLICY_3,
SetAsDefault=True)
versions = conn.list_policy_versions(
PolicyArn="arn:aws:iam::123456789012:policy/TestSetDefaultPolicyVersion")
versions.get('Versions')[0].get('Document').should.equal(json.loads(MOCK_POLICY))
versions.get('Versions')[0].get('IsDefaultVersion').shouldnt.be.ok
versions.get('Versions')[1].get('Document').should.equal(json.loads(MOCK_POLICY_2))
versions.get('Versions')[1].get('IsDefaultVersion').shouldnt.be.ok
versions.get('Versions')[2].get('Document').should.equal(json.loads(MOCK_POLICY_3))
versions.get('Versions')[2].get('IsDefaultVersion').should.be.ok
@mock_iam
@ -320,10 +410,21 @@ def test_get_policy():
conn = boto3.client('iam', region_name='us-east-1')
response = conn.create_policy(
PolicyName="TestGetPolicy",
PolicyDocument='{"some":"policy"}')
PolicyDocument=MOCK_POLICY)
policy = conn.get_policy(
PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicy")
response['Policy']['Arn'].should.equal("arn:aws:iam::123456789012:policy/TestGetPolicy")
policy['Policy']['Arn'].should.equal("arn:aws:iam::123456789012:policy/TestGetPolicy")
@mock_iam
def test_get_aws_managed_policy():
conn = boto3.client('iam', region_name='us-east-1')
managed_policy_arn = 'arn:aws:iam::aws:policy/IAMUserChangePassword'
managed_policy_create_date = datetime.strptime("2016-11-15T00:25:16+00:00", "%Y-%m-%dT%H:%M:%S+00:00")
policy = conn.get_policy(
PolicyArn=managed_policy_arn)
policy['Policy']['Arn'].should.equal(managed_policy_arn)
policy['Policy']['CreateDate'].replace(tzinfo=None).should.equal(managed_policy_create_date)
@mock_iam
@ -331,10 +432,10 @@ def test_get_policy_version():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_policy(
PolicyName="TestGetPolicyVersion",
PolicyDocument='{"some":"policy"}')
PolicyDocument=MOCK_POLICY)
version = conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicyVersion",
PolicyDocument='{"some":"policy"}')
PolicyDocument=MOCK_POLICY)
with assert_raises(ClientError):
conn.get_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicyVersion",
@ -342,7 +443,40 @@ def test_get_policy_version():
retrieved = conn.get_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicyVersion",
VersionId=version.get('PolicyVersion').get('VersionId'))
retrieved.get('PolicyVersion').get('Document').should.equal({'some': 'policy'})
retrieved.get('PolicyVersion').get('Document').should.equal(json.loads(MOCK_POLICY))
retrieved.get('PolicyVersion').get('IsDefaultVersion').shouldnt.be.ok
@mock_iam
def test_get_aws_managed_policy_version():
conn = boto3.client('iam', region_name='us-east-1')
managed_policy_arn = 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
managed_policy_version_create_date = datetime.strptime("2015-04-09T15:03:43+00:00", "%Y-%m-%dT%H:%M:%S+00:00")
with assert_raises(ClientError):
conn.get_policy_version(
PolicyArn=managed_policy_arn,
VersionId='v2-does-not-exist')
retrieved = conn.get_policy_version(
PolicyArn=managed_policy_arn,
VersionId="v1")
retrieved['PolicyVersion']['CreateDate'].replace(tzinfo=None).should.equal(managed_policy_version_create_date)
retrieved['PolicyVersion']['Document'].should.be.an(dict)
@mock_iam
def test_get_aws_managed_policy_v4_version():
conn = boto3.client('iam', region_name='us-east-1')
managed_policy_arn = 'arn:aws:iam::aws:policy/job-function/SystemAdministrator'
managed_policy_version_create_date = datetime.strptime("2018-10-08T21:33:45+00:00", "%Y-%m-%dT%H:%M:%S+00:00")
with assert_raises(ClientError):
conn.get_policy_version(
PolicyArn=managed_policy_arn,
VersionId='v2-does-not-exist')
retrieved = conn.get_policy_version(
PolicyArn=managed_policy_arn,
VersionId="v4")
retrieved['PolicyVersion']['CreateDate'].replace(tzinfo=None).should.equal(managed_policy_version_create_date)
retrieved['PolicyVersion']['Document'].should.be.an(dict)
@mock_iam
@ -353,22 +487,24 @@ def test_list_policy_versions():
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions")
conn.create_policy(
PolicyName="TestListPolicyVersions",
PolicyDocument='{"first":"policy"}')
PolicyDocument=MOCK_POLICY)
versions = conn.list_policy_versions(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions")
versions.get('Versions')[0].get('VersionId').should.equal('v1')
versions.get('Versions')[0].get('IsDefaultVersion').should.be.ok
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions",
PolicyDocument='{"second":"policy"}')
PolicyDocument=MOCK_POLICY_2)
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions",
PolicyDocument='{"third":"policy"}')
PolicyDocument=MOCK_POLICY_3)
versions = conn.list_policy_versions(
PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions")
print(versions.get('Versions'))
versions.get('Versions')[1].get('Document').should.equal({'second': 'policy'})
versions.get('Versions')[2].get('Document').should.equal({'third': 'policy'})
versions.get('Versions')[1].get('Document').should.equal(json.loads(MOCK_POLICY_2))
versions.get('Versions')[1].get('IsDefaultVersion').shouldnt.be.ok
versions.get('Versions')[2].get('Document').should.equal(json.loads(MOCK_POLICY_3))
versions.get('Versions')[2].get('IsDefaultVersion').shouldnt.be.ok
@mock_iam
@ -376,10 +512,10 @@ def test_delete_policy_version():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_policy(
PolicyName="TestDeletePolicyVersion",
PolicyDocument='{"first":"policy"}')
PolicyDocument=MOCK_POLICY)
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
PolicyDocument='{"second":"policy"}')
PolicyDocument=MOCK_POLICY)
with assert_raises(ClientError):
conn.delete_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
@ -392,6 +528,21 @@ def test_delete_policy_version():
len(versions.get('Versions')).should.equal(1)
@mock_iam
def test_delete_default_policy_version():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_policy(
PolicyName="TestDeletePolicyVersion",
PolicyDocument=MOCK_POLICY)
conn.create_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
PolicyDocument=MOCK_POLICY_2)
with assert_raises(ClientError):
conn.delete_policy_version(
PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion",
VersionId='v1')
@mock_iam_deprecated()
def test_create_user():
conn = boto.connect_iam()
@ -446,22 +597,20 @@ def test_list_users():
@mock_iam()
def test_user_policies():
policy_name = 'UserManagedPolicy'
policy_document = "{'mypolicy': 'test'}"
user_name = 'my-user'
conn = boto3.client('iam', region_name='us-east-1')
conn.create_user(UserName=user_name)
conn.put_user_policy(
UserName=user_name,
PolicyName=policy_name,
PolicyDocument=policy_document
PolicyDocument=MOCK_POLICY
)
policy_doc = conn.get_user_policy(
UserName=user_name,
PolicyName=policy_name
)
test = policy_document in policy_doc['PolicyDocument']
test.should.equal(True)
policy_doc['PolicyDocument'].should.equal(json.loads(MOCK_POLICY))
policies = conn.list_user_policies(UserName=user_name)
len(policies['PolicyNames']).should.equal(1)
@ -497,13 +646,17 @@ def test_delete_login_profile():
conn.delete_login_profile('my-user')
@mock_iam_deprecated()
@mock_iam()
def test_create_access_key():
conn = boto.connect_iam()
with assert_raises(BotoServerError):
conn.create_access_key('my-user')
conn.create_user('my-user')
conn.create_access_key('my-user')
conn = boto3.client('iam', region_name='us-east-1')
with assert_raises(ClientError):
conn.create_access_key(UserName='my-user')
conn.create_user(UserName='my-user')
access_key = conn.create_access_key(UserName='my-user')["AccessKey"]
(datetime.utcnow() - access_key["CreateDate"].replace(tzinfo=None)).seconds.should.be.within(0, 10)
access_key["AccessKeyId"].should.have.length_of(20)
access_key["SecretAccessKey"].should.have.length_of(40)
assert access_key["AccessKeyId"].startswith("AKIA")
@mock_iam_deprecated()
@ -622,7 +775,7 @@ def test_managed_policy():
conn = boto.connect_iam()
conn.create_policy(policy_name='UserManagedPolicy',
policy_document={'mypolicy': 'test'},
policy_document=MOCK_POLICY,
path='/mypolicy/',
description='my user managed policy')
@ -723,7 +876,7 @@ def test_attach_detach_user_policy():
policy_name = 'UserAttachedPolicy'
policy = iam.create_policy(PolicyName=policy_name,
PolicyDocument='{"mypolicy": "test"}',
PolicyDocument=MOCK_POLICY,
Path='/mypolicy/',
Description='my user attached policy')
@ -779,7 +932,6 @@ def test_get_access_key_last_used():
@mock_iam
def test_get_account_authorization_details():
import json
test_policy = json.dumps({
"Version": "2012-10-17",
"Statement": [
@ -1211,7 +1363,6 @@ def test_update_role():
@mock_iam()
def test_list_entities_for_policy():
import json
test_policy = json.dumps({
"Version": "2012-10-17",
"Statement": [

20
tests/test_iam/test_iam_groups.py
View file

@ -10,6 +10,18 @@ from nose.tools import assert_raises
from boto.exception import BotoServerError
from moto import mock_iam, mock_iam_deprecated
MOCK_POLICY = """
{
"Version": "2012-10-17",
"Statement":
{
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": "arn:aws:s3:::example_bucket"
}
}
"""
@mock_iam_deprecated()
def test_create_group():
@ -101,7 +113,7 @@ def test_get_groups_for_user():
def test_put_group_policy():
conn = boto.connect_iam()
conn.create_group('my-group')
conn.put_group_policy('my-group', 'my-policy', '{"some": "json"}')
conn.put_group_policy('my-group', 'my-policy', MOCK_POLICY)
@mock_iam
@ -131,7 +143,7 @@ def test_get_group_policy():
with assert_raises(BotoServerError):
conn.get_group_policy('my-group', 'my-policy')
conn.put_group_policy('my-group', 'my-policy', '{"some": "json"}')
conn.put_group_policy('my-group', 'my-policy', MOCK_POLICY)
conn.get_group_policy('my-group', 'my-policy')
@ -141,7 +153,7 @@ def test_get_all_group_policies():
conn.create_group('my-group')
policies = conn.get_all_group_policies('my-group')['list_group_policies_response']['list_group_policies_result']['policy_names']
assert policies == []
conn.put_group_policy('my-group', 'my-policy', '{"some": "json"}')
conn.put_group_policy('my-group', 'my-policy', MOCK_POLICY)
policies = conn.get_all_group_policies('my-group')['list_group_policies_response']['list_group_policies_result']['policy_names']
assert policies == ['my-policy']
@ -151,5 +163,5 @@ def test_list_group_policies():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_group(GroupName='my-group')
conn.list_group_policies(GroupName='my-group')['PolicyNames'].should.be.empty
conn.put_group_policy(GroupName='my-group', PolicyName='my-policy', PolicyDocument='{"some": "json"}')
conn.put_group_policy(GroupName='my-group', PolicyName='my-policy', PolicyDocument=MOCK_POLICY)
conn.list_group_policies(GroupName='my-group')['PolicyNames'].should.equal(['my-policy'])

1861
tests/test_iam/test_iam_policies.py Normal file
View file

File diff suppressed because it is too large Load diff

4
tests/test_kinesis/test_kinesis.py
View file

@ -15,7 +15,7 @@ from moto import mock_kinesis, mock_kinesis_deprecated
def test_create_cluster():
conn = boto.kinesis.connect_to_region("us-west-2")
conn.create_stream("my_stream", 2)
conn.create_stream("my_stream", 3)
stream_response = conn.describe_stream("my_stream")
@ -27,7 +27,7 @@ def test_create_cluster():
stream["StreamStatus"].should.equal("ACTIVE")
shards = stream['Shards']
shards.should.have.length_of(2)
shards.should.have.length_of(3)
@mock_kinesis_deprecated

26
tests/test_route53/test_route53.py
View file

@ -110,6 +110,7 @@ def test_rrset():
changes = ResourceRecordSets(conn, zoneid)
changes.add_change("DELETE", "foo.bar.testdns.aws.com", "A")
changes.add_change("DELETE", "foo.bar.testdns.aws.com", "TXT")
changes.commit()
changes = ResourceRecordSets(conn, zoneid)
@ -123,12 +124,12 @@ def test_rrset():
rrsets.should.have.length_of(2)
rrsets = conn.get_all_rrsets(
zoneid, name="foo.bar.testdns.aws.com", type="A")
zoneid, name="bar.foo.testdns.aws.com", type="A")
rrsets.should.have.length_of(1)
rrsets[0].resource_records[0].should.equal('1.2.3.4')
rrsets[0].resource_records[0].should.equal('5.6.7.8')
rrsets = conn.get_all_rrsets(
zoneid, name="bar.foo.testdns.aws.com", type="A")
zoneid, name="foo.bar.testdns.aws.com", type="A")
rrsets.should.have.length_of(2)
resource_records = [rr for rr_set in rrsets for rr in rr_set.resource_records]
resource_records.should.contain('1.2.3.4')
@ -617,7 +618,7 @@ def test_change_resource_record_sets_crud_valid():
})
cname_alias_record_detail.should_not.contain('ResourceRecords')
# Delete record.
# Delete record with wrong type.
delete_payload = {
'Comment': 'delete prod.redis.db',
'Changes': [
@ -632,6 +633,23 @@ def test_change_resource_record_sets_crud_valid():
}
conn.change_resource_record_sets(HostedZoneId=hosted_zone_id, ChangeBatch=delete_payload)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response['ResourceRecordSets']).should.equal(1)
# Delete record.
delete_payload = {
'Comment': 'delete prod.redis.db',
'Changes': [
{
'Action': 'DELETE',
'ResourceRecordSet': {
'Name': 'prod.redis.db',
'Type': 'A',
}
}
]
}
conn.change_resource_record_sets(HostedZoneId=hosted_zone_id, ChangeBatch=delete_payload)
response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
len(response['ResourceRecordSets']).should.equal(0)

12
tests/test_sns/test_topics_boto3.py
View file

@ -32,6 +32,18 @@ def test_create_and_delete_topic():
topics = topics_json["Topics"]
topics.should.have.length_of(0)
@mock_sns
def test_create_topic_with_attributes():
conn = boto3.client("sns", region_name="us-east-1")
conn.create_topic(Name='some-topic-with-attribute', Attributes={'DisplayName': 'test-topic'})
topics_json = conn.list_topics()
topic_arn = topics_json["Topics"][0]['TopicArn']
attributes = conn.get_topic_attributes(TopicArn=topic_arn)['Attributes']
attributes['DisplayName'].should.equal('test-topic')
@mock_sns
def test_create_topic_should_be_indempodent():
conn = boto3.client("sns", region_name="us-east-1")