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

Merge #913.

Browse source
This commit is contained in:
Steve Pulec 2017-05-10 21:58:42 -04:00
commit 0adebeed24
36 changed files with 669 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

24
tests/test_autoscaling/test_autoscaling.py
View file

@ -115,6 +115,30 @@ def test_create_autoscaling_groups_defaults():
list(group.tags).should.equal([])
@mock_autoscaling
def test_list_many_autoscaling_groups():
conn = boto3.client('autoscaling', region_name='us-east-1')
conn.create_launch_configuration(LaunchConfigurationName='TestLC')
for i in range(51):
conn.create_auto_scaling_group(AutoScalingGroupName='TestGroup%d' % i,
MinSize=1,
MaxSize=2,
LaunchConfigurationName='TestLC')
response = conn.describe_auto_scaling_groups()
groups = response["AutoScalingGroups"]
marker = response["NextToken"]
groups.should.have.length_of(50)
marker.should.equal(groups[-1]['AutoScalingGroupName'])
response2 = conn.describe_auto_scaling_groups(NextToken=marker)
groups.extend(response2["AutoScalingGroups"])
groups.should.have.length_of(51)
assert 'NextToken' not in response2.keys()
@mock_autoscaling_deprecated
def test_autoscaling_group_describe_filter():
conn = boto.connect_autoscale()

21
tests/test_autoscaling/test_launch_configurations.py
View file

@ -1,11 +1,13 @@
from __future__ import unicode_literals
import boto
import boto3
from boto.ec2.autoscale.launchconfig import LaunchConfiguration
from boto.ec2.blockdevicemapping import BlockDeviceType, BlockDeviceMapping
import sure # noqa
from moto import mock_autoscaling_deprecated
from moto import mock_autoscaling
from tests.helpers import requires_boto_gte
@ -208,6 +210,25 @@ def test_launch_configuration_describe_filter():
conn.get_all_launch_configurations().should.have.length_of(3)
@mock_autoscaling
def test_launch_configuration_describe_paginated():
conn = boto3.client('autoscaling', region_name='us-east-1')
for i in range(51):
conn.create_launch_configuration(LaunchConfigurationName='TestLC%d' % i)
response = conn.describe_launch_configurations()
lcs = response["LaunchConfigurations"]
marker = response["NextToken"]
lcs.should.have.length_of(50)
marker.should.equal(lcs[-1]['LaunchConfigurationName'])
response2 = conn.describe_launch_configurations(NextToken=marker)
lcs.extend(response2["LaunchConfigurations"])
lcs.should.have.length_of(51)
assert 'NextToken' not in response2.keys()
@mock_autoscaling_deprecated
def test_launch_configuration_delete():
conn = boto.connect_autoscale()

20
tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
View file

@ -144,6 +144,26 @@ def test_create_stack_from_s3_url():
'TemplateBody'].should.equal(dummy_template)
@mock_cloudformation
def test_describe_stack_pagination():
conn = boto3.client('cloudformation', region_name='us-east-1')
for i in range(100):
conn.create_stack(
StackName="test_stack",
TemplateBody=dummy_template_json,
)
resp = conn.describe_stacks()
stacks = resp['Stacks']
stacks.should.have.length_of(50)
next_token = resp['NextToken']
next_token.should_not.be.none
resp2 = conn.describe_stacks(NextToken=next_token)
stacks.extend(resp2['Stacks'])
stacks.should.have.length_of(100)
assert 'NextToken' not in resp2.keys()
@mock_cloudformation
def test_describe_stack_resources():
cf_conn = boto3.client('cloudformation', region_name='us-east-1')

13
tests/test_datapipeline/test_datapipeline.py
View file

@ -170,6 +170,19 @@ def test_listing_pipelines():
})
@mock_datapipeline_deprecated
def test_listing_paginated_pipelines():
conn = boto.datapipeline.connect_to_region("us-west-2")
for i in range(100):
conn.create_pipeline("mypipeline%d" % i, "some-unique-id%d" % i)
response = conn.list_pipelines()
response["hasMoreResults"].should.be(True)
response["marker"].should.equal(response["pipelineIdList"][-1]['id'])
response["pipelineIdList"].should.have.length_of(50)
# testing a helper function
def test_remove_capitalization_of_dict_keys():
result = remove_capitalization_of_dict_keys(

85
tests/test_dynamodb2/test_dynamodb.py
View file

@ -2,11 +2,13 @@ from __future__ import unicode_literals, print_function
import six
import boto
import boto3
import sure # noqa
import requests
from moto import mock_dynamodb2, mock_dynamodb2_deprecated
from moto.dynamodb2 import dynamodb_backend2
from boto.exception import JSONResponseError
from botocore.exceptions import ClientError
from tests.helpers import requires_boto_gte
import tests.backport_assert_raises
from nose.tools import assert_raises
@ -64,3 +66,86 @@ def test_describe_missing_table():
aws_secret_access_key="sk")
with assert_raises(JSONResponseError):
conn.describe_table('messages')
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_list_table_tags():
name = 'TestTable'
conn = boto3.client('dynamodb',
region_name='us-west-2',
aws_access_key_id="ak",
aws_secret_access_key="sk")
conn.create_table(TableName=name,
KeySchema=[{'AttributeName':'id','KeyType':'HASH'}],
AttributeDefinitions=[{'AttributeName':'id','AttributeType':'S'}],
ProvisionedThroughput={'ReadCapacityUnits':5,'WriteCapacityUnits':5})
table_description = conn.describe_table(TableName=name)
arn = table_description['Table']['TableArn']
tags = [{'Key':'TestTag', 'Value': 'TestValue'}]
conn.tag_resource(ResourceArn=arn,
Tags=tags)
resp = conn.list_tags_of_resource(ResourceArn=arn)
assert resp["Tags"] == tags
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_list_table_tags_empty():
name = 'TestTable'
conn = boto3.client('dynamodb',
region_name='us-west-2',
aws_access_key_id="ak",
aws_secret_access_key="sk")
conn.create_table(TableName=name,
KeySchema=[{'AttributeName':'id','KeyType':'HASH'}],
AttributeDefinitions=[{'AttributeName':'id','AttributeType':'S'}],
ProvisionedThroughput={'ReadCapacityUnits':5,'WriteCapacityUnits':5})
table_description = conn.describe_table(TableName=name)
arn = table_description['Table']['TableArn']
tags = [{'Key':'TestTag', 'Value': 'TestValue'}]
# conn.tag_resource(ResourceArn=arn,
# Tags=tags)
resp = conn.list_tags_of_resource(ResourceArn=arn)
assert resp["Tags"] == []
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_list_table_tags_paginated():
name = 'TestTable'
conn = boto3.client('dynamodb',
region_name='us-west-2',
aws_access_key_id="ak",
aws_secret_access_key="sk")
conn.create_table(TableName=name,
KeySchema=[{'AttributeName':'id','KeyType':'HASH'}],
AttributeDefinitions=[{'AttributeName':'id','AttributeType':'S'}],
ProvisionedThroughput={'ReadCapacityUnits':5,'WriteCapacityUnits':5})
table_description = conn.describe_table(TableName=name)
arn = table_description['Table']['TableArn']
for i in range(11):
tags = [{'Key':'TestTag%d' % i, 'Value': 'TestValue'}]
conn.tag_resource(ResourceArn=arn,
Tags=tags)
resp = conn.list_tags_of_resource(ResourceArn=arn)
assert len(resp["Tags"]) == 10
assert 'NextToken' in resp.keys()
resp2 = conn.list_tags_of_resource(ResourceArn=arn,
NextToken=resp['NextToken'])
assert len(resp2["Tags"]) == 1
assert 'NextToken' not in resp2.keys()
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_list_not_found_table_tags():
conn = boto3.client('dynamodb',
region_name='us-west-2',
aws_access_key_id="ak",
aws_secret_access_key="sk")
arn = 'DymmyArn'
try:
conn.list_tags_of_resource(ResourceArn=arn)
except ClientError as exception:
assert exception.response['Error']['Code'] == "ResourceNotFoundException"

6
tests/test_dynamodb2/test_dynamodb_table_with_range_key.py
View file

@ -77,13 +77,14 @@ def test_create_table():
'TableSizeBytes': 0,
'TableName': 'messages',
'TableStatus': 'ACTIVE',
'TableArn': 'arn:aws:dynamodb:us-east-1:123456789011:table/messages',
'KeySchema': [
{'KeyType': 'HASH', 'AttributeName': 'forum_name'},
{'KeyType': 'RANGE', 'AttributeName': 'subject'}
],
'LocalSecondaryIndexes': [],
'ItemCount': 0, 'CreationDateTime': 1326499200.0,
'GlobalSecondaryIndexes': [],
'GlobalSecondaryIndexes': []
}
}
table.describe().should.equal(expected)
@ -109,6 +110,7 @@ def test_create_table_with_local_index():
'TableSizeBytes': 0,
'TableName': 'messages',
'TableStatus': 'ACTIVE',
'TableArn': 'arn:aws:dynamodb:us-east-1:123456789011:table/messages',
'KeySchema': [
{'KeyType': 'HASH', 'AttributeName': 'forum_name'},
{'KeyType': 'RANGE', 'AttributeName': 'subject'}
@ -125,7 +127,7 @@ def test_create_table_with_local_index():
],
'ItemCount': 0,
'CreationDateTime': 1326499200.0,
'GlobalSecondaryIndexes': [],
'GlobalSecondaryIndexes': []
}
}
table.describe().should.equal(expected)

1
tests/test_dynamodb2/test_dynamodb_table_without_range_key.py
View file

@ -44,6 +44,7 @@ def test_create_table():
'TableSizeBytes': 0,
'TableName': 'messages',
'TableStatus': 'ACTIVE',
'TableArn': 'arn:aws:dynamodb:us-east-1:123456789011:table/messages',
'KeySchema': [
{'KeyType': 'HASH', 'AttributeName': 'forum_name'}
],

129
tests/test_ec2/test_amis.py
View file

@ -4,17 +4,18 @@ import tests.backport_assert_raises # noqa
from nose.tools import assert_raises
import boto
import boto3
import boto.ec2
import boto3
from boto.exception import EC2ResponseError, EC2ResponseError
import sure # noqa
from moto import mock_emr_deprecated, mock_ec2
from moto import mock_ec2_deprecated, mock_ec2
from tests.helpers import requires_boto_gte
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_create_and_delete():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
@ -75,7 +76,7 @@ def test_ami_create_and_delete():
@requires_boto_gte("2.14.0")
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_copy():
conn = boto.ec2.connect_to_region("us-west-1")
reservation = conn.run_instances('ami-1234abcd')
@ -134,7 +135,7 @@ def test_ami_copy():
cm.exception.request_id.should_not.be.none
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_tagging():
conn = boto.connect_vpc('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
@ -161,7 +162,7 @@ def test_ami_tagging():
image.tags["a key"].should.equal("some value")
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_create_from_missing_instance():
conn = boto.connect_ec2('the_key', 'the_secret')
args = ["i-abcdefg", "test-ami", "this is a test ami"]
@ -173,7 +174,7 @@ def test_ami_create_from_missing_instance():
cm.exception.request_id.should_not.be.none
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_pulls_attributes_from_instance():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
@ -185,7 +186,7 @@ def test_ami_pulls_attributes_from_instance():
image.kernel_id.should.equal('test-kernel')
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_filters():
conn = boto.connect_ec2('the_key', 'the_secret')
@ -242,7 +243,7 @@ def test_ami_filters():
set([ami.id for ami in amis_by_nonpublic]).should.equal(set([imageA.id]))
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_filtering_via_tag():
conn = boto.connect_vpc('the_key', 'the_secret')
@ -268,7 +269,7 @@ def test_ami_filtering_via_tag():
set([ami.id for ami in amis_by_tagB]).should.equal(set([imageB.id]))
@mock_emr_deprecated
@mock_ec2_deprecated
def test_getting_missing_ami():
conn = boto.connect_ec2('the_key', 'the_secret')
@ -279,7 +280,7 @@ def test_getting_missing_ami():
cm.exception.request_id.should_not.be.none
@mock_emr_deprecated
@mock_ec2_deprecated
def test_getting_malformed_ami():
conn = boto.connect_ec2('the_key', 'the_secret')
@ -290,7 +291,7 @@ def test_getting_malformed_ami():
cm.exception.request_id.should_not.be.none
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_attribute_group_permissions():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
@ -350,7 +351,7 @@ def test_ami_attribute_group_permissions():
**REMOVE_GROUP_ARGS).should_not.throw(EC2ResponseError)
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_attribute_user_permissions():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
@ -422,7 +423,107 @@ def test_ami_attribute_user_permissions():
**REMOVE_USERS_ARGS).should_not.throw(EC2ResponseError)
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_describe_executable_users():
conn = boto3.client('ec2', region_name='us-east-1')
ec2 = boto3.resource('ec2', 'us-east-1')
ec2.create_instances(ImageId='',
MinCount=1,
MaxCount=1)
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name','Values': ['running']}])
instance_id = response['Reservations'][0]['Instances'][0]['InstanceId']
image_id = conn.create_image(InstanceId=instance_id,
Name='TestImage',)['ImageId']
USER1 = '123456789011'
ADD_USER_ARGS = {'ImageId': image_id,
'Attribute': 'launchPermission',
'OperationType': 'add',
'UserIds': [USER1]}
# Add users and get no images
conn.modify_image_attribute(**ADD_USER_ARGS)
attributes = conn.describe_image_attribute(ImageId=image_id,
Attribute='LaunchPermissions',
DryRun=False)
attributes['LaunchPermissions'].should.have.length_of(1)
attributes['LaunchPermissions'][0]['UserId'].should.equal(USER1)
images = conn.describe_images(ExecutableUsers=[USER1])['Images']
images.should.have.length_of(1)
images[0]['ImageId'].should.equal(image_id)
@mock_ec2_deprecated
def test_ami_describe_executable_users_negative():
conn = boto3.client('ec2', region_name='us-east-1')
ec2 = boto3.resource('ec2', 'us-east-1')
ec2.create_instances(ImageId='',
MinCount=1,
MaxCount=1)
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name','Values': ['running']}])
instance_id = response['Reservations'][0]['Instances'][0]['InstanceId']
image_id = conn.create_image(InstanceId=instance_id,
Name='TestImage')['ImageId']
USER1 = '123456789011'
USER2 = '113355789012'
ADD_USER_ARGS = {'ImageId': image_id,
'Attribute': 'launchPermission',
'OperationType': 'add',
'UserIds': [USER1]}
# Add users and get no images
conn.modify_image_attribute(**ADD_USER_ARGS)
attributes = conn.describe_image_attribute(ImageId=image_id,
Attribute='LaunchPermissions',
DryRun=False)
attributes['LaunchPermissions'].should.have.length_of(1)
attributes['LaunchPermissions'][0]['UserId'].should.equal(USER1)
images = conn.describe_images(ExecutableUsers=[USER2])['Images']
images.should.have.length_of(0)
@mock_ec2_deprecated
def test_ami_describe_executable_users_and_filter():
conn = boto3.client('ec2', region_name='us-east-1')
ec2 = boto3.resource('ec2', 'us-east-1')
ec2.create_instances(ImageId='',
MinCount=1,
MaxCount=1)
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name','Values': ['running']}])
instance_id = response['Reservations'][0]['Instances'][0]['InstanceId']
image_id = conn.create_image(InstanceId=instance_id,
Name='ImageToDelete',)['ImageId']
USER1 = '123456789011'
ADD_USER_ARGS = {'ImageId': image_id,
'Attribute': 'launchPermission',
'OperationType': 'add',
'UserIds': [USER1]}
# Add users and get no images
conn.modify_image_attribute(**ADD_USER_ARGS)
attributes = conn.describe_image_attribute(ImageId=image_id,
Attribute='LaunchPermissions',
DryRun=False)
attributes['LaunchPermissions'].should.have.length_of(1)
attributes['LaunchPermissions'][0]['UserId'].should.equal(USER1)
images = conn.describe_images(ExecutableUsers=[USER1],
Filters=[{'Name': 'state', 'Values': ['available']}])['Images']
images.should.have.length_of(1)
images[0]['ImageId'].should.equal(image_id)
@mock_ec2_deprecated
def test_ami_attribute_user_and_group_permissions():
"""
Boto supports adding/removing both users and groups at the same time.
@ -477,7 +578,7 @@ def test_ami_attribute_user_and_group_permissions():
image.is_public.should.equal(False)
@mock_emr_deprecated
@mock_ec2_deprecated
def test_ami_attribute_error_cases():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')

5
tests/test_ec2/test_elastic_block_store.py
View file

@ -336,6 +336,11 @@ def test_snapshot_filters():
set([snap.id for snap in snapshots_by_volume_id]
).should.equal(set([snapshot1.id, snapshot2.id]))
snapshots_by_status = conn.get_all_snapshots(
filters={'status': 'completed'})
set([snap.id for snap in snapshots_by_status]
).should.equal(set([snapshot1.id, snapshot2.id, snapshot3.id]))
snapshots_by_volume_size = conn.get_all_snapshots(
filters={'volume-size': volume1.size})
set([snap.id for snap in snapshots_by_volume_size]

37
tests/test_ec2/test_instances.py
View file

@ -7,12 +7,13 @@ import base64
import datetime
import boto
import boto3
from boto.ec2.instance import Reservation, InstanceAttribute
from boto.exception import EC2ResponseError, EC2ResponseError
from freezegun import freeze_time
import sure # noqa
from moto import mock_ec2_deprecated
from moto import mock_ec2_deprecated, mock_ec2
from tests.helpers import requires_boto_gte
@ -157,6 +158,26 @@ def test_get_instances_by_id():
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_get_paginated_instances():
image_id = 'ami-1234abcd'
client = boto3.client('ec2', region_name='us-east-1')
conn = boto3.resource('ec2', 'us-east-1')
for i in range(100):
conn.create_instances(ImageId=image_id,
MinCount=1,
MaxCount=1)
resp = client.describe_instances(MaxResults=50)
reservations = resp['Reservations']
reservations.should.have.length_of(50)
next_token = resp['NextToken']
next_token.should_not.be.none
resp2 = client.describe_instances(NextToken=next_token)
reservations.extend(resp2['Reservations'])
reservations.should.have.length_of(100)
assert 'NextToken' not in resp2.keys()
@mock_ec2_deprecated
def test_get_instances_filtering_by_state():
conn = boto.connect_ec2()
@ -337,6 +358,20 @@ def test_get_instances_filtering_by_architecture():
reservations[0].instances.should.have.length_of(1)
@mock_ec2
def test_get_instances_filtering_by_image_id():
image_id = 'ami-1234abcd'
client = boto3.client('ec2', region_name='us-east-1')
conn = boto3.resource('ec2', 'us-east-1')
conn.create_instances(ImageId=image_id,
MinCount=1,
MaxCount=1)
reservations = client.describe_instances(Filters=[{'Name': 'image-id',
'Values': [image_id]}])['Reservations']
reservations[0]['Instances'].should.have.length_of(1)
@mock_ec2_deprecated
def test_get_instances_filtering_by_tag():
conn = boto.connect_ec2()

21
tests/test_elb/test_elb.py
View file

@ -109,6 +109,27 @@ def test_create_and_delete_boto3_support():
'LoadBalancerDescriptions']).should.have.length_of(0)
@mock_elb
def test_describe_paginated_balancers():
client = boto3.client('elb', region_name='us-east-1')
for i in range(51):
client.create_load_balancer(
LoadBalancerName='my-lb%d' % i,
Listeners=[
{'Protocol': 'tcp', 'LoadBalancerPort': 80, 'InstancePort': 8080}],
AvailabilityZones=['us-east-1a', 'us-east-1b']
)
resp = client.describe_load_balancers()
resp['LoadBalancerDescriptions'].should.have.length_of(50)
resp['NextMarker'].should.equal(resp['LoadBalancerDescriptions'][-1]['LoadBalancerName'])
resp2 = client.describe_load_balancers(Marker=resp['NextMarker'])
resp2['LoadBalancerDescriptions'].should.have.length_of(1)
assert 'NextToken' not in resp2.keys()
@mock_elb_deprecated
def test_add_listener():
conn = boto.connect_elb()

12
tests/test_emr/test_emr_boto3.py
View file

@ -127,6 +127,18 @@ def test_describe_cluster():
cl['VisibleToAllUsers'].should.equal(True)
@mock_emr
def test_describe_cluster_not_found():
conn = boto3.client('emr', region_name='us-east-1')
raised = False
try:
cluster = conn.describe_cluster(ClusterId='DummyId')
except ClientError as e:
if e.response['Error']['Code'] == "ResourceNotFoundException":
raised = True
raised.should.equal(True)
@mock_emr
def test_describe_job_flows():
client = boto3.client('emr', region_name='us-east-1')

22
tests/test_kinesis/test_kinesis.py
View file

@ -2,9 +2,10 @@ from __future__ import unicode_literals
import boto.kinesis
from boto.kinesis.exceptions import ResourceNotFoundException, InvalidArgumentException
import boto3
import sure # noqa
from moto import mock_kinesis_deprecated
from moto import mock_kinesis, mock_kinesis_deprecated
@mock_kinesis_deprecated
@ -51,6 +52,25 @@ def test_list_and_delete_stream():
"not-a-stream").should.throw(ResourceNotFoundException)
@mock_kinesis
def test_list_many_streams():
conn = boto3.client('kinesis', region_name="us-west-2")
for i in range(11):
conn.create_stream(StreamName="stream%d" % i, ShardCount=1)
resp = conn.list_streams()
stream_names = resp["StreamNames"]
has_more_streams = resp["HasMoreStreams"]
stream_names.should.have.length_of(10)
has_more_streams.should.be(True)
resp2 = conn.list_streams(ExclusiveStartStreamName=stream_names[-1])
stream_names = resp2["StreamNames"]
has_more_streams = resp2["HasMoreStreams"]
stream_names.should.have.length_of(1)
has_more_streams.should.equal(False)
@mock_kinesis_deprecated
def test_basic_shard_iterator():
conn = boto.kinesis.connect_to_region("us-west-2")

23
tests/test_rds/test_rds.py
View file

@ -1,11 +1,12 @@
from __future__ import unicode_literals
import boto3
import boto.rds
import boto.vpc
from boto.exception import BotoServerError
import sure # noqa
from moto import mock_ec2_deprecated, mock_rds_deprecated
from moto import mock_ec2_deprecated, mock_rds_deprecated, mock_rds
from tests.helpers import disable_on_py3
@ -45,6 +46,26 @@ def test_get_databases():
databases[0].id.should.equal("db-master-1")
@disable_on_py3()
@mock_rds
def test_get_databases_paginated():
conn = boto3.client('rds', region_name="us-west-2")
for i in range(51):
conn.create_db_instance(AllocatedStorage=5,
Port=5432,
DBInstanceIdentifier='rds%d' % i,
DBInstanceClass='db.t1.micro',
Engine='postgres')
resp = conn.describe_db_instances()
resp["DBInstances"].should.have.length_of(50)
resp["Marker"].should.equal(resp["DBInstances"][-1]['DBInstanceIdentifier'])
resp2 = conn.describe_db_instances(Marker=resp["Marker"])
resp2["DBInstances"].should.have.length_of(1)
@mock_rds_deprecated
def test_describe_non_existant_database():
conn = boto.rds.connect_to_region("us-west-2")

19
tests/test_rds2/test_rds2.py
View file

@ -65,6 +65,25 @@ def test_get_databases():
'arn:aws:rds:us-west-2:1234567890:db:db-master-1')
@disable_on_py3()
@mock_rds2
def test_get_databases_paginated():
conn = boto3.client('rds', region_name="us-west-2")
for i in range(51):
conn.create_db_instance(AllocatedStorage=5,
Port=5432,
DBInstanceIdentifier='rds%d' % i,
DBInstanceClass='db.t1.micro',
Engine='postgres')
resp = conn.describe_db_instances()
resp["DBInstances"].should.have.length_of(50)
resp["Marker"].should.equal(resp["DBInstances"][-1]['DBInstanceIdentifier'])
resp2 = conn.describe_db_instances(Marker=resp["Marker"])
resp2["DBInstances"].should.have.length_of(1)
@disable_on_py3()
@mock_rds2
def test_describe_non_existant_database():