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

Merge branch 'master' into support-iterator-type-at-after-sequence

Browse source
This commit is contained in:
Daniel Guerrero 2019-08-27 16:06:11 -05:00
commit a068a56972
60 changed files with 5796 additions and 482 deletions
Download patch file Download diff file Expand all files Collapse all files

26
tests/test_acm/test_acm.py
View file

@ -74,6 +74,31 @@ def test_list_certificates():
resp['CertificateSummaryList'][0]['DomainName'].should.equal(SERVER_COMMON_NAME)
@mock_acm
def test_list_certificates_by_status():
client = boto3.client('acm', region_name='eu-central-1')
issued_arn = _import_cert(client)
pending_arn = client.request_certificate(DomainName='google.com')['CertificateArn']
resp = client.list_certificates()
len(resp['CertificateSummaryList']).should.equal(2)
resp = client.list_certificates(CertificateStatuses=['EXPIRED', 'INACTIVE'])
len(resp['CertificateSummaryList']).should.equal(0)
resp = client.list_certificates(CertificateStatuses=['PENDING_VALIDATION'])
len(resp['CertificateSummaryList']).should.equal(1)
resp['CertificateSummaryList'][0]['CertificateArn'].should.equal(pending_arn)
resp = client.list_certificates(CertificateStatuses=['ISSUED'])
len(resp['CertificateSummaryList']).should.equal(1)
resp['CertificateSummaryList'][0]['CertificateArn'].should.equal(issued_arn)
resp = client.list_certificates(CertificateStatuses=['ISSUED', 'PENDING_VALIDATION'])
len(resp['CertificateSummaryList']).should.equal(2)
arns = {cert['CertificateArn'] for cert in resp['CertificateSummaryList']}
arns.should.contain(issued_arn)
arns.should.contain(pending_arn)
@mock_acm
def test_get_invalid_certificate():
client = boto3.client('acm', region_name='eu-central-1')
@ -291,6 +316,7 @@ def test_request_certificate():
)
resp.should.contain('CertificateArn')
arn = resp['CertificateArn']
arn.should.match(r"arn:aws:acm:eu-central-1:\d{12}:certificate/")
resp = client.request_certificate(
DomainName='google.com',

21
tests/test_apigateway/test_apigateway.py
View file

@ -988,13 +988,30 @@ def test_api_keys():
apikey['name'].should.equal(apikey_name)
len(apikey['value']).should.equal(40)
apikey_name = 'TESTKEY3'
payload = {'name': apikey_name }
response = client.create_api_key(**payload)
apikey_id = response['id']
patch_operations = [
{'op': 'replace', 'path': '/name', 'value': 'TESTKEY3_CHANGE'},
{'op': 'replace', 'path': '/customerId', 'value': '12345'},
{'op': 'replace', 'path': '/description', 'value': 'APIKEY UPDATE TEST'},
{'op': 'replace', 'path': '/enabled', 'value': 'false'},
]
response = client.update_api_key(apiKey=apikey_id, patchOperations=patch_operations)
response['name'].should.equal('TESTKEY3_CHANGE')
response['customerId'].should.equal('12345')
response['description'].should.equal('APIKEY UPDATE TEST')
response['enabled'].should.equal(False)
response = client.get_api_keys()
len(response['items']).should.equal(2)
len(response['items']).should.equal(3)
client.delete_api_key(apiKey=apikey_id)
response = client.get_api_keys()
len(response['items']).should.equal(1)
len(response['items']).should.equal(2)
@mock_apigateway
def test_usage_plans():

314
tests/test_awslambda/test_lambda.py
View file

@ -1,6 +1,7 @@
from __future__ import unicode_literals
import base64
import uuid
import botocore.client
import boto3
import hashlib
@ -11,11 +12,12 @@ import zipfile
import sure # noqa
from freezegun import freeze_time
from moto import mock_lambda, mock_s3, mock_ec2, mock_sns, mock_logs, settings
from moto import mock_lambda, mock_s3, mock_ec2, mock_sns, mock_logs, settings, mock_sqs
from nose.tools import assert_raises
from botocore.exceptions import ClientError
_lambda_region = 'us-west-2'
boto3.setup_default_session(region_name=_lambda_region)
def _process_lambda(func_str):
@ -59,6 +61,13 @@ def lambda_handler(event, context):
"""
return _process_lambda(pfunc)
def get_test_zip_file4():
pfunc = """
def lambda_handler(event, context):
raise Exception('I failed!')
"""
return _process_lambda(pfunc)
@mock_lambda
def test_list_functions():
@ -933,3 +942,306 @@ def test_list_versions_by_function_for_nonexistent_function():
versions = conn.list_versions_by_function(FunctionName='testFunction')
assert len(versions['Versions']) == 0
@mock_logs
@mock_lambda
@mock_sqs
def test_create_event_source_mapping():
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func['FunctionArn'],
)
assert response['EventSourceArn'] == queue.attributes['QueueArn']
assert response['FunctionArn'] == func['FunctionArn']
assert response['State'] == 'Enabled'
@mock_logs
@mock_lambda
@mock_sqs
def test_invoke_function_from_sqs():
logs_conn = boto3.client("logs")
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func['FunctionArn'],
)
assert response['EventSourceArn'] == queue.attributes['QueueArn']
assert response['State'] == 'Enabled'
sqs_client = boto3.client('sqs')
sqs_client.send_message(QueueUrl=queue.url, MessageBody='test')
start = time.time()
while (time.time() - start) < 30:
result = logs_conn.describe_log_streams(logGroupName='/aws/lambda/testFunction')
log_streams = result.get('logStreams')
if not log_streams:
time.sleep(1)
continue
assert len(log_streams) == 1
result = logs_conn.get_log_events(logGroupName='/aws/lambda/testFunction', logStreamName=log_streams[0]['logStreamName'])
for event in result.get('events'):
if event['message'] == 'get_test_zip_file3 success':
return
time.sleep(1)
assert False, "Test Failed"
@mock_logs
@mock_lambda
@mock_sqs
def test_invoke_function_from_sqs_exception():
logs_conn = boto3.client("logs")
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file4(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func['FunctionArn'],
)
assert response['EventSourceArn'] == queue.attributes['QueueArn']
assert response['State'] == 'Enabled'
entries = []
for i in range(3):
body = {
"uuid": str(uuid.uuid4()),
"test": "test_{}".format(i),
}
entry = {
'Id': str(i),
'MessageBody': json.dumps(body)
}
entries.append(entry)
queue.send_messages(Entries=entries)
start = time.time()
while (time.time() - start) < 30:
result = logs_conn.describe_log_streams(logGroupName='/aws/lambda/testFunction')
log_streams = result.get('logStreams')
if not log_streams:
time.sleep(1)
continue
assert len(log_streams) >= 1
result = logs_conn.get_log_events(logGroupName='/aws/lambda/testFunction', logStreamName=log_streams[0]['logStreamName'])
for event in result.get('events'):
if 'I failed!' in event['message']:
messages = queue.receive_messages(MaxNumberOfMessages=10)
# Verify messages are still visible and unprocessed
assert len(messages) is 3
return
time.sleep(1)
assert False, "Test Failed"
@mock_logs
@mock_lambda
@mock_sqs
def test_list_event_source_mappings():
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func['FunctionArn'],
)
mappings = conn.list_event_source_mappings(EventSourceArn='123')
assert len(mappings['EventSourceMappings']) == 0
mappings = conn.list_event_source_mappings(EventSourceArn=queue.attributes['QueueArn'])
assert len(mappings['EventSourceMappings']) == 1
assert mappings['EventSourceMappings'][0]['UUID'] == response['UUID']
assert mappings['EventSourceMappings'][0]['FunctionArn'] == func['FunctionArn']
@mock_lambda
@mock_sqs
def test_get_event_source_mapping():
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func['FunctionArn'],
)
mapping = conn.get_event_source_mapping(UUID=response['UUID'])
assert mapping['UUID'] == response['UUID']
assert mapping['FunctionArn'] == func['FunctionArn']
conn.get_event_source_mapping.when.called_with(UUID='1')\
.should.throw(botocore.client.ClientError)
@mock_lambda
@mock_sqs
def test_update_event_source_mapping():
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func1 = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
func2 = conn.create_function(
FunctionName='testFunction2',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func1['FunctionArn'],
)
assert response['FunctionArn'] == func1['FunctionArn']
assert response['BatchSize'] == 10
assert response['State'] == 'Enabled'
mapping = conn.update_event_source_mapping(
UUID=response['UUID'],
Enabled=False,
BatchSize=15,
FunctionName='testFunction2'
)
assert mapping['UUID'] == response['UUID']
assert mapping['FunctionArn'] == func2['FunctionArn']
assert mapping['State'] == 'Disabled'
@mock_lambda
@mock_sqs
def test_delete_event_source_mapping():
sqs = boto3.resource('sqs')
queue = sqs.create_queue(QueueName="test-sqs-queue1")
conn = boto3.client('lambda')
func1 = conn.create_function(
FunctionName='testFunction',
Runtime='python2.7',
Role='test-iam-role',
Handler='lambda_function.lambda_handler',
Code={
'ZipFile': get_test_zip_file3(),
},
Description='test lambda function',
Timeout=3,
MemorySize=128,
Publish=True,
)
response = conn.create_event_source_mapping(
EventSourceArn=queue.attributes['QueueArn'],
FunctionName=func1['FunctionArn'],
)
assert response['FunctionArn'] == func1['FunctionArn']
assert response['BatchSize'] == 10
assert response['State'] == 'Enabled'
response = conn.delete_event_source_mapping(UUID=response['UUID'])
assert response['State'] == 'Deleting'
conn.get_event_source_mapping.when.called_with(UUID=response['UUID'])\
.should.throw(botocore.client.ClientError)

81
tests/test_batch/test_batch.py
View file

@ -642,6 +642,87 @@ def test_describe_task_definition():
len(resp['jobDefinitions']).should.equal(3)
@mock_logs
@mock_ec2
@mock_ecs
@mock_iam
@mock_batch
def test_submit_job_by_name():
ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()
vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)
compute_name = 'test_compute_env'
resp = batch_client.create_compute_environment(
computeEnvironmentName=compute_name,
type='UNMANAGED',
state='ENABLED',
serviceRole=iam_arn
)
arn = resp['computeEnvironmentArn']
resp = batch_client.create_job_queue(
jobQueueName='test_job_queue',
state='ENABLED',
priority=123,
computeEnvironmentOrder=[
{
'order': 123,
'computeEnvironment': arn
},
]
)
queue_arn = resp['jobQueueArn']
job_definition_name = 'sleep10'
batch_client.register_job_definition(
jobDefinitionName=job_definition_name,
type='container',
containerProperties={
'image': 'busybox',
'vcpus': 1,
'memory': 128,
'command': ['sleep', '10']
}
)
batch_client.register_job_definition(
jobDefinitionName=job_definition_name,
type='container',
containerProperties={
'image': 'busybox',
'vcpus': 1,
'memory': 256,
'command': ['sleep', '10']
}
)
resp = batch_client.register_job_definition(
jobDefinitionName=job_definition_name,
type='container',
containerProperties={
'image': 'busybox',
'vcpus': 1,
'memory': 512,
'command': ['sleep', '10']
}
)
job_definition_arn = resp['jobDefinitionArn']
resp = batch_client.submit_job(
jobName='test1',
jobQueue=queue_arn,
jobDefinition=job_definition_name
)
job_id = resp['jobId']
resp_jobs = batch_client.describe_jobs(jobs=[job_id])
# batch_client.terminate_job(jobId=job_id)
len(resp_jobs['jobs']).should.equal(1)
resp_jobs['jobs'][0]['jobId'].should.equal(job_id)
resp_jobs['jobs'][0]['jobQueue'].should.equal(queue_arn)
resp_jobs['jobs'][0]['jobDefinition'].should.equal(job_definition_arn)
# SLOW TESTS
@expected_failure
@mock_logs

14
tests/test_cloudformation/test_cloudformation_stack_crud.py
View file

@ -593,9 +593,11 @@ def test_create_stack_lambda_and_dynamodb():
}
},
"func1version": {
"Type": "AWS::Lambda::LambdaVersion",
"Properties" : {
"Version": "v1.2.3"
"Type": "AWS::Lambda::Version",
"Properties": {
"FunctionName": {
"Ref": "func1"
}
}
},
"tab1": {
@ -618,8 +620,10 @@ def test_create_stack_lambda_and_dynamodb():
},
"func1mapping": {
"Type": "AWS::Lambda::EventSourceMapping",
"Properties" : {
"FunctionName": "v1.2.3",
"Properties": {
"FunctionName": {
"Ref": "func1"
},
"EventSourceArn": "arn:aws:dynamodb:region:XXXXXX:table/tab1/stream/2000T00:00:00.000",
"StartingPosition": "0",
"BatchSize": 100,

520
tests/test_config/test_config.py
View file

@ -123,6 +123,526 @@ def test_put_configuration_recorder():
assert "maximum number of configuration recorders: 1 is reached." in ce.exception.response['Error']['Message']
@mock_config
def test_put_configuration_aggregator():
client = boto3.client('config', region_name='us-west-2')
# With too many aggregation sources:
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
'111111111111',
'222222222222'
],
'AwsRegions': [
'us-east-1',
'us-west-2'
]
},
{
'AccountIds': [
'012345678910',
'111111111111',
'222222222222'
],
'AwsRegions': [
'us-east-1',
'us-west-2'
]
}
]
)
assert 'Member must have length less than or equal to 1' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# With an invalid region config (no regions defined):
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
'111111111111',
'222222222222'
],
'AllAwsRegions': False
}
]
)
assert 'Your request does not specify any regions' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
OrganizationAggregationSource={
'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole'
}
)
assert 'Your request does not specify any regions' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'
# With both region flags defined:
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
'111111111111',
'222222222222'
],
'AwsRegions': [
'us-east-1',
'us-west-2'
],
'AllAwsRegions': True
}
]
)
assert 'You must choose one of these options' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
OrganizationAggregationSource={
'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',
'AwsRegions': [
'us-east-1',
'us-west-2'
],
'AllAwsRegions': True
}
)
assert 'You must choose one of these options' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'
# Name too long:
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='a' * 257,
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
]
)
assert 'configurationAggregatorName' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Too many tags (>50):
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
],
Tags=[{'Key': '{}'.format(x), 'Value': '{}'.format(x)} for x in range(0, 51)]
)
assert 'Member must have length less than or equal to 50' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Tag key is too big (>128 chars):
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
],
Tags=[{'Key': 'a' * 129, 'Value': 'a'}]
)
assert 'Member must have length less than or equal to 128' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Tag value is too big (>256 chars):
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
],
Tags=[{'Key': 'tag', 'Value': 'a' * 257}]
)
assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Duplicate Tags:
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
],
Tags=[{'Key': 'a', 'Value': 'a'}, {'Key': 'a', 'Value': 'a'}]
)
assert 'Duplicate tag keys found.' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidInput'
# Invalid characters in the tag key:
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
],
Tags=[{'Key': '!', 'Value': 'a'}]
)
assert 'Member must satisfy regular expression pattern:' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# If it contains both the AccountAggregationSources and the OrganizationAggregationSource
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': False
}
],
OrganizationAggregationSource={
'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',
'AllAwsRegions': False
}
)
assert 'AccountAggregationSource and the OrganizationAggregationSource' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'
# If it contains neither:
with assert_raises(ClientError) as ce:
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
)
assert 'AccountAggregationSource or the OrganizationAggregationSource' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'
# Just make one:
account_aggregation_source = {
'AccountIds': [
'012345678910',
'111111111111',
'222222222222'
],
'AwsRegions': [
'us-east-1',
'us-west-2'
],
'AllAwsRegions': False
}
result = client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[account_aggregation_source],
)
assert result['ConfigurationAggregator']['ConfigurationAggregatorName'] == 'testing'
assert result['ConfigurationAggregator']['AccountAggregationSources'] == [account_aggregation_source]
assert 'arn:aws:config:us-west-2:123456789012:config-aggregator/config-aggregator-' in \
result['ConfigurationAggregator']['ConfigurationAggregatorArn']
assert result['ConfigurationAggregator']['CreationTime'] == result['ConfigurationAggregator']['LastUpdatedTime']
# Update the existing one:
original_arn = result['ConfigurationAggregator']['ConfigurationAggregatorArn']
account_aggregation_source.pop('AwsRegions')
account_aggregation_source['AllAwsRegions'] = True
result = client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[account_aggregation_source]
)
assert result['ConfigurationAggregator']['ConfigurationAggregatorName'] == 'testing'
assert result['ConfigurationAggregator']['AccountAggregationSources'] == [account_aggregation_source]
assert result['ConfigurationAggregator']['ConfigurationAggregatorArn'] == original_arn
# Make an org one:
result = client.put_configuration_aggregator(
ConfigurationAggregatorName='testingOrg',
OrganizationAggregationSource={
'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',
'AwsRegions': ['us-east-1', 'us-west-2']
}
)
assert result['ConfigurationAggregator']['ConfigurationAggregatorName'] == 'testingOrg'
assert result['ConfigurationAggregator']['OrganizationAggregationSource'] == {
'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',
'AwsRegions': [
'us-east-1',
'us-west-2'
],
'AllAwsRegions': False
}
@mock_config
def test_describe_configuration_aggregators():
client = boto3.client('config', region_name='us-west-2')
# Without any config aggregators:
assert not client.describe_configuration_aggregators()['ConfigurationAggregators']
# Make 10 config aggregators:
for x in range(0, 10):
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing{}'.format(x),
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
]
)
# Describe with an incorrect name:
with assert_raises(ClientError) as ce:
client.describe_configuration_aggregators(ConfigurationAggregatorNames=['DoesNotExist'])
assert 'The configuration aggregator does not exist.' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationAggregatorException'
# Error describe with more than 1 item in the list:
with assert_raises(ClientError) as ce:
client.describe_configuration_aggregators(ConfigurationAggregatorNames=['testing0', 'DoesNotExist'])
assert 'At least one of the configuration aggregators does not exist.' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationAggregatorException'
# Get the normal list:
result = client.describe_configuration_aggregators()
assert not result.get('NextToken')
assert len(result['ConfigurationAggregators']) == 10
# Test filtered list:
agg_names = ['testing0', 'testing1', 'testing2']
result = client.describe_configuration_aggregators(ConfigurationAggregatorNames=agg_names)
assert not result.get('NextToken')
assert len(result['ConfigurationAggregators']) == 3
assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == agg_names
# Test Pagination:
result = client.describe_configuration_aggregators(Limit=4)
assert len(result['ConfigurationAggregators']) == 4
assert result['NextToken'] == 'testing4'
assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == \
['testing{}'.format(x) for x in range(0, 4)]
result = client.describe_configuration_aggregators(Limit=4, NextToken='testing4')
assert len(result['ConfigurationAggregators']) == 4
assert result['NextToken'] == 'testing8'
assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == \
['testing{}'.format(x) for x in range(4, 8)]
result = client.describe_configuration_aggregators(Limit=4, NextToken='testing8')
assert len(result['ConfigurationAggregators']) == 2
assert not result.get('NextToken')
assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == \
['testing{}'.format(x) for x in range(8, 10)]
# Test Pagination with Filtering:
result = client.describe_configuration_aggregators(ConfigurationAggregatorNames=['testing2', 'testing4'], Limit=1)
assert len(result['ConfigurationAggregators']) == 1
assert result['NextToken'] == 'testing4'
assert result['ConfigurationAggregators'][0]['ConfigurationAggregatorName'] == 'testing2'
result = client.describe_configuration_aggregators(ConfigurationAggregatorNames=['testing2', 'testing4'], Limit=1, NextToken='testing4')
assert not result.get('NextToken')
assert result['ConfigurationAggregators'][0]['ConfigurationAggregatorName'] == 'testing4'
# Test with an invalid filter:
with assert_raises(ClientError) as ce:
client.describe_configuration_aggregators(NextToken='WRONG')
assert 'The nextToken provided is invalid' == ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidNextTokenException'
@mock_config
def test_put_aggregation_authorization():
client = boto3.client('config', region_name='us-west-2')
# Too many tags (>50):
with assert_raises(ClientError) as ce:
client.put_aggregation_authorization(
AuthorizedAccountId='012345678910',
AuthorizedAwsRegion='us-west-2',
Tags=[{'Key': '{}'.format(x), 'Value': '{}'.format(x)} for x in range(0, 51)]
)
assert 'Member must have length less than or equal to 50' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Tag key is too big (>128 chars):
with assert_raises(ClientError) as ce:
client.put_aggregation_authorization(
AuthorizedAccountId='012345678910',
AuthorizedAwsRegion='us-west-2',
Tags=[{'Key': 'a' * 129, 'Value': 'a'}]
)
assert 'Member must have length less than or equal to 128' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Tag value is too big (>256 chars):
with assert_raises(ClientError) as ce:
client.put_aggregation_authorization(
AuthorizedAccountId='012345678910',
AuthorizedAwsRegion='us-west-2',
Tags=[{'Key': 'tag', 'Value': 'a' * 257}]
)
assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Duplicate Tags:
with assert_raises(ClientError) as ce:
client.put_aggregation_authorization(
AuthorizedAccountId='012345678910',
AuthorizedAwsRegion='us-west-2',
Tags=[{'Key': 'a', 'Value': 'a'}, {'Key': 'a', 'Value': 'a'}]
)
assert 'Duplicate tag keys found.' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidInput'
# Invalid characters in the tag key:
with assert_raises(ClientError) as ce:
client.put_aggregation_authorization(
AuthorizedAccountId='012345678910',
AuthorizedAwsRegion='us-west-2',
Tags=[{'Key': '!', 'Value': 'a'}]
)
assert 'Member must satisfy regular expression pattern:' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'ValidationException'
# Put a normal one there:
result = client.put_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-east-1',
Tags=[{'Key': 'tag', 'Value': 'a'}])
assert result['AggregationAuthorization']['AggregationAuthorizationArn'] == 'arn:aws:config:us-west-2:123456789012:' \
'aggregation-authorization/012345678910/us-east-1'
assert result['AggregationAuthorization']['AuthorizedAccountId'] == '012345678910'
assert result['AggregationAuthorization']['AuthorizedAwsRegion'] == 'us-east-1'
assert isinstance(result['AggregationAuthorization']['CreationTime'], datetime)
creation_date = result['AggregationAuthorization']['CreationTime']
# And again:
result = client.put_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-east-1')
assert result['AggregationAuthorization']['AggregationAuthorizationArn'] == 'arn:aws:config:us-west-2:123456789012:' \
'aggregation-authorization/012345678910/us-east-1'
assert result['AggregationAuthorization']['AuthorizedAccountId'] == '012345678910'
assert result['AggregationAuthorization']['AuthorizedAwsRegion'] == 'us-east-1'
assert result['AggregationAuthorization']['CreationTime'] == creation_date
@mock_config
def test_describe_aggregation_authorizations():
client = boto3.client('config', region_name='us-west-2')
# With no aggregation authorizations:
assert not client.describe_aggregation_authorizations()['AggregationAuthorizations']
# Make 10 account authorizations:
for i in range(0, 10):
client.put_aggregation_authorization(AuthorizedAccountId='{}'.format(str(i) * 12), AuthorizedAwsRegion='us-west-2')
result = client.describe_aggregation_authorizations()
assert len(result['AggregationAuthorizations']) == 10
assert not result.get('NextToken')
for i in range(0, 10):
assert result['AggregationAuthorizations'][i]['AuthorizedAccountId'] == str(i) * 12
# Test Pagination:
result = client.describe_aggregation_authorizations(Limit=4)
assert len(result['AggregationAuthorizations']) == 4
assert result['NextToken'] == ('4' * 12) + '/us-west-2'
assert [auth['AuthorizedAccountId'] for auth in result['AggregationAuthorizations']] == ['{}'.format(str(x) * 12) for x in range(0, 4)]
result = client.describe_aggregation_authorizations(Limit=4, NextToken=('4' * 12) + '/us-west-2')
assert len(result['AggregationAuthorizations']) == 4
assert result['NextToken'] == ('8' * 12) + '/us-west-2'
assert [auth['AuthorizedAccountId'] for auth in result['AggregationAuthorizations']] == ['{}'.format(str(x) * 12) for x in range(4, 8)]
result = client.describe_aggregation_authorizations(Limit=4, NextToken=('8' * 12) + '/us-west-2')
assert len(result['AggregationAuthorizations']) == 2
assert not result.get('NextToken')
assert [auth['AuthorizedAccountId'] for auth in result['AggregationAuthorizations']] == ['{}'.format(str(x) * 12) for x in range(8, 10)]
# Test with an invalid filter:
with assert_raises(ClientError) as ce:
client.describe_aggregation_authorizations(NextToken='WRONG')
assert 'The nextToken provided is invalid' == ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'InvalidNextTokenException'
@mock_config
def test_delete_aggregation_authorization():
client = boto3.client('config', region_name='us-west-2')
client.put_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-west-2')
# Delete it:
client.delete_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-west-2')
# Verify that none are there:
assert not client.describe_aggregation_authorizations()['AggregationAuthorizations']
# Try it again -- nothing should happen:
client.delete_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-west-2')
@mock_config
def test_delete_configuration_aggregator():
client = boto3.client('config', region_name='us-west-2')
client.put_configuration_aggregator(
ConfigurationAggregatorName='testing',
AccountAggregationSources=[
{
'AccountIds': [
'012345678910',
],
'AllAwsRegions': True
}
]
)
client.delete_configuration_aggregator(ConfigurationAggregatorName='testing')
# And again to confirm that it's deleted:
with assert_raises(ClientError) as ce:
client.delete_configuration_aggregator(ConfigurationAggregatorName='testing')
assert 'The configuration aggregator does not exist.' in ce.exception.response['Error']['Message']
assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationAggregatorException'
@mock_config
def test_describe_configurations():
client = boto3.client('config', region_name='us-west-2')

21
tests/test_core/test_auth.py
View file

@ -273,6 +273,27 @@ def test_access_denied_with_denying_policy():
)
@set_initial_no_auth_action_count(3)
@mock_sts
def test_get_caller_identity_allowed_with_denying_policy():
user_name = 'test-user'
inline_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Deny",
"Action": "sts:GetCallerIdentity",
"Resource": "*"
}
]
}
access_key = create_user_with_access_key_and_inline_policy(user_name, inline_policy_document)
client = boto3.client('sts', region_name='us-east-1',
aws_access_key_id=access_key['AccessKeyId'],
aws_secret_access_key=access_key['SecretAccessKey'])
client.get_caller_identity().should.be.a(dict)
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_allowed_with_wildcard_action():

52
tests/test_dynamodb2/test_dynamodb.py
View file

@ -2141,3 +2141,55 @@ def test_scan_by_non_exists_index():
ex.exception.response['Error']['Message'].should.equal(
'The table does not have the specified index: non_exists_index'
)
@mock_dynamodb2
def test_batch_items_returns_all():
dynamodb = _create_user_table()
returned_items = dynamodb.batch_get_item(RequestItems={
'users': {
'Keys': [{
'username': {'S': 'user0'}
}, {
'username': {'S': 'user1'}
}, {
'username': {'S': 'user2'}
}, {
'username': {'S': 'user3'}
}],
'ConsistentRead': True
}
})['Responses']['users']
assert len(returned_items) == 3
assert [item['username']['S'] for item in returned_items] == ['user1', 'user2', 'user3']
@mock_dynamodb2
def test_batch_items_should_throw_exception_for_duplicate_request():
client = _create_user_table()
with assert_raises(ClientError) as ex:
client.batch_get_item(RequestItems={
'users': {
'Keys': [{
'username': {'S': 'user0'}
}, {
'username': {'S': 'user0'}
}],
'ConsistentRead': True
}})
ex.exception.response['Error']['Code'].should.equal('ValidationException')
ex.exception.response['Error']['Message'].should.equal('Provided list of item keys contains duplicates')
def _create_user_table():
client = boto3.client('dynamodb', region_name='us-east-1')
client.create_table(
TableName='users',
KeySchema=[{'AttributeName': 'username', 'KeyType': 'HASH'}],
AttributeDefinitions=[{'AttributeName': 'username', 'AttributeType': 'S'}],
ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
)
client.put_item(TableName='users', Item={'username': {'S': 'user1'}, 'foo': {'S': 'bar'}})
client.put_item(TableName='users', Item={'username': {'S': 'user2'}, 'foo': {'S': 'bar'}})
client.put_item(TableName='users', Item={'username': {'S': 'user3'}, 'foo': {'S': 'bar'}})
return client

415
tests/test_ec2/test_launch_templates.py Normal file
View file

@ -0,0 +1,415 @@
import boto3
import sure # noqa
from nose.tools import assert_raises
from botocore.client import ClientError
from moto import mock_ec2
@mock_ec2
def test_launch_template_create():
cli = boto3.client("ec2", region_name="us-east-1")
resp = cli.create_launch_template(
LaunchTemplateName="test-template",
# the absolute minimum needed to create a template without other resources
LaunchTemplateData={
"TagSpecifications": [{
"ResourceType": "instance",
"Tags": [{
"Key": "test",
"Value": "value",
}],
}],
},
)
resp.should.have.key("LaunchTemplate")
lt = resp["LaunchTemplate"]
lt["LaunchTemplateName"].should.equal("test-template")
lt["DefaultVersionNumber"].should.equal(1)
lt["LatestVersionNumber"].should.equal(1)
with assert_raises(ClientError) as ex:
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"TagSpecifications": [{
"ResourceType": "instance",
"Tags": [{
"Key": "test",
"Value": "value",
}],
}],
},
)
str(ex.exception).should.equal(
'An error occurred (InvalidLaunchTemplateName.AlreadyExistsException) when calling the CreateLaunchTemplate operation: Launch template name already in use.')
@mock_ec2
def test_describe_launch_template_versions():
template_data = {
"ImageId": "ami-abc123",
"DisableApiTermination": False,
"TagSpecifications": [{
"ResourceType": "instance",
"Tags": [{
"Key": "test",
"Value": "value",
}],
}],
"SecurityGroupIds": [
"sg-1234",
"sg-ab5678",
],
}
cli = boto3.client("ec2", region_name="us-east-1")
create_resp = cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData=template_data)
# test using name
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template",
Versions=['1'])
templ = resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]
templ.should.equal(template_data)
# test using id
resp = cli.describe_launch_template_versions(
LaunchTemplateId=create_resp["LaunchTemplate"]["LaunchTemplateId"],
Versions=['1'])
templ = resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]
templ.should.equal(template_data)
@mock_ec2
def test_create_launch_template_version():
cli = boto3.client("ec2", region_name="us-east-1")
create_resp = cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
version_resp = cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
version_resp.should.have.key("LaunchTemplateVersion")
version = version_resp["LaunchTemplateVersion"]
version["DefaultVersion"].should.equal(False)
version["LaunchTemplateId"].should.equal(create_resp["LaunchTemplate"]["LaunchTemplateId"])
version["VersionDescription"].should.equal("new ami")
version["VersionNumber"].should.equal(2)
@mock_ec2
def test_create_launch_template_version_by_id():
cli = boto3.client("ec2", region_name="us-east-1")
create_resp = cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
version_resp = cli.create_launch_template_version(
LaunchTemplateId=create_resp["LaunchTemplate"]["LaunchTemplateId"],
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
version_resp.should.have.key("LaunchTemplateVersion")
version = version_resp["LaunchTemplateVersion"]
version["DefaultVersion"].should.equal(False)
version["LaunchTemplateId"].should.equal(create_resp["LaunchTemplate"]["LaunchTemplateId"])
version["VersionDescription"].should.equal("new ami")
version["VersionNumber"].should.equal(2)
@mock_ec2
def test_describe_launch_template_versions_with_multiple_versions():
cli = boto3.client("ec2", region_name="us-east-1")
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template")
resp["LaunchTemplateVersions"].should.have.length_of(2)
resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]["ImageId"].should.equal("ami-abc123")
resp["LaunchTemplateVersions"][1]["LaunchTemplateData"]["ImageId"].should.equal("ami-def456")
@mock_ec2
def test_describe_launch_template_versions_with_versions_option():
cli = boto3.client("ec2", region_name="us-east-1")
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-hij789"
},
VersionDescription="new ami, again")
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template",
Versions=["2", "3"])
resp["LaunchTemplateVersions"].should.have.length_of(2)
resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]["ImageId"].should.equal("ami-def456")
resp["LaunchTemplateVersions"][1]["LaunchTemplateData"]["ImageId"].should.equal("ami-hij789")
@mock_ec2
def test_describe_launch_template_versions_with_min():
cli = boto3.client("ec2", region_name="us-east-1")
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-hij789"
},
VersionDescription="new ami, again")
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template",
MinVersion="2")
resp["LaunchTemplateVersions"].should.have.length_of(2)
resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]["ImageId"].should.equal("ami-def456")
resp["LaunchTemplateVersions"][1]["LaunchTemplateData"]["ImageId"].should.equal("ami-hij789")
@mock_ec2
def test_describe_launch_template_versions_with_max():
cli = boto3.client("ec2", region_name="us-east-1")
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-hij789"
},
VersionDescription="new ami, again")
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template",
MaxVersion="2")
resp["LaunchTemplateVersions"].should.have.length_of(2)
resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]["ImageId"].should.equal("ami-abc123")
resp["LaunchTemplateVersions"][1]["LaunchTemplateData"]["ImageId"].should.equal("ami-def456")
@mock_ec2
def test_describe_launch_template_versions_with_min_and_max():
cli = boto3.client("ec2", region_name="us-east-1")
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-def456"
},
VersionDescription="new ami")
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-hij789"
},
VersionDescription="new ami, again")
cli.create_launch_template_version(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-345abc"
},
VersionDescription="new ami, because why not")
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template",
MinVersion="2",
MaxVersion="3")
resp["LaunchTemplateVersions"].should.have.length_of(2)
resp["LaunchTemplateVersions"][0]["LaunchTemplateData"]["ImageId"].should.equal("ami-def456")
resp["LaunchTemplateVersions"][1]["LaunchTemplateData"]["ImageId"].should.equal("ami-hij789")
@mock_ec2
def test_describe_launch_templates():
cli = boto3.client("ec2", region_name="us-east-1")
lt_ids = []
r = cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
lt_ids.append(r["LaunchTemplate"]["LaunchTemplateId"])
r = cli.create_launch_template(
LaunchTemplateName="test-template2",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
lt_ids.append(r["LaunchTemplate"]["LaunchTemplateId"])
# general call, all templates
resp = cli.describe_launch_templates()
resp.should.have.key("LaunchTemplates")
resp["LaunchTemplates"].should.have.length_of(2)
resp["LaunchTemplates"][0]["LaunchTemplateName"].should.equal("test-template")
resp["LaunchTemplates"][1]["LaunchTemplateName"].should.equal("test-template2")
# filter by names
resp = cli.describe_launch_templates(
LaunchTemplateNames=["test-template2", "test-template"])
resp.should.have.key("LaunchTemplates")
resp["LaunchTemplates"].should.have.length_of(2)
resp["LaunchTemplates"][0]["LaunchTemplateName"].should.equal("test-template2")
resp["LaunchTemplates"][1]["LaunchTemplateName"].should.equal("test-template")
# filter by ids
resp = cli.describe_launch_templates(LaunchTemplateIds=lt_ids)
resp.should.have.key("LaunchTemplates")
resp["LaunchTemplates"].should.have.length_of(2)
resp["LaunchTemplates"][0]["LaunchTemplateName"].should.equal("test-template")
resp["LaunchTemplates"][1]["LaunchTemplateName"].should.equal("test-template2")
@mock_ec2
def test_describe_launch_templates_with_filters():
cli = boto3.client("ec2", region_name="us-east-1")
r = cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
cli.create_tags(
Resources=[r["LaunchTemplate"]["LaunchTemplateId"]],
Tags=[
{"Key": "tag1", "Value": "a value"},
{"Key": "another-key", "Value": "this value"},
])
cli.create_launch_template(
LaunchTemplateName="no-tags",
LaunchTemplateData={
"ImageId": "ami-abc123"
})
resp = cli.describe_launch_templates(Filters=[{
"Name": "tag:tag1", "Values": ["a value"]
}])
resp["LaunchTemplates"].should.have.length_of(1)
resp["LaunchTemplates"][0]["LaunchTemplateName"].should.equal("test-template")
resp = cli.describe_launch_templates(Filters=[{
"Name": "launch-template-name", "Values": ["no-tags"]
}])
resp["LaunchTemplates"].should.have.length_of(1)
resp["LaunchTemplates"][0]["LaunchTemplateName"].should.equal("no-tags")
@mock_ec2
def test_create_launch_template_with_tag_spec():
cli = boto3.client("ec2", region_name="us-east-1")
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={"ImageId": "ami-abc123"},
TagSpecifications=[{
"ResourceType": "instance",
"Tags": [
{"Key": "key", "Value": "value"}
]
}],
)
resp = cli.describe_launch_template_versions(
LaunchTemplateName="test-template",
Versions=["1"])
version = resp["LaunchTemplateVersions"][0]
version["LaunchTemplateData"].should.have.key("TagSpecifications")
version["LaunchTemplateData"]["TagSpecifications"].should.have.length_of(1)
version["LaunchTemplateData"]["TagSpecifications"][0].should.equal({
"ResourceType": "instance",
"Tags": [
{"Key": "key", "Value": "value"}
]
})

3
tests/test_ecs/test_ecs_boto3.py
View file

@ -1,4 +1,5 @@
from __future__ import unicode_literals
from datetime import datetime
from copy import deepcopy
@ -477,6 +478,8 @@ def test_describe_services():
response['services'][0]['deployments'][0]['pendingCount'].should.equal(2)
response['services'][0]['deployments'][0]['runningCount'].should.equal(0)
response['services'][0]['deployments'][0]['status'].should.equal('PRIMARY')
(datetime.now() - response['services'][0]['deployments'][0]["createdAt"].replace(tzinfo=None)).seconds.should.be.within(0, 10)
(datetime.now() - response['services'][0]['deployments'][0]["updatedAt"].replace(tzinfo=None)).seconds.should.be.within(0, 10)
@mock_ecs

129
tests/test_elbv2/test_elbv2.py
View file

@ -1811,3 +1811,132 @@ def test_redirect_action_listener_rule_cloudformation():
'Port': '443', 'Protocol': 'HTTPS', 'StatusCode': 'HTTP_301',
}
},])
@mock_elbv2
@mock_ec2
def test_cognito_action_listener_rule():
conn = boto3.client('elbv2', region_name='us-east-1')
ec2 = boto3.resource('ec2', region_name='us-east-1')
security_group = ec2.create_security_group(
GroupName='a-security-group', Description='First One')
vpc = ec2.create_vpc(CidrBlock='172.28.7.0/24', InstanceTenancy='default')
subnet1 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock='172.28.7.192/26',
AvailabilityZone='us-east-1a')
subnet2 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock='172.28.7.128/26',
AvailabilityZone='us-east-1b')
response = conn.create_load_balancer(
Name='my-lb',
Subnets=[subnet1.id, subnet2.id],
SecurityGroups=[security_group.id],
Scheme='internal',
Tags=[{'Key': 'key_name', 'Value': 'a_value'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
action = {
'Type': 'authenticate-cognito',
'AuthenticateCognitoConfig': {
'UserPoolArn': 'arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_ABCD1234',
'UserPoolClientId': 'abcd1234abcd',
'UserPoolDomain': 'testpool',
}
}
response = conn.create_listener(LoadBalancerArn=load_balancer_arn,
Protocol='HTTP',
Port=80,
DefaultActions=[action])
listener = response.get('Listeners')[0]
listener.get('DefaultActions')[0].should.equal(action)
listener_arn = listener.get('ListenerArn')
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
describe_rules_response['Rules'][0]['Actions'][0].should.equal(action)
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn, ])
describe_listener_actions = describe_listener_response['Listeners'][0]['DefaultActions'][0]
describe_listener_actions.should.equal(action)
@mock_elbv2
@mock_cloudformation
def test_cognito_action_listener_rule_cloudformation():
cnf_conn = boto3.client('cloudformation', region_name='us-east-1')
elbv2_client = boto3.client('elbv2', region_name='us-east-1')
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "ECS Cluster Test CloudFormation",
"Resources": {
"testVPC": {
"Type": "AWS::EC2::VPC",
"Properties": {
"CidrBlock": "10.0.0.0/16",
},
},
"subnet1": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.0.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"subnet2": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.1.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"testLb": {
"Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
"Properties": {
"Name": "my-lb",
"Subnets": [{"Ref": "subnet1"}, {"Ref": "subnet2"}],
"Type": "application",
"SecurityGroups": [],
}
},
"testListener": {
"Type": "AWS::ElasticLoadBalancingV2::Listener",
"Properties": {
"LoadBalancerArn": {"Ref": "testLb"},
"Port": 80,
"Protocol": "HTTP",
"DefaultActions": [{
"Type": "authenticate-cognito",
"AuthenticateCognitoConfig": {
'UserPoolArn': 'arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_ABCD1234',
'UserPoolClientId': 'abcd1234abcd',
'UserPoolDomain': 'testpool',
}
}]
}
}
}
}
template_json = json.dumps(template)
cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json)
describe_load_balancers_response = elbv2_client.describe_load_balancers(Names=['my-lb',])
load_balancer_arn = describe_load_balancers_response['LoadBalancers'][0]['LoadBalancerArn']
describe_listeners_response = elbv2_client.describe_listeners(LoadBalancerArn=load_balancer_arn)
describe_listeners_response['Listeners'].should.have.length_of(1)
describe_listeners_response['Listeners'][0]['DefaultActions'].should.equal([{
'Type': 'authenticate-cognito',
"AuthenticateCognitoConfig": {
'UserPoolArn': 'arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_ABCD1234',
'UserPoolClientId': 'abcd1234abcd',
'UserPoolDomain': 'testpool',
}
},])

86
tests/test_iam/test_iam.py
View file

@ -944,7 +944,8 @@ def test_get_account_authorization_details():
})
conn = boto3.client('iam', region_name='us-east-1')
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/")
boundary = 'arn:aws:iam::123456789012:policy/boundary'
conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/", Description='testing', PermissionsBoundary=boundary)
conn.create_user(Path='/', UserName='testUser')
conn.create_group(Path='/', GroupName='testGroup')
conn.create_policy(
@ -985,6 +986,11 @@ def test_get_account_authorization_details():
assert len(result['GroupDetailList']) == 0
assert len(result['Policies']) == 0
assert len(result['RoleDetailList'][0]['InstanceProfileList']) == 1
assert result['RoleDetailList'][0]['InstanceProfileList'][0]['Roles'][0]['Description'] == 'testing'
assert result['RoleDetailList'][0]['InstanceProfileList'][0]['Roles'][0]['PermissionsBoundary'] == {
'PermissionsBoundaryType': 'PermissionsBoundaryPolicy',
'PermissionsBoundaryArn': 'arn:aws:iam::123456789012:policy/boundary'
}
assert len(result['RoleDetailList'][0]['Tags']) == 2
assert len(result['RoleDetailList'][0]['RolePolicyList']) == 1
assert len(result['RoleDetailList'][0]['AttachedManagedPolicies']) == 1
@ -1151,6 +1157,79 @@ def test_delete_saml_provider():
assert not resp['Certificates']
@mock_iam()
def test_create_role_with_tags():
"""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="{}", Tags=[
{
'Key': 'somekey',
'Value': 'somevalue'
},
{
'Key': 'someotherkey',
'Value': 'someothervalue'
}
], Description='testing')
# 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'
assert role['Description'] == 'testing'
# Empty is good:
conn.create_role(RoleName="my-role2", AssumeRolePolicyDocument="{}", Tags=[
{
'Key': 'somekey',
'Value': ''
}
])
tags = conn.list_role_tags(RoleName='my-role2')
assert len(tags['Tags']) == 1
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.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", 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.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", 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.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", 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.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", 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.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", 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.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=[{'Key': 'NOWAY!', 'Value': ''}])
assert 'Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+' \
in ce.exception.response['Error']['Message']
@mock_iam()
def test_tag_role():
"""Tests both the tag_role and get_role_tags capability"""
@ -1338,6 +1417,7 @@ def test_update_role_description():
assert response['Role']['RoleName'] == 'my-role'
@mock_iam()
def test_update_role():
conn = boto3.client('iam', region_name='us-east-1')
@ -1349,6 +1429,7 @@ def test_update_role():
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')
@ -1443,6 +1524,8 @@ def test_create_role_no_path():
resp = conn.create_role(RoleName='my-role', AssumeRolePolicyDocument='some policy', Description='test')
resp.get('Role').get('Arn').should.equal('arn:aws:iam::123456789012:role/my-role')
resp.get('Role').should_not.have.key('PermissionsBoundary')
resp.get('Role').get('Description').should.equal('test')
@mock_iam()
def test_create_role_with_permissions_boundary():
@ -1454,6 +1537,7 @@ def test_create_role_with_permissions_boundary():
'PermissionsBoundaryArn': boundary
}
resp.get('Role').get('PermissionsBoundary').should.equal(expected)
resp.get('Role').get('Description').should.equal('test')
invalid_boundary_arn = 'arn:aws:iam::123456789:not_a_boundary'
with assert_raises(ClientError):

1
tests/test_kms/test_kms.py
View file

@ -191,6 +191,7 @@ def test_decrypt():
conn = boto.kms.connect_to_region('us-west-2')
response = conn.decrypt('ZW5jcnlwdG1l'.encode('utf-8'))
response['Plaintext'].should.equal(b'encryptme')
response['KeyId'].should.equal('key_id')
@mock_kms_deprecated

60
tests/test_logs/test_logs.py
View file

@ -162,3 +162,63 @@ def test_delete_retention_policy():
response = conn.delete_log_group(logGroupName=log_group_name)
@mock_logs
def test_get_log_events():
conn = boto3.client('logs', 'us-west-2')
log_group_name = 'test'
log_stream_name = 'stream'
conn.create_log_group(logGroupName=log_group_name)
conn.create_log_stream(
logGroupName=log_group_name,
logStreamName=log_stream_name
)
events = [{'timestamp': x, 'message': str(x)} for x in range(20)]
conn.put_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
logEvents=events
)
resp = conn.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
limit=10)
resp['events'].should.have.length_of(10)
resp.should.have.key('nextForwardToken')
resp.should.have.key('nextBackwardToken')
for i in range(10):
resp['events'][i]['timestamp'].should.equal(i)
resp['events'][i]['message'].should.equal(str(i))
next_token = resp['nextForwardToken']
resp = conn.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=next_token,
limit=10)
resp['events'].should.have.length_of(10)
resp.should.have.key('nextForwardToken')
resp.should.have.key('nextBackwardToken')
resp['nextForwardToken'].should.equal(next_token)
for i in range(10):
resp['events'][i]['timestamp'].should.equal(i+10)
resp['events'][i]['message'].should.equal(str(i+10))
resp = conn.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
nextToken=resp['nextBackwardToken'],
limit=10)
resp['events'].should.have.length_of(10)
resp.should.have.key('nextForwardToken')
resp.should.have.key('nextBackwardToken')
for i in range(10):
resp['events'][i]['timestamp'].should.equal(i)
resp['events'][i]['message'].should.equal(str(i))

1
tests/test_organizations/organizations_test_utils.py
View file

@ -1,7 +1,6 @@
from __future__ import unicode_literals
import six
import sure # noqa
import datetime
from moto.organizations import utils

31
tests/test_organizations/test_organizations_boto3.py
View file

@ -3,7 +3,6 @@ from __future__ import unicode_literals
import boto3
import json
import six
import sure # noqa
from botocore.exceptions import ClientError
from nose.tools import assert_raises
@ -27,6 +26,25 @@ def test_create_organization():
validate_organization(response)
response['Organization']['FeatureSet'].should.equal('ALL')
response = client.list_accounts()
len(response['Accounts']).should.equal(1)
response['Accounts'][0]['Name'].should.equal('master')
response['Accounts'][0]['Id'].should.equal(utils.MASTER_ACCOUNT_ID)
response['Accounts'][0]['Email'].should.equal(utils.MASTER_ACCOUNT_EMAIL)
response = client.list_policies(Filter='SERVICE_CONTROL_POLICY')
len(response['Policies']).should.equal(1)
response['Policies'][0]['Name'].should.equal('FullAWSAccess')
response['Policies'][0]['Id'].should.equal(utils.DEFAULT_POLICY_ID)
response['Policies'][0]['AwsManaged'].should.equal(True)
response = client.list_targets_for_policy(PolicyId=utils.DEFAULT_POLICY_ID)
len(response['Targets']).should.equal(2)
root_ou = [t for t in response['Targets'] if t['Type'] == 'ROOT'][0]
root_ou['Name'].should.equal('Root')
master_account = [t for t in response['Targets'] if t['Type'] == 'ACCOUNT'][0]
master_account['Name'].should.equal('master')
@mock_organizations
def test_describe_organization():
@ -177,11 +195,11 @@ def test_list_accounts():
response = client.list_accounts()
response.should.have.key('Accounts')
accounts = response['Accounts']
len(accounts).should.equal(5)
len(accounts).should.equal(6)
for account in accounts:
validate_account(org, account)
accounts[3]['Name'].should.equal(mockname + '3')
accounts[2]['Email'].should.equal(mockname + '2' + '@' + mockdomain)
accounts[4]['Name'].should.equal(mockname + '3')
accounts[3]['Email'].should.equal(mockname + '2' + '@' + mockdomain)
@mock_organizations
@ -291,8 +309,10 @@ def test_list_children():
response02 = client.list_children(ParentId=root_id, ChildType='ORGANIZATIONAL_UNIT')
response03 = client.list_children(ParentId=ou01_id, ChildType='ACCOUNT')
response04 = client.list_children(ParentId=ou01_id, ChildType='ORGANIZATIONAL_UNIT')
response01['Children'][0]['Id'].should.equal(account01_id)
response01['Children'][0]['Id'].should.equal(utils.MASTER_ACCOUNT_ID)
response01['Children'][0]['Type'].should.equal('ACCOUNT')
response01['Children'][1]['Id'].should.equal(account01_id)
response01['Children'][1]['Type'].should.equal('ACCOUNT')
response02['Children'][0]['Id'].should.equal(ou01_id)
response02['Children'][0]['Type'].should.equal('ORGANIZATIONAL_UNIT')
response03['Children'][0]['Id'].should.equal(account02_id)
@ -591,4 +611,3 @@ def test_list_targets_for_policy_exception():
ex.operation_name.should.equal('ListTargetsForPolicy')
ex.response['Error']['Code'].should.equal('400')
ex.response['Error']['Message'].should.contain('InvalidInputException')

1
tests/test_redshift/test_redshift.py
View file

@ -36,6 +36,7 @@ def test_create_cluster_boto3():
response['Cluster']['NodeType'].should.equal('ds2.xlarge')
create_time = response['Cluster']['ClusterCreateTime']
create_time.should.be.lower_than(datetime.datetime.now(create_time.tzinfo))
create_time.should.be.greater_than(datetime.datetime.now(create_time.tzinfo) - datetime.timedelta(minutes=1))
@mock_redshift

42
tests/test_s3/test_s3_storageclass.py
View file

@ -1,16 +1,12 @@
from __future__ import unicode_literals
import boto
import boto3
from boto.exception import S3CreateError, S3ResponseError
from boto.s3.lifecycle import Lifecycle, Transition, Expiration, Rule
import sure # noqa
from botocore.exceptions import ClientError
from datetime import datetime
from nose.tools import assert_raises
from moto import mock_s3_deprecated, mock_s3
from moto import mock_s3
@mock_s3
@ -41,6 +37,18 @@ def test_s3_storage_class_infrequent_access():
D['Contents'][0]["StorageClass"].should.equal("STANDARD_IA")
@mock_s3
def test_s3_storage_class_intelligent_tiering():
s3 = boto3.client("s3")
s3.create_bucket(Bucket="Bucket")
s3.put_object(Bucket="Bucket", Key="my_key_infrequent", Body="my_value_infrequent", StorageClass="INTELLIGENT_TIERING")
objects = s3.list_objects(Bucket="Bucket")
objects['Contents'][0]["StorageClass"].should.equal("INTELLIGENT_TIERING")
@mock_s3
def test_s3_storage_class_copy():
s3 = boto3.client("s3")
@ -90,6 +98,7 @@ def test_s3_invalid_storage_class():
e.response["Error"]["Code"].should.equal("InvalidStorageClass")
e.response["Error"]["Message"].should.equal("The storage class you specified is not valid")
@mock_s3
def test_s3_default_storage_class():
s3 = boto3.client("s3")
@ -103,4 +112,27 @@ def test_s3_default_storage_class():
list_of_objects["Contents"][0]["StorageClass"].should.equal("STANDARD")
@mock_s3
def test_s3_copy_object_error_for_glacier_storage_class():
s3 = boto3.client("s3")
s3.create_bucket(Bucket="Bucket")
s3.put_object(Bucket="Bucket", Key="First_Object", Body="Body", StorageClass="GLACIER")
with assert_raises(ClientError) as exc:
s3.copy_object(CopySource={"Bucket": "Bucket", "Key": "First_Object"}, Bucket="Bucket", Key="Second_Object")
exc.exception.response["Error"]["Code"].should.equal("ObjectNotInActiveTierError")
@mock_s3
def test_s3_copy_object_error_for_deep_archive_storage_class():
s3 = boto3.client("s3")
s3.create_bucket(Bucket="Bucket")
s3.put_object(Bucket="Bucket", Key="First_Object", Body="Body", StorageClass="DEEP_ARCHIVE")
with assert_raises(ClientError) as exc:
s3.copy_object(CopySource={"Bucket": "Bucket", "Key": "First_Object"}, Bucket="Bucket", Key="Second_Object")
exc.exception.response["Error"]["Code"].should.equal("ObjectNotInActiveTierError")

110
tests/test_sts/test_sts.py
View file

@ -8,7 +8,9 @@ from freezegun import freeze_time
from nose.tools import assert_raises
import sure # noqa
from moto import mock_sts, mock_sts_deprecated
from moto import mock_sts, mock_sts_deprecated, mock_iam, settings
from moto.iam.models import ACCOUNT_ID
from moto.sts.responses import MAX_FEDERATION_TOKEN_POLICY_LENGTH
@ -29,7 +31,8 @@ def test_get_session_token():
@mock_sts_deprecated
def test_get_federation_token():
conn = boto.connect_sts()
token = conn.get_federation_token(duration=123, name="Bob")
token_name = "Bob"
token = conn.get_federation_token(duration=123, name=token_name)
token.credentials.expiration.should.equal('2012-01-01T12:02:03.000Z')
token.credentials.session_token.should.equal(
@ -38,15 +41,17 @@ def test_get_federation_token():
token.credentials.secret_key.should.equal(
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY")
token.federated_user_arn.should.equal(
"arn:aws:sts::123456789012:federated-user/Bob")
token.federated_user_id.should.equal("123456789012:Bob")
"arn:aws:sts::{account_id}:federated-user/{token_name}".format(account_id=ACCOUNT_ID, token_name=token_name))
token.federated_user_id.should.equal(str(ACCOUNT_ID) + ":" + token_name)
@freeze_time("2012-01-01 12:00:00")
@mock_sts_deprecated
@mock_sts
def test_assume_role():
conn = boto.connect_sts()
client = boto3.client(
"sts", region_name='us-east-1')
session_name = "session-name"
policy = json.dumps({
"Statement": [
{
@ -61,20 +66,25 @@ def test_assume_role():
},
]
})
s3_role = "arn:aws:iam::123456789012:role/test-role"
role = conn.assume_role(s3_role, "session-name",
policy, duration_seconds=123)
role_name = "test-role"
s3_role = "arn:aws:iam::{account_id}:role/{role_name}".format(account_id=ACCOUNT_ID, role_name=role_name)
assume_role_response = client.assume_role(RoleArn=s3_role, RoleSessionName=session_name,
Policy=policy, DurationSeconds=900)
credentials = role.credentials
credentials.expiration.should.equal('2012-01-01T12:02:03.000Z')
credentials.session_token.should.have.length_of(356)
assert credentials.session_token.startswith("FQoGZXIvYXdzE")
credentials.access_key.should.have.length_of(20)
assert credentials.access_key.startswith("ASIA")
credentials.secret_key.should.have.length_of(40)
credentials = assume_role_response['Credentials']
if not settings.TEST_SERVER_MODE:
credentials['Expiration'].isoformat().should.equal('2012-01-01T12:15:00+00:00')
credentials['SessionToken'].should.have.length_of(356)
assert credentials['SessionToken'].startswith("FQoGZXIvYXdzE")
credentials['AccessKeyId'].should.have.length_of(20)
assert credentials['AccessKeyId'].startswith("ASIA")
credentials['SecretAccessKey'].should.have.length_of(40)
role.user.arn.should.equal("arn:aws:iam::123456789012:role/test-role")
role.user.assume_role_id.should.contain("session-name")
assume_role_response['AssumedRoleUser']['Arn'].should.equal("arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID, role_name=role_name, session_name=session_name))
assert assume_role_response['AssumedRoleUser']['AssumedRoleId'].startswith("AROA")
assert assume_role_response['AssumedRoleUser']['AssumedRoleId'].endswith(":" + session_name)
assume_role_response['AssumedRoleUser']['AssumedRoleId'].should.have.length_of(21 + 1 + len(session_name))
@freeze_time("2012-01-01 12:00:00")
@ -96,9 +106,11 @@ def test_assume_role_with_web_identity():
},
]
})
s3_role = "arn:aws:iam::123456789012:role/test-role"
role_name = "test-role"
s3_role = "arn:aws:iam::{account_id}:role/{role_name}".format(account_id=ACCOUNT_ID, role_name=role_name)
session_name = "session-name"
role = conn.assume_role_with_web_identity(
s3_role, "session-name", policy, duration_seconds=123)
s3_role, session_name, policy, duration_seconds=123)
credentials = role.credentials
credentials.expiration.should.equal('2012-01-01T12:02:03.000Z')
@ -108,18 +120,68 @@ def test_assume_role_with_web_identity():
assert credentials.access_key.startswith("ASIA")
credentials.secret_key.should.have.length_of(40)
role.user.arn.should.equal("arn:aws:iam::123456789012:role/test-role")
role.user.arn.should.equal("arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID, role_name=role_name, session_name=session_name))
role.user.assume_role_id.should.contain("session-name")
@mock_sts
def test_get_caller_identity():
def test_get_caller_identity_with_default_credentials():
identity = boto3.client(
"sts", region_name='us-east-1').get_caller_identity()
identity['Arn'].should.equal('arn:aws:sts::123456789012:user/moto')
identity['Arn'].should.equal('arn:aws:sts::{account_id}:user/moto'.format(account_id=ACCOUNT_ID))
identity['UserId'].should.equal('AKIAIOSFODNN7EXAMPLE')
identity['Account'].should.equal('123456789012')
identity['Account'].should.equal(str(ACCOUNT_ID))
@mock_sts
@mock_iam
def test_get_caller_identity_with_iam_user_credentials():
iam_client = boto3.client("iam", region_name='us-east-1')
iam_user_name = "new-user"
iam_user = iam_client.create_user(UserName=iam_user_name)['User']
access_key = iam_client.create_access_key(UserName=iam_user_name)['AccessKey']
identity = boto3.client(
"sts", region_name='us-east-1', aws_access_key_id=access_key['AccessKeyId'],
aws_secret_access_key=access_key['SecretAccessKey']).get_caller_identity()
identity['Arn'].should.equal(iam_user['Arn'])
identity['UserId'].should.equal(iam_user['UserId'])
identity['Account'].should.equal(str(ACCOUNT_ID))
@mock_sts
@mock_iam
def test_get_caller_identity_with_assumed_role_credentials():
iam_client = boto3.client("iam", region_name='us-east-1')
sts_client = boto3.client("sts", region_name='us-east-1')
iam_role_name = "new-user"
trust_policy_document = {
"Version": "2012-10-17",
"Statement": {
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::{account_id}:root".format(account_id=ACCOUNT_ID)},
"Action": "sts:AssumeRole"
}
}
iam_role_arn = iam_client.role_arn = iam_client.create_role(
RoleName=iam_role_name,
AssumeRolePolicyDocument=json.dumps(trust_policy_document)
)['Role']['Arn']
session_name = "new-session"
assumed_role = sts_client.assume_role(RoleArn=iam_role_arn,
RoleSessionName=session_name)
access_key = assumed_role['Credentials']
identity = boto3.client(
"sts", region_name='us-east-1', aws_access_key_id=access_key['AccessKeyId'],
aws_secret_access_key=access_key['SecretAccessKey']).get_caller_identity()
identity['Arn'].should.equal(assumed_role['AssumedRoleUser']['Arn'])
identity['UserId'].should.equal(assumed_role['AssumedRoleUser']['AssumedRoleId'])
identity['Account'].should.equal(str(ACCOUNT_ID))
@mock_sts