CloudWatch DescribeAlarm filters.

Adds support for filtering by:
action_prefix
alarm_name_prefix
alarm_names

And throw NotImplementedError when filtering by:
state_value
This commit is contained in:
Ilya Sukhanov 2015-07-22 21:08:39 -04:00
commit 9bf5c2e706
3 changed files with 104 additions and 23 deletions

View file

@ -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)

View file

@ -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):
alarms = cloudwatch_backend.get_all_alarms()
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)