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

Lints.

Browse source
This commit is contained in:
Steve Pulec 2017-02-23 21:37:43 -05:00
commit f37bad0e00
260 changed files with 6363 additions and 3766 deletions
Download patch file Download diff file Expand all files Collapse all files

178
tests/test_ec2/test_security_groups.py
View file

@ -20,25 +20,30 @@ def test_create_and_describe_security_group():
conn = boto.connect_ec2('the_key', 'the_secret')
with assert_raises(EC2ResponseError) as ex:
security_group = conn.create_security_group('test security group', 'this is a test security group', dry_run=True)
security_group = conn.create_security_group(
'test security group', 'this is a test security group', dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the CreateSecurityGroup operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the CreateSecurityGroup operation: Request would have succeeded, but DryRun flag is set')
security_group = conn.create_security_group('test security group', 'this is a test security group')
security_group = conn.create_security_group(
'test security group', 'this is a test security group')
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 should throw an error
with assert_raises(EC2ResponseError) as cm:
conn.create_security_group('test security group', 'this is a test security group')
conn.create_security_group(
'test security group', 'this is a test security group')
cm.exception.code.should.equal('InvalidGroup.Duplicate')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
all_groups = conn.get_all_security_groups()
all_groups.should.have.length_of(3) # The default group gets created automatically
# The default group gets created automatically
all_groups.should.have.length_of(3)
group_names = [group.name for group in all_groups]
set(group_names).should.equal(set(["default", "test security group"]))
@ -66,16 +71,19 @@ def test_default_security_group():
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 = 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
# Trying to create another group with the same name in the same VPC should
# throw an error
with assert_raises(EC2ResponseError) as cm:
conn.create_security_group('test security group', 'this is a test security group', vpc_id)
conn.create_security_group(
'test security group', 'this is a test security group', vpc_id)
cm.exception.code.should.equal('InvalidGroup.Duplicate')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@ -94,8 +102,10 @@ def test_create_two_security_groups_with_same_name_in_different_vpc():
vpc_id = 'vpc-5300000c'
vpc_id2 = 'vpc-5300000d'
conn.create_security_group('test security group', 'this is a test security group', vpc_id)
conn.create_security_group('test security group', 'this is a test security group', vpc_id2)
conn.create_security_group(
'test security group', 'this is a test security group', vpc_id)
conn.create_security_group(
'test security group', 'this is a test security group', vpc_id2)
all_groups = conn.get_all_security_groups()
@ -125,7 +135,8 @@ def test_deleting_security_groups():
conn.delete_security_group('test2', dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the DeleteSecurityGroup operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the DeleteSecurityGroup operation: Request would have succeeded, but DryRun flag is set')
conn.delete_security_group('test2')
conn.get_all_security_groups().should.have.length_of(3)
@ -151,65 +162,83 @@ def test_authorize_ip_range_and_revoke():
security_group = conn.create_security_group('test', 'test')
with assert_raises(EC2ResponseError) as ex:
success = security_group.authorize(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
success = security_group.authorize(
ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the GrantSecurityGroupIngress operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the GrantSecurityGroupIngress operation: Request would have succeeded, but DryRun flag is set')
success = security_group.authorize(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
success = security_group.authorize(
ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
assert success.should.be.true
security_group = conn.get_all_security_groups(groupnames=['test'])[0]
int(security_group.rules[0].to_port).should.equal(2222)
security_group.rules[0].grants[0].cidr_ip.should.equal("123.123.123.123/32")
security_group.rules[0].grants[
0].cidr_ip.should.equal("123.123.123.123/32")
# Wrong Cidr should throw error
with assert_raises(EC2ResponseError) as cm:
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.122/32")
security_group.revoke(ip_protocol="tcp", from_port="22",
to_port="2222", cidr_ip="123.123.123.122/32")
cm.exception.code.should.equal('InvalidPermission.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Actually revoke
with assert_raises(EC2ResponseError) as ex:
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
security_group.revoke(ip_protocol="tcp", from_port="22",
to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the RevokeSecurityGroupIngress operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the RevokeSecurityGroupIngress operation: Request would have succeeded, but DryRun flag is set')
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
security_group.revoke(ip_protocol="tcp", from_port="22",
to_port="2222", cidr_ip="123.123.123.123/32")
security_group = conn.get_all_security_groups()[0]
security_group.rules.should.have.length_of(0)
# Test for egress as well
egress_security_group = conn.create_security_group('testegress', 'testegress', vpc_id='vpc-3432589')
egress_security_group = conn.create_security_group(
'testegress', 'testegress', vpc_id='vpc-3432589')
with assert_raises(EC2ResponseError) as ex:
success = conn.authorize_security_group_egress(egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
success = conn.authorize_security_group_egress(
egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the GrantSecurityGroupEgress operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the GrantSecurityGroupEgress operation: Request would have succeeded, but DryRun flag is set')
success = conn.authorize_security_group_egress(egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
success = conn.authorize_security_group_egress(
egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
assert success.should.be.true
egress_security_group = conn.get_all_security_groups(groupnames='testegress')[0]
egress_security_group = conn.get_all_security_groups(
groupnames='testegress')[0]
# There are two egress rules associated with the security group:
# the default outbound rule and the new one
int(egress_security_group.rules_egress[1].to_port).should.equal(2222)
egress_security_group.rules_egress[1].grants[0].cidr_ip.should.equal("123.123.123.123/32")
egress_security_group.rules_egress[1].grants[
0].cidr_ip.should.equal("123.123.123.123/32")
# Wrong Cidr should throw error
egress_security_group.revoke.when.called_with(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.122/32").should.throw(EC2ResponseError)
egress_security_group.revoke.when.called_with(
ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.122/32").should.throw(EC2ResponseError)
# Actually revoke
with assert_raises(EC2ResponseError) as ex:
conn.revoke_security_group_egress(egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
conn.revoke_security_group_egress(
egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32", dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the RevokeSecurityGroupEgress operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the RevokeSecurityGroupEgress operation: Request would have succeeded, but DryRun flag is set')
conn.revoke_security_group_egress(egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
conn.revoke_security_group_egress(
egress_security_group.id, "tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
egress_security_group = conn.get_all_security_groups()[0]
# There is still the default outbound rule
@ -223,24 +252,30 @@ def test_authorize_other_group_and_revoke():
other_security_group = conn.create_security_group('other', 'other')
wrong_group = conn.create_security_group('wrong', 'wrong')
success = security_group.authorize(ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
success = security_group.authorize(
ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
assert success.should.be.true
security_group = [group for group in conn.get_all_security_groups() if group.name == 'test'][0]
security_group = [
group for group in conn.get_all_security_groups() if group.name == 'test'][0]
int(security_group.rules[0].to_port).should.equal(2222)
security_group.rules[0].grants[0].group_id.should.equal(other_security_group.id)
security_group.rules[0].grants[
0].group_id.should.equal(other_security_group.id)
# Wrong source group should throw error
with assert_raises(EC2ResponseError) as cm:
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", src_group=wrong_group)
security_group.revoke(ip_protocol="tcp", from_port="22",
to_port="2222", src_group=wrong_group)
cm.exception.code.should.equal('InvalidPermission.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Actually revoke
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
security_group.revoke(ip_protocol="tcp", from_port="22",
to_port="2222", src_group=other_security_group)
security_group = [group for group in conn.get_all_security_groups() if group.name == 'test'][0]
security_group = [
group for group in conn.get_all_security_groups() if group.name == 'test'][0]
security_group.rules.should.have.length_of(0)
@ -250,8 +285,10 @@ def test_authorize_other_group_egress_and_revoke():
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
sg01 = ec2.create_security_group(GroupName='sg01', Description='Test security group sg01', VpcId=vpc.id)
sg02 = ec2.create_security_group(GroupName='sg02', Description='Test security group sg02', VpcId=vpc.id)
sg01 = ec2.create_security_group(
GroupName='sg01', Description='Test security group sg01', VpcId=vpc.id)
sg02 = ec2.create_security_group(
GroupName='sg02', Description='Test security group sg02', VpcId=vpc.id)
ip_permission = {
'IpProtocol': 'tcp',
@ -278,27 +315,33 @@ def test_authorize_group_in_vpc():
security_group = conn.create_security_group('test1', 'test1', vpc_id)
other_security_group = conn.create_security_group('test2', 'test2', vpc_id)
success = security_group.authorize(ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
success = security_group.authorize(
ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
success.should.be.true
# Check that the rule is accurate
security_group = [group for group in conn.get_all_security_groups() if group.name == 'test1'][0]
security_group = [
group for group in conn.get_all_security_groups() if group.name == 'test1'][0]
int(security_group.rules[0].to_port).should.equal(2222)
security_group.rules[0].grants[0].group_id.should.equal(other_security_group.id)
security_group.rules[0].grants[
0].group_id.should.equal(other_security_group.id)
# Now remove the rule
success = security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
success = security_group.revoke(
ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)
success.should.be.true
# And check that it gets revoked
security_group = [group for group in conn.get_all_security_groups() if group.name == 'test1'][0]
security_group = [
group for group in conn.get_all_security_groups() if group.name == 'test1'][0]
security_group.rules.should.have.length_of(0)
@mock_ec2_deprecated
def test_get_all_security_groups():
conn = boto.connect_ec2()
sg1 = conn.create_security_group(name='test1', description='test1', vpc_id='vpc-mjm05d27')
sg1 = conn.create_security_group(
name='test1', description='test1', vpc_id='vpc-mjm05d27')
conn.create_security_group(name='test2', description='test2')
resp = conn.get_all_security_groups(groupnames=['test1'])
@ -326,7 +369,8 @@ def test_authorize_bad_cidr_throws_invalid_parameter_value():
conn = boto.connect_ec2('the_key', 'the_secret')
security_group = conn.create_security_group('test', 'test')
with assert_raises(EC2ResponseError) as cm:
security_group.authorize(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123")
security_group.authorize(
ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123")
cm.exception.code.should.equal('InvalidParameterValue')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@ -343,7 +387,8 @@ def test_security_group_tagging():
sg.add_tag("Test", "Tag", dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal('An error occurred (DryRunOperation) when calling the CreateTags operation: Request would have succeeded, but DryRun flag is set')
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the CreateTags operation: Request would have succeeded, but DryRun flag is set')
sg.add_tag("Test", "Tag")
@ -362,7 +407,8 @@ def test_security_group_tag_filtering():
sg = conn.create_security_group("test-sg", "Test SG")
sg.add_tag("test-tag", "test-value")
groups = conn.get_all_security_groups(filters={"tag:test-tag": "test-value"})
groups = conn.get_all_security_groups(
filters={"tag:test-tag": "test-value"})
groups.should.have.length_of(1)
@ -507,18 +553,18 @@ def test_sec_group_rule_limit_vpc():
cm.exception.error_code.should.equal('RulesPerSecurityGroupLimitExceeded')
'''
Boto3
'''
@mock_ec2
def test_add_same_rule_twice_throws_error():
ec2 = boto3.resource('ec2', region_name='us-west-1')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
sg = ec2.create_security_group(GroupName='sg1', Description='Test security group sg1', VpcId=vpc.id)
sg = ec2.create_security_group(
GroupName='sg1', Description='Test security group sg1', VpcId=vpc.id)
ip_permissions = [
{
@ -541,13 +587,18 @@ def test_security_group_tagging_boto3():
sg = conn.create_security_group(GroupName="test-sg", Description="Test SG")
with assert_raises(ClientError) as ex:
conn.create_tags(Resources=[sg['GroupId']], Tags=[{'Key': 'Test', 'Value': 'Tag'}], DryRun=True)
conn.create_tags(Resources=[sg['GroupId']], Tags=[
{'Key': 'Test', 'Value': 'Tag'}], DryRun=True)
ex.exception.response['Error']['Code'].should.equal('DryRunOperation')
ex.exception.response['ResponseMetadata']['HTTPStatusCode'].should.equal(400)
ex.exception.response['Error']['Message'].should.equal('An error occurred (DryRunOperation) when calling the CreateTags operation: Request would have succeeded, but DryRun flag is set')
ex.exception.response['ResponseMetadata'][
'HTTPStatusCode'].should.equal(400)
ex.exception.response['Error']['Message'].should.equal(
'An error occurred (DryRunOperation) when calling the CreateTags operation: Request would have succeeded, but DryRun flag is set')
conn.create_tags(Resources=[sg['GroupId']], Tags=[{'Key': 'Test', 'Value': 'Tag'}])
describe = conn.describe_security_groups(Filters=[{'Name': 'tag-value', 'Values': ['Tag']}])
conn.create_tags(Resources=[sg['GroupId']], Tags=[
{'Key': 'Test', 'Value': 'Tag'}])
describe = conn.describe_security_groups(
Filters=[{'Name': 'tag-value', 'Values': ['Tag']}])
tag = describe["SecurityGroups"][0]['Tags'][0]
tag['Value'].should.equal("Tag")
tag['Key'].should.equal("Test")
@ -559,9 +610,12 @@ def test_authorize_and_revoke_in_bulk():
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
sg01 = ec2.create_security_group(GroupName='sg01', Description='Test security group sg01', VpcId=vpc.id)
sg02 = ec2.create_security_group(GroupName='sg02', Description='Test security group sg02', VpcId=vpc.id)
sg03 = ec2.create_security_group(GroupName='sg03', Description='Test security group sg03')
sg01 = ec2.create_security_group(
GroupName='sg01', Description='Test security group sg01', VpcId=vpc.id)
sg02 = ec2.create_security_group(
GroupName='sg02', Description='Test security group sg02', VpcId=vpc.id)
sg03 = ec2.create_security_group(
GroupName='sg03', Description='Test security group sg03')
ip_permissions = [
{
@ -611,15 +665,19 @@ def test_authorize_and_revoke_in_bulk():
for ip_permission in expected_ip_permissions:
sg01.ip_permissions_egress.shouldnt.contain(ip_permission)
@mock_ec2_deprecated
def test_get_all_security_groups_filter_with_same_vpc_id():
conn = boto.connect_ec2('the_key', 'the_secret')
vpc_id = 'vpc-5300000c'
security_group = conn.create_security_group('test1', 'test1', vpc_id=vpc_id)
security_group2 = conn.create_security_group('test2', 'test2', vpc_id=vpc_id)
security_group = conn.create_security_group(
'test1', 'test1', vpc_id=vpc_id)
security_group2 = conn.create_security_group(
'test2', 'test2', vpc_id=vpc_id)
security_group.vpc_id.should.equal(vpc_id)
security_group2.vpc_id.should.equal(vpc_id)
security_groups = conn.get_all_security_groups(group_ids=[security_group.id], filters={'vpc-id': [vpc_id]})
security_groups = conn.get_all_security_groups(
group_ids=[security_group.id], filters={'vpc-id': [vpc_id]})
security_groups.should.have.length_of(1)