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

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

Browse source 
Conflicts:
	moto/s3/models.py
	moto/s3/responses.py
This commit is contained in:
Konstantinos Koukopoulos 2013-09-30 11:02:25 +03:00
commit f25caa872d
113 changed files with 4360 additions and 1446 deletions
Download patch file Download diff file Expand all files Collapse all files

0
tests/__init__.py Normal file
View file

19
tests/helpers.py Normal file
View file

@ -0,0 +1,19 @@
import boto
from nose.plugins.skip import SkipTest
def version_tuple(v):
return tuple(map(int, (v.split("."))))
class requires_boto_gte(object):
"""Decorator for requiring boto version greater than or equal to 'version'"""
def __init__(self, version):
self.version = version
def __call__(self, test):
boto_version = version_tuple(boto.__version__)
required = version_tuple(self.version)
if boto_version >= required:
return test()
raise SkipTest

306
tests/test_autoscaling/test_autoscaling.py Normal file
View file

@ -0,0 +1,306 @@
import boto
from boto.ec2.autoscale.launchconfig import LaunchConfiguration
from boto.ec2.autoscale.group import AutoScalingGroup
import sure # noqa
from moto import mock_autoscaling, mock_ec2
from tests.helpers import requires_boto_gte
@mock_autoscaling
def test_create_autoscaling_group():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
availability_zones=['us-east-1c', 'us-east-1b'],
default_cooldown=60,
desired_capacity=2,
health_check_period=100,
health_check_type="EC2",
max_size=2,
min_size=2,
launch_config=config,
load_balancers=["test_lb"],
placement_group="test_placement",
vpc_zone_identifier='subnet-1234abcd',
termination_policies=["OldestInstance", "NewestInstance"],
)
conn.create_auto_scaling_group(group)
group = conn.get_all_groups()[0]
group.name.should.equal('tester_group')
set(group.availability_zones).should.equal(set(['us-east-1c', 'us-east-1b']))
group.desired_capacity.should.equal(2)
group.max_size.should.equal(2)
group.min_size.should.equal(2)
group.vpc_zone_identifier.should.equal('subnet-1234abcd')
group.launch_config_name.should.equal('tester')
group.default_cooldown.should.equal(60)
group.health_check_period.should.equal(100)
group.health_check_type.should.equal("EC2")
list(group.load_balancers).should.equal(["test_lb"])
group.placement_group.should.equal("test_placement")
list(group.termination_policies).should.equal(["OldestInstance", "NewestInstance"])
@mock_autoscaling
def test_create_autoscaling_groups_defaults():
""" Test with the minimum inputs and check that all of the proper defaults
are assigned for the other attributes """
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
max_size=2,
min_size=2,
launch_config=config,
)
conn.create_auto_scaling_group(group)
group = conn.get_all_groups()[0]
group.name.should.equal('tester_group')
group.max_size.should.equal(2)
group.min_size.should.equal(2)
group.launch_config_name.should.equal('tester')
# Defaults
list(group.availability_zones).should.equal([])
group.desired_capacity.should.equal(2)
group.vpc_zone_identifier.should.equal('')
group.default_cooldown.should.equal(300)
group.health_check_period.should.equal(None)
group.health_check_type.should.equal("EC2")
list(group.load_balancers).should.equal([])
group.placement_group.should.equal(None)
list(group.termination_policies).should.equal([])
@mock_autoscaling
def test_autoscaling_group_describe_filter():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
max_size=2,
min_size=2,
launch_config=config,
)
conn.create_auto_scaling_group(group)
group.name = 'tester_group2'
conn.create_auto_scaling_group(group)
group.name = 'tester_group3'
conn.create_auto_scaling_group(group)
conn.get_all_groups(names=['tester_group', 'tester_group2']).should.have.length_of(2)
conn.get_all_groups().should.have.length_of(3)
@mock_autoscaling
def test_autoscaling_update():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
availability_zones=['us-east-1c', 'us-east-1b'],
desired_capacity=2,
max_size=2,
min_size=2,
launch_config=config,
vpc_zone_identifier='subnet-1234abcd',
)
conn.create_auto_scaling_group(group)
group = conn.get_all_groups()[0]
group.vpc_zone_identifier.should.equal('subnet-1234abcd')
group.vpc_zone_identifier = 'subnet-5678efgh'
group.update()
group = conn.get_all_groups()[0]
group.vpc_zone_identifier.should.equal('subnet-5678efgh')
@mock_autoscaling
def test_autoscaling_group_delete():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
max_size=2,
min_size=2,
launch_config=config,
)
conn.create_auto_scaling_group(group)
conn.get_all_groups().should.have.length_of(1)
conn.delete_auto_scaling_group('tester_group')
conn.get_all_groups().should.have.length_of(0)
@mock_ec2
@mock_autoscaling
def test_autoscaling_group_describe_instances():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
max_size=2,
min_size=2,
launch_config=config,
)
conn.create_auto_scaling_group(group)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(2)
instances[0].launch_config_name.should.equal('tester')
autoscale_instance_ids = [instance.instance_id for instance in instances]
ec2_conn = boto.connect_ec2()
reservations = ec2_conn.get_all_instances()
instances = reservations[0].instances
instances.should.have.length_of(2)
instance_ids = [instance.id for instance in instances]
set(autoscale_instance_ids).should.equal(set(instance_ids))
@requires_boto_gte("2.8")
@mock_autoscaling
def test_set_desired_capacity_up():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
availability_zones=['us-east-1c', 'us-east-1b'],
desired_capacity=2,
max_size=2,
min_size=2,
launch_config=config,
vpc_zone_identifier='subnet-1234abcd',
)
conn.create_auto_scaling_group(group)
group = conn.get_all_groups()[0]
group.desired_capacity.should.equal(2)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(2)
conn.set_desired_capacity("tester_group", 3)
group = conn.get_all_groups()[0]
group.desired_capacity.should.equal(3)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(3)
@requires_boto_gte("2.8")
@mock_autoscaling
def test_set_desired_capacity_down():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
availability_zones=['us-east-1c', 'us-east-1b'],
desired_capacity=2,
max_size=2,
min_size=2,
launch_config=config,
vpc_zone_identifier='subnet-1234abcd',
)
conn.create_auto_scaling_group(group)
group = conn.get_all_groups()[0]
group.desired_capacity.should.equal(2)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(2)
conn.set_desired_capacity("tester_group", 1)
group = conn.get_all_groups()[0]
group.desired_capacity.should.equal(1)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(1)
@requires_boto_gte("2.8")
@mock_autoscaling
def test_set_desired_capacity_the_same():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
availability_zones=['us-east-1c', 'us-east-1b'],
desired_capacity=2,
max_size=2,
min_size=2,
launch_config=config,
vpc_zone_identifier='subnet-1234abcd',
)
conn.create_auto_scaling_group(group)
group = conn.get_all_groups()[0]
group.desired_capacity.should.equal(2)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(2)
conn.set_desired_capacity("tester_group", 2)
group = conn.get_all_groups()[0]
group.desired_capacity.should.equal(2)
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(2)

125
tests/test_autoscaling/test_launch_configurations.py Normal file
View file

@ -0,0 +1,125 @@
import boto
from boto.ec2.autoscale.launchconfig import LaunchConfiguration
import sure # noqa
from moto import mock_autoscaling
from tests.helpers import requires_boto_gte
@mock_autoscaling
def test_create_launch_configuration():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
key_name='the_keys',
security_groups=["default", "default2"],
user_data="This is some user_data",
instance_monitoring=True,
instance_profile_name='arn:aws:iam::123456789012:instance-profile/testing',
spot_price=0.1,
)
conn.create_launch_configuration(config)
launch_config = conn.get_all_launch_configurations()[0]
launch_config.name.should.equal('tester')
launch_config.image_id.should.equal('ami-abcd1234')
launch_config.instance_type.should.equal('m1.small')
launch_config.key_name.should.equal('the_keys')
set(launch_config.security_groups).should.equal(set(['default', 'default2']))
launch_config.user_data.should.equal("This is some user_data")
launch_config.instance_monitoring.enabled.should.equal('true')
launch_config.instance_profile_name.should.equal('arn:aws:iam::123456789012:instance-profile/testing')
launch_config.spot_price.should.equal(0.1)
@requires_boto_gte("2.12")
@mock_autoscaling
def test_create_launch_configuration_for_2_12():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
ebs_optimized=True,
)
conn.create_launch_configuration(config)
launch_config = conn.get_all_launch_configurations()[0]
launch_config.ebs_optimized.should.equal(True)
@mock_autoscaling
def test_create_launch_configuration_defaults():
""" Test with the minimum inputs and check that all of the proper defaults
are assigned for the other attributes """
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
launch_config = conn.get_all_launch_configurations()[0]
launch_config.name.should.equal('tester')
launch_config.image_id.should.equal('ami-abcd1234')
launch_config.instance_type.should.equal('m1.small')
# Defaults
launch_config.key_name.should.equal('')
list(launch_config.security_groups).should.equal([])
launch_config.user_data.should.equal("")
launch_config.instance_monitoring.enabled.should.equal('false')
launch_config.instance_profile_name.should.equal(None)
launch_config.spot_price.should.equal(None)
launch_config.ebs_optimized.should.equal(False)
@requires_boto_gte("2.12")
@mock_autoscaling
def test_create_launch_configuration_defaults_for_2_12():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
)
conn.create_launch_configuration(config)
launch_config = conn.get_all_launch_configurations()[0]
launch_config.ebs_optimized.should.equal(False)
@mock_autoscaling
def test_launch_configuration_describe_filter():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
config.name = 'tester2'
conn.create_launch_configuration(config)
config.name = 'tester3'
conn.create_launch_configuration(config)
conn.get_all_launch_configurations(names=['tester', 'tester2']).should.have.length_of(2)
conn.get_all_launch_configurations().should.have.length_of(3)
@mock_autoscaling
def test_launch_configuration_delete():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
conn.get_all_launch_configurations().should.have.length_of(1)
conn.delete_launch_configuration('tester')
conn.get_all_launch_configurations().should.have.length_of(0)

186
tests/test_autoscaling/test_policies.py Normal file
View file

@ -0,0 +1,186 @@
import boto
from boto.ec2.autoscale.launchconfig import LaunchConfiguration
from boto.ec2.autoscale.group import AutoScalingGroup
from boto.ec2.autoscale.policy import ScalingPolicy
import sure # noqa
from moto import mock_autoscaling
def setup_autoscale_group():
conn = boto.connect_autoscale()
config = LaunchConfiguration(
name='tester',
image_id='ami-abcd1234',
instance_type='m1.small',
)
conn.create_launch_configuration(config)
group = AutoScalingGroup(
name='tester_group',
max_size=2,
min_size=2,
launch_config=config,
)
conn.create_auto_scaling_group(group)
return group
@mock_autoscaling
def test_create_policy():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ExactCapacity',
as_name='tester_group',
scaling_adjustment=3,
cooldown=60,
)
conn.create_scaling_policy(policy)
policy = conn.get_all_policies()[0]
policy.name.should.equal('ScaleUp')
policy.adjustment_type.should.equal('ExactCapacity')
policy.as_name.should.equal('tester_group')
policy.scaling_adjustment.should.equal(3)
policy.cooldown.should.equal(60)
@mock_autoscaling
def test_create_policy_default_values():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ExactCapacity',
as_name='tester_group',
scaling_adjustment=3,
)
conn.create_scaling_policy(policy)
policy = conn.get_all_policies()[0]
policy.name.should.equal('ScaleUp')
# Defaults
policy.cooldown.should.equal(300)
@mock_autoscaling
def test_update_policy():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ExactCapacity',
as_name='tester_group',
scaling_adjustment=3,
)
conn.create_scaling_policy(policy)
policy = conn.get_all_policies()[0]
policy.scaling_adjustment.should.equal(3)
# Now update it by creating another with the same name
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ExactCapacity',
as_name='tester_group',
scaling_adjustment=2,
)
conn.create_scaling_policy(policy)
policy = conn.get_all_policies()[0]
policy.scaling_adjustment.should.equal(2)
@mock_autoscaling
def test_delete_policy():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ExactCapacity',
as_name='tester_group',
scaling_adjustment=3,
)
conn.create_scaling_policy(policy)
conn.get_all_policies().should.have.length_of(1)
conn.delete_policy('ScaleUp')
conn.get_all_policies().should.have.length_of(0)
@mock_autoscaling
def test_execute_policy_exact_capacity():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ExactCapacity',
as_name='tester_group',
scaling_adjustment=3,
)
conn.create_scaling_policy(policy)
conn.execute_policy("ScaleUp")
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(3)
@mock_autoscaling
def test_execute_policy_positive_change_in_capacity():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='ChangeInCapacity',
as_name='tester_group',
scaling_adjustment=3,
)
conn.create_scaling_policy(policy)
conn.execute_policy("ScaleUp")
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(5)
@mock_autoscaling
def test_execute_policy_percent_change_in_capacity():
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='PercentChangeInCapacity',
as_name='tester_group',
scaling_adjustment=50,
)
conn.create_scaling_policy(policy)
conn.execute_policy("ScaleUp")
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(3)
@mock_autoscaling
def test_execute_policy_small_percent_change_in_capacity():
""" http://docs.aws.amazon.com/AutoScaling/latest/DeveloperGuide/as-scale-based-on-demand.html
If PercentChangeInCapacity returns a value between 0 and 1,
Auto Scaling will round it off to 1."""
setup_autoscale_group()
conn = boto.connect_autoscale()
policy = ScalingPolicy(
name='ScaleUp',
adjustment_type='PercentChangeInCapacity',
as_name='tester_group',
scaling_adjustment=1,
)
conn.create_scaling_policy(policy)
conn.execute_policy("ScaleUp")
instances = list(conn.get_all_autoscaling_instances())
instances.should.have.length_of(3)

16
tests/test_autoscaling/test_server.py Normal file
View file

@ -0,0 +1,16 @@
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("autoscaling")
def test_describe_autoscaling_groups():
test_client = server.app.test_client()
res = test_client.get('/?Action=DescribeLaunchConfigurations')
res.data.should.contain('<DescribeLaunchConfigurationsResponse')
res.data.should.contain('<LaunchConfigurations>')

12
tests/test_core/test_decorator_calls.py
View file

@ -1,6 +1,6 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2
@ -11,7 +11,7 @@ Test the different ways that the decorator can be used
@mock_ec2
def test_basic_connect():
conn = boto.connect_ec2()
boto.connect_ec2()
@mock_ec2
@ -42,3 +42,11 @@ def test_decorator_start_and_stop():
mock.stop()
conn.get_all_instances.when.called_with().should.throw(EC2ResponseError)
@mock_ec2
def test_decorater_wrapped_gets_set():
"""
Moto decorator's __wrapped__ should get set to the tests function
"""
test_decorater_wrapped_gets_set.__wrapped__.__name__.should.equal('test_decorater_wrapped_gets_set')

36
tests/test_core/test_instance_metadata.py Normal file
View file

@ -0,0 +1,36 @@
import requests
from moto import mock_ec2
@mock_ec2
def test_latest_meta_data():
res = requests.get("http://169.254.169.254/latest/meta-data/")
res.content.should.equal("iam")
@mock_ec2
def test_meta_data_iam():
res = requests.get("http://169.254.169.254/latest/meta-data/iam")
json_response = res.json()
default_role = json_response['security-credentials']['default-role']
default_role.should.contain('AccessKeyId')
default_role.should.contain('SecretAccessKey')
default_role.should.contain('Token')
default_role.should.contain('Expiration')
@mock_ec2
def test_meta_data_security_credentials():
res = requests.get("http://169.254.169.254/latest/meta-data/iam/security-credentials/")
res.content.should.equal("default-role")
@mock_ec2
def test_meta_data_default_role():
res = requests.get("http://169.254.169.254/latest/meta-data/iam/security-credentials/default-role")
json_response = res.json()
json_response.should.contain('AccessKeyId')
json_response.should.contain('SecretAccessKey')
json_response.should.contain('Token')
json_response.should.contain('Expiration')

14
tests/test_core/test_server.py
View file

@ -1,12 +1,12 @@
from mock import patch
import sure # flake8: noqa
import sure # noqa
from moto.server import main
def test_wrong_arguments():
try:
main(["name", "test1", "test2"])
main(["name", "test1", "test2", "test3"])
assert False, ("main() when called with the incorrect number of args"
" should raise a system exit")
except SystemExit:
@ -15,5 +15,11 @@ def test_wrong_arguments():
@patch('moto.server.app.run')
def test_right_arguments(app_run):
main(["name", "s3"])
app_run.assert_called_once_with()
main(["s3"])
app_run.assert_called_once_with(host='0.0.0.0', port=5000)
@patch('moto.server.app.run')
def test_port_argument(app_run):
main(["s3", "--port", "8080"])
app_run.assert_called_once_with(host='0.0.0.0', port=8080)

2
tests/test_core/test_url_mapping.py
View file

@ -1,4 +1,4 @@
import sure # flake8: noqa
import sure # noqa
from moto.core.utils import convert_regex_to_flask_path

13
tests/test_dynamodb/test_dynamodb.py
View file

@ -1,12 +1,10 @@
import boto
import sure # flake8: noqa
from freezegun import freeze_time
import sure # noqa
import requests
from moto import mock_dynamodb
from moto.dynamodb import dynamodb_backend
from boto.dynamodb import condition
from boto.exception import DynamoDBResponseError
@ -43,3 +41,12 @@ def test_sts_handler():
res = requests.post("https://sts.amazonaws.com/", data={"GetSessionToken": ""})
res.ok.should.be.ok
res.text.should.contain("SecretAccessKey")
@mock_dynamodb
def test_dynamodb_with_connect_to_region():
# this will work if connected with boto.connect_dynamodb()
dynamodb = boto.dynamodb.connect_to_region('us-west-2')
schema = dynamodb.create_schema('column1', str(), 'column2', int())
dynamodb.create_table('table1', schema, 200, 200)

38
tests/test_dynamodb/test_dynamodb_table_with_range_key.py
View file

@ -1,11 +1,11 @@
import boto
import sure # flake8: noqa
import sure # noqa
from freezegun import freeze_time
from moto import mock_dynamodb
from moto.dynamodb import dynamodb_backend
from boto.dynamodb import condition
from boto.dynamodb.exceptions import DynamoDBKeyNotFoundError, DynamoDBValidationError
from boto.exception import DynamoDBResponseError
@ -101,6 +101,8 @@ def test_item_add_and_describe_and_update():
)
item.put()
table.has_item("LOLCat Forum", "Check this out!").should.equal(True)
returned_item = table.get_item(
hash_key='LOLCat Forum',
range_key='Check this out!',
@ -150,7 +152,8 @@ def test_get_missing_item():
table.get_item.when.called_with(
hash_key='tester',
range_key='other',
).should.throw(DynamoDBResponseError)
).should.throw(DynamoDBKeyNotFoundError)
table.has_item("foobar", "more").should.equal(False)
@mock_dynamodb
@ -163,7 +166,31 @@ def test_get_item_with_undeclared_table():
'HashKeyElement': {'S': 'tester'},
'RangeKeyElement': {'S': 'test-range'},
},
).should.throw(DynamoDBResponseError)
).should.throw(DynamoDBKeyNotFoundError)
@mock_dynamodb
def test_get_item_without_range_key():
conn = boto.connect_dynamodb()
message_table_schema = conn.create_schema(
hash_key_name="test_hash",
hash_key_proto_value=int,
range_key_name="test_range",
range_key_proto_value=int,
)
table = conn.create_table(
name='messages',
schema=message_table_schema,
read_units=10,
write_units=10
)
hash_key = 3241526475
range_key = 1234567890987
new_item = table.new_item(hash_key=hash_key, range_key=range_key)
new_item.put()
table.get_item.when.called_with(hash_key=hash_key).should.throw(DynamoDBValidationError)
@mock_dynamodb
@ -473,5 +500,6 @@ def test_batch_read():
item.put()
items = table.batch_get_item([('the-key', '123'), ('another-key', '789')])
count = len([item for item in items])
# Iterate through so that batch_item gets called
count = len([x for x in items])
count.should.equal(2)

13
tests/test_dynamodb/test_dynamodb_table_without_range_key.py
View file

@ -1,11 +1,11 @@
import boto
import sure # flake8: noqa
import sure # noqa
from freezegun import freeze_time
from moto import mock_dynamodb
from moto.dynamodb import dynamodb_backend
from boto.dynamodb import condition
from boto.dynamodb.exceptions import DynamoDBKeyNotFoundError
from boto.exception import DynamoDBResponseError
@ -137,7 +137,7 @@ def test_get_missing_item():
table.get_item.when.called_with(
hash_key='tester',
).should.throw(DynamoDBResponseError)
).should.throw(DynamoDBKeyNotFoundError)
@mock_dynamodb
@ -149,7 +149,7 @@ def test_get_item_with_undeclared_table():
key={
'HashKeyElement': {'S': 'tester'},
},
).should.throw(DynamoDBResponseError)
).should.throw(DynamoDBKeyNotFoundError)
@mock_dynamodb
@ -411,5 +411,6 @@ def test_batch_read():
item.put()
items = table.batch_get_item([('the-key1'), ('another-key')])
count = len([item for item in items])
count.should.equal(2)
# Iterate through so that batch_item gets called
count = len([x for x in items])
count.should.have.equal(2)

2
tests/test_dynamodb/test_server.py
View file

@ -1,4 +1,4 @@
import sure # flake8: noqa
import sure # noqa
import moto.server as server

2
tests/test_ec2/test_amazon_dev_pay.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_amis.py
View file

@ -1,7 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_availability_zones_and_regions.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_customer_gateways.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_dhcp_options.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_elastic_block_store.py
View file

@ -1,6 +1,6 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_elastic_ip_addresses.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_elastic_network_interfaces.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_general.py
View file

@ -1,6 +1,6 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

77
tests/test_ec2/test_instances.py
View file

@ -1,13 +1,16 @@
import base64
import boto
from boto.ec2.instance import Reservation, InstanceAttribute
import sure # flake8: noqa
from boto.exception import EC2ResponseError
import sure # noqa
from moto import mock_ec2
################ Test Readme ###############
def add_servers(ami_id, count):
conn = boto.connect_ec2('the_key', 'the_secret')
conn = boto.connect_ec2()
for index in range(count):
conn.run_instances(ami_id)
@ -16,7 +19,7 @@ def add_servers(ami_id, count):
def test_add_servers():
add_servers('ami-1234abcd', 2)
conn = boto.connect_ec2('the_key', 'the_secret')
conn = boto.connect_ec2()
reservations = conn.get_all_instances()
assert len(reservations) == 2
instance1 = reservations[0].instances[0]
@ -48,6 +51,61 @@ def test_instance_launch_and_terminate():
instance.state.should.equal('shutting-down')
@mock_ec2
def test_get_instances_by_id():
conn = boto.connect_ec2()
reservation = conn.run_instances('ami-1234abcd', min_count=2)
instance1, instance2 = reservation.instances
reservations = conn.get_all_instances(instance_ids=[instance1.id])
reservations.should.have.length_of(1)
reservation = reservations[0]
reservation.instances.should.have.length_of(1)
reservation.instances[0].id.should.equal(instance1.id)
reservations = conn.get_all_instances(instance_ids=[instance1.id, instance2.id])
reservations.should.have.length_of(1)
reservation = reservations[0]
reservation.instances.should.have.length_of(2)
instance_ids = [instance.id for instance in reservation.instances]
instance_ids.should.equal([instance1.id, instance2.id])
# Call get_all_instances with a bad id should raise an error
conn.get_all_instances.when.called_with(instance_ids=[instance1.id, "i-1234abcd"]).should.throw(
EC2ResponseError,
"The instance ID 'i-1234abcd' does not exist"
)
@mock_ec2
def test_get_instances_filtering_by_state():
conn = boto.connect_ec2()
reservation = conn.run_instances('ami-1234abcd', min_count=3)
instance1, instance2, instance3 = reservation.instances
conn.terminate_instances([instance1.id])
reservations = conn.get_all_instances(filters={'instance-state-name': 'pending'})
reservations.should.have.length_of(1)
# Since we terminated instance1, only instance2 and instance3 should be returned
instance_ids = [instance.id for instance in reservations[0].instances]
set(instance_ids).should.equal(set([instance2.id, instance3.id]))
reservations = conn.get_all_instances([instance2.id], filters={'instance-state-name': 'pending'})
reservations.should.have.length_of(1)
instance_ids = [instance.id for instance in reservations[0].instances]
instance_ids.should.equal([instance2.id])
reservations = conn.get_all_instances([instance2.id], filters={'instance-state-name': 'terminating'})
list(reservations).should.equal([])
# get_all_instances should still return all 3
reservations = conn.get_all_instances()
reservations[0].instances.should.have.length_of(3)
conn.get_all_instances.when.called_with(filters={'not-implemented-filter': 'foobar'}).should.throw(NotImplementedError)
@mock_ec2
def test_instance_start_and_stop():
conn = boto.connect_ec2('the_key', 'the_secret')
@ -98,3 +156,16 @@ def test_instance_attribute_user_data():
instance_attribute = instance.get_attribute("userData")
instance_attribute.should.be.a(InstanceAttribute)
instance_attribute.get("userData").should.equal("this is my user data")
@mock_ec2
def test_user_data_with_run_instance():
user_data = "some user data"
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd', user_data=user_data)
instance = reservation.instances[0]
instance_attribute = instance.get_attribute("userData")
instance_attribute.should.be.a(InstanceAttribute)
decoded_user_data = base64.decodestring(instance_attribute.get("userData"))
decoded_user_data.should.equal("some user data")

2
tests/test_ec2/test_internet_gateways.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_ip_addresses.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_key_pairs.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_monitoring.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_network_acls.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_placement_groups.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_reserved_instances.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_route_tables.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_security_groups.py
View file

@ -1,6 +1,6 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_server.py
View file

@ -1,5 +1,5 @@
import re
import sure # flake8: noqa
import sure # noqa
import moto.server as server

96
tests/test_ec2/test_spot_instances.py
View file

@ -1,9 +1,99 @@
import datetime
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2
from moto.core.utils import iso_8601_datetime
@mock_ec2
def test_spot_instances():
pass
def test_request_spot_instances():
conn = boto.connect_ec2()
conn.create_security_group('group1', 'description')
conn.create_security_group('group2', 'description')
start = iso_8601_datetime(datetime.datetime(2013, 1, 1))
end = iso_8601_datetime(datetime.datetime(2013, 1, 2))
request = conn.request_spot_instances(
price=0.5, image_id='ami-abcd1234', count=1, type='one-time',
valid_from=start, valid_until=end, launch_group="the-group",
availability_zone_group='my-group', key_name="test",
security_groups=['group1', 'group2'], user_data="some test data",
instance_type='m1.small', placement='us-east-1c',
kernel_id="test-kernel", ramdisk_id="test-ramdisk",
monitoring_enabled=True, subnet_id="subnet123",
)
requests = conn.get_all_spot_instance_requests()
requests.should.have.length_of(1)
request = requests[0]
request.state.should.equal("open")
request.price.should.equal(0.5)
request.launch_specification.image_id.should.equal('ami-abcd1234')
request.type.should.equal('one-time')
request.valid_from.should.equal(start)
request.valid_until.should.equal(end)
request.launch_group.should.equal("the-group")
request.availability_zone_group.should.equal('my-group')
request.launch_specification.key_name.should.equal("test")
security_group_names = [group.name for group in request.launch_specification.groups]
set(security_group_names).should.equal(set(['group1', 'group2']))
request.launch_specification.instance_type.should.equal('m1.small')
request.launch_specification.placement.should.equal('us-east-1c')
request.launch_specification.kernel.should.equal("test-kernel")
request.launch_specification.ramdisk.should.equal("test-ramdisk")
request.launch_specification.subnet_id.should.equal("subnet123")
@mock_ec2
def test_request_spot_instances_default_arguments():
"""
Test that moto set the correct default arguments
"""
conn = boto.connect_ec2()
request = conn.request_spot_instances(
price=0.5, image_id='ami-abcd1234',
)
requests = conn.get_all_spot_instance_requests()
requests.should.have.length_of(1)
request = requests[0]
request.state.should.equal("open")
request.price.should.equal(0.5)
request.launch_specification.image_id.should.equal('ami-abcd1234')
request.type.should.equal('one-time')
request.valid_from.should.equal(None)
request.valid_until.should.equal(None)
request.launch_group.should.equal(None)
request.availability_zone_group.should.equal(None)
request.launch_specification.key_name.should.equal(None)
security_group_names = [group.name for group in request.launch_specification.groups]
security_group_names.should.equal(["default"])
request.launch_specification.instance_type.should.equal('m1.small')
request.launch_specification.placement.should.equal(None)
request.launch_specification.kernel.should.equal(None)
request.launch_specification.ramdisk.should.equal(None)
request.launch_specification.subnet_id.should.equal(None)
@mock_ec2
def test_cancel_spot_instance_request():
conn = boto.connect_ec2()
conn.request_spot_instances(
price=0.5, image_id='ami-abcd1234',
)
requests = conn.get_all_spot_instance_requests()
requests.should.have.length_of(1)
conn.cancel_spot_instance_requests([requests[0].id])
requests = conn.get_all_spot_instance_requests()
requests.should.have.length_of(0)

2
tests/test_ec2/test_subnets.py
View file

@ -1,6 +1,6 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

18
tests/test_ec2/test_tags.py
View file

@ -1,5 +1,7 @@
import itertools
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2
@ -19,3 +21,17 @@ def test_instance_launch_and_terminate():
instance.remove_tag("a key")
conn.get_all_tags().should.have.length_of(0)
@mock_ec2
def test_instance_launch_and_retrieve_all_instances():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
instance = reservation.instances[0]
instance.add_tag("a key", "some value")
chain = itertools.chain.from_iterable
existing_instances = list(chain([res.instances for res in conn.get_all_instances()]))
existing_instances.should.have.length_of(1)
existing_instance = existing_instances[0]
existing_instance.tags["a key"].should.equal("some value")

2
tests/test_ec2/test_virtual_private_gateways.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_vm_export.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_vm_import.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_vpcs.py
View file

@ -1,6 +1,6 @@
import boto
from boto.exception import EC2ResponseError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_vpn_connections.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

2
tests/test_ec2/test_windows.py
View file

@ -1,5 +1,5 @@
import boto
import sure # flake8: noqa
import sure # noqa
from moto import mock_ec2

124
tests/test_elb/test_elb.py Normal file
View file

@ -0,0 +1,124 @@
import boto
from boto.ec2.elb import HealthCheck
import sure # noqa
from moto import mock_elb, mock_ec2
@mock_elb
def test_create_load_balancer():
conn = boto.connect_elb()
zones = ['us-east-1a', 'us-east-1b']
ports = [(80, 8080, 'http'), (443, 8443, 'tcp')]
conn.create_load_balancer('my-lb', zones, ports)
balancers = conn.get_all_load_balancers()
balancer = balancers[0]
balancer.name.should.equal("my-lb")
set(balancer.availability_zones).should.equal(set(['us-east-1a', 'us-east-1b']))
listener1 = balancer.listeners[0]
listener1.load_balancer_port.should.equal(80)
listener1.instance_port.should.equal(8080)
listener1.protocol.should.equal("HTTP")
listener2 = balancer.listeners[1]
listener2.load_balancer_port.should.equal(443)
listener2.instance_port.should.equal(8443)
listener2.protocol.should.equal("TCP")
@mock_elb
def test_get_load_balancers_by_name():
conn = boto.connect_elb()
zones = ['us-east-1a', 'us-east-1b']
ports = [(80, 8080, 'http'), (443, 8443, 'tcp')]
conn.create_load_balancer('my-lb1', zones, ports)
conn.create_load_balancer('my-lb2', zones, ports)
conn.create_load_balancer('my-lb3', zones, ports)
conn.get_all_load_balancers().should.have.length_of(3)
conn.get_all_load_balancers(load_balancer_names=['my-lb1']).should.have.length_of(1)
conn.get_all_load_balancers(load_balancer_names=['my-lb1', 'my-lb2']).should.have.length_of(2)
@mock_elb
def test_delete_load_balancer():
conn = boto.connect_elb()
zones = ['us-east-1a']
ports = [(80, 8080, 'http'), (443, 8443, 'tcp')]
conn.create_load_balancer('my-lb', zones, ports)
balancers = conn.get_all_load_balancers()
balancers.should.have.length_of(1)
conn.delete_load_balancer("my-lb")
balancers = conn.get_all_load_balancers()
balancers.should.have.length_of(0)
@mock_elb
def test_create_health_check():
conn = boto.connect_elb()
hc = HealthCheck(
interval=20,
healthy_threshold=3,
unhealthy_threshold=5,
target='HTTP:8080/health',
timeout=23,
)
ports = [(80, 8080, 'http'), (443, 8443, 'tcp')]
lb = conn.create_load_balancer('my-lb', [], ports)
lb.configure_health_check(hc)
balancer = conn.get_all_load_balancers()[0]
health_check = balancer.health_check
health_check.interval.should.equal(20)
health_check.healthy_threshold.should.equal(3)
health_check.unhealthy_threshold.should.equal(5)
health_check.target.should.equal('HTTP:8080/health')
health_check.timeout.should.equal(23)
@mock_ec2
@mock_elb
def test_register_instances():
ec2_conn = boto.connect_ec2()
reservation = ec2_conn.run_instances('ami-1234abcd', 2)
instance_id1 = reservation.instances[0].id
instance_id2 = reservation.instances[1].id
conn = boto.connect_elb()
ports = [(80, 8080, 'http'), (443, 8443, 'tcp')]
lb = conn.create_load_balancer('my-lb', [], ports)
lb.register_instances([instance_id1, instance_id2])
balancer = conn.get_all_load_balancers()[0]
instance_ids = [instance.id for instance in balancer.instances]
set(instance_ids).should.equal(set([instance_id1, instance_id2]))
@mock_ec2
@mock_elb
def test_deregister_instances():
ec2_conn = boto.connect_ec2()
reservation = ec2_conn.run_instances('ami-1234abcd', 2)
instance_id1 = reservation.instances[0].id
instance_id2 = reservation.instances[1].id
conn = boto.connect_elb()
ports = [(80, 8080, 'http'), (443, 8443, 'tcp')]
lb = conn.create_load_balancer('my-lb', [], ports)
lb.register_instances([instance_id1, instance_id2])
balancer = conn.get_all_load_balancers()[0]
balancer.instances.should.have.length_of(2)
balancer.deregister_instances([instance_id1])
balancer.instances.should.have.length_of(1)
balancer.instances[0].id.should.equal(instance_id2)

15
tests/test_elb/test_server.py Normal file
View file

@ -0,0 +1,15 @@
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("elb")
def test_elb_describe_instances():
test_client = server.app.test_client()
res = test_client.get('/?Action=DescribeLoadBalancers')
res.data.should.contain('DescribeLoadBalancersResponse')

292
tests/test_emr/test_emr.py Normal file
View file

@ -0,0 +1,292 @@
import boto
from boto.emr.instance_group import InstanceGroup
from boto.emr.step import StreamingStep
import sure # noqa
from moto import mock_emr
from tests.helpers import requires_boto_gte
@mock_emr
def test_create_job_flow():
conn = boto.connect_emr()
step1 = StreamingStep(
name='My wordcount example',
mapper='s3n://elasticmapreduce/samples/wordcount/wordSplitter.py',
reducer='aggregate',
input='s3n://elasticmapreduce/samples/wordcount/input',
output='s3n://output_bucket/output/wordcount_output'
)
step2 = StreamingStep(
name='My wordcount example2',
mapper='s3n://elasticmapreduce/samples/wordcount/wordSplitter2.py',
reducer='aggregate',
input='s3n://elasticmapreduce/samples/wordcount/input2',
output='s3n://output_bucket/output/wordcount_output2'
)
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
master_instance_type='m1.medium',
slave_instance_type='m1.small',
steps=[step1, step2],
)
job_flow = conn.describe_jobflow(job_id)
job_flow.state.should.equal('STARTING')
job_flow.jobflowid.should.equal(job_id)
job_flow.name.should.equal('My jobflow')
job_flow.masterinstancetype.should.equal('m1.medium')
job_flow.slaveinstancetype.should.equal('m1.small')
job_flow.loguri.should.equal('s3://some_bucket/jobflow_logs')
job_flow.visibletoallusers.should.equal('False')
int(job_flow.normalizedinstancehours).should.equal(0)
job_step = job_flow.steps[0]
job_step.name.should.equal('My wordcount example')
job_step.state.should.equal('STARTING')
args = [arg.value for arg in job_step.args]
args.should.equal([
'-mapper',
's3n://elasticmapreduce/samples/wordcount/wordSplitter.py',
'-reducer',
'aggregate',
'-input',
's3n://elasticmapreduce/samples/wordcount/input',
'-output',
's3n://output_bucket/output/wordcount_output',
])
job_step2 = job_flow.steps[1]
job_step2.name.should.equal('My wordcount example2')
job_step2.state.should.equal('PENDING')
args = [arg.value for arg in job_step2.args]
args.should.equal([
'-mapper',
's3n://elasticmapreduce/samples/wordcount/wordSplitter2.py',
'-reducer',
'aggregate',
'-input',
's3n://elasticmapreduce/samples/wordcount/input2',
'-output',
's3n://output_bucket/output/wordcount_output2',
])
@requires_boto_gte("2.8")
@mock_emr
def test_create_job_flow_with_new_params():
# Test that run_jobflow works with newer params
conn = boto.connect_emr()
conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
master_instance_type='m1.medium',
slave_instance_type='m1.small',
job_flow_role='some-role-arn',
steps=[],
)
@mock_emr
def test_create_job_flow_visible_to_all_users():
conn = boto.connect_emr()
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
job_flow_role='some-role-arn',
steps=[],
visible_to_all_users=True,
)
job_flow = conn.describe_jobflow(job_id)
job_flow.visibletoallusers.should.equal('True')
@mock_emr
def test_terminate_job_flow():
conn = boto.connect_emr()
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
steps=[]
)
flow = conn.describe_jobflows()[0]
flow.state.should.equal('STARTING')
conn.terminate_jobflow(job_id)
flow = conn.describe_jobflows()[0]
flow.state.should.equal('TERMINATED')
@mock_emr
def test_add_steps_to_flow():
conn = boto.connect_emr()
step1 = StreamingStep(
name='My wordcount example',
mapper='s3n://elasticmapreduce/samples/wordcount/wordSplitter.py',
reducer='aggregate',
input='s3n://elasticmapreduce/samples/wordcount/input',
output='s3n://output_bucket/output/wordcount_output'
)
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
steps=[step1]
)
job_flow = conn.describe_jobflow(job_id)
job_flow.state.should.equal('STARTING')
job_flow.jobflowid.should.equal(job_id)
job_flow.name.should.equal('My jobflow')
job_flow.loguri.should.equal('s3://some_bucket/jobflow_logs')
step2 = StreamingStep(
name='My wordcount example2',
mapper='s3n://elasticmapreduce/samples/wordcount/wordSplitter2.py',
reducer='aggregate',
input='s3n://elasticmapreduce/samples/wordcount/input2',
output='s3n://output_bucket/output/wordcount_output2'
)
conn.add_jobflow_steps(job_id, [step2])
job_flow = conn.describe_jobflow(job_id)
job_step = job_flow.steps[0]
job_step.name.should.equal('My wordcount example')
job_step.state.should.equal('STARTING')
args = [arg.value for arg in job_step.args]
args.should.equal([
'-mapper',
's3n://elasticmapreduce/samples/wordcount/wordSplitter.py',
'-reducer',
'aggregate',
'-input',
's3n://elasticmapreduce/samples/wordcount/input',
'-output',
's3n://output_bucket/output/wordcount_output',
])
job_step2 = job_flow.steps[1]
job_step2.name.should.equal('My wordcount example2')
job_step2.state.should.equal('PENDING')
args = [arg.value for arg in job_step2.args]
args.should.equal([
'-mapper',
's3n://elasticmapreduce/samples/wordcount/wordSplitter2.py',
'-reducer',
'aggregate',
'-input',
's3n://elasticmapreduce/samples/wordcount/input2',
'-output',
's3n://output_bucket/output/wordcount_output2',
])
@mock_emr
def test_create_instance_groups():
conn = boto.connect_emr()
step1 = StreamingStep(
name='My wordcount example',
mapper='s3n://elasticmapreduce/samples/wordcount/wordSplitter.py',
reducer='aggregate',
input='s3n://elasticmapreduce/samples/wordcount/input',
output='s3n://output_bucket/output/wordcount_output'
)
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
steps=[step1],
)
instance_group = InstanceGroup(6, 'TASK', 'c1.medium', 'SPOT', 'spot-0.07', '0.07')
instance_group = conn.add_instance_groups(job_id, [instance_group])
instance_group_id = instance_group.instancegroupids
job_flow = conn.describe_jobflows()[0]
int(job_flow.instancecount).should.equal(6)
instance_group = job_flow.instancegroups[0]
instance_group.instancegroupid.should.equal(instance_group_id)
int(instance_group.instancerunningcount).should.equal(6)
instance_group.instancerole.should.equal('TASK')
instance_group.instancetype.should.equal('c1.medium')
instance_group.market.should.equal('SPOT')
instance_group.name.should.equal('spot-0.07')
instance_group.bidprice.should.equal('0.07')
@mock_emr
def test_modify_instance_groups():
conn = boto.connect_emr()
step1 = StreamingStep(
name='My wordcount example',
mapper='s3n://elasticmapreduce/samples/wordcount/wordSplitter.py',
reducer='aggregate',
input='s3n://elasticmapreduce/samples/wordcount/input',
output='s3n://output_bucket/output/wordcount_output'
)
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
steps=[step1]
)
instance_group1 = InstanceGroup(6, 'TASK', 'c1.medium', 'SPOT', 'spot-0.07', '0.07')
instance_group2 = InstanceGroup(6, 'TASK', 'c1.medium', 'SPOT', 'spot-0.07', '0.07')
instance_group = conn.add_instance_groups(job_id, [instance_group1, instance_group2])
instance_group_ids = instance_group.instancegroupids.split(",")
job_flow = conn.describe_jobflows()[0]
int(job_flow.instancecount).should.equal(12)
instance_group = job_flow.instancegroups[0]
int(instance_group.instancerunningcount).should.equal(6)
conn.modify_instance_groups(instance_group_ids, [2, 3])
job_flow = conn.describe_jobflows()[0]
int(job_flow.instancecount).should.equal(5)
instance_group1 = [
group for group
in job_flow.instancegroups
if group.instancegroupid == instance_group_ids[0]
][0]
int(instance_group1.instancerunningcount).should.equal(2)
instance_group2 = [
group for group
in job_flow.instancegroups
if group.instancegroupid == instance_group_ids[1]
][0]
int(instance_group2.instancerunningcount).should.equal(3)
@mock_emr
def test_set_visible_to_all_users():
conn = boto.connect_emr()
job_id = conn.run_jobflow(
name='My jobflow',
log_uri='s3://some_bucket/jobflow_logs',
job_flow_role='some-role-arn',
steps=[],
visible_to_all_users=False,
)
job_flow = conn.describe_jobflow(job_id)
job_flow.visibletoallusers.should.equal('False')
conn.set_visible_to_all_users(job_id, True)
job_flow = conn.describe_jobflow(job_id)
job_flow.visibletoallusers.should.equal('True')
conn.set_visible_to_all_users(job_id, False)
job_flow = conn.describe_jobflow(job_id)
job_flow.visibletoallusers.should.equal('False')

16
tests/test_emr/test_server.py Normal file
View file

@ -0,0 +1,16 @@
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("emr")
def test_describe_jobflows():
test_client = server.app.test_client()
res = test_client.get('/?Action=DescribeJobFlows')
res.data.should.contain('<DescribeJobFlowsResult>')
res.data.should.contain('<JobFlows>')

165
tests/test_s3/test_s3.py
View file

@ -4,9 +4,10 @@ from io import BytesIO
import boto
from boto.exception import S3ResponseError
from boto.s3.key import Key
from freezegun import freeze_time
import requests
import sure # flake8: noqa
import sure # noqa
from moto import mock_s3
@ -83,6 +84,31 @@ def test_empty_key():
bucket.get_key("the-key").get_contents_as_string().should.equal('')
@mock_s3
def test_empty_key_set_on_existing_key():
conn = boto.connect_s3('the_key', 'the_secret')
bucket = conn.create_bucket("foobar")
key = Key(bucket)
key.key = "the-key"
key.set_contents_from_string("foobar")
bucket.get_key("the-key").get_contents_as_string().should.equal('foobar')
key.set_contents_from_string("")
bucket.get_key("the-key").get_contents_as_string().should.equal('')
@mock_s3
def test_large_key_save():
conn = boto.connect_s3('the_key', 'the_secret')
bucket = conn.create_bucket("foobar")
key = Key(bucket)
key.key = "the-key"
key.set_contents_from_string("foobar" * 100000)
bucket.get_key("the-key").get_contents_as_string().should.equal('foobar' * 100000)
@mock_s3
def test_copy_key():
conn = boto.connect_s3('the_key', 'the_secret')
@ -98,39 +124,31 @@ def test_copy_key():
@mock_s3
def test_get_all_keys():
def test_set_metadata():
conn = boto.connect_s3('the_key', 'the_secret')
bucket = conn.create_bucket("foobar")
key = Key(bucket)
key.key = 'the-key'
key.set_metadata('md', 'Metadatastring')
key.set_contents_from_string("Testval")
bucket.get_key('the-key').get_metadata('md').should.equal('Metadatastring')
@freeze_time("2012-01-01 12:00:00")
@mock_s3
def test_last_modified():
# See https://github.com/boto/boto/issues/466
conn = boto.connect_s3()
bucket = conn.create_bucket("foobar")
key = Key(bucket)
key.key = "the-key"
key.set_contents_from_string("some value")
key2 = Key(bucket)
key2.key = "folder/some-stuff"
key2.set_contents_from_string("some value")
rs = bucket.get_all_keys()
rs[0].last_modified.should.equal('2012-01-01T12:00:00Z')
key3 = Key(bucket)
key3.key = "folder/more-folder/foobar"
key3.set_contents_from_string("some value")
key4 = Key(bucket)
key4.key = "a-key"
key4.set_contents_from_string("some value")
keys = bucket.get_all_keys()
keys.should.have.length_of(3)
keys[0].name.should.equal("a-key")
keys[1].name.should.equal("the-key")
# Prefix
keys[2].name.should.equal("folder")
keys = bucket.get_all_keys(prefix="folder/")
keys.should.have.length_of(2)
keys[0].name.should.equal("folder/some-stuff")
keys[1].name.should.equal("folder/more-folder")
bucket.get_key("the-key").last_modified.should.equal('Sun, 01 Jan 2012 12:00:00 GMT')
@mock_s3
@ -177,11 +195,102 @@ def test_get_all_buckets():
buckets.should.have.length_of(2)
@mock_s3
def test_post_to_bucket():
conn = boto.connect_s3('the_key', 'the_secret')
bucket = conn.create_bucket("foobar")
requests.post("https://foobar.s3.amazonaws.com/", {
'key': 'the-key',
'file': 'nothing'
})
bucket.get_key('the-key').get_contents_as_string().should.equal('nothing')
@mock_s3
def test_post_with_metadata_to_bucket():
conn = boto.connect_s3('the_key', 'the_secret')
bucket = conn.create_bucket("foobar")
requests.post("https://foobar.s3.amazonaws.com/", {
'key': 'the-key',
'file': 'nothing',
'x-amz-meta-test': 'metadata'
})
bucket.get_key('the-key').get_metadata('test').should.equal('metadata')
@mock_s3
def test_bucket_method_not_implemented():
requests.post.when.called_with("https://foobar.s3.amazonaws.com/").should.throw(NotImplementedError)
requests.patch.when.called_with("https://foobar.s3.amazonaws.com/").should.throw(NotImplementedError)
@mock_s3
def test_key_method_not_implemented():
requests.post.when.called_with("https://foobar.s3.amazonaws.com/foo").should.throw(NotImplementedError)
@mock_s3
def test_bucket_name_with_dot():
conn = boto.connect_s3()
bucket = conn.create_bucket('firstname.lastname')
k = Key(bucket, 'somekey')
k.set_contents_from_string('somedata')
@mock_s3
def test_key_with_special_characters():
conn = boto.connect_s3()
bucket = conn.create_bucket('test_bucket_name')
key = Key(bucket, 'test_list_keys_2/x?y')
key.set_contents_from_string('value1')
key_list = bucket.list('test_list_keys_2/', '/')
keys = [x for x in key_list]
keys[0].name.should.equal("test_list_keys_2/x?y")
@mock_s3
def test_bucket_key_listing_order():
conn = boto.connect_s3()
bucket = conn.create_bucket('test_bucket')
prefix = 'toplevel/'
def store(name):
k = Key(bucket, prefix + name)
k.set_contents_from_string('somedata')
names = ['x/key', 'y.key1', 'y.key2', 'y.key3', 'x/y/key', 'x/y/z/key']
for name in names:
store(name)
delimiter = None
keys = [x.name for x in bucket.list(prefix, delimiter)]
keys.should.equal([
'toplevel/x/key', 'toplevel/x/y/key', 'toplevel/x/y/z/key',
'toplevel/y.key1', 'toplevel/y.key2', 'toplevel/y.key3'
])
delimiter = '/'
keys = [x.name for x in bucket.list(prefix, delimiter)]
keys.should.equal([
'toplevel/y.key1', 'toplevel/y.key2', 'toplevel/y.key3', 'toplevel/x/'
])
# Test delimiter with no prefix
delimiter = '/'
keys = [x.name for x in bucket.list(prefix=None, delimiter=delimiter)]
keys.should.equal(['toplevel'])
delimiter = None
keys = [x.name for x in bucket.list(prefix + 'x', delimiter)]
keys.should.equal([u'toplevel/x/key', u'toplevel/x/y/key', u'toplevel/x/y/z/key'])
delimiter = '/'
keys = [x.name for x in bucket.list(prefix + 'x', delimiter)]
keys.should.equal([u'toplevel/x/'])

14
tests/test_s3/test_s3_utils.py Normal file
View file

@ -0,0 +1,14 @@
from sure import expect
from moto.s3.utils import bucket_name_from_url
def test_base_url():
expect(bucket_name_from_url('https://s3.amazonaws.com/')).should.equal(None)
def test_localhost_bucket():
expect(bucket_name_from_url('https://wfoobar.localhost:5000/abc')).should.equal("wfoobar")
def test_localhost_without_bucket():
expect(bucket_name_from_url('https://www.localhost:5000/def')).should.equal(None)

17
tests/test_s3/test_server.py
View file

@ -1,4 +1,4 @@
import sure # flake8: noqa
import sure # noqa
import moto.server as server
@ -33,3 +33,18 @@ def test_s3_server_bucket_create():
res = test_client.get('/bar', 'http://foobar.localhost:5000/')
res.status_code.should.equal(200)
res.data.should.equal("test value")
def test_s3_server_post_to_bucket():
test_client = server.app.test_client()
res = test_client.put('/', 'http://foobar.localhost:5000/')
res.status_code.should.equal(200)
test_client.post('/', "https://foobar.localhost:5000/", data={
'key': 'the-key',
'file': 'nothing'
})
res = test_client.get('/the-key', 'http://foobar.localhost:5000/')
res.status_code.should.equal(200)
res.data.should.equal("nothing")

2
tests/test_ses/test_server.py
View file

@ -1,4 +1,4 @@
import sure # flake8: noqa
import sure # noqa
import moto.server as server

33
tests/test_ses/test_ses.py
View file

@ -3,7 +3,7 @@ import email
import boto
from boto.exception import BotoServerError
import sure # flake8: noqa
import sure # noqa
from moto import mock_ses
@ -44,7 +44,8 @@ def test_delete_identity():
def test_send_email():
conn = boto.connect_ses('the_key', 'the_secret')
conn.send_email.when.called_with("test@example.com", "test subject",
conn.send_email.when.called_with(
"test@example.com", "test subject",
"test body", "test_to@example.com").should.throw(BotoServerError)
conn.verify_email_identity("test@example.com")
@ -53,7 +54,21 @@ def test_send_email():
send_quota = conn.get_send_quota()
sent_count = int(send_quota['GetSendQuotaResponse']['GetSendQuotaResult']['SentLast24Hours'])
sent_count.should.equal(1)
@mock_ses
def test_send_html_email():
conn = boto.connect_ses('the_key', 'the_secret')
conn.send_email.when.called_with(
"test@example.com", "test subject",
"<span>test body</span>", "test_to@example.com", format="html").should.throw(BotoServerError)
conn.verify_email_identity("test@example.com")
conn.send_email("test@example.com", "test subject", "<span>test body</span>", "test_to@example.com", format="html")
send_quota = conn.get_send_quota()
sent_count = int(send_quota['GetSendQuotaResponse']['GetSendQuotaResult']['SentLast24Hours'])
sent_count.should.equal(1)
@mock_ses
def test_send_raw_email():
@ -74,12 +89,18 @@ def test_send_raw_email():
part.add_header('Content-Disposition', 'attachment; filename=test.txt')
message.attach(part)
conn.send_raw_email.when.called_with(source=message['From'], raw_message=message.as_string(),
destinations=message['To']).should.throw(BotoServerError)
conn.send_raw_email.when.called_with(
source=message['From'],
raw_message=message.as_string(),
destinations=message['To']
).should.throw(BotoServerError)
conn.verify_email_identity("test@example.com")
conn.send_raw_email(source=message['From'], raw_message=message.as_string(),
destinations=message['To'])
conn.send_raw_email(
source=message['From'],
raw_message=message.as_string(),
destinations=message['To']
)
send_quota = conn.get_send_quota()
sent_count = int(send_quota['GetSendQuotaResponse']['GetSendQuotaResult']['SentLast24Hours'])

2
tests/test_sqs/test_server.py
View file

@ -1,5 +1,5 @@
import re
import sure # flake8: noqa
import sure # noqa
import moto.server as server

45
tests/test_sqs/test_sqs.py
View file

@ -2,7 +2,7 @@ import boto
from boto.exception import SQSError
from boto.sqs.message import RawMessage
import requests
import sure # flake8: noqa
import sure # noqa
from moto import mock_sqs
@ -18,6 +18,19 @@ def test_create_queue():
all_queues[0].get_timeout().should.equal(60)
@mock_sqs
def test_get_queue():
conn = boto.connect_sqs('the_key', 'the_secret')
conn.create_queue("test-queue", visibility_timeout=60)
queue = conn.get_queue("test-queue")
queue.name.should.equal("test-queue")
queue.get_timeout().should.equal(60)
nonexisting_queue = conn.get_queue("nonexisting_queue")
nonexisting_queue.should.be.none
@mock_sqs
def test_delete_queue():
conn = boto.connect_sqs('the_key', 'the_secret')
@ -130,3 +143,33 @@ def test_delete_batch_operation():
@mock_sqs
def test_sqs_method_not_implemented():
requests.post.when.called_with("https://sqs.amazonaws.com/?Action=[foobar]").should.throw(NotImplementedError)
@mock_sqs
def test_queue_attributes():
conn = boto.connect_sqs('the_key', 'the_secret')
queue_name = 'test-queue'
visibility_timeout = 60
queue = conn.create_queue(queue_name, visibility_timeout=visibility_timeout)
attributes = queue.get_attributes()
attributes['QueueArn'].should.look_like(
'arn:aws:sqs:sqs.us-east-1:123456789012:%s' % queue_name)
attributes['VisibilityTimeout'].should.look_like(str(visibility_timeout))
attribute_names = queue.get_attributes().keys()
attribute_names.should.contain('ApproximateNumberOfMessagesNotVisible')
attribute_names.should.contain('MessageRetentionPeriod')
attribute_names.should.contain('ApproximateNumberOfMessagesDelayed')
attribute_names.should.contain('MaximumMessageSize')
attribute_names.should.contain('CreatedTimestamp')
attribute_names.should.contain('ApproximateNumberOfMessages')
attribute_names.should.contain('ReceiveMessageWaitTimeSeconds')
attribute_names.should.contain('DelaySeconds')
attribute_names.should.contain('VisibilityTimeout')
attribute_names.should.contain('LastModifiedTimestamp')
attribute_names.should.contain('QueueArn')

16
tests/test_sts/test_server.py Normal file
View file

@ -0,0 +1,16 @@
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("sts")
def test_sts_get_session_token():
test_client = server.app.test_client()
res = test_client.get('/?Action=GetSessionToken')
res.status_code.should.equal(200)
res.data.should.contain("SessionToken")
res.data.should.contain("AccessKeyId")

51
tests/test_sts/test_sts.py Normal file
View file

@ -0,0 +1,51 @@
import json
import boto
from freezegun import freeze_time
import sure # noqa
from moto import mock_sts
@freeze_time("2012-01-01 12:00:00")
@mock_sts
def test_get_session_token():
conn = boto.connect_sts()
token = conn.get_session_token(duration=123)
token.expiration.should.equal('2012-01-01T12:02:03Z')
token.session_token.should.equal("AQoEXAMPLEH4aoAH0gNCAPyJxz4BlCFFxWNE1OPTgk5TthT+FvwqnKwRcOIfrRh3c/LTo6UDdyJwOOvEVPvLXCrrrUtdnniCEXAMPLE/IvU1dYUg2RVAJBanLiHb4IgRmpRV3zrkuWJOgQs8IZZaIv2BXIa2R4OlgkBN9bkUDNCJiBeb/AXlzBBko7b15fjrBs2+cTQtpZ3CYWFXG8C5zqx37wnOE49mRl/+OtkIKGO7fAE")
token.access_key.should.equal("AKIAIOSFODNN7EXAMPLE")
token.secret_key.should.equal("wJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY")
@freeze_time("2012-01-01 12:00:00")
@mock_sts
def test_assume_role():
conn = boto.connect_sts()
policy = json.dumps({
"Statement": [
{
"Sid": "Stmt13690092345534",
"Action": [
"S3:ListBucket"
],
"Effect": "Allow",
"Resource": [
"arn:aws:s3:::foobar-tester"
]
},
]
})
s3_role = "arn:aws:iam::123456789012:role/test-role"
role = conn.assume_role(s3_role, "session-name", policy, duration_seconds=123)
credentials = role.credentials
credentials.expiration.should.equal('2012-01-01T12:02:03Z')
credentials.session_token.should.equal("BQoEXAMPLEH4aoAH0gNCAPyJxz4BlCFFxWNE1OPTgk5TthT+FvwqnKwRcOIfrRh3c/LTo6UDdyJwOOvEVPvLXCrrrUtdnniCEXAMPLE/IvU1dYUg2RVAJBanLiHb4IgRmpRV3zrkuWJOgQs8IZZaIv2BXIa2R4OlgkBN9bkUDNCJiBeb/AXlzBBko7b15fjrBs2+cTQtpZ3CYWFXG8C5zqx37wnOE49mRl/+OtkIKGO7fAE")
credentials.access_key.should.equal("AKIAIOSFODNN7EXAMPLE")
credentials.secret_key.should.equal("aJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY")
role.user.arn.should.equal("arn:aws:iam::123456789012:role/test-role")
role.user.assume_role_id.should.contain("session-name")