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

Merge branch 'master' into sns_regions

Browse source
This commit is contained in:
Ilya Sukhanov 2015-08-03 10:22:55 -04:00
commit 9904193d66
21 changed files with 477 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

40
tests/test_cloudformation/test_server.py
View file

@ -1 +1,41 @@
from __future__ import unicode_literals
import json
from six.moves.urllib.parse import urlencode
import re
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
def test_cloudformation_server_get():
backend = server.create_backend_app("cloudformation")
stack_name = 'test stack'
test_client = backend.test_client()
template_body = {
"Resources": {},
}
res = test_client.get(
'/?{0}'.format(
urlencode({
"Action": "CreateStack",
"StackName": stack_name,
"TemplateBody": json.dumps(template_body)
})
),
headers={"Host": "cloudformation.us-east-1.amazonaws.com"}
)
stack_id = json.loads(res.data.decode("utf-8"))["CreateStackResponse"]["CreateStackResult"]["StackId"]
res = test_client.get(
'/?Action=ListStacks',
headers={"Host": "cloudformation.us-east-1.amazonaws.com"}
)
stacks = re.search("<StackId>(.*)</StackId>", res.data.decode('utf-8'))
list_stack_id = stacks.groups()[0]
assert stack_id == list_stack_id

128
tests/test_dynamodb2/test_dynamodb_table_with_range_key.py
View file

@ -1,6 +1,8 @@
from __future__ import unicode_literals
import boto
import boto3
from boto3.dynamodb.conditions import Key
import sure # noqa
from freezegun import freeze_time
from moto import mock_dynamodb2
@ -253,31 +255,31 @@ def test_query():
table.count().should.equal(4)
results = table.query(forum_name__eq='the-key', subject__gt='1', consistent=True)
results = table.query_2(forum_name__eq='the-key', subject__gt='1', consistent=True)
expected = ["123", "456", "789"]
for index, item in enumerate(results):
item["subject"].should.equal(expected[index])
results = table.query(forum_name__eq="the-key", subject__gt='1', reverse=True)
results = table.query_2(forum_name__eq="the-key", subject__gt='1', reverse=True)
for index, item in enumerate(results):
item["subject"].should.equal(expected[len(expected) - 1 - index])
results = table.query(forum_name__eq='the-key', subject__gt='1', consistent=True)
results = table.query_2(forum_name__eq='the-key', subject__gt='1', consistent=True)
sum(1 for _ in results).should.equal(3)
results = table.query(forum_name__eq='the-key', subject__gt='234', consistent=True)
results = table.query_2(forum_name__eq='the-key', subject__gt='234', consistent=True)
sum(1 for _ in results).should.equal(2)
results = table.query(forum_name__eq='the-key', subject__gt='9999')
results = table.query_2(forum_name__eq='the-key', subject__gt='9999')
sum(1 for _ in results).should.equal(0)
results = table.query(forum_name__eq='the-key', subject__beginswith='12')
results = table.query_2(forum_name__eq='the-key', subject__beginswith='12')
sum(1 for _ in results).should.equal(1)
results = table.query(forum_name__eq='the-key', subject__beginswith='7')
results = table.query_2(forum_name__eq='the-key', subject__beginswith='7')
sum(1 for _ in results).should.equal(1)
results = table.query(forum_name__eq='the-key', subject__between=['567', '890'])
results = table.query_2(forum_name__eq='the-key', subject__between=['567', '890'])
sum(1 for _ in results).should.equal(1)
@ -558,7 +560,6 @@ def test_lookup():
@mock_dynamodb2
def test_failed_overwrite():
from decimal import Decimal
table = Table.create('messages', schema=[
HashKey('id'),
RangeKey('range'),
@ -567,19 +568,19 @@ def test_failed_overwrite():
'write': 3,
})
data1 = {'id': '123', 'range': 'abc', 'data':'678'}
data1 = {'id': '123', 'range': 'abc', 'data': '678'}
table.put_item(data=data1)
data2 = {'id': '123', 'range': 'abc', 'data':'345'}
table.put_item(data=data2, overwrite = True)
data2 = {'id': '123', 'range': 'abc', 'data': '345'}
table.put_item(data=data2, overwrite=True)
data3 = {'id': '123', 'range': 'abc', 'data':'812'}
data3 = {'id': '123', 'range': 'abc', 'data': '812'}
table.put_item.when.called_with(data=data3).should.throw(ConditionalCheckFailedException)
returned_item = table.lookup('123', 'abc')
dict(returned_item).should.equal(data2)
data4 = {'id': '123', 'range': 'ghi', 'data':812}
data4 = {'id': '123', 'range': 'ghi', 'data': 812}
table.put_item(data=data4)
returned_item = table.lookup('123', 'ghi')
@ -593,7 +594,7 @@ def test_conflicting_writes():
RangeKey('range'),
])
item_data = {'id': '123', 'range':'abc', 'data':'678'}
item_data = {'id': '123', 'range': 'abc', 'data': '678'}
item1 = Item(table, item_data)
item2 = Item(table, item_data)
item1.save()
@ -603,3 +604,100 @@ def test_conflicting_writes():
item1.save()
item2.save.when.called_with().should.throw(ConditionalCheckFailedException)
"""
boto3
"""
@mock_dynamodb2
def test_boto3_conditions():
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
table = dynamodb.create_table(
TableName='users',
KeySchema=[
{
'AttributeName': 'forum_name',
'KeyType': 'HASH'
},
{
'AttributeName': 'subject',
'KeyType': 'RANGE'
},
],
AttributeDefinitions=[
{
'AttributeName': 'forum_name',
'AttributeType': 'S'
},
{
'AttributeName': 'subject',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
table = dynamodb.Table('users')
table.put_item(Item={
'forum_name': 'the-key',
'subject': '123'
})
table.put_item(Item={
'forum_name': 'the-key',
'subject': '456'
})
table.put_item(Item={
'forum_name': 'the-key',
'subject': '789'
})
# Test a query returning all items
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").gt('1'),
ScanIndexForward=True,
)
expected = ["123", "456", "789"]
for index, item in enumerate(results['Items']):
item["subject"].should.equal(expected[index])
# Return all items again, but in reverse
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").gt('1'),
ScanIndexForward=False,
)
for index, item in enumerate(reversed(results['Items'])):
item["subject"].should.equal(expected[index])
# Filter the subjects to only return some of the results
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").gt('234'),
ConsistentRead=True,
)
results['Count'].should.equal(2)
# Filter to return no results
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").gt('9999')
)
results['Count'].should.equal(0)
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").begins_with('12')
)
results['Count'].should.equal(1)
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").begins_with('7')
)
results['Count'].should.equal(1)
results = table.query(
KeyConditionExpression=Key('forum_name').eq('the-key') & Key("subject").between('567', '890')
)
results['Count'].should.equal(1)

67
tests/test_dynamodb2/test_dynamodb_table_without_range_key.py
View file

@ -1,6 +1,8 @@
from __future__ import unicode_literals
import boto
import boto3
from boto3.dynamodb.conditions import Key
import sure # noqa
from freezegun import freeze_time
from boto.exception import JSONResponseError
@ -135,14 +137,6 @@ def test_item_put_without_table():
).should.throw(JSONResponseError)
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_get_missing_item():
table = create_table()
table.get_item.when.called_with(test_hash=3241526475).should.throw(JSONResponseError)
@requires_boto_gte("2.9")
@mock_dynamodb2
def test_get_item_with_undeclared_table():
@ -449,7 +443,6 @@ def test_update_item_set():
@mock_dynamodb2
def test_failed_overwrite():
from decimal import Decimal
table = Table.create('messages', schema=[
HashKey('id'),
], throughput={
@ -457,19 +450,19 @@ def test_failed_overwrite():
'write': 3,
})
data1 = {'id': '123', 'data':'678'}
data1 = {'id': '123', 'data': '678'}
table.put_item(data=data1)
data2 = {'id': '123', 'data':'345'}
table.put_item(data=data2, overwrite = True)
data2 = {'id': '123', 'data': '345'}
table.put_item(data=data2, overwrite=True)
data3 = {'id': '123', 'data':'812'}
data3 = {'id': '123', 'data': '812'}
table.put_item.when.called_with(data=data3).should.throw(ConditionalCheckFailedException)
returned_item = table.lookup('123')
dict(returned_item).should.equal(data2)
data4 = {'id': '124', 'data':812}
data4 = {'id': '124', 'data': 812}
table.put_item(data=data4)
returned_item = table.lookup('124')
@ -482,7 +475,7 @@ def test_conflicting_writes():
HashKey('id'),
])
item_data = {'id': '123', 'data':'678'}
item_data = {'id': '123', 'data': '678'}
item1 = Item(table, item_data)
item2 = Item(table, item_data)
item1.save()
@ -491,4 +484,46 @@ def test_conflicting_writes():
item2['data'] = '912'
item1.save()
item2.save.when.called_with().should.throw(ConditionalCheckFailedException)
item2.save.when.called_with().should.throw(ConditionalCheckFailedException)
"""
boto3
"""
@mock_dynamodb2
def test_boto3_conditions():
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
# Create the DynamoDB table.
table = dynamodb.create_table(
TableName='users',
KeySchema=[
{
'AttributeName': 'username',
'KeyType': 'HASH'
},
],
AttributeDefinitions=[
{
'AttributeName': 'username',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
table = dynamodb.Table('users')
table.put_item(Item={'username': 'johndoe'})
table.put_item(Item={'username': 'janedoe'})
response = table.query(
KeyConditionExpression=Key('username').eq('johndoe')
)
response['Count'].should.equal(1)
response['Items'].should.have.length_of(1)
response['Items'][0].should.equal({"username": "johndoe"})

33
tests/test_ec2/test_elastic_block_store.py
View file

@ -33,6 +33,19 @@ def test_create_and_delete_volume():
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_filter_volume_by_id():
conn = boto.connect_ec2('the_key', 'the_secret')
volume1 = conn.create_volume(80, "us-east-1a")
volume2 = conn.create_volume(36, "us-east-1b")
volume3 = conn.create_volume(20, "us-east-1c")
vol1 = conn.get_all_volumes(volume_ids=volume3.id)
vol1.should.have.length_of(1)
vol1[0].size.should.equal(20)
vol1[0].zone.should.equal('us-east-1c')
vol2 = conn.get_all_volumes(volume_ids=[volume1.id, volume2.id])
vol2.should.have.length_of(2)
@mock_ec2
def test_volume_attach_and_detach():
@ -85,6 +98,7 @@ def test_create_snapshot():
snapshots = conn.get_all_snapshots()
snapshots.should.have.length_of(1)
snapshots[0].description.should.equal('a test snapshot')
snapshots[0].start_time.should_not.be.none
# Create snapshot without description
snapshot = volume.create_snapshot()
@ -100,6 +114,25 @@ def test_create_snapshot():
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_filter_snapshot_by_id():
conn = boto.connect_ec2('the_key', 'the_secret')
volume1 = conn.create_volume(36, "us-east-1a")
snap1 = volume1.create_snapshot('a test snapshot 1')
volume2 = conn.create_volume(42, 'us-east-1a')
snap2 = volume2.create_snapshot('a test snapshot 2')
volume3 = conn.create_volume(84, 'us-east-1a')
snap3 = volume3.create_snapshot('a test snapshot 3')
snapshots1 = conn.get_all_snapshots(snapshot_ids=snap2.id)
snapshots1.should.have.length_of(1)
snapshots1[0].volume_id.should.equal(volume2.id)
snapshots1[0].region.name.should.equal('us-east-1')
snapshots2 = conn.get_all_snapshots(snapshot_ids=[snap2.id, snap3.id])
snapshots2.should.have.length_of(2)
for s in snapshots2:
s.start_time.should_not.be.none
s.volume_id.should.be.within([volume2.id, volume3.id])
s.region.name.should.equal('us-east-1')
@mock_ec2
def test_snapshot_attribute():

31
tests/test_ec2/test_instances.py
View file

@ -53,7 +53,7 @@ def test_instance_launch_and_terminate():
instances.should.have.length_of(1)
instances[0].id.should.equal(instance.id)
instances[0].state.should.equal('running')
instances[0].launch_time.should.equal("2014-01-01T05:00:00")
instances[0].launch_time.should.equal("2014-01-01T05:00:00Z")
instances[0].vpc_id.should.equal(None)
root_device_name = instances[0].root_device_name
@ -66,6 +66,35 @@ def test_instance_launch_and_terminate():
instance = reservations[0].instances[0]
instance.state.should.equal('terminated')
@freeze_time("2014-01-01 05:00:00")
@mock_ec2
def test_instance_attach_volume():
conn = boto.connect_ec2('the_key', 'the_secret')
reservation = conn.run_instances('ami-1234abcd')
instance = reservation.instances[0]
vol1 = conn.create_volume(size=36, zone=conn.region.name)
vol1.attach(instance.id, "/dev/sda1")
vol1.update()
vol2 = conn.create_volume(size=65, zone=conn.region.name)
vol2.attach(instance.id, "/dev/sdb1")
vol2.update()
vol3 = conn.create_volume(size=130, zone=conn.region.name)
vol3.attach(instance.id, "/dev/sdc1")
vol3.update()
reservations = conn.get_all_instances()
instance = reservations[0].instances[0]
instance.block_device_mapping.should.have.length_of(3)
for v in conn.get_all_volumes(volume_ids=[instance.block_device_mapping['/dev/sdc1'].volume_id]):
v.attach_data.instance_id.should.equal(instance.id)
v.attach_data.attach_time.should.equal(instance.launch_time) # can do due to freeze_time decorator.
v.create_time.should.equal(instance.launch_time) # can do due to freeze_time decorator.
v.region.name.should.equal(instance.region.name)
v.status.should.equal('in-use')
@mock_ec2
def test_get_instances_by_id():

16
tests/test_s3/test_s3.py
View file

@ -8,6 +8,7 @@ from io import BytesIO
import json
import boto
import boto3
from boto.exception import S3CreateError, S3ResponseError
from boto.s3.connection import S3Connection
from boto.s3.key import Key
@ -869,3 +870,18 @@ def test_policy():
bucket = conn.get_bucket(bucket_name)
bucket.get_policy().decode('utf-8').should.equal(policy)
"""
boto3
"""
@mock_s3
def test_boto3_bucket_create():
s3 = boto3.resource('s3', region_name='us-east-1')
s3.create_bucket(Bucket="blah")
s3.Object('blah', 'hello.txt').put(Body="some text")
s3.Object('blah', 'hello.txt').get()['Body'].read().decode("utf-8").should.equal("some text")

15
tests/test_sqs/test_sqs.py
View file

@ -1,5 +1,6 @@
from __future__ import unicode_literals
import boto
import boto3
from boto.exception import SQSError
from boto.sqs.message import RawMessage, Message
@ -462,3 +463,17 @@ def test_delete_message_after_visibility_timeout():
m1_retrieved.delete()
assert new_queue.count() == 0
"""
boto3
"""
@mock_sqs
def test_boto3_message_send():
sqs = boto3.resource('sqs', region_name='us-east-1')
queue = sqs.create_queue(QueueName="blah")
queue.send_message(MessageBody="derp")
messages = queue.receive_messages()
messages.should.have.length_of(1)