Merge pull request #382 from IlyaSukhanov/cloudwatch_alarm_filters

CloudWatch DescribeAlarm filters.

moto/cloudwatch/models.py

@@ -51,6 +51,42 @@ class CloudWatchBackend(BaseBackend):
     def get_all_alarms(self):
         return self.alarms.values()
 
+    @staticmethod
+    def _list_element_starts_with(items, needle):
+        """True of any of the list elements starts with needle"""
+        for item in items:
+            if item.startswith(needle):
+                return True
+        return False
+
+    def get_alarms_by_action_prefix(self, action_prefix):
+        return [
+            alarm
+            for alarm in self.alarms.values()
+            if CloudWatchBackend._list_element_starts_with(
+                alarm.alarm_actions, action_prefix
+            )
+        ]
+
+    def get_alarms_by_alarm_name_prefix(self, name_prefix):
+        return [
+            alarm
+            for alarm in self.alarms.values()
+            if alarm.name.startswith(name_prefix)
+        ]
+
+    def get_alarms_by_alarm_names(self, alarm_names):
+        return [
+            alarm
+            for alarm in self.alarms.values()
+            if alarm.name in alarm_names
+        ]
+
+    def get_alarms_by_state_value(self, state):
+        raise NotImplementedError(
+            "DescribeAlarm by state is not implemented in moto."
+        )
+
     def delete_alarms(self, alarm_names):
         for alarm_name in alarm_names:
             self.alarms.pop(alarm_name, None)

moto/cloudwatch/responses.py

@@ -1,5 +1,6 @@
 from moto.core.responses import BaseResponse
 from .models import cloudwatch_backend
+import logging
 
 
 class CloudWatchResponse(BaseResponse):
@@ -28,7 +29,22 @@ class CloudWatchResponse(BaseResponse):
         return template.render(alarm=alarm)
 
     def describe_alarms(self):
+        action_prefix = self._get_param('ActionPrefix')
+        alarm_name_prefix = self._get_param('AlarmNamePrefix')
+        alarm_names = self._get_multi_param('AlarmNames.member')
+        state_value = self._get_param('StateValue')
+
+        if action_prefix:
+            alarms = cloudwatch_backend.get_alarms_by_action_prefix(action_prefix)
+        elif alarm_name_prefix:
+            alarms = cloudwatch_backend.get_alarms_by_alarm_name_prefix(alarm_name_prefix)
+        elif alarm_names:
+            alarms = cloudwatch_backend.get_alarms_by_alarm_names(alarm_names)
+        elif state_value:
+            alarms = cloudwatch_backend.get_alarms_by_state_value(state_value)
+        else :
             alarms = cloudwatch_backend.get_all_alarms()
+
         template = self.response_template(DESCRIBE_ALARMS_TEMPLATE)
         return template.render(alarms=alarms)
 

setup.py

@@ -21,7 +21,7 @@ if sys.version_info < (2, 7):
 
 setup(
     name='moto',
-    version='0.4.6',
+    version='0.4.7',
     description='A library that allows your python tests to easily'
                 ' mock out the boto library',
     author='Steve Pulec',

tests/test_cloudwatch/test_cloudwatch.py

@@ -4,13 +4,10 @@ import sure  # noqa
 
 from moto import mock_cloudwatch
 
-
+def alarm_fixture(name="tester", action=None):
-@mock_cloudwatch
+    action = action or ['arn:alarm']
-def test_create_alarm():
+    return MetricAlarm(
-    conn = boto.connect_cloudwatch()
+        name=name,
-
-    alarm = MetricAlarm(
-        name='tester',
         comparison='>=',
         threshold=2.0,
         period=60,
@@ -18,11 +15,17 @@ def test_create_alarm():
         statistic='Average',
         description='A test',
         dimensions={'InstanceId': ['i-0123456,i-0123457']},
-        alarm_actions=['arn:alarm'],
+        alarm_actions=action,
         ok_actions=['arn:ok'],
         insufficient_data_actions=['arn:insufficient'],
         unit='Seconds',
     )
+
+@mock_cloudwatch
+def test_create_alarm():
+    conn = boto.connect_cloudwatch()
+
+    alarm = alarm_fixture()
     conn.create_alarm(alarm)
 
     alarms = conn.describe_alarms()
@@ -46,20 +49,10 @@ def test_create_alarm():
 def test_delete_alarm():
     conn = boto.connect_cloudwatch()
 
-    alarm = MetricAlarm(
+    alarms = conn.describe_alarms()
-        name='tester',
+    alarms.should.have.length_of(0)
-        comparison='>=',
+
-        threshold=2.0,
+    alarm = alarm_fixture()
-        period=60,
-        evaluation_periods=5,
-        statistic='Average',
-        description='A test',
-        dimensions={'InstanceId': ['i-0123456,i-0123457']},
-        alarm_actions=['arn:alarm'],
-        ok_actions=['arn:ok'],
-        insufficient_data_actions=['arn:insufficient'],
-        unit='Seconds',
-    )
     conn.create_alarm(alarm)
 
     alarms = conn.describe_alarms()
@@ -88,3 +81,39 @@ def test_put_metric_data():
     metric.namespace.should.equal('tester')
     metric.name.should.equal('metric')
     dict(metric.dimensions).should.equal({'InstanceId': ['i-0123456,i-0123457']})
+
+
+@mock_cloudwatch
+def test_describe_alarms():
+    conn = boto.connect_cloudwatch()
+
+    alarms = conn.describe_alarms()
+    alarms.should.have.length_of(0)
+
+    conn.create_alarm(alarm_fixture(name="nfoobar", action="afoobar"))
+    conn.create_alarm(alarm_fixture(name="nfoobaz", action="afoobaz"))
+    conn.create_alarm(alarm_fixture(name="nbarfoo", action="abarfoo"))
+    conn.create_alarm(alarm_fixture(name="nbazfoo", action="abazfoo"))
+
+    alarms = conn.describe_alarms()
+    alarms.should.have.length_of(4)
+    alarms = conn.describe_alarms(alarm_name_prefix="nfoo")
+    alarms.should.have.length_of(2)
+    alarms = conn.describe_alarms(alarm_names=["nfoobar", "nbarfoo", "nbazfoo"])
+    alarms.should.have.length_of(3)
+    alarms = conn.describe_alarms(action_prefix="afoo")
+    alarms.should.have.length_of(2)
+
+    for alarm in conn.describe_alarms():
+        alarm.delete()
+
+    alarms = conn.describe_alarms()
+    alarms.should.have.length_of(0)
+
+@mock_cloudwatch
+def test_describe_state_value_unimplemented():
+    conn = boto.connect_cloudwatch()
+
+    conn.describe_alarms()
+    conn.describe_alarms.when.called_with(state_value="foo").should.throw(NotImplementedError)
+