Merge pull request #170 from DreadPirateShawn/DescribeSubnetsFiltering

DescribeSubnets: Added support for 'filters' parameter.

moto/ec2/models.py

@@ -880,6 +880,19 @@ class Subnet(TaggedEC2Instance):
     def physical_resource_id(self):
         return self.id
 
+    def get_filter_value(self, filter_name):
+        if filter_name in ['cidr', 'cidrBlock', 'cidr-block']:
+            return self.cidr_block
+        elif filter_name == 'vpc-id':
+            return self.vpc_id
+        elif filter_name == 'subnet-id':
+            return self.id
+        else:
+            msg = "The filter '{0}' for DescribeSubnets has not been" \
+                  " implemented in Moto yet. Feel free to open an issue at" \
+                  " https://github.com/spulec/moto/issues".format(filter_name)
+            raise NotImplementedError(msg)
+
 
 class SubnetBackend(object):
     def __init__(self):
@@ -892,8 +905,14 @@ class SubnetBackend(object):
         self.subnets[subnet_id] = subnet
         return subnet
 
-    def get_all_subnets(self):
-        return self.subnets.values()
+    def get_all_subnets(self, filters=None):
+        subnets = self.subnets.values()
+
+        if filters:
+            for (_filter, _filter_value) in filters.iteritems():
+                subnets = [ subnet for subnet in subnets if subnet.get_filter_value(_filter) in _filter_value ]
+
+        return subnets
 
     def delete_subnet(self, subnet_id):
         deleted = self.subnets.pop(subnet_id, None)

moto/ec2/responses/subnets.py

@@ -2,6 +2,7 @@ from jinja2 import Template
 
 from moto.core.responses import BaseResponse
 from moto.ec2.models import ec2_backend
+from moto.ec2.utils import filters_from_querystring
 
 
 class Subnets(BaseResponse):
@@ -19,7 +20,8 @@ class Subnets(BaseResponse):
         return template.render(subnet=subnet)
 
     def describe_subnets(self):
-        subnets = ec2_backend.get_all_subnets()
+        filters = filters_from_querystring(self.querystring)
+        subnets = ec2_backend.get_all_subnets(filters)
         template = Template(DESCRIBE_SUBNETS_RESPONSE)
         return template.render(subnets=subnets)
 

tests/test_ec2/test_subnets.py

@@ -46,3 +46,47 @@ def test_subnet_tagging():
     subnet = conn.get_all_subnets()[0]
     subnet.tags.should.have.length_of(1)
     subnet.tags["a key"].should.equal("some value")
+
+
+@mock_ec2
+def test_get_subnets_filtering():
+    conn = boto.connect_vpc('the_key', 'the_secret')
+    vpcA = conn.create_vpc("10.0.0.0/16")
+    subnetA = conn.create_subnet(vpcA.id, "10.0.0.0/24")
+    vpcB = conn.create_vpc("10.0.0.0/16")
+    subnetB1 = conn.create_subnet(vpcB.id, "10.0.0.0/24")
+    subnetB2 = conn.create_subnet(vpcB.id, "10.0.1.0/24")
+
+    all_subnets = conn.get_all_subnets()
+    all_subnets.should.have.length_of(3)
+
+    # Filter by VPC ID
+    subnets_by_vpc = conn.get_all_subnets(filters={'vpc-id': vpcB.id})
+    subnets_by_vpc.should.have.length_of(2)
+    set([subnet.id for subnet in subnets_by_vpc]).should.equal(set([subnetB1.id,subnetB2.id]))
+
+    # Filter by CIDR variations
+    subnets_by_cidr1 = conn.get_all_subnets(filters={'cidr': "10.0.0.0/24"})
+    subnets_by_cidr1.should.have.length_of(2)
+    set([subnet.id for subnet in subnets_by_cidr1]).should.equal(set([subnetA.id,subnetB1.id]))
+
+    subnets_by_cidr2 = conn.get_all_subnets(filters={'cidr-block': "10.0.0.0/24"})
+    subnets_by_cidr2.should.have.length_of(2)
+    set([subnet.id for subnet in subnets_by_cidr2]).should.equal(set([subnetA.id,subnetB1.id]))
+
+    subnets_by_cidr3 = conn.get_all_subnets(filters={'cidrBlock': "10.0.0.0/24"})
+    subnets_by_cidr3.should.have.length_of(2)
+    set([subnet.id for subnet in subnets_by_cidr3]).should.equal(set([subnetA.id,subnetB1.id]))
+
+    # Filter by VPC ID and CIDR
+    subnets_by_vpc_and_cidr = conn.get_all_subnets(filters={'vpc-id': vpcB.id, 'cidr': "10.0.0.0/24"})
+    subnets_by_vpc_and_cidr.should.have.length_of(1)
+    set([subnet.id for subnet in subnets_by_vpc_and_cidr]).should.equal(set([subnetB1.id]))
+
+    # Filter by subnet ID
+    subnets_by_id = conn.get_all_subnets(filters={'subnet-id': subnetA.id})
+    subnets_by_id.should.have.length_of(1)
+    set([subnet.id for subnet in subnets_by_id]).should.equal(set([subnetA.id]))
+
+    # Unsupported filter
+    conn.get_all_subnets.when.called_with(filters={'not-implemented-filter': 'foobar'}).should.throw(NotImplementedError)