Fix : Added implementation for CloudWatch Describe Metric for Alarm (#3148)

* Fix : added implementation for CloudWatch Describe Metric for Alarm

* Linting

Co-authored-by: Bert Blommers <info@bertblommers.nl>
This commit is contained in:
usmangani1 2020-07-18 15:17:53 +05:30 committed by GitHub
parent bed769a387
commit 09b764148c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 100 additions and 1 deletions

View File

@ -161,9 +161,23 @@ class CloudWatchResponse(BaseResponse):
def describe_alarm_history(self):
raise NotImplementedError()
@staticmethod
def filter_alarms(alarms, metric_name, namespace):
metric_filtered_alarms = []
for alarm in alarms:
if alarm.metric_name == metric_name and alarm.namespace == namespace:
metric_filtered_alarms.append(alarm)
return metric_filtered_alarms
@amzn_request_id
def describe_alarms_for_metric(self):
raise NotImplementedError()
alarms = self.cloudwatch_backend.get_all_alarms()
namespace = self._get_param("Namespace")
metric_name = self._get_param("MetricName")
filtered_alarms = self.filter_alarms(alarms, metric_name, namespace)
template = self.response_template(DESCRIBE_METRIC_ALARMS_TEMPLATE)
return template.render(alarms=filtered_alarms)
@amzn_request_id
def disable_alarm_actions(self):
@ -282,6 +296,57 @@ DESCRIBE_ALARMS_TEMPLATE = """<DescribeAlarmsResponse xmlns="http://monitoring.a
</DescribeAlarmsResult>
</DescribeAlarmsResponse>"""
DESCRIBE_METRIC_ALARMS_TEMPLATE = """<DescribeAlarmsForMetricResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<DescribeAlarmsForMetricResult>
<MetricAlarms>
{% for alarm in alarms %}
<member>
<ActionsEnabled>{{ alarm.actions_enabled }}</ActionsEnabled>
<AlarmActions>
{% for action in alarm.alarm_actions %}
<member>{{ action }}</member>
{% endfor %}
</AlarmActions>
<AlarmArn>{{ alarm.arn }}</AlarmArn>
<AlarmConfigurationUpdatedTimestamp>{{ alarm.configuration_updated_timestamp }}</AlarmConfigurationUpdatedTimestamp>
<AlarmDescription>{{ alarm.description }}</AlarmDescription>
<AlarmName>{{ alarm.name }}</AlarmName>
<ComparisonOperator>{{ alarm.comparison_operator }}</ComparisonOperator>
<Dimensions>
{% for dimension in alarm.dimensions %}
<member>
<Name>{{ dimension.name }}</Name>
<Value>{{ dimension.value }}</Value>
</member>
{% endfor %}
</Dimensions>
<EvaluationPeriods>{{ alarm.evaluation_periods }}</EvaluationPeriods>
<InsufficientDataActions>
{% for action in alarm.insufficient_data_actions %}
<member>{{ action }}</member>
{% endfor %}
</InsufficientDataActions>
<MetricName>{{ alarm.metric_name }}</MetricName>
<Namespace>{{ alarm.namespace }}</Namespace>
<OKActions>
{% for action in alarm.ok_actions %}
<member>{{ action }}</member>
{% endfor %}
</OKActions>
<Period>{{ alarm.period }}</Period>
<StateReason>{{ alarm.state_reason }}</StateReason>
<StateReasonData>{{ alarm.state_reason_data }}</StateReasonData>
<StateUpdatedTimestamp>{{ alarm.state_updated_timestamp }}</StateUpdatedTimestamp>
<StateValue>{{ alarm.state_value }}</StateValue>
<Statistic>{{ alarm.statistic }}</Statistic>
<Threshold>{{ alarm.threshold }}</Threshold>
<Unit>{{ alarm.unit }}</Unit>
</member>
{% endfor %}
</MetricAlarms>
</DescribeAlarmsForMetricResult>
</DescribeAlarmsForMetricResponse>"""
DELETE_METRIC_ALARMS_TEMPLATE = """<DeleteMetricAlarmResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<ResponseMetadata>
<RequestId>

View File

@ -127,6 +127,22 @@ def test_describe_alarms():
alarms.should.have.length_of(0)
@mock_cloudwatch_deprecated
def test_describe_alarms_for_metric():
conn = boto.connect_cloudwatch()
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_for_metric("nbarfoo_metric", "nbarfoo_namespace")
alarms.should.have.length_of(1)
alarms = conn.describe_alarms_for_metric("nbazfoo_metric", "nbazfoo_namespace")
alarms.should.have.length_of(1)
@mock_cloudwatch_deprecated
def test_get_metric_statistics():
conn = boto.connect_cloudwatch()

View File

@ -123,6 +123,24 @@ def test_delete_invalid_alarm():
e.exception.response["Error"]["Code"].should.equal("ResourceNotFound")
@mock_cloudwatch
def test_describe_alarms_for_metric():
conn = boto3.client("cloudwatch", region_name="eu-central-1")
conn.put_metric_alarm(
AlarmName="testalarm1",
MetricName="cpu",
Namespace="blah",
Period=10,
EvaluationPeriods=5,
Statistic="Average",
Threshold=2,
ComparisonOperator="GreaterThanThreshold",
ActionsEnabled=True,
)
alarms = conn.describe_alarms_for_metric(MetricName="cpu", Namespace="blah")
alarms.get("MetricAlarms").should.have.length_of(1)
@mock_cloudwatch
def test_alarm_state():
client = boto3.client("cloudwatch", region_name="eu-central-1")