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

Improved support for VPC Address filtering

Browse source
This commit is contained in:
Nuwan Goonasekera 2017-09-16 01:49:57 +05:30
commit 8776129816
3 changed files with 129 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

81
tests/test_ec2/test_elastic_ip_addresses.py
View file

@ -402,3 +402,84 @@ def test_eip_describe_none():
cm.exception.code.should.equal('InvalidAddress.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_eip_filters():
service = boto3.resource('ec2', region_name='us-west-1')
client = boto3.client('ec2', region_name='us-west-1')
vpc_res = client.create_vpc(CidrBlock='10.0.0.0/24')
subnet_res = client.create_subnet(
VpcId=vpc_res['Vpc']['VpcId'], CidrBlock='10.0.0.0/24')
def create_inst_with_eip():
instance = service.create_instances(**{
'InstanceType': 't2.micro',
'ImageId': 'ami-test',
'MinCount': 1,
'MaxCount': 1,
'SubnetId': subnet_res['Subnet']['SubnetId']
})[0]
allocation_id = client.allocate_address(Domain='vpc')['AllocationId']
_ = client.associate_address(
InstanceId=instance.id,
AllocationId=allocation_id,
AllowReassociation=False)
instance.load()
address = service.VpcAddress(allocation_id)
address.load()
return instance, address
inst1, eip1 = create_inst_with_eip()
inst2, eip2 = create_inst_with_eip()
inst3, eip3 = create_inst_with_eip()
# Param search by AllocationId
addresses = list(service.vpc_addresses.filter(AllocationIds=[eip2.allocation_id]))
len(addresses).should.be.equal(1)
addresses[0].public_ip.should.equal(eip2.public_ip)
inst2.public_ip_address.should.equal(addresses[0].public_ip)
# Param search by PublicIp
addresses = list(service.vpc_addresses.filter(PublicIps=[eip3.public_ip]))
len(addresses).should.be.equal(1)
addresses[0].public_ip.should.equal(eip3.public_ip)
inst3.public_ip_address.should.equal(addresses[0].public_ip)
# Param search by Filter
def check_vpc_filter_valid(filter_name, filter_values):
addresses = list(service.vpc_addresses.filter(
Filters=[{'Name': filter_name,
'Values': filter_values}]))
len(addresses).should.equal(2)
ips = [addr.public_ip for addr in addresses]
set(ips).should.equal(set([eip1.public_ip, eip2.public_ip]))
ips.should.contain(inst1.public_ip_address)
def check_vpc_filter_invalid(filter_name):
addresses = list(service.vpc_addresses.filter(
Filters=[{'Name': filter_name,
'Values': ['dummy1', 'dummy2']}]))
len(addresses).should.equal(0)
def check_vpc_filter(filter_name, filter_values):
check_vpc_filter_valid(filter_name, filter_values)
check_vpc_filter_invalid(filter_name)
check_vpc_filter('allocation-id', [eip1.allocation_id, eip2.allocation_id])
check_vpc_filter('association-id', [eip1.association_id, eip2.association_id])
check_vpc_filter('instance-id', [inst1.id, inst2.id])
check_vpc_filter(
'network-interface-id',
[inst1.network_interfaces_attribute[0].get('NetworkInterfaceId'),
inst2.network_interfaces_attribute[0].get('NetworkInterfaceId')])
check_vpc_filter(
'private-ip-address',
[inst1.network_interfaces_attribute[0].get('PrivateIpAddress'),
inst2.network_interfaces_attribute[0].get('PrivateIpAddress')])
check_vpc_filter('public-ip', [inst1.public_ip_address, inst2.public_ip_address])
# all the ips are in a VPC
addresses = list(service.vpc_addresses.filter(
Filters=[{'Name': 'domain', 'Values': ['vpc']}]))
len(addresses).should.equal(3)