CloudWatch: Improve typing/docs (getmoto#7560)

kpiekarczyk-hbu · Apr 4, 2024 · 7289a6c · 7289a6c
1 parent 65f88e7
commit 7289a6c
Show file tree

Hide file tree

Showing 5 changed files with 35 additions and 37 deletions.
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
@@ -1039,15 +1039,15 @@
 
 ## cloudwatch
 <details>
-<summary>36% implemented</summary>
+<summary>39% implemented</summary>
 
 - [X] delete_alarms
 - [ ] delete_anomaly_detector
 - [X] delete_dashboards
 - [ ] delete_insight_rules
 - [ ] delete_metric_stream
 - [ ] describe_alarm_history
-- [ ] describe_alarms
+- [X] describe_alarms
 - [ ] describe_alarms_for_metric
 - [ ] describe_anomaly_detectors
 - [ ] describe_insight_rules

diff --git a/docs/docs/services/cloudwatch.rst b/docs/docs/services/cloudwatch.rst
@@ -20,7 +20,7 @@ cloudwatch
 - [ ] delete_insight_rules
 - [ ] delete_metric_stream
 - [ ] describe_alarm_history
-- [ ] describe_alarms
+- [X] describe_alarms
 - [ ] describe_alarms_for_metric
 - [ ] describe_anomaly_detectors
 - [ ] describe_insight_rules

diff --git a/moto/cloudwatch/models.py b/moto/cloudwatch/models.py
@@ -119,32 +119,32 @@ def __init__(
         name: str,
         namespace: str,
         metric_name: str,
-        metric_data_queries: List[MetricDataQuery],
+        metric_data_queries: Optional[List[MetricDataQuery]],
         comparison_operator: str,
         evaluation_periods: int,
-        datapoints_to_alarm: int,
+        datapoints_to_alarm: Optional[int],
         period: int,
         threshold: float,
         statistic: str,
-        extended_statistic: str,
+        extended_statistic: Optional[str],
         description: str,
         dimensions: List[Dict[str, str]],
         alarm_actions: List[str],
-        ok_actions: List[str],
-        insufficient_data_actions: List[str],
-        unit: str,
+        ok_actions: Optional[List[str]],
+        insufficient_data_actions: Optional[List[str]],
+        unit: Optional[str],
         actions_enabled: bool,
-        treat_missing_data: str,
-        evaluate_low_sample_count_percentile: str,
-        threshold_metric_id: str,
-        rule: str,
+        treat_missing_data: Optional[str],
+        evaluate_low_sample_count_percentile: Optional[str],
+        threshold_metric_id: Optional[str],
+        rule: Optional[str],
     ):
         self.region_name = region_name
         self.name = name
         self.alarm_arn = make_arn_for_alarm(region_name, account_id, name)
         self.namespace = namespace
         self.metric_name = metric_name
-        self.metric_data_queries = metric_data_queries
+        self.metric_data_queries = metric_data_queries or []
         self.comparison_operator = comparison_operator
         self.evaluation_periods = evaluation_periods
         self.datapoints_to_alarm = datapoints_to_alarm
@@ -158,8 +158,8 @@ def __init__(
         ]
         self.actions_enabled = True if actions_enabled is None else actions_enabled
         self.alarm_actions = alarm_actions
-        self.ok_actions = ok_actions
-        self.insufficient_data_actions = insufficient_data_actions
+        self.ok_actions = ok_actions or []
+        self.insufficient_data_actions = insufficient_data_actions or []
         self.unit = unit
         self.configuration_updated_timestamp = iso_8601_datetime_with_nanoseconds()
         self.treat_missing_data = treat_missing_data
@@ -457,26 +457,26 @@ def put_metric_alarm(
         name: str,
         namespace: str,
         metric_name: str,
-        metric_data_queries: List[MetricDataQuery],
         comparison_operator: str,
         evaluation_periods: int,
-        datapoints_to_alarm: int,
         period: int,
         threshold: float,
         statistic: str,
-        extended_statistic: str,
         description: str,
         dimensions: List[Dict[str, str]],
         alarm_actions: List[str],
-        ok_actions: List[str],
-        insufficient_data_actions: List[str],
-        unit: str,
-        actions_enabled: bool,
-        treat_missing_data: str,
-        evaluate_low_sample_count_percentile: str,
-        threshold_metric_id: str,
-        rule: str,
-        tags: List[Dict[str, str]],
+        metric_data_queries: Optional[List[MetricDataQuery]] = None,
+        datapoints_to_alarm: Optional[int] = None,
+        extended_statistic: Optional[str] = None,
+        ok_actions: Optional[List[str]] = None,
+        insufficient_data_actions: Optional[List[str]] = None,
+        unit: Optional[str] = None,
+        actions_enabled: bool = True,
+        treat_missing_data: Optional[str] = None,
+        evaluate_low_sample_count_percentile: Optional[str] = None,
+        threshold_metric_id: Optional[str] = None,
+        rule: Optional[str] = None,
+        tags: Optional[List[Dict[str, str]]] = None,
     ) -> FakeAlarm:
         if extended_statistic and not extended_statistic.startswith("p"):
             raise InvalidParameterValue(
@@ -519,11 +519,12 @@ def put_metric_alarm(
         )
 
         self.alarms[name] = alarm
-        self.tagger.tag_resource(alarm.alarm_arn, tags)
+        if tags:
+            self.tagger.tag_resource(alarm.alarm_arn, tags)
 
         return alarm
 
-    def get_all_alarms(self) -> Iterable[FakeAlarm]:
+    def describe_alarms(self) -> Iterable[FakeAlarm]:
         return self.alarms.values()
 
     @staticmethod
@@ -889,7 +890,7 @@ def list_tags_for_resource(self, arn: str) -> Dict[str, str]:
     def tag_resource(self, arn: str, tags: List[Dict[str, str]]) -> None:
         # From boto3:
         # Currently, the only CloudWatch resources that can be tagged are alarms and Contributor Insights rules.
-        all_arns = [alarm.alarm_arn for alarm in self.get_all_alarms()]
+        all_arns = [alarm.alarm_arn for alarm in self.describe_alarms()]
         if arn not in all_arns:
             raise ResourceNotFoundException
 

diff --git a/moto/cloudwatch/responses.py b/moto/cloudwatch/responses.py
@@ -147,7 +147,7 @@ def describe_alarms(self) -> str:
         elif state_value:
             alarms = self.cloudwatch_backend.get_alarms_by_state_value(state_value)
         else:
-            alarms = self.cloudwatch_backend.get_all_alarms()
+            alarms = self.cloudwatch_backend.describe_alarms()
 
         metric_alarms = [a for a in alarms if a.rule is None]
         composite_alarms = [a for a in alarms if a.rule is not None]
@@ -244,9 +244,6 @@ def delete_dashboards(self) -> Union[str, ERROR_RESPONSE]:
         template = self.response_template(DELETE_DASHBOARD_TEMPLATE)
         return template.render()
 
-    def describe_alarm_history(self) -> None:
-        raise NotImplementedError()
-
     @staticmethod
     def filter_alarms(
         alarms: Iterable[FakeAlarm], metric_name: str, namespace: str
@@ -259,7 +256,7 @@ def filter_alarms(
         return metric_filtered_alarms
 
     def describe_alarms_for_metric(self) -> str:
-        alarms = self.cloudwatch_backend.get_all_alarms()
+        alarms = self.cloudwatch_backend.describe_alarms()
         namespace = self._get_param("Namespace")
         metric_name = self._get_param("MetricName")
         filtered_alarms = self.filter_alarms(alarms, metric_name, namespace)

diff --git a/tests/test_cloudwatch/test_cloudwatch_alarms.py b/tests/test_cloudwatch/test_cloudwatch_alarms.py
@@ -181,7 +181,7 @@ def test_describe_alarms():
     ][0]
 
     assert single_metric_alarm["MetricName"] == "cpu"
-    assert "Metrics" not in single_metric_alarm
+    assert single_metric_alarm["Metrics"] == []
     assert single_metric_alarm["Namespace"] == "blah"
     assert single_metric_alarm["Period"] == 10
     assert single_metric_alarm["EvaluationPeriods"] == 5