Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #608 from yannlambret/add-filtering-to-network-interfaces

Browse Source 
Extend filtering support for elastic network interfaces
This commit is contained in:
Steve Pulec 2016-05-08 16:41:22 -04:00
commit 5415091d14
3 changed files with 152 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
moto/ec2/models.py
View File

@ -152,7 +152,7 @@ class TaggedEC2Resource(object):
return [tag['value'] for tag in tags]
class NetworkInterface(object):
class NetworkInterface(TaggedEC2Resource):
def __init__(self, ec2_backend, subnet, private_ip_address, device_index=0,
public_ip_auto_assign=True, group_ids=None):
self.ec2_backend = ec2_backend
@ -236,6 +236,27 @@ class NetworkInterface(object):
def physical_resource_id(self):
return self.id
def get_filter_value(self, filter_name):
if filter_name == 'network-interface-id':
return self.id
elif filter_name in ('addresses.private-ip-address', 'private-ip-address'):
return self.private_ip_address
elif filter_name == 'subnet-id':
return self.subnet.id
elif filter_name == 'vpc-id':
return self.subnet.vpc_id
elif filter_name == 'group-id':
return [group.id for group in self._group_set]
filter_value = super(NetworkInterface, self).get_filter_value(filter_name)
if filter_value is None:
self.ec2_backend.raise_not_implemented_error(
"The filter '{0}' for DescribeNetworkInterfaces".format(filter_name)
)
return filter_value
class NetworkInterfaceBackend(object):
def __init__(self):
@ -301,6 +322,18 @@ class NetworkInterfaceBackend(object):
group = self.get_security_group_from_id(group_id)
eni._group_set = [group]
def get_all_network_interfaces(self, eni_ids=None, filters=None):
enis = self.enis.values()
if eni_ids:
enis = [eni for eni in enis if eni.id in eni_ids]
if len(enis) != len(eni_ids):
invalid_id = list(set(eni_ids).difference(set([eni.id for eni in enis])))[0]
raise InvalidNetworkInterfaceIdError(invalid_id)
return generic_filter(filters, enis)
class Instance(BotoInstance, TaggedEC2Resource):
def __init__(self, ec2_backend, image_id, user_data, security_groups, **kwargs):

24
moto/ec2/responses/elastic_network_interfaces.py
View File

@ -23,14 +23,9 @@ class ElasticNetworkInterfaces(BaseResponse):
raise NotImplementedError('ElasticNetworkInterfaces(AmazonVPC).describe_network_interface_attribute is not yet implemented')
def describe_network_interfaces(self):
# Partially implemented. Supports only network-interface-id and group-id filters
eni_ids = sequence_from_querystring('NetworkInterfaceId', self.querystring)
filters = filters_from_querystring(self.querystring)
eni_ids = self._get_multi_param('NetworkInterfaceId.')
if 'network-interface-id' not in filters and eni_ids:
# Network interfaces can be filtered by passing the 'network-interface-id'
# filter or by passing the NetworkInterfaceId parameter
filters['network-interface-id'] = eni_ids
enis = self.ec2_backend.describe_network_interfaces(filters)
enis = self.ec2_backend.get_all_network_interfaces(eni_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_INTERFACES_RESPONSE)
return template.render(enis=enis)
@ -112,7 +107,11 @@ DESCRIBE_NETWORK_INTERFACES_RESPONSE = """<DescribeNetworkInterfacesResponse xml
<description>Primary network interface</description>
<ownerId>190610284047</ownerId>
<requesterManaged>false</requesterManaged>
<status>in-use</status>
{% if eni.attachment_id %}
<status>in-use</status>
{% else %}
<status>available</status>
{% endif %}
<macAddress>0e:a3:a7:7b:95:a7</macAddress>
{% if eni.private_ip_address %}
<privateIpAddress>{{ eni.private_ip_address }}</privateIpAddress>
@ -129,6 +128,14 @@ DESCRIBE_NETWORK_INTERFACES_RESPONSE = """<DescribeNetworkInterfacesResponse xml
</item>
{% endfor %}
</groupSet>
<tagSet>
{% for tag in eni.get_tags() %}
<item>
<key>{{ tag.key }}</key>
<value>{{ tag.value }}</value>
</item>
{% endfor %}
</tagSet>
{% if eni.instance %}
<attachment>
<attachmentId>{{ eni.attachment_id }}</attachmentId>
@ -145,7 +152,6 @@ DESCRIBE_NETWORK_INTERFACES_RESPONSE = """<DescribeNetworkInterfacesResponse xml
<publicDnsName>ec2-54-200-86-47.us-west-2.compute.amazonaws.com</publicDnsName>
<ipOwnerId>amazon</ipOwnerId>
</association>
<tagSet/>
{% if eni.private_ip_address %}
<privateIpAddressesSet>
<item>

104
tests/test_ec2/test_elastic_network_interfaces.py
View File

@ -3,6 +3,7 @@ from __future__ import unicode_literals
import tests.backport_assert_raises
from nose.tools import assert_raises
import boto3
import boto
import boto.cloudformation
import boto.ec2
@ -140,7 +141,7 @@ def test_elastic_network_interfaces_filtering():
set([eni.id for eni in enis_by_id]).should.equal(set([eni1.id]))
# Filter by Security Group
enis_by_group = conn.get_all_network_interfaces(filters={'group-id':security_group1.id})
enis_by_group = conn.get_all_network_interfaces(filters={'group-id': security_group1.id})
enis_by_group.should.have.length_of(2)
set([eni.id for eni in enis_by_group]).should.equal(set([eni1.id, eni2.id]))
@ -153,6 +154,107 @@ def test_elastic_network_interfaces_filtering():
conn.get_all_network_interfaces.when.called_with(filters={'not-implemented-filter': 'foobar'}).should.throw(NotImplementedError)
@mock_ec2
def test_elastic_network_interfaces_get_by_tag_name():
ec2 = boto3.resource('ec2', region_name='us-west-2')
ec2_client = boto3.client('ec2', region_name='us-west-2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-2a')
eni1 = ec2.create_network_interface(SubnetId=subnet.id, PrivateIpAddress='10.0.10.5')
eni1.create_tags(Tags=[{'Key': 'Name', 'Value': 'eni1'}])
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter('network_interface_available')
waiter.wait(NetworkInterfaceIds=[eni1.id])
filters = [{'Name': 'tag:Name', 'Values': ['eni1']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(1)
filters = [{'Name': 'tag:Name', 'Values': ['wrong-name']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(0)
@mock_ec2
def test_elastic_network_interfaces_get_by_private_ip():
ec2 = boto3.resource('ec2', region_name='us-west-2')
ec2_client = boto3.client('ec2', region_name='us-west-2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-2a')
eni1 = ec2.create_network_interface(SubnetId=subnet.id, PrivateIpAddress='10.0.10.5')
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter('network_interface_available')
waiter.wait(NetworkInterfaceIds=[eni1.id])
filters = [{'Name': 'private-ip-address', 'Values': ['10.0.10.5']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(1)
filters = [{'Name': 'private-ip-address', 'Values': ['10.0.10.10']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(0)
filters = [{'Name': 'addresses.private-ip-address', 'Values': ['10.0.10.5']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(1)
filters = [{'Name': 'addresses.private-ip-address', 'Values': ['10.0.10.10']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(0)
@mock_ec2
def test_elastic_network_interfaces_get_by_vpc_id():
ec2 = boto3.resource('ec2', region_name='us-west-2')
ec2_client = boto3.client('ec2', region_name='us-west-2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-2a')
eni1 = ec2.create_network_interface(SubnetId=subnet.id, PrivateIpAddress='10.0.10.5')
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter('network_interface_available')
waiter.wait(NetworkInterfaceIds=[eni1.id])
filters = [{'Name': 'vpc-id', 'Values': [subnet.vpc_id]}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(1)
filters = [{'Name': 'vpc-id', 'Values': ['vpc-aaaa1111']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(0)
@mock_ec2
def test_elastic_network_interfaces_get_by_subnet_id():
ec2 = boto3.resource('ec2', region_name='us-west-2')
ec2_client = boto3.client('ec2', region_name='us-west-2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-2a')
eni1 = ec2.create_network_interface(SubnetId=subnet.id, PrivateIpAddress='10.0.10.5')
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter('network_interface_available')
waiter.wait(NetworkInterfaceIds=[eni1.id])
filters = [{'Name': 'subnet-id', 'Values': [subnet.id]}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(1)
filters = [{'Name': 'subnet-id', 'Values': ['subnet-aaaa1111']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(0)
@mock_ec2
@mock_cloudformation
def test_elastic_network_interfaces_cloudformation():