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

Merge branch 'master' into dynamov2_no_indexes

Browse source
This commit is contained in:
Sorin 2014-01-07 11:10:24 +02:00
commit 776e1bc65a
57 changed files with 368 additions and 203 deletions
Download patch file Download diff file Expand all files Collapse all files

5
tests/test_autoscaling/test_server.py
View file

@ -5,11 +5,12 @@ 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()
backend = server.create_backend_app("autoscaling")
test_client = backend.test_client()
res = test_client.get('/?Action=DescribeLaunchConfigurations')
res.data.should.contain('<DescribeLaunchConfigurationsResponse')

28
tests/test_core/test_nested.py Normal file
View file

@ -0,0 +1,28 @@
import unittest
from boto.sqs.connection import SQSConnection
from boto.sqs.message import Message
from boto.ec2 import EC2Connection
from moto import mock_sqs, mock_ec2
class TestNestedDecorators(unittest.TestCase):
@mock_sqs
def setup_sqs_queue(self):
conn = SQSConnection()
q = conn.create_queue('some-queue')
m = Message()
m.set_body('This is my first message.')
q.write(m)
self.assertEqual(q.count(), 1)
@mock_ec2
def test_nested(self):
self.setup_sqs_queue()
conn = EC2Connection()
conn.run_instances('ami-123456')

16
tests/test_core/test_server.py
View file

@ -13,13 +13,17 @@ def test_wrong_arguments():
pass
@patch('moto.server.app.run')
def test_right_arguments(app_run):
@patch('moto.server.run_simple')
def test_right_arguments(run_simple):
main(["s3"])
app_run.assert_called_once_with(host='0.0.0.0', port=5000)
func_call = run_simple.call_args[0]
func_call[0].should.equal("0.0.0.0")
func_call[1].should.equal(5000)
@patch('moto.server.app.run')
def test_port_argument(app_run):
@patch('moto.server.run_simple')
def test_port_argument(run_simple):
main(["s3", "--port", "8080"])
app_run.assert_called_once_with(host='0.0.0.0', port=8080)
func_call = run_simple.call_args[0]
func_call[0].should.equal("0.0.0.0")
func_call[1].should.equal(8080)

5
tests/test_dynamodb/test_server.py
View file

@ -5,11 +5,12 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("dynamodb")
def test_table_list():
test_client = server.app.test_client()
backend = server.create_backend_app("dynamodb")
test_client = backend.test_client()
res = test_client.get('/')
res.status_code.should.equal(404)

47
tests/test_ec2/test_security_groups.py
View file

@ -20,6 +20,42 @@ def test_create_and_describe_security_group():
all_groups.should.have.length_of(1)
all_groups[0].name.should.equal('test security group')
@mock_ec2
def test_create_and_describe_vpc_security_group():
conn = boto.connect_ec2('the_key', 'the_secret')
vpc_id = 'vpc-5300000c'
security_group = conn.create_security_group('test security group', 'this is a test security group', vpc_id=vpc_id)
security_group.vpc_id.should.equal(vpc_id)
security_group.name.should.equal('test security group')
security_group.description.should.equal('this is a test security group')
# Trying to create another group with the same name in the same VPC should throw an error
conn.create_security_group.when.called_with('test security group', 'this is a test security group', vpc_id).should.throw(EC2ResponseError)
all_groups = conn.get_all_security_groups()
all_groups[0].vpc_id.should.equal(vpc_id)
all_groups.should.have.length_of(1)
all_groups[0].name.should.equal('test security group')
@mock_ec2
def test_create_two_security_groups_with_same_name_in_different_vpc():
conn = boto.connect_ec2('the_key', 'the_secret')
vpc_id = 'vpc-5300000c'
vpc_id2 = 'vpc-5300000d'
sg1 = conn.create_security_group('test security group', 'this is a test security group', vpc_id)
sg2 = conn.create_security_group('test security group', 'this is a test security group', vpc_id2)
all_groups = conn.get_all_security_groups()
all_groups.should.have.length_of(2)
all_groups[0].name.should.equal('test security group')
all_groups[1].name.should.equal('test security group')
@mock_ec2
def test_deleting_security_groups():
@ -37,9 +73,18 @@ def test_deleting_security_groups():
conn.get_all_security_groups().should.have.length_of(1)
# Delete by group id
conn.delete_security_group(security_group1.id)
conn.delete_security_group(group_id=security_group1.id)
conn.get_all_security_groups().should.have.length_of(0)
@mock_ec2
def test_delete_security_group_in_vpc():
conn = boto.connect_ec2('the_key', 'the_secret')
vpc_id = "vpc-12345"
security_group1 = conn.create_security_group('test1', 'test1', vpc_id)
# this should not throw an exception
conn.delete_security_group(group_id=security_group1.id)
@mock_ec2
def test_authorize_ip_range_and_revoke():

5
tests/test_ec2/test_server.py
View file

@ -6,11 +6,12 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("ec2")
def test_ec2_server_get():
test_client = server.app.test_client()
backend = server.create_backend_app("ec2")
test_client = backend.test_client()
res = test_client.get('/?Action=RunInstances&ImageId=ami-60a54009')
groups = re.search("<instanceId>(.*)</instanceId>", res.data)

5
tests/test_elb/test_server.py
View file

@ -5,11 +5,12 @@ 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()
backend = server.create_backend_app("elb")
test_client = backend.test_client()
res = test_client.get('/?Action=DescribeLoadBalancers')
res.data.should.contain('DescribeLoadBalancersResponse')

5
tests/test_emr/test_server.py
View file

@ -5,11 +5,12 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("emr")
def test_describe_jobflows():
test_client = server.app.test_client()
backend = server.create_backend_app("emr")
test_client = backend.test_client()
res = test_client.get('/?Action=DescribeJobFlows')
res.data.should.contain('<DescribeJobFlowsResult>')

13
tests/test_s3/test_server.py
View file

@ -5,18 +5,21 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("s3")
def test_s3_server_get():
test_client = server.app.test_client()
backend = server.create_backend_app("s3")
test_client = backend.test_client()
res = test_client.get('/')
res.data.should.contain('ListAllMyBucketsResult')
def test_s3_server_bucket_create():
test_client = server.app.test_client()
backend = server.create_backend_app("s3")
test_client = backend.test_client()
res = test_client.put('/', 'http://foobar.localhost:5000/')
res.status_code.should.equal(200)
@ -36,7 +39,9 @@ def test_s3_server_bucket_create():
def test_s3_server_post_to_bucket():
test_client = server.app.test_client()
backend = server.create_backend_app("s3")
test_client = backend.test_client()
res = test_client.put('/', 'http://foobar.localhost:5000/')
res.status_code.should.equal(200)

13
tests/test_s3bucket_path/test_bucket_path_server.py
View file

@ -5,18 +5,21 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("s3bucket_path")
def test_s3_server_get():
test_client = server.app.test_client()
backend = server.create_backend_app("s3bucket_path")
test_client = backend.test_client()
res = test_client.get('/')
res.data.should.contain('ListAllMyBucketsResult')
def test_s3_server_bucket_create():
test_client = server.app.test_client()
backend = server.create_backend_app("s3bucket_path")
test_client = backend.test_client()
res = test_client.put('/foobar', 'http://localhost:5000')
res.status_code.should.equal(200)
@ -36,7 +39,9 @@ def test_s3_server_bucket_create():
def test_s3_server_post_to_bucket():
test_client = server.app.test_client()
backend = server.create_backend_app("s3bucket_path")
test_client = backend.test_client()
res = test_client.put('/foobar', 'http://localhost:5000/')
res.status_code.should.equal(200)

5
tests/test_ses/test_server.py
View file

@ -5,10 +5,11 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("ses")
def test_ses_list_identities():
test_client = server.app.test_client()
backend = server.create_backend_app("ses")
test_client = backend.test_client()
res = test_client.get('/?Action=ListIdentities')
res.data.should.contain("ListIdentitiesResponse")

5
tests/test_sqs/test_server.py
View file

@ -6,11 +6,12 @@ import moto.server as server
'''
Test the different server responses
'''
server.configure_urls("sqs")
def test_sqs_list_identities():
test_client = server.app.test_client()
backend = server.create_backend_app("sqs")
test_client = backend.test_client()
res = test_client.get('/?Action=ListQueues')
res.data.should.contain("ListQueuesResponse")

5
tests/test_sts/test_server.py
View file

@ -5,11 +5,12 @@ 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()
backend = server.create_backend_app("sts")
test_client = backend.test_client()
res = test_client.get('/?Action=GetSessionToken')
res.status_code.should.equal(200)
res.data.should.contain("SessionToken")