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

Merge remote-tracking branch 'upstream/master' into feat/cognitoidp-add-update-identity-provider

Browse source
This commit is contained in:
Robert Jensen 2019-03-12 17:37:07 -04:00
commit c7d3e1f8b6
52 changed files with 3081 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

91
tests/test_awslambda/test_lambda.py
View file

@ -12,6 +12,8 @@ import sure # noqa
from freezegun import freeze_time
from moto import mock_lambda, mock_s3, mock_ec2, mock_sns, mock_logs, settings
from nose.tools import assert_raises
from botocore.exceptions import ClientError
_lambda_region = 'us-west-2'
@ -397,6 +399,11 @@ def test_get_function():
result = conn.get_function(FunctionName='testFunction', Qualifier='$LATEST')
result['Configuration']['Version'].should.equal('$LATEST')
# Test get function when can't find function name
with assert_raises(ClientError):
conn.get_function(FunctionName='junk', Qualifier='$LATEST')
@mock_lambda
@mock_s3
@ -819,3 +826,87 @@ def get_function_policy():
assert isinstance(response['Policy'], str)
res = json.loads(response['Policy'])
assert res['Statement'][0]['Action'] == 'lambda:InvokeFunction'
@mock_lambda
@mock_s3
def test_list_versions_by_function():
s3_conn = boto3.client('s3', 'us-west-2')
s3_conn.create_bucket(Bucket='test-bucket')
zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
conn = boto3.client('lambda', 'us-west-2')
conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'S3Bucket': 'test-bucket',
'S3Key': 'test.zip',
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
conn.publish_version(FunctionName='testFunction')
versions = conn.list_versions_by_function(FunctionName='testFunction')
assert versions['Versions'][0]['FunctionArn'] == 'arn:aws:lambda:us-west-2:123456789012:function:testFunction:$LATEST'
@mock_lambda
@mock_s3
def test_create_function_with_already_exists():
s3_conn = boto3.client('s3', 'us-west-2')
s3_conn.create_bucket(Bucket='test-bucket')
zip_content = get_test_zip_file2()
s3_conn.put_object(Bucket='test-bucket', Key='test.zip', Body=zip_content)
conn = boto3.client('lambda', 'us-west-2')
conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'S3Bucket': 'test-bucket',
'S3Key': 'test.zip',
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'S3Bucket': 'test-bucket',
'S3Key': 'test.zip',
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
assert response['FunctionName'] == 'testFunction'
@mock_lambda
@mock_s3
def test_list_versions_by_function_for_nonexistent_function():
conn = boto3.client('lambda', 'us-west-2')
versions = conn.list_versions_by_function(FunctionName='testFunction')
assert len(versions['Versions']) == 0

59
tests/test_batch/test_batch.py
View file

@ -323,6 +323,54 @@ def test_create_job_queue():
resp.should.contain('jobQueues')
len(resp['jobQueues']).should.equal(0)
# Create job queue which already exists
try:
resp = batch_client.create_job_queue(
jobQueueName='test_job_queue',
state='ENABLED',
priority=123,
computeEnvironmentOrder=[
{
'order': 123,
'computeEnvironment': arn
},
]
)
except ClientError as err:
err.response['Error']['Code'].should.equal('ClientException')
# Create job queue with incorrect state
try:
resp = batch_client.create_job_queue(
jobQueueName='test_job_queue2',
state='JUNK',
priority=123,
computeEnvironmentOrder=[
{
'order': 123,
'computeEnvironment': arn
},
]
)
except ClientError as err:
err.response['Error']['Code'].should.equal('ClientException')
# Create job queue with no compute env
try:
resp = batch_client.create_job_queue(
jobQueueName='test_job_queue3',
state='JUNK',
priority=123,
computeEnvironmentOrder=[
]
)
except ClientError as err:
err.response['Error']['Code'].should.equal('ClientException')
@mock_ec2
@mock_ecs
@ -397,6 +445,17 @@ def test_update_job_queue():
len(resp['jobQueues']).should.equal(1)
resp['jobQueues'][0]['priority'].should.equal(5)
batch_client.update_job_queue(
jobQueue='test_job_queue',
priority=5
)
resp = batch_client.describe_job_queues()
resp.should.contain('jobQueues')
len(resp['jobQueues']).should.equal(1)
resp['jobQueues'][0]['priority'].should.equal(5)
@mock_ec2
@mock_ecs

58
tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
View file

@ -399,6 +399,32 @@ def test_create_change_set_from_s3_url():
assert 'arn:aws:cloudformation:us-east-1:123456789:stack/NewStack' in response['StackId']
@mock_cloudformation
def test_describe_change_set():
cf_conn = boto3.client('cloudformation', region_name='us-east-1')
cf_conn.create_change_set(
StackName='NewStack',
TemplateBody=dummy_template_json,
ChangeSetName='NewChangeSet',
ChangeSetType='CREATE',
)
stack = cf_conn.describe_change_set(ChangeSetName="NewChangeSet")
stack['ChangeSetName'].should.equal('NewChangeSet')
stack['StackName'].should.equal('NewStack')
cf_conn.create_change_set(
StackName='NewStack',
TemplateBody=dummy_update_template_json,
ChangeSetName='NewChangeSet2',
ChangeSetType='UPDATE',
)
stack = cf_conn.describe_change_set(ChangeSetName="NewChangeSet2")
stack['ChangeSetName'].should.equal('NewChangeSet2')
stack['StackName'].should.equal('NewStack')
stack['Changes'].should.have.length_of(2)
@mock_cloudformation
def test_execute_change_set_w_arn():
cf_conn = boto3.client('cloudformation', region_name='us-east-1')
@ -420,7 +446,7 @@ def test_execute_change_set_w_name():
ChangeSetName='NewChangeSet',
ChangeSetType='CREATE',
)
cf_conn.execute_change_set(ChangeSetName='NewStack', StackName='NewStack')
cf_conn.execute_change_set(ChangeSetName='NewChangeSet', StackName='NewStack')
@mock_cloudformation
@ -489,6 +515,20 @@ def test_describe_stack_by_stack_id():
stack_by_id['StackName'].should.equal("test_stack")
@mock_cloudformation
def test_list_change_sets():
cf_conn = boto3.client('cloudformation', region_name='us-east-1')
cf_conn.create_change_set(
StackName='NewStack2',
TemplateBody=dummy_template_json,
ChangeSetName='NewChangeSet2',
ChangeSetType='CREATE',
)
change_set = cf_conn.list_change_sets(StackName='NewStack2')['Summaries'][0]
change_set['StackName'].should.equal('NewStack2')
change_set['ChangeSetName'].should.equal('NewChangeSet2')
@mock_cloudformation
def test_list_stacks():
cf = boto3.resource('cloudformation', region_name='us-east-1')
@ -521,6 +561,22 @@ def test_delete_stack_from_resource():
list(cf.stacks.all()).should.have.length_of(0)
@mock_cloudformation
@mock_ec2
def test_delete_change_set():
cf_conn = boto3.client('cloudformation', region_name='us-east-1')
cf_conn.create_change_set(
StackName='NewStack',
TemplateBody=dummy_template_json,
ChangeSetName='NewChangeSet',
ChangeSetType='CREATE',
)
cf_conn.list_change_sets(StackName='NewStack')['Summaries'].should.have.length_of(1)
cf_conn.delete_change_set(ChangeSetName='NewChangeSet', StackName='NewStack')
cf_conn.list_change_sets(StackName='NewStack')['Summaries'].should.have.length_of(0)
@mock_cloudformation
@mock_ec2
def test_delete_stack_by_name():

491
tests/test_config/test_config.py Normal file
View file

@ -0,0 +1,491 @@
from datetime import datetime, timedelta
import boto3
from botocore.exceptions import ClientError
from nose.tools import assert_raises
from moto.config import mock_config
@mock_config
def test_put_configuration_recorder():
client = boto3.client('config', region_name='us-west-2')
# Try without a name supplied:
with assert_raises(ClientError) as ce:
client.put_configuration_recorder(ConfigurationRecorder={'roleARN': 'somearn'})
assert ce.exception.response['Error']['Code'] == 'InvalidConfigurationRecorderNameException'
assert 'is not valid, blank string.' in ce.exception.response['Error']['Message']
# Try with a really long name:
with assert_raises(ClientError) as ce:
client.put_configuration_recorder(ConfigurationRecorder={'name': 'a' * 257, 'roleARN': 'somearn'})
assert ce.exception.response['Error']['Code'] == 'ValidationException'
assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']
# With resource types and flags set to True:
bad_groups = [
{'allSupported': True, 'includeGlobalResourceTypes': True, 'resourceTypes': ['item']},
{'allSupported': False, 'includeGlobalResourceTypes': True, 'resourceTypes': ['item']},
{'allSupported': True, 'includeGlobalResourceTypes': False, 'resourceTypes': ['item']},
{'allSupported': False, 'includeGlobalResourceTypes': False, 'resourceTypes': []},
{'includeGlobalResourceTypes': False, 'resourceTypes': []},
{'includeGlobalResourceTypes': True},
{'resourceTypes': []},
{}
]
for bg in bad_groups:
with assert_raises(ClientError) as ce:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'default',
'roleARN': 'somearn',
'recordingGroup': bg
})
assert ce.exception.response['Error']['Code'] == 'InvalidRecordingGroupException'
assert ce.exception.response['Error']['Message'] == 'The recording group provided is not valid'
# With an invalid Resource Type:
with assert_raises(ClientError) as ce:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'default',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
# 2 good, and 2 bad:
'resourceTypes': ['AWS::EC2::Volume', 'LOLNO', 'AWS::EC2::VPC', 'LOLSTILLNO']
}
})
assert ce.exception.response['Error']['Code'] == 'ValidationException'
assert "2 validation error detected: Value '['LOLNO', 'LOLSTILLNO']" in str(ce.exception.response['Error']['Message'])
assert 'AWS::EC2::Instance' in ce.exception.response['Error']['Message']
# Create a proper one:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
result = client.describe_configuration_recorders()['ConfigurationRecorders']
assert len(result) == 1
assert result[0]['name'] == 'testrecorder'
assert result[0]['roleARN'] == 'somearn'
assert not result[0]['recordingGroup']['allSupported']
assert not result[0]['recordingGroup']['includeGlobalResourceTypes']
assert len(result[0]['recordingGroup']['resourceTypes']) == 2
assert 'AWS::EC2::Volume' in result[0]['recordingGroup']['resourceTypes'] \
and 'AWS::EC2::VPC' in result[0]['recordingGroup']['resourceTypes']
# Now update the configuration recorder:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': True,
'includeGlobalResourceTypes': True
}
})
result = client.describe_configuration_recorders()['ConfigurationRecorders']
assert len(result) == 1
assert result[0]['name'] == 'testrecorder'
assert result[0]['roleARN'] == 'somearn'
assert result[0]['recordingGroup']['allSupported']
assert result[0]['recordingGroup']['includeGlobalResourceTypes']
assert len(result[0]['recordingGroup']['resourceTypes']) == 0
# With a default recording group (i.e. lacking one)
client.put_configuration_recorder(ConfigurationRecorder={'name': 'testrecorder', 'roleARN': 'somearn'})
result = client.describe_configuration_recorders()['ConfigurationRecorders']
assert len(result) == 1
assert result[0]['name'] == 'testrecorder'
assert result[0]['roleARN'] == 'somearn'
assert result[0]['recordingGroup']['allSupported']
assert not result[0]['recordingGroup']['includeGlobalResourceTypes']
assert not result[0]['recordingGroup'].get('resourceTypes')
# Can currently only have exactly 1 Config Recorder in an account/region:
with assert_raises(ClientError) as ce:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'someotherrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
}
})
assert ce.exception.response['Error']['Code'] == 'MaxNumberOfConfigurationRecordersExceededException'
assert "maximum number of configuration recorders: 1 is reached." in ce.exception.response['Error']['Message']
@mock_config
def test_describe_configurations():
client = boto3.client('config', region_name='us-west-2')
# Without any configurations:
result = client.describe_configuration_recorders()
assert not result['ConfigurationRecorders']
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
result = client.describe_configuration_recorders()['ConfigurationRecorders']
assert len(result) == 1
assert result[0]['name'] == 'testrecorder'
assert result[0]['roleARN'] == 'somearn'
assert not result[0]['recordingGroup']['allSupported']
assert not result[0]['recordingGroup']['includeGlobalResourceTypes']
assert len(result[0]['recordingGroup']['resourceTypes']) == 2
assert 'AWS::EC2::Volume' in result[0]['recordingGroup']['resourceTypes'] \
and 'AWS::EC2::VPC' in result[0]['recordingGroup']['resourceTypes']
# Specify an incorrect name:
with assert_raises(ClientError) as ce:
client.describe_configuration_recorders(ConfigurationRecorderNames=['wrong'])
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'
assert 'wrong' in ce.exception.response['Error']['Message']
# And with both a good and wrong name:
with assert_raises(ClientError) as ce:
client.describe_configuration_recorders(ConfigurationRecorderNames=['testrecorder', 'wrong'])
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'
assert 'wrong' in ce.exception.response['Error']['Message']
@mock_config
def test_delivery_channels():
client = boto3.client('config', region_name='us-west-2')
# Try without a config recorder:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={})
assert ce.exception.response['Error']['Code'] == 'NoAvailableConfigurationRecorderException'
assert ce.exception.response['Error']['Message'] == 'Configuration recorder is not available to ' \
'put delivery channel.'
# Create a config recorder to continue testing:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
# Try without a name supplied:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={})
assert ce.exception.response['Error']['Code'] == 'InvalidDeliveryChannelNameException'
assert 'is not valid, blank string.' in ce.exception.response['Error']['Message']
# Try with a really long name:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={'name': 'a' * 257})
assert ce.exception.response['Error']['Code'] == 'ValidationException'
assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']
# Without specifying a bucket name:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel'})
assert ce.exception.response['Error']['Code'] == 'NoSuchBucketException'
assert ce.exception.response['Error']['Message'] == 'Cannot find a S3 bucket with an empty bucket name.'
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': ''})
assert ce.exception.response['Error']['Code'] == 'NoSuchBucketException'
assert ce.exception.response['Error']['Message'] == 'Cannot find a S3 bucket with an empty bucket name.'
# With an empty string for the S3 key prefix:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={
'name': 'testchannel', 's3BucketName': 'somebucket', 's3KeyPrefix': ''})
assert ce.exception.response['Error']['Code'] == 'InvalidS3KeyPrefixException'
assert 'empty s3 key prefix.' in ce.exception.response['Error']['Message']
# With an empty string for the SNS ARN:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={
'name': 'testchannel', 's3BucketName': 'somebucket', 'snsTopicARN': ''})
assert ce.exception.response['Error']['Code'] == 'InvalidSNSTopicARNException'
assert 'The sns topic arn' in ce.exception.response['Error']['Message']
# With an invalid delivery frequency:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={
'name': 'testchannel',
's3BucketName': 'somebucket',
'configSnapshotDeliveryProperties': {'deliveryFrequency': 'WRONG'}
})
assert ce.exception.response['Error']['Code'] == 'InvalidDeliveryFrequency'
assert 'WRONG' in ce.exception.response['Error']['Message']
assert 'TwentyFour_Hours' in ce.exception.response['Error']['Message']
# Create a proper one:
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})
result = client.describe_delivery_channels()['DeliveryChannels']
assert len(result) == 1
assert len(result[0].keys()) == 2
assert result[0]['name'] == 'testchannel'
assert result[0]['s3BucketName'] == 'somebucket'
# Overwrite it with another proper configuration:
client.put_delivery_channel(DeliveryChannel={
'name': 'testchannel',
's3BucketName': 'somebucket',
'snsTopicARN': 'sometopicarn',
'configSnapshotDeliveryProperties': {'deliveryFrequency': 'TwentyFour_Hours'}
})
result = client.describe_delivery_channels()['DeliveryChannels']
assert len(result) == 1
assert len(result[0].keys()) == 4
assert result[0]['name'] == 'testchannel'
assert result[0]['s3BucketName'] == 'somebucket'
assert result[0]['snsTopicARN'] == 'sometopicarn'
assert result[0]['configSnapshotDeliveryProperties']['deliveryFrequency'] == 'TwentyFour_Hours'
# Can only have 1:
with assert_raises(ClientError) as ce:
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel2', 's3BucketName': 'somebucket'})
assert ce.exception.response['Error']['Code'] == 'MaxNumberOfDeliveryChannelsExceededException'
assert 'because the maximum number of delivery channels: 1 is reached.' in ce.exception.response['Error']['Message']
@mock_config
def test_describe_delivery_channels():
client = boto3.client('config', region_name='us-west-2')
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
# Without any channels:
result = client.describe_delivery_channels()
assert not result['DeliveryChannels']
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})
result = client.describe_delivery_channels()['DeliveryChannels']
assert len(result) == 1
assert len(result[0].keys()) == 2
assert result[0]['name'] == 'testchannel'
assert result[0]['s3BucketName'] == 'somebucket'
# Overwrite it with another proper configuration:
client.put_delivery_channel(DeliveryChannel={
'name': 'testchannel',
's3BucketName': 'somebucket',
'snsTopicARN': 'sometopicarn',
'configSnapshotDeliveryProperties': {'deliveryFrequency': 'TwentyFour_Hours'}
})
result = client.describe_delivery_channels()['DeliveryChannels']
assert len(result) == 1
assert len(result[0].keys()) == 4
assert result[0]['name'] == 'testchannel'
assert result[0]['s3BucketName'] == 'somebucket'
assert result[0]['snsTopicARN'] == 'sometopicarn'
assert result[0]['configSnapshotDeliveryProperties']['deliveryFrequency'] == 'TwentyFour_Hours'
# Specify an incorrect name:
with assert_raises(ClientError) as ce:
client.describe_delivery_channels(DeliveryChannelNames=['wrong'])
assert ce.exception.response['Error']['Code'] == 'NoSuchDeliveryChannelException'
assert 'wrong' in ce.exception.response['Error']['Message']
# And with both a good and wrong name:
with assert_raises(ClientError) as ce:
client.describe_delivery_channels(DeliveryChannelNames=['testchannel', 'wrong'])
assert ce.exception.response['Error']['Code'] == 'NoSuchDeliveryChannelException'
assert 'wrong' in ce.exception.response['Error']['Message']
@mock_config
def test_start_configuration_recorder():
client = boto3.client('config', region_name='us-west-2')
# Without a config recorder:
with assert_raises(ClientError) as ce:
client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'
# Make the config recorder;
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
# Without a delivery channel:
with assert_raises(ClientError) as ce:
client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')
assert ce.exception.response['Error']['Code'] == 'NoAvailableDeliveryChannelException'
# Make the delivery channel:
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})
# Start it:
client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')
# Verify it's enabled:
result = client.describe_configuration_recorder_status()['ConfigurationRecordersStatus']
lower_bound = (datetime.utcnow() - timedelta(minutes=5))
assert result[0]['recording']
assert result[0]['lastStatus'] == 'PENDING'
assert lower_bound < result[0]['lastStartTime'].replace(tzinfo=None) <= datetime.utcnow()
assert lower_bound < result[0]['lastStatusChangeTime'].replace(tzinfo=None) <= datetime.utcnow()
@mock_config
def test_stop_configuration_recorder():
client = boto3.client('config', region_name='us-west-2')
# Without a config recorder:
with assert_raises(ClientError) as ce:
client.stop_configuration_recorder(ConfigurationRecorderName='testrecorder')
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'
# Make the config recorder;
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
# Make the delivery channel for creation:
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})
# Start it:
client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')
client.stop_configuration_recorder(ConfigurationRecorderName='testrecorder')
# Verify it's disabled:
result = client.describe_configuration_recorder_status()['ConfigurationRecordersStatus']
lower_bound = (datetime.utcnow() - timedelta(minutes=5))
assert not result[0]['recording']
assert result[0]['lastStatus'] == 'PENDING'
assert lower_bound < result[0]['lastStartTime'].replace(tzinfo=None) <= datetime.utcnow()
assert lower_bound < result[0]['lastStopTime'].replace(tzinfo=None) <= datetime.utcnow()
assert lower_bound < result[0]['lastStatusChangeTime'].replace(tzinfo=None) <= datetime.utcnow()
@mock_config
def test_describe_configuration_recorder_status():
client = boto3.client('config', region_name='us-west-2')
# Without any:
result = client.describe_configuration_recorder_status()
assert not result['ConfigurationRecordersStatus']
# Make the config recorder;
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
# Without specifying a config recorder:
result = client.describe_configuration_recorder_status()['ConfigurationRecordersStatus']
assert len(result) == 1
assert result[0]['name'] == 'testrecorder'
assert not result[0]['recording']
# With a proper name:
result = client.describe_configuration_recorder_status(
ConfigurationRecorderNames=['testrecorder'])['ConfigurationRecordersStatus']
assert len(result) == 1
assert result[0]['name'] == 'testrecorder'
assert not result[0]['recording']
# Invalid name:
with assert_raises(ClientError) as ce:
client.describe_configuration_recorder_status(ConfigurationRecorderNames=['testrecorder', 'wrong'])
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'
assert 'wrong' in ce.exception.response['Error']['Message']
@mock_config
def test_delete_configuration_recorder():
client = boto3.client('config', region_name='us-west-2')
# Make the config recorder;
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
# Delete it:
client.delete_configuration_recorder(ConfigurationRecorderName='testrecorder')
# Try again -- it should be deleted:
with assert_raises(ClientError) as ce:
client.delete_configuration_recorder(ConfigurationRecorderName='testrecorder')
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'
@mock_config
def test_delete_delivery_channel():
client = boto3.client('config', region_name='us-west-2')
# Need a recorder to test the constraint on recording being enabled:
client.put_configuration_recorder(ConfigurationRecorder={
'name': 'testrecorder',
'roleARN': 'somearn',
'recordingGroup': {
'allSupported': False,
'includeGlobalResourceTypes': False,
'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']
}
})
client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})
client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')
# With the recorder enabled:
with assert_raises(ClientError) as ce:
client.delete_delivery_channel(DeliveryChannelName='testchannel')
assert ce.exception.response['Error']['Code'] == 'LastDeliveryChannelDeleteFailedException'
assert 'because there is a running configuration recorder.' in ce.exception.response['Error']['Message']
# Stop recording:
client.stop_configuration_recorder(ConfigurationRecorderName='testrecorder')
# Try again:
client.delete_delivery_channel(DeliveryChannelName='testchannel')
# Verify:
with assert_raises(ClientError) as ce:
client.delete_delivery_channel(DeliveryChannelName='testchannel')
assert ce.exception.response['Error']['Code'] == 'NoSuchDeliveryChannelException'

107
tests/test_dynamodb2/test_dynamodb.py
View file

@ -1505,3 +1505,110 @@ def test_dynamodb_streams_2():
assert 'LatestStreamLabel' in resp['TableDescription']
assert 'LatestStreamArn' in resp['TableDescription']
@mock_dynamodb2
def test_condition_expressions():
client = boto3.client('dynamodb', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
client.create_table(
TableName='test1',
AttributeDefinitions=[{'AttributeName': 'client', 'AttributeType': 'S'}, {'AttributeName': 'app', 'AttributeType': 'S'}],
KeySchema=[{'AttributeName': 'client', 'KeyType': 'HASH'}, {'AttributeName': 'app', 'KeyType': 'RANGE'}],
ProvisionedThroughput={'ReadCapacityUnits': 123, 'WriteCapacityUnits': 123}
)
client.put_item(
TableName='test1',
Item={
'client': {'S': 'client1'},
'app': {'S': 'app1'},
'match': {'S': 'match'},
'existing': {'S': 'existing'},
}
)
client.put_item(
TableName='test1',
Item={
'client': {'S': 'client1'},
'app': {'S': 'app1'},
'match': {'S': 'match'},
'existing': {'S': 'existing'},
},
ConditionExpression='attribute_exists(#existing) AND attribute_not_exists(#nonexistent) AND #match = :match',
ExpressionAttributeNames={
'#existing': 'existing',
'#nonexistent': 'nope',
'#match': 'match',
},
ExpressionAttributeValues={
':match': {'S': 'match'}
}
)
client.put_item(
TableName='test1',
Item={
'client': {'S': 'client1'},
'app': {'S': 'app1'},
'match': {'S': 'match'},
'existing': {'S': 'existing'},
},
ConditionExpression='NOT(attribute_exists(#nonexistent1) AND attribute_exists(#nonexistent2))',
ExpressionAttributeNames={
'#nonexistent1': 'nope',
'#nonexistent2': 'nope2'
}
)
with assert_raises(client.exceptions.ConditionalCheckFailedException):
client.put_item(
TableName='test1',
Item={
'client': {'S': 'client1'},
'app': {'S': 'app1'},
'match': {'S': 'match'},
'existing': {'S': 'existing'},
},
ConditionExpression='attribute_exists(#nonexistent1) AND attribute_exists(#nonexistent2)',
ExpressionAttributeNames={
'#nonexistent1': 'nope',
'#nonexistent2': 'nope2'
}
)
with assert_raises(client.exceptions.ConditionalCheckFailedException):
client.put_item(
TableName='test1',
Item={
'client': {'S': 'client1'},
'app': {'S': 'app1'},
'match': {'S': 'match'},
'existing': {'S': 'existing'},
},
ConditionExpression='NOT(attribute_not_exists(#nonexistent1) AND attribute_not_exists(#nonexistent2))',
ExpressionAttributeNames={
'#nonexistent1': 'nope',
'#nonexistent2': 'nope2'
}
)
with assert_raises(client.exceptions.ConditionalCheckFailedException):
client.put_item(
TableName='test1',
Item={
'client': {'S': 'client1'},
'app': {'S': 'app1'},
'match': {'S': 'match'},
'existing': {'S': 'existing'},
},
ConditionExpression='attribute_exists(#existing) AND attribute_not_exists(#nonexistent) AND #match = :match',
ExpressionAttributeNames={
'#existing': 'existing',
'#nonexistent': 'nope',
'#match': 'match',
},
ExpressionAttributeValues={
':match': {'S': 'match2'}
}
)

41
tests/test_dynamodb2/test_dynamodb_table_without_range_key.py
View file

@ -750,6 +750,47 @@ def test_boto3_update_item_conditions_pass_because_expect_exists_by_compare_to_n
returned_item = table.get_item(Key={'username': 'johndoe'})
assert dict(returned_item)['Item']['foo'].should.equal("baz")
@mock_dynamodb2
def test_boto3_update_settype_item_with_conditions():
class OrderedSet(set):
"""A set with predictable iteration order"""
def __init__(self, values):
super(OrderedSet, self).__init__(values)
self.__ordered_values = values
def __iter__(self):
return iter(self.__ordered_values)
table = _create_user_table()
table.put_item(Item={'username': 'johndoe'})
table.update_item(
Key={'username': 'johndoe'},
UpdateExpression='SET foo=:new_value',
ExpressionAttributeValues={
':new_value': OrderedSet(['hello', 'world']),
},
)
table.update_item(
Key={'username': 'johndoe'},
UpdateExpression='SET foo=:new_value',
ExpressionAttributeValues={
':new_value': set(['baz']),
},
Expected={
'foo': {
'ComparisonOperator': 'EQ',
'AttributeValueList': [
OrderedSet(['world', 'hello']), # Opposite order to original
],
}
},
)
returned_item = table.get_item(Key={'username': 'johndoe'})
assert dict(returned_item)['Item']['foo'].should.equal(set(['baz']))
@mock_dynamodb2
def test_boto3_put_item_conditions_pass():
table = _create_user_table()

24
tests/test_ec2/test_elastic_block_store.py
View file

@ -589,6 +589,18 @@ def test_volume_tag_escaping():
dict(snaps[0].tags).should.equal({'key': '</closed>'})
@mock_ec2
def test_volume_property_hidden_when_no_tags_exist():
ec2_client = boto3.client('ec2', region_name='us-east-1')
volume_response = ec2_client.create_volume(
Size=10,
AvailabilityZone='us-east-1a'
)
volume_response.get('Tags').should.equal(None)
@freeze_time
@mock_ec2
def test_copy_snapshot():
@ -602,26 +614,26 @@ def test_copy_snapshot():
create_snapshot_response = ec2_client.create_snapshot(
VolumeId=volume_response['VolumeId']
)
copy_snapshot_response = dest_ec2_client.copy_snapshot(
SourceSnapshotId=create_snapshot_response['SnapshotId'],
SourceRegion="eu-west-1"
)
ec2 = boto3.resource('ec2', region_name='eu-west-1')
dest_ec2 = boto3.resource('ec2', region_name='eu-west-2')
source = ec2.Snapshot(create_snapshot_response['SnapshotId'])
dest = dest_ec2.Snapshot(copy_snapshot_response['SnapshotId'])
attribs = ['data_encryption_key_id', 'encrypted',
'kms_key_id', 'owner_alias', 'owner_id',
'progress', 'state', 'state_message',
'tags', 'volume_id', 'volume_size']
for attrib in attribs:
getattr(source, attrib).should.equal(getattr(dest, attrib))
# Copy from non-existent source ID.
with assert_raises(ClientError) as cm:
create_snapshot_error = ec2_client.create_snapshot(

15
tests/test_ec2/test_instances.py
View file

@ -1254,3 +1254,18 @@ 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
client = boto3.client('ec2', region_name='us-east-1')
client.run_instances(ImageId='ami-1234abcd',
MinCount=instance_count,
MaxCount=instance_count)
reservations = client.describe_instances()['Reservations']
reservations[0]['Instances'].should.have.length_of(instance_count)
instances = reservations[0]['Instances']
for i in range(0, instance_count):
instances[i]['AmiLaunchIndex'].should.be(i)

43
tests/test_ec2/test_network_acls.py
View file

@ -1,8 +1,9 @@
from __future__ import unicode_literals
import boto
import boto3
import sure # noqa
from moto import mock_ec2_deprecated
from moto import mock_ec2_deprecated, mock_ec2
@mock_ec2_deprecated
@ -173,3 +174,43 @@ def test_network_acl_tagging():
if na.id == network_acl.id)
test_network_acl.tags.should.have.length_of(1)
test_network_acl.tags["a key"].should.equal("some value")
@mock_ec2
def test_new_subnet_in_new_vpc_associates_with_default_network_acl():
ec2 = boto3.resource('ec2', region_name='us-west-1')
new_vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
new_vpc.reload()
subnet = ec2.create_subnet(VpcId=new_vpc.id, CidrBlock='10.0.0.0/24')
subnet.reload()
new_vpcs_default_network_acl = next(iter(new_vpc.network_acls.all()), None)
new_vpcs_default_network_acl.reload()
new_vpcs_default_network_acl.vpc_id.should.equal(new_vpc.id)
new_vpcs_default_network_acl.associations.should.have.length_of(1)
new_vpcs_default_network_acl.associations[0]['SubnetId'].should.equal(subnet.id)
@mock_ec2
def test_default_network_acl_default_entries():
ec2 = boto3.resource('ec2', region_name='us-west-1')
default_network_acl = next(iter(ec2.network_acls.all()), None)
default_network_acl.is_default.should.be.ok
default_network_acl.entries.should.have.length_of(4)
unique_entries = []
for entry in default_network_acl.entries:
entry['CidrBlock'].should.equal('0.0.0.0/0')
entry['Protocol'].should.equal('-1')
entry['RuleNumber'].should.be.within([100, 32767])
entry['RuleAction'].should.be.within(['allow', 'deny'])
assert type(entry['Egress']) is bool
if entry['RuleAction'] == 'allow':
entry['RuleNumber'].should.be.equal(100)
else:
entry['RuleNumber'].should.be.equal(32767)
if entry not in unique_entries:
unique_entries.append(entry)
unique_entries.should.have.length_of(4)

44
tests/test_ec2/test_spot_fleet.py
View file

@ -54,7 +54,7 @@ def spot_config(subnet_id, allocation_strategy="lowestPrice"):
},
'EbsOptimized': False,
'WeightedCapacity': 2.0,
'SpotPrice': '0.13'
'SpotPrice': '0.13',
}, {
'ImageId': 'ami-123',
'KeyName': 'my-key',
@ -148,6 +148,48 @@ def test_create_diversified_spot_fleet():
instances[0]['InstanceId'].should.contain("i-")
@mock_ec2
def test_create_spot_fleet_request_with_tag_spec():
conn = boto3.client("ec2", region_name='us-west-2')
subnet_id = get_subnet_id(conn)
tag_spec = [
{
'ResourceType': 'instance',
'Tags': [
{
'Key': 'tag-1',
'Value': 'foo',
},
{
'Key': 'tag-2',
'Value': 'bar',
},
]
},
]
config = spot_config(subnet_id)
config['LaunchSpecifications'][0]['TagSpecifications'] = tag_spec
spot_fleet_res = conn.request_spot_fleet(
SpotFleetRequestConfig=config
)
spot_fleet_id = spot_fleet_res['SpotFleetRequestId']
spot_fleet_requests = conn.describe_spot_fleet_requests(
SpotFleetRequestIds=[spot_fleet_id])['SpotFleetRequestConfigs']
spot_fleet_config = spot_fleet_requests[0]['SpotFleetRequestConfig']
spot_fleet_config['LaunchSpecifications'][0]['TagSpecifications'][0][
'ResourceType'].should.equal('instance')
for tag in tag_spec[0]['Tags']:
spot_fleet_config['LaunchSpecifications'][0]['TagSpecifications'][0]['Tags'].should.contain(tag)
instance_res = conn.describe_spot_fleet_instances(
SpotFleetRequestId=spot_fleet_id)
instances = conn.describe_instances(InstanceIds=[i['InstanceId'] for i in instance_res['ActiveInstances']])
for instance in instances['Reservations'][0]['Instances']:
for tag in tag_spec[0]['Tags']:
instance['Tags'].should.contain(tag)
@mock_ec2
def test_cancel_spot_fleet_request():
conn = boto3.client("ec2", region_name='us-west-2')

9
tests/test_events/test_events.py
View file

@ -1,5 +1,4 @@
import random
import boto3
import json
@ -7,7 +6,6 @@ from moto.events import mock_events
from botocore.exceptions import ClientError
from nose.tools import assert_raises
RULES = [
{'Name': 'test1', 'ScheduleExpression': 'rate(5 minutes)'},
{'Name': 'test2', 'ScheduleExpression': 'rate(1 minute)'},
@ -109,6 +107,13 @@ def test_enable_disable_rule():
rule = client.describe_rule(Name=rule_name)
assert(rule['State'] == 'ENABLED')
# Test invalid name
try:
client.enable_rule(Name='junk')
except ClientError as ce:
assert ce.response['Error']['Code'] == 'ResourceNotFoundException'
@mock_events
def test_list_rule_names_by_target():

357
tests/test_iam/test_iam.py
View file

@ -306,6 +306,7 @@ 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')
@ -579,6 +580,7 @@ def test_get_credential_report():
'get_credential_report_result']['content'].encode('ascii')).decode('ascii')
report.should.match(r'.*my-user.*')
@mock_iam
def test_boto3_get_credential_report():
conn = boto3.client('iam', region_name='us-east-1')
@ -757,6 +759,17 @@ 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": [
{
"Action": "s3:ListBucket",
"Resource": "*",
"Effect": "Allow",
}
]
})
conn = boto3.client('iam', region_name='us-east-1')
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/")
conn.create_user(Path='/', UserName='testUser')
@ -764,21 +777,34 @@ def test_get_account_authorization_details():
conn.create_policy(
PolicyName='testPolicy',
Path='/',
PolicyDocument=json.dumps({
"Version": "2012-10-17",
"Statement": [
{
"Action": "s3:ListBucket",
"Resource": "*",
"Effect": "Allow",
}
]
}),
PolicyDocument=test_policy,
Description='Test Policy'
)
# Attach things to the user and group:
conn.put_user_policy(UserName='testUser', PolicyName='testPolicy', PolicyDocument=test_policy)
conn.put_group_policy(GroupName='testGroup', PolicyName='testPolicy', PolicyDocument=test_policy)
conn.attach_user_policy(UserName='testUser', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy')
conn.attach_group_policy(GroupName='testGroup', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy')
conn.add_user_to_group(UserName='testUser', GroupName='testGroup')
# Add things to the role:
conn.create_instance_profile(InstanceProfileName='ipn')
conn.add_role_to_instance_profile(InstanceProfileName='ipn', RoleName='my-role')
conn.tag_role(RoleName='my-role', Tags=[
{
'Key': 'somekey',
'Value': 'somevalue'
},
{
'Key': 'someotherkey',
'Value': 'someothervalue'
}
])
conn.put_role_policy(RoleName='my-role', PolicyName='test-policy', PolicyDocument=test_policy)
conn.attach_role_policy(RoleName='my-role', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy')
result = conn.get_account_authorization_details(Filter=['Role'])
assert len(result['RoleDetailList']) == 1
@ -786,24 +812,41 @@ def test_get_account_authorization_details():
assert len(result['GroupDetailList']) == 0
assert len(result['Policies']) == 0
assert len(result['RoleDetailList'][0]['InstanceProfileList']) == 1
assert len(result['RoleDetailList'][0]['Tags']) == 2
assert len(result['RoleDetailList'][0]['RolePolicyList']) == 1
assert len(result['RoleDetailList'][0]['AttachedManagedPolicies']) == 1
assert result['RoleDetailList'][0]['AttachedManagedPolicies'][0]['PolicyName'] == 'testPolicy'
assert result['RoleDetailList'][0]['AttachedManagedPolicies'][0]['PolicyArn'] == \
'arn:aws:iam::123456789012:policy/testPolicy'
result = conn.get_account_authorization_details(Filter=['User'])
assert len(result['RoleDetailList']) == 0
assert len(result['UserDetailList']) == 1
assert len(result['UserDetailList'][0]['GroupList']) == 1
assert len(result['UserDetailList'][0]['AttachedManagedPolicies']) == 1
assert len(result['GroupDetailList']) == 0
assert len(result['Policies']) == 0
assert result['UserDetailList'][0]['AttachedManagedPolicies'][0]['PolicyName'] == 'testPolicy'
assert result['UserDetailList'][0]['AttachedManagedPolicies'][0]['PolicyArn'] == \
'arn:aws:iam::123456789012:policy/testPolicy'
result = conn.get_account_authorization_details(Filter=['Group'])
assert len(result['RoleDetailList']) == 0
assert len(result['UserDetailList']) == 0
assert len(result['GroupDetailList']) == 1
assert len(result['GroupDetailList'][0]['GroupPolicyList']) == 1
assert len(result['GroupDetailList'][0]['AttachedManagedPolicies']) == 1
assert len(result['Policies']) == 0
assert result['GroupDetailList'][0]['AttachedManagedPolicies'][0]['PolicyName'] == 'testPolicy'
assert result['GroupDetailList'][0]['AttachedManagedPolicies'][0]['PolicyArn'] == \
'arn:aws:iam::123456789012:policy/testPolicy'
result = conn.get_account_authorization_details(Filter=['LocalManagedPolicy'])
assert len(result['RoleDetailList']) == 0
assert len(result['UserDetailList']) == 0
assert len(result['GroupDetailList']) == 0
assert len(result['Policies']) == 1
assert len(result['Policies'][0]['PolicyVersionList']) == 1
# Check for greater than 1 since this should always be greater than one but might change.
# See iam/aws_managed_policies.py
@ -872,6 +915,7 @@ def test_signing_certs():
with assert_raises(ClientError):
client.delete_signing_certificate(UserName='notauser', CertificateId=cert_id)
@mock_iam()
def test_create_saml_provider():
conn = boto3.client('iam', region_name='us-east-1')
@ -881,6 +925,7 @@ def test_create_saml_provider():
)
response['SAMLProviderArn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider")
@mock_iam()
def test_get_saml_provider():
conn = boto3.client('iam', region_name='us-east-1')
@ -893,6 +938,7 @@ def test_get_saml_provider():
)
response['SAMLMetadataDocument'].should.equal('a' * 1024)
@mock_iam()
def test_list_saml_providers():
conn = boto3.client('iam', region_name='us-east-1')
@ -903,6 +949,7 @@ def test_list_saml_providers():
response = conn.list_saml_providers()
response['SAMLProviderList'][0]['Arn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider")
@mock_iam()
def test_delete_saml_provider():
conn = boto3.client('iam', region_name='us-east-1')
@ -929,3 +976,293 @@ def test_delete_saml_provider():
# Verify that it's not in the list:
resp = conn.list_signing_certificates(UserName='testing')
assert not resp['Certificates']
@mock_iam()
def test_tag_role():
"""Tests both the tag_role and get_role_tags capability"""
conn = boto3.client('iam', region_name='us-east-1')
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="{}")
# Get without tags:
role = conn.get_role(RoleName='my-role')['Role']
assert not role.get('Tags')
# With proper tag values:
conn.tag_role(RoleName='my-role', Tags=[
{
'Key': 'somekey',
'Value': 'somevalue'
},
{
'Key': 'someotherkey',
'Value': 'someothervalue'
}
])
# Get role:
role = conn.get_role(RoleName='my-role')['Role']
assert len(role['Tags']) == 2
assert role['Tags'][0]['Key'] == 'somekey'
assert role['Tags'][0]['Value'] == 'somevalue'
assert role['Tags'][1]['Key'] == 'someotherkey'
assert role['Tags'][1]['Value'] == 'someothervalue'
# Same -- but for list_role_tags:
tags = conn.list_role_tags(RoleName='my-role')
assert len(tags['Tags']) == 2
assert role['Tags'][0]['Key'] == 'somekey'
assert role['Tags'][0]['Value'] == 'somevalue'
assert role['Tags'][1]['Key'] == 'someotherkey'
assert role['Tags'][1]['Value'] == 'someothervalue'
assert not tags['IsTruncated']
assert not tags.get('Marker')
# Test pagination:
tags = conn.list_role_tags(RoleName='my-role', MaxItems=1)
assert len(tags['Tags']) == 1
assert tags['IsTruncated']
assert tags['Tags'][0]['Key'] == 'somekey'
assert tags['Tags'][0]['Value'] == 'somevalue'
assert tags['Marker'] == '1'
tags = conn.list_role_tags(RoleName='my-role', Marker=tags['Marker'])
assert len(tags['Tags']) == 1
assert tags['Tags'][0]['Key'] == 'someotherkey'
assert tags['Tags'][0]['Value'] == 'someothervalue'
assert not tags['IsTruncated']
assert not tags.get('Marker')
# Test updating an existing tag:
conn.tag_role(RoleName='my-role', Tags=[
{
'Key': 'somekey',
'Value': 'somenewvalue'
}
])
tags = conn.list_role_tags(RoleName='my-role')
assert len(tags['Tags']) == 2
assert tags['Tags'][0]['Key'] == 'somekey'
assert tags['Tags'][0]['Value'] == 'somenewvalue'
# Empty is good:
conn.tag_role(RoleName='my-role', Tags=[
{
'Key': 'somekey',
'Value': ''
}
])
tags = conn.list_role_tags(RoleName='my-role')
assert len(tags['Tags']) == 2
assert tags['Tags'][0]['Key'] == 'somekey'
assert tags['Tags'][0]['Value'] == ''
# Test creating tags with invalid values:
# With more than 50 tags:
with assert_raises(ClientError) as ce:
too_many_tags = list(map(lambda x: {'Key': str(x), 'Value': str(x)}, range(0, 51)))
conn.tag_role(RoleName='my-role', Tags=too_many_tags)
assert 'failed to satisfy constraint: Member must have length less than or equal to 50.' \
in ce.exception.response['Error']['Message']
# With a duplicate tag:
with assert_raises(ClientError) as ce:
conn.tag_role(RoleName='my-role', Tags=[{'Key': '0', 'Value': ''}, {'Key': '0', 'Value': ''}])
assert 'Duplicate tag keys found. Please note that Tag keys are case insensitive.' \
in ce.exception.response['Error']['Message']
# Duplicate tag with different casing:
with assert_raises(ClientError) as ce:
conn.tag_role(RoleName='my-role', Tags=[{'Key': 'a', 'Value': ''}, {'Key': 'A', 'Value': ''}])
assert 'Duplicate tag keys found. Please note that Tag keys are case insensitive.' \
in ce.exception.response['Error']['Message']
# With a really big key:
with assert_raises(ClientError) as ce:
conn.tag_role(RoleName='my-role', Tags=[{'Key': '0' * 129, 'Value': ''}])
assert 'Member must have length less than or equal to 128.' in ce.exception.response['Error']['Message']
# With a really big value:
with assert_raises(ClientError) as ce:
conn.tag_role(RoleName='my-role', Tags=[{'Key': '0', 'Value': '0' * 257}])
assert 'Member must have length less than or equal to 256.' in ce.exception.response['Error']['Message']
# With an invalid character:
with assert_raises(ClientError) as ce:
conn.tag_role(RoleName='my-role', Tags=[{'Key': 'NOWAY!', 'Value': ''}])
assert 'Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+' \
in ce.exception.response['Error']['Message']
# With a role that doesn't exist:
with assert_raises(ClientError):
conn.tag_role(RoleName='notarole', Tags=[{'Key': 'some', 'Value': 'value'}])
@mock_iam
def test_untag_role():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="{}")
# With proper tag values:
conn.tag_role(RoleName='my-role', Tags=[
{
'Key': 'somekey',
'Value': 'somevalue'
},
{
'Key': 'someotherkey',
'Value': 'someothervalue'
}
])
# Remove them:
conn.untag_role(RoleName='my-role', TagKeys=['somekey'])
tags = conn.list_role_tags(RoleName='my-role')
assert len(tags['Tags']) == 1
assert tags['Tags'][0]['Key'] == 'someotherkey'
assert tags['Tags'][0]['Value'] == 'someothervalue'
# And again:
conn.untag_role(RoleName='my-role', TagKeys=['someotherkey'])
tags = conn.list_role_tags(RoleName='my-role')
assert not tags['Tags']
# Test removing tags with invalid values:
# With more than 50 tags:
with assert_raises(ClientError) as ce:
conn.untag_role(RoleName='my-role', TagKeys=[str(x) for x in range(0, 51)])
assert 'failed to satisfy constraint: Member must have length less than or equal to 50.' \
in ce.exception.response['Error']['Message']
assert 'tagKeys' in ce.exception.response['Error']['Message']
# With a really big key:
with assert_raises(ClientError) as ce:
conn.untag_role(RoleName='my-role', TagKeys=['0' * 129])
assert 'Member must have length less than or equal to 128.' in ce.exception.response['Error']['Message']
assert 'tagKeys' in ce.exception.response['Error']['Message']
# With an invalid character:
with assert_raises(ClientError) as ce:
conn.untag_role(RoleName='my-role', TagKeys=['NOWAY!'])
assert 'Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+' \
in ce.exception.response['Error']['Message']
assert 'tagKeys' in ce.exception.response['Error']['Message']
# With a role that doesn't exist:
with assert_raises(ClientError):
conn.untag_role(RoleName='notarole', TagKeys=['somevalue'])
@mock_iam()
def test_update_role_description():
conn = boto3.client('iam', region_name='us-east-1')
with assert_raises(ClientError):
conn.delete_role(RoleName="my-role")
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/")
response = conn.update_role_description(RoleName="my-role", Description="test")
assert response['Role']['RoleName'] == 'my-role'
@mock_iam()
def test_update_role():
conn = boto3.client('iam', region_name='us-east-1')
with assert_raises(ClientError):
conn.delete_role(RoleName="my-role")
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/")
response = conn.update_role_description(RoleName="my-role", Description="test")
assert response['Role']['RoleName'] == 'my-role'
@mock_iam()
def test_update_role():
conn = boto3.client('iam', region_name='us-east-1')
with assert_raises(ClientError):
conn.delete_role(RoleName="my-role")
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/")
response = conn.update_role(RoleName="my-role", Description="test")
assert len(response.keys()) == 1
@mock_iam()
def test_list_entities_for_policy():
import json
test_policy = json.dumps({
"Version": "2012-10-17",
"Statement": [
{
"Action": "s3:ListBucket",
"Resource": "*",
"Effect": "Allow",
}
]
})
conn = boto3.client('iam', region_name='us-east-1')
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/")
conn.create_user(Path='/', UserName='testUser')
conn.create_group(Path='/', GroupName='testGroup')
conn.create_policy(
PolicyName='testPolicy',
Path='/',
PolicyDocument=test_policy,
Description='Test Policy'
)
# Attach things to the user and group:
conn.put_user_policy(UserName='testUser', PolicyName='testPolicy', PolicyDocument=test_policy)
conn.put_group_policy(GroupName='testGroup', PolicyName='testPolicy', PolicyDocument=test_policy)
conn.attach_user_policy(UserName='testUser', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy')
conn.attach_group_policy(GroupName='testGroup', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy')
conn.add_user_to_group(UserName='testUser', GroupName='testGroup')
# Add things to the role:
conn.create_instance_profile(InstanceProfileName='ipn')
conn.add_role_to_instance_profile(InstanceProfileName='ipn', RoleName='my-role')
conn.tag_role(RoleName='my-role', Tags=[
{
'Key': 'somekey',
'Value': 'somevalue'
},
{
'Key': 'someotherkey',
'Value': 'someothervalue'
}
])
conn.put_role_policy(RoleName='my-role', PolicyName='test-policy', PolicyDocument=test_policy)
conn.attach_role_policy(RoleName='my-role', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy')
response = conn.list_entities_for_policy(
PolicyArn='arn:aws:iam::123456789012:policy/testPolicy',
EntityFilter='Role'
)
assert response['PolicyRoles'] == [{'RoleName': 'my-role'}]
response = conn.list_entities_for_policy(
PolicyArn='arn:aws:iam::123456789012:policy/testPolicy',
EntityFilter='User',
)
assert response['PolicyUsers'] == [{'UserName': 'testUser'}]
response = conn.list_entities_for_policy(
PolicyArn='arn:aws:iam::123456789012:policy/testPolicy',
EntityFilter='Group',
)
assert response['PolicyGroups'] == [{'GroupName': 'testGroup'}]
response = conn.list_entities_for_policy(
PolicyArn='arn:aws:iam::123456789012:policy/testPolicy',
EntityFilter='LocalManagedPolicy',
)
assert response['PolicyGroups'] == [{'GroupName': 'testGroup'}]
assert response['PolicyUsers'] == [{'UserName': 'testUser'}]
assert response['PolicyRoles'] == [{'RoleName': 'my-role'}]

58
tests/test_kms/test_kms.py
View file

@ -1,6 +1,5 @@
from __future__ import unicode_literals
import os, re
import boto3
import boto.kms
from boto.exception import JSONResponseError
@ -717,3 +716,60 @@ def test_cancel_key_deletion():
assert result["KeyMetadata"]["Enabled"] == False
assert result["KeyMetadata"]["KeyState"] == 'Disabled'
assert 'DeletionDate' not in result["KeyMetadata"]
@mock_kms
def test_update_key_description():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='old_description')
key_id = key['KeyMetadata']['KeyId']
result = client.update_key_description(KeyId=key_id, Description='new_description')
assert 'ResponseMetadata' in result
@mock_kms
def test_tag_resource():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='cancel-key-deletion')
response = client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId']
)
keyid = response['KeyId']
response = client.tag_resource(
KeyId=keyid,
Tags=[
{
'TagKey': 'string',
'TagValue': 'string'
},
]
)
# Shouldn't have any data, just header
assert len(response.keys()) == 1
@mock_kms
def test_list_resource_tags():
client = boto3.client('kms', region_name='us-east-1')
key = client.create_key(Description='cancel-key-deletion')
response = client.schedule_key_deletion(
KeyId=key['KeyMetadata']['KeyId']
)
keyid = response['KeyId']
response = client.tag_resource(
KeyId=keyid,
Tags=[
{
'TagKey': 'string',
'TagValue': 'string'
},
]
)
response = client.list_resource_tags(KeyId=keyid)
assert response['Tags'][0]['TagKey'] == 'string'
assert response['Tags'][0]['TagValue'] == 'string'

8
tests/test_packages/__init__.py Normal file
View file

@ -0,0 +1,8 @@
from __future__ import unicode_literals
import logging
# Disable extra logging for tests
logging.getLogger('boto').setLevel(logging.CRITICAL)
logging.getLogger('boto3').setLevel(logging.CRITICAL)
logging.getLogger('botocore').setLevel(logging.CRITICAL)
logging.getLogger('nose').setLevel(logging.CRITICAL)

37
tests/test_packages/test_httpretty.py Normal file
View file

@ -0,0 +1,37 @@
# #!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import mock
from moto.packages.httpretty.core import HTTPrettyRequest, fake_gethostname, fake_gethostbyname
def test_parse_querystring():
core = HTTPrettyRequest(headers='test test HTTP/1.1')
qs = 'test test'
response = core.parse_querystring(qs)
assert response == {}
def test_parse_request_body():
core = HTTPrettyRequest(headers='test test HTTP/1.1')
qs = 'test'
response = core.parse_request_body(qs)
assert response == 'test'
def test_fake_gethostname():
response = fake_gethostname()
assert response == 'localhost'
def test_fake_gethostbyname():
host = 'test'
response = fake_gethostbyname(host=host)
assert response == '127.0.0.1'

20
tests/test_route53/test_route53.py
View file

@ -98,6 +98,16 @@ def test_rrset():
rrsets.should.have.length_of(1)
rrsets[0].resource_records[0].should.equal('5.6.7.8')
changes = ResourceRecordSets(conn, zoneid)
change = changes.add_change("UPSERT", "foo.bar.testdns.aws.com", "TXT")
change.add_value("foo")
changes.commit()
rrsets = conn.get_all_rrsets(zoneid)
rrsets.should.have.length_of(2)
rrsets[0].resource_records[0].should.equal('5.6.7.8')
rrsets[1].resource_records[0].should.equal('foo')
changes = ResourceRecordSets(conn, zoneid)
changes.add_change("DELETE", "foo.bar.testdns.aws.com", "A")
changes.commit()
@ -520,7 +530,7 @@ def test_change_resource_record_sets_crud_valid():
# Create A Record.
a_record_endpoint_payload = {
'Comment': 'create A record prod.redis.db',
'Comment': 'Create A record prod.redis.db',
'Changes': [
{
'Action': 'CREATE',
@ -545,15 +555,15 @@ def test_change_resource_record_sets_crud_valid():
a_record_detail['TTL'].should.equal(10)
a_record_detail['ResourceRecords'].should.equal([{'Value': '127.0.0.1'}])
# Update type to CNAME
# Update A Record.
cname_record_endpoint_payload = {
'Comment': 'Update to CNAME prod.redis.db',
'Comment': 'Update A record prod.redis.db',
'Changes': [
{
'Action': 'UPSERT',
'ResourceRecordSet': {
'Name': 'prod.redis.db.',
'Type': 'CNAME',
'Type': 'A',
'TTL': 60,
'ResourceRecords': [{
'Value': '192.168.1.1'
@ -568,7 +578,7 @@ def test_change_resource_record_sets_crud_valid():
len(response['ResourceRecordSets']).should.equal(1)
cname_record_detail = response['ResourceRecordSets'][0]
cname_record_detail['Name'].should.equal('prod.redis.db.')
cname_record_detail['Type'].should.equal('CNAME')
cname_record_detail['Type'].should.equal('A')
cname_record_detail['TTL'].should.equal(60)
cname_record_detail['ResourceRecords'].should.equal([{'Value': '192.168.1.1'}])

150
tests/test_s3/test_s3.py
View file

@ -418,6 +418,22 @@ def test_copy_key():
"new-key").get_contents_as_string().should.equal(b"some value")
@mock_s3_deprecated
def test_copy_key_with_unicode():
conn = boto.connect_s3('the_key', 'the_secret')
bucket = conn.create_bucket("foobar")
key = Key(bucket)
key.key = "the-unicode-💩-key"
key.set_contents_from_string("some value")
bucket.copy_key('new-key', 'foobar', 'the-unicode-💩-key')
bucket.get_key(
"the-unicode-💩-key").get_contents_as_string().should.equal(b"some value")
bucket.get_key(
"new-key").get_contents_as_string().should.equal(b"some value")
@mock_s3_deprecated
def test_copy_key_with_version():
conn = boto.connect_s3('the_key', 'the_secret')
@ -428,7 +444,12 @@ def test_copy_key_with_version():
key.set_contents_from_string("some value")
key.set_contents_from_string("another value")
bucket.copy_key('new-key', 'foobar', 'the-key', src_version_id='0')
key = [
key.version_id
for key in bucket.get_all_versions()
if not key.is_latest
][0]
bucket.copy_key('new-key', 'foobar', 'the-key', src_version_id=key)
bucket.get_key(
"the-key").get_contents_as_string().should.equal(b"another value")
@ -802,16 +823,19 @@ def test_key_version():
bucket = conn.create_bucket('foobar')
bucket.configure_versioning(versioning=True)
versions = []
key = Key(bucket)
key.key = 'the-key'
key.version_id.should.be.none
key.set_contents_from_string('some string')
key.version_id.should.equal('0')
versions.append(key.version_id)
key.set_contents_from_string('some string')
key.version_id.should.equal('1')
versions.append(key.version_id)
set(versions).should.have.length_of(2)
key = bucket.get_key('the-key')
key.version_id.should.equal('1')
key.version_id.should.equal(versions[-1])
@mock_s3_deprecated
@ -820,23 +844,25 @@ def test_list_versions():
bucket = conn.create_bucket('foobar')
bucket.configure_versioning(versioning=True)
key_versions = []
key = Key(bucket, 'the-key')
key.version_id.should.be.none
key.set_contents_from_string("Version 1")
key.version_id.should.equal('0')
key_versions.append(key.version_id)
key.set_contents_from_string("Version 2")
key.version_id.should.equal('1')
key_versions.append(key.version_id)
key_versions.should.have.length_of(2)
versions = list(bucket.list_versions())
versions.should.have.length_of(2)
versions[0].name.should.equal('the-key')
versions[0].version_id.should.equal('0')
versions[0].version_id.should.equal(key_versions[0])
versions[0].get_contents_as_string().should.equal(b"Version 1")
versions[1].name.should.equal('the-key')
versions[1].version_id.should.equal('1')
versions[1].version_id.should.equal(key_versions[1])
versions[1].get_contents_as_string().should.equal(b"Version 2")
key = Key(bucket, 'the2-key')
@ -1467,16 +1493,22 @@ def test_boto3_head_object_with_versioning():
s3.Object('blah', 'hello.txt').put(Body=old_content)
s3.Object('blah', 'hello.txt').put(Body=new_content)
versions = list(s3.Bucket('blah').object_versions.all())
latest = list(filter(lambda item: item.is_latest, versions))[0]
oldest = list(filter(lambda item: not item.is_latest, versions))[0]
head_object = s3.Object('blah', 'hello.txt').meta.client.head_object(
Bucket='blah', Key='hello.txt')
head_object['VersionId'].should.equal('1')
head_object['VersionId'].should.equal(latest.id)
head_object['ContentLength'].should.equal(len(new_content))
old_head_object = s3.Object('blah', 'hello.txt').meta.client.head_object(
Bucket='blah', Key='hello.txt', VersionId='0')
old_head_object['VersionId'].should.equal('0')
Bucket='blah', Key='hello.txt', VersionId=oldest.id)
old_head_object['VersionId'].should.equal(oldest.id)
old_head_object['ContentLength'].should.equal(len(old_content))
old_head_object['VersionId'].should_not.equal(head_object['VersionId'])
@mock_s3
def test_boto3_copy_object_with_versioning():
@ -1491,9 +1523,6 @@ def test_boto3_copy_object_with_versioning():
obj1_version = client.get_object(Bucket='blah', Key='test1')['VersionId']
obj2_version = client.get_object(Bucket='blah', Key='test2')['VersionId']
# Versions should be the same
obj1_version.should.equal(obj2_version)
client.copy_object(CopySource={'Bucket': 'blah', 'Key': 'test1'}, Bucket='blah', Key='test2')
obj2_version_new = client.get_object(Bucket='blah', Key='test2')['VersionId']
@ -2491,6 +2520,75 @@ def test_boto3_list_object_versions():
response['Body'].read().should.equal(items[-1])
@mock_s3
def test_boto3_list_object_versions_with_versioning_disabled():
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'mybucket'
key = 'key-with-versions'
s3.create_bucket(Bucket=bucket_name)
items = (six.b('v1'), six.b('v2'))
for body in items:
s3.put_object(
Bucket=bucket_name,
Key=key,
Body=body
)
response = s3.list_object_versions(
Bucket=bucket_name
)
# One object version should be returned
len(response['Versions']).should.equal(1)
response['Versions'][0]['Key'].should.equal(key)
# The version id should be the string null
response['Versions'][0]['VersionId'].should.equal('null')
# Test latest object version is returned
response = s3.get_object(Bucket=bucket_name, Key=key)
response['Body'].read().should.equal(items[-1])
@mock_s3
def test_boto3_list_object_versions_with_versioning_enabled_late():
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'mybucket'
key = 'key-with-versions'
s3.create_bucket(Bucket=bucket_name)
items = (six.b('v1'), six.b('v2'))
s3.put_object(
Bucket=bucket_name,
Key=key,
Body=six.b('v1')
)
s3.put_bucket_versioning(
Bucket=bucket_name,
VersioningConfiguration={
'Status': 'Enabled'
}
)
s3.put_object(
Bucket=bucket_name,
Key=key,
Body=six.b('v2')
)
response = s3.list_object_versions(
Bucket=bucket_name
)
# Two object versions should be returned
len(response['Versions']).should.equal(2)
keys = set([item['Key'] for item in response['Versions']])
keys.should.equal({key})
# There should still be a null version id.
versionsId = set([item['VersionId'] for item in response['Versions']])
versionsId.should.contain('null')
# Test latest object version is returned
response = s3.get_object(Bucket=bucket_name, Key=key)
response['Body'].read().should.equal(items[-1])
@mock_s3
def test_boto3_bad_prefix_list_object_versions():
s3 = boto3.client('s3', region_name='us-east-1')
@ -2547,18 +2645,25 @@ def test_boto3_delete_markers():
Bucket=bucket_name,
Key=key
)
e.response['Error']['Code'].should.equal('404')
e.exception.response['Error']['Code'].should.equal('NoSuchKey')
response = s3.list_object_versions(
Bucket=bucket_name
)
response['Versions'].should.have.length_of(2)
response['DeleteMarkers'].should.have.length_of(1)
s3.delete_object(
Bucket=bucket_name,
Key=key,
VersionId='2'
VersionId=response['DeleteMarkers'][0]['VersionId']
)
response = s3.get_object(
Bucket=bucket_name,
Key=key
)
response['Body'].read().should.equal(items[-1])
response = s3.list_object_versions(
Bucket=bucket_name
)
@ -2567,10 +2672,8 @@ def test_boto3_delete_markers():
# 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')
latest['VersionId'].should_not.equal(oldest['VersionId'])
# Double check the name is still unicode
latest['Key'].should.equal('key-with-versions-and-unicode-ó')
@ -2615,12 +2718,12 @@ def test_boto3_multiple_delete_markers():
s3.delete_object(
Bucket=bucket_name,
Key=key,
VersionId='2'
VersionId=response['DeleteMarkers'][0]['VersionId']
)
s3.delete_object(
Bucket=bucket_name,
Key=key,
VersionId='3'
VersionId=response['DeleteMarkers'][1]['VersionId']
)
response = s3.get_object(
@ -2636,8 +2739,7 @@ def test_boto3_multiple_delete_markers():
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')
latest['VersionId'].should_not.equal(oldest['VersionId'])
# Double check the name is still unicode
latest['Key'].should.equal('key-with-versions-and-unicode-ó')