Filtering implementation for DHCP options sets

moto/ec2/models.py

@@ -2599,6 +2599,34 @@ class DHCPOptionsSet(TaggedEC2Resource):
         self.id = random_dhcp_option_id()
         self.vpc = None
 
+    def get_filter_value(self, filter_name):
+        """
+        API Version 2015-10-01 defines the following filters for DescribeDhcpOptions:
+
+        * dhcp-options-id
+        * key
+        * value
+        * tag:key=value
+        * tag-key
+        * tag-value
+
+        Taken from: http://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeDhcpOptions.html
+        """
+        if filter_name == 'dhcp-options-id':
+            return self.id
+        elif filter_name == 'key':
+            return list(self._options.keys())
+        elif filter_name == 'value':
+            values = [item for item in list(self._options.values()) if item]
+            return itertools.chain(*values)
+
+        filter_value = super(DHCPOptionsSet, self).get_filter_value(filter_name)
+
+        if filter_value is None:
+            self.ec2_backend.raise_not_implemented_error("The filter '{0}' for DescribeDhcpOptions".format(filter_name))
+
+        return filter_value
+
     @property
     def options(self):
         return self._options

moto/ec2/responses/dhcp_options.py

@@ -49,9 +49,8 @@ class DHCPOptions(BaseResponse):
 
     def describe_dhcp_options(self):
         dhcp_opt_ids = sequence_from_querystring("DhcpOptionsId", self.querystring)
-        if filters_from_querystring(self.querystring):
-            raise NotImplementedError("Filtering not supported in describe_dhcp_options.")
-        dhcp_opts = self.ec2_backend.get_all_dhcp_options(dhcp_opt_ids, None)
+        filters = filters_from_querystring(self.querystring)
+        dhcp_opts = self.ec2_backend.get_all_dhcp_options(dhcp_opt_ids, filters)
         template = self.response_template(DESCRIBE_DHCP_OPTIONS_RESPONSE)
         return template.render(dhcp_options=dhcp_opts)
 

tests/test_ec2/test_dhcp_options.py

@@ -3,6 +3,7 @@ from __future__ import unicode_literals
 import tests.backport_assert_raises
 from nose.tools import assert_raises
 
+import boto3
 import boto
 from boto.exception import EC2ResponseError
 
@@ -191,3 +192,127 @@ def test_dhcp_tagging():
     dhcp_option = conn.get_all_dhcp_options()[0]
     dhcp_option.tags.should.have.length_of(1)
     dhcp_option.tags["a key"].should.equal("some value")
+
+
+@mock_ec2
+def test_dhcp_options_get_by_tag():
+    conn = boto.connect_vpc('the_key', 'the_secret')
+
+    dhcp1 = conn.create_dhcp_options('example.com', ['10.0.10.2'])
+    dhcp1.add_tag('Name', 'TestDhcpOptions1')
+    dhcp1.add_tag('test-tag', 'test-value')
+
+    dhcp2 = conn.create_dhcp_options('example.com', ['10.0.20.2'])
+    dhcp2.add_tag('Name', 'TestDhcpOptions2')
+    dhcp2.add_tag('test-tag', 'test-value')
+
+    filters = {'tag:Name': 'TestDhcpOptions1', 'tag:test-tag': 'test-value'}
+    dhcp_options_sets = conn.get_all_dhcp_options(filters=filters)
+
+    dhcp_options_sets.should.have.length_of(1)
+    dhcp_options_sets[0].options['domain-name'][0].should.be.equal('example.com')
+    dhcp_options_sets[0].options['domain-name-servers'][0].should.be.equal('10.0.10.2')
+    dhcp_options_sets[0].tags['Name'].should.equal('TestDhcpOptions1')
+    dhcp_options_sets[0].tags['test-tag'].should.equal('test-value')
+
+    filters = {'tag:Name': 'TestDhcpOptions2', 'tag:test-tag': 'test-value'}
+    dhcp_options_sets = conn.get_all_dhcp_options(filters=filters)
+
+    dhcp_options_sets.should.have.length_of(1)
+    dhcp_options_sets[0].options['domain-name'][0].should.be.equal('example.com')
+    dhcp_options_sets[0].options['domain-name-servers'][0].should.be.equal('10.0.20.2')
+    dhcp_options_sets[0].tags['Name'].should.equal('TestDhcpOptions2')
+    dhcp_options_sets[0].tags['test-tag'].should.equal('test-value')
+
+    filters = {'tag:test-tag': 'test-value'}
+    dhcp_options_sets = conn.get_all_dhcp_options(filters=filters)
+
+    dhcp_options_sets.should.have.length_of(2)
+
+
+@mock_ec2
+def test_dhcp_options_get_by_id():
+    conn = boto.connect_vpc('the_key', 'the_secret')
+
+    dhcp1 = conn.create_dhcp_options('test1.com', ['10.0.10.2'])
+    dhcp1.add_tag('Name', 'TestDhcpOptions1')
+    dhcp1.add_tag('test-tag', 'test-value')
+    dhcp1_id = dhcp1.id
+
+    dhcp2 = conn.create_dhcp_options('test2.com', ['10.0.20.2'])
+    dhcp2.add_tag('Name', 'TestDhcpOptions2')
+    dhcp2.add_tag('test-tag', 'test-value')
+    dhcp2_id = dhcp2.id
+
+    dhcp_options_sets = conn.get_all_dhcp_options()
+    dhcp_options_sets.should.have.length_of(2)
+
+    dhcp_options_sets = conn.get_all_dhcp_options(filters={'dhcp-options-id': dhcp1_id})
+
+    dhcp_options_sets.should.have.length_of(1)
+    dhcp_options_sets[0].options['domain-name'][0].should.be.equal('test1.com')
+    dhcp_options_sets[0].options['domain-name-servers'][0].should.be.equal('10.0.10.2')
+
+    dhcp_options_sets = conn.get_all_dhcp_options(filters={'dhcp-options-id': dhcp2_id})
+
+    dhcp_options_sets.should.have.length_of(1)
+    dhcp_options_sets[0].options['domain-name'][0].should.be.equal('test2.com')
+    dhcp_options_sets[0].options['domain-name-servers'][0].should.be.equal('10.0.20.2')
+
+
+@mock_ec2
+def test_dhcp_options_get_by_value_filter():
+    ec2 = boto3.resource('ec2', region_name='us-west-1')
+
+    ec2.create_dhcp_options(DhcpConfigurations=[
+        {'Key': 'domain-name', 'Values': ['example.com']},
+        {'Key': 'domain-name-servers', 'Values': ['10.0.10.2']}
+    ])
+
+    ec2.create_dhcp_options(DhcpConfigurations=[
+        {'Key': 'domain-name', 'Values': ['example.com']},
+        {'Key': 'domain-name-servers', 'Values': ['10.0.20.2']}
+    ])
+
+    ec2.create_dhcp_options(DhcpConfigurations=[
+        {'Key': 'domain-name', 'Values': ['example.com']},
+        {'Key': 'domain-name-servers', 'Values': ['10.0.30.2']}
+    ])
+
+    filters = [{'Name': 'value', 'Values': ['10.0.10.2']}]
+    dhcp_options_sets = list(ec2.dhcp_options_sets.filter(Filters=filters))
+    dhcp_options_sets.should.have.length_of(1)
+
+
+@mock_ec2
+def test_dhcp_options_get_by_key_filter():
+    ec2 = boto3.resource('ec2', region_name='us-west-1')
+
+    ec2.create_dhcp_options(DhcpConfigurations=[
+        {'Key': 'domain-name', 'Values': ['example.com']},
+        {'Key': 'domain-name-servers', 'Values': ['10.0.10.2']}
+    ])
+
+    ec2.create_dhcp_options(DhcpConfigurations=[
+        {'Key': 'domain-name', 'Values': ['example.com']},
+        {'Key': 'domain-name-servers', 'Values': ['10.0.20.2']}
+    ])
+
+    ec2.create_dhcp_options(DhcpConfigurations=[
+        {'Key': 'domain-name', 'Values': ['example.com']},
+        {'Key': 'domain-name-servers', 'Values': ['10.0.30.2']}
+    ])
+
+    filters = [{'Name': 'key', 'Values': ['domain-name']}]
+    dhcp_options_sets = list(ec2.dhcp_options_sets.filter(Filters=filters))
+    dhcp_options_sets.should.have.length_of(3)
+
+
+@mock_ec2
+def test_dhcp_options_get_by_invalid_filter():
+    conn = boto.connect_vpc('the_key', 'the_secret')
+
+    conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
+    filters = {'invalid-filter': 'invalid-value'}
+
+    conn.get_all_dhcp_options.when.called_with(filters=filters).should.throw(NotImplementedError)