diff --git a/moto/cloudwatch/models.py b/moto/cloudwatch/models.py index 635d86d40..612c39ce8 100644 --- a/moto/cloudwatch/models.py +++ b/moto/cloudwatch/models.py @@ -438,6 +438,12 @@ class CloudWatchBackend(BaseBackend): self.alarms.pop(alarm_name, None) def put_metric_data(self, namespace, metric_data): + for i, metric in enumerate(metric_data): + if metric.get("Value") == "NaN": + raise InvalidParameterValue( + f"The value NaN for parameter MetricData.member.{i + 1}.Value is invalid." + ) + for metric_member in metric_data: # Preserve "datetime" for get_metric_statistics comparisons timestamp = metric_member.get("Timestamp") diff --git a/tests/test_cloudwatch/test_cloudwatch_alarms.py b/tests/test_cloudwatch/test_cloudwatch_alarms.py new file mode 100644 index 000000000..503872833 --- /dev/null +++ b/tests/test_cloudwatch/test_cloudwatch_alarms.py @@ -0,0 +1,250 @@ +import boto3 +import sure # noqa + +from moto import mock_cloudwatch +from moto.core import ACCOUNT_ID + + +@mock_cloudwatch +def test_create_alarm(): + region = "eu-west-1" + cloudwatch = boto3.client("cloudwatch", region) + + name = "tester" + cloudwatch.put_metric_alarm( + AlarmActions=["arn:alarm"], + AlarmDescription="A test", + AlarmName=name, + ComparisonOperator="GreaterThanOrEqualToThreshold", + Dimensions=[{"Name": "InstanceId", "Value": "i-0123457"}], + EvaluationPeriods=5, + InsufficientDataActions=["arn:insufficient"], + Namespace="{0}_namespace".format(name), + MetricName="{0}_metric".format(name), + OKActions=["arn:ok"], + Period=60, + Statistic="Average", + Threshold=2, + Unit="Seconds", + ) + + alarms = cloudwatch.describe_alarms()["MetricAlarms"] + alarms.should.have.length_of(1) + alarm = alarms[0] + alarm.should.have.key("AlarmName").equal("tester") + alarm.should.have.key("Namespace").equal("tester_namespace") + alarm.should.have.key("MetricName").equal("tester_metric") + alarm.should.have.key("ComparisonOperator").equal("GreaterThanOrEqualToThreshold") + alarm.should.have.key("Threshold").equal(2.0) + alarm.should.have.key("Period").equal(60) + alarm.should.have.key("EvaluationPeriods").equal(5) + alarm.should.have.key("Statistic").should.equal("Average") + alarm.should.have.key("AlarmDescription").equal("A test") + alarm.should.have.key("Dimensions").equal( + [{"Name": "InstanceId", "Value": "i-0123457"}] + ) + alarm.should.have.key("AlarmActions").equal(["arn:alarm"]) + alarm.should.have.key("OKActions").equal(["arn:ok"]) + alarm.should.have.key("InsufficientDataActions").equal(["arn:insufficient"]) + alarm.should.have.key("Unit").equal("Seconds") + alarm.should.have.key("AlarmArn").equal( + "arn:aws:cloudwatch:{}:{}:alarm:{}".format(region, ACCOUNT_ID, name) + ) + + +@mock_cloudwatch +def test_delete_alarm(): + cloudwatch = boto3.client("cloudwatch", region_name="eu-central-1") + + alarms = cloudwatch.describe_alarms()["MetricAlarms"] + alarms.should.have.length_of(0) + + name = "tester" + cloudwatch.put_metric_alarm( + AlarmActions=["arn:alarm"], + AlarmDescription="A test", + AlarmName=name, + ComparisonOperator="GreaterThanOrEqualToThreshold", + Dimensions=[{"Name": "InstanceId", "Value": "i-0123457"}], + EvaluationPeriods=5, + InsufficientDataActions=["arn:insufficient"], + Namespace="{0}_namespace".format(name), + MetricName="{0}_metric".format(name), + OKActions=["arn:ok"], + Period=60, + Statistic="Average", + Threshold=2, + Unit="Seconds", + ) + + alarms = cloudwatch.describe_alarms()["MetricAlarms"] + alarms.should.have.length_of(1) + + cloudwatch.delete_alarms(AlarmNames=[name]) + + alarms = cloudwatch.describe_alarms()["MetricAlarms"] + alarms.should.have.length_of(0) + + +@mock_cloudwatch +def test_delete_alarms_without_error(): + # given + cloudwatch = boto3.client("cloudwatch", "eu-west-1") + + # when/then + cloudwatch.delete_alarms(AlarmNames=["not-exists"]) + + +@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) + + assert "testalarm1" in alarms.get("MetricAlarms")[0].get("AlarmArn") + + +@mock_cloudwatch +def test_describe_alarms(): + 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, + ) + metric_data_queries = [ + { + "Id": "metricA", + "Expression": "metricB + metricC", + "Label": "metricA", + "ReturnData": True, + }, + { + "Id": "metricB", + "MetricStat": { + "Metric": { + "Namespace": "ns", + "MetricName": "metricB", + "Dimensions": [{"Name": "Name", "Value": "B"}], + }, + "Period": 60, + "Stat": "Sum", + }, + "ReturnData": False, + }, + { + "Id": "metricC", + "MetricStat": { + "Metric": { + "Namespace": "AWS/Lambda", + "MetricName": "metricC", + "Dimensions": [{"Name": "Name", "Value": "C"}], + }, + "Period": 60, + "Stat": "Sum", + "Unit": "Seconds", + }, + "ReturnData": False, + }, + ] + conn.put_metric_alarm( + AlarmName="testalarm2", + EvaluationPeriods=1, + DatapointsToAlarm=1, + Metrics=metric_data_queries, + ComparisonOperator="GreaterThanThreshold", + Threshold=1.0, + ) + alarms = conn.describe_alarms() + metric_alarms = alarms.get("MetricAlarms") + metric_alarms.should.have.length_of(2) + single_metric_alarm = [ + alarm for alarm in metric_alarms if alarm["AlarmName"] == "testalarm1" + ][0] + multiple_metric_alarm = [ + alarm for alarm in metric_alarms if alarm["AlarmName"] == "testalarm2" + ][0] + + single_metric_alarm["MetricName"].should.equal("cpu") + single_metric_alarm.shouldnt.have.property("Metrics") + single_metric_alarm["Namespace"].should.equal("blah") + single_metric_alarm["Period"].should.equal(10) + single_metric_alarm["EvaluationPeriods"].should.equal(5) + single_metric_alarm["Statistic"].should.equal("Average") + single_metric_alarm["ComparisonOperator"].should.equal("GreaterThanThreshold") + single_metric_alarm["Threshold"].should.equal(2) + + multiple_metric_alarm.shouldnt.have.property("MetricName") + multiple_metric_alarm["EvaluationPeriods"].should.equal(1) + multiple_metric_alarm["DatapointsToAlarm"].should.equal(1) + multiple_metric_alarm["Metrics"].should.equal(metric_data_queries) + multiple_metric_alarm["ComparisonOperator"].should.equal("GreaterThanThreshold") + multiple_metric_alarm["Threshold"].should.equal(1.0) + + +@mock_cloudwatch +def test_alarm_state(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + + client.put_metric_alarm( + AlarmName="testalarm1", + MetricName="cpu", + Namespace="blah", + Period=10, + EvaluationPeriods=5, + Statistic="Average", + Threshold=2, + ComparisonOperator="GreaterThanThreshold", + ActionsEnabled=True, + ) + client.put_metric_alarm( + AlarmName="testalarm2", + MetricName="cpu", + Namespace="blah", + Period=10, + EvaluationPeriods=5, + Statistic="Average", + Threshold=2, + ComparisonOperator="GreaterThanThreshold", + ) + + # This is tested implicitly as if it doesnt work the rest will die + client.set_alarm_state( + AlarmName="testalarm1", + StateValue="ALARM", + StateReason="testreason", + StateReasonData='{"some": "json_data"}', + ) + + resp = client.describe_alarms(StateValue="ALARM") + len(resp["MetricAlarms"]).should.equal(1) + resp["MetricAlarms"][0]["AlarmName"].should.equal("testalarm1") + resp["MetricAlarms"][0]["StateValue"].should.equal("ALARM") + resp["MetricAlarms"][0]["ActionsEnabled"].should.equal(True) + + resp = client.describe_alarms(StateValue="OK") + len(resp["MetricAlarms"]).should.equal(1) + resp["MetricAlarms"][0]["AlarmName"].should.equal("testalarm2") + resp["MetricAlarms"][0]["StateValue"].should.equal("OK") + resp["MetricAlarms"][0]["ActionsEnabled"].should.equal(False) + + # Just for sanity + resp = client.describe_alarms() + len(resp["MetricAlarms"]).should.equal(2) diff --git a/tests/test_cloudwatch/test_cloudwatch_boto3.py b/tests/test_cloudwatch/test_cloudwatch_boto3.py index 147b6e5a2..04ce052c5 100644 --- a/tests/test_cloudwatch/test_cloudwatch_boto3.py +++ b/tests/test_cloudwatch/test_cloudwatch_boto3.py @@ -1,345 +1,20 @@ -from datetime import datetime, timedelta -from operator import itemgetter - import boto3 -from botocore.exceptions import ClientError -from dateutil.tz import tzutc -from freezegun import freeze_time import pytest -from uuid import uuid4 import pytz import sure # noqa +from botocore.exceptions import ClientError +from datetime import datetime, timedelta +from dateutil.tz import tzutc +from decimal import Decimal +from freezegun import freeze_time +from operator import itemgetter +from uuid import uuid4 + from moto import mock_cloudwatch -from moto.cloudwatch.utils import make_arn_for_alarm from moto.core import ACCOUNT_ID -@mock_cloudwatch -def test_put_list_dashboard(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' - - client.put_dashboard(DashboardName="test1", DashboardBody=widget) - resp = client.list_dashboards() - - len(resp["DashboardEntries"]).should.equal(1) - - -@mock_cloudwatch -def test_put_list_prefix_nomatch_dashboard(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' - - client.put_dashboard(DashboardName="test1", DashboardBody=widget) - resp = client.list_dashboards(DashboardNamePrefix="nomatch") - - len(resp["DashboardEntries"]).should.equal(0) - - -@mock_cloudwatch -def test_delete_dashboard(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' - - client.put_dashboard(DashboardName="test1", DashboardBody=widget) - client.put_dashboard(DashboardName="test2", DashboardBody=widget) - client.put_dashboard(DashboardName="test3", DashboardBody=widget) - client.delete_dashboards(DashboardNames=["test2", "test1"]) - - resp = client.list_dashboards(DashboardNamePrefix="test3") - len(resp["DashboardEntries"]).should.equal(1) - - -@mock_cloudwatch -def test_delete_dashboard_fail(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' - - client.put_dashboard(DashboardName="test1", DashboardBody=widget) - client.put_dashboard(DashboardName="test2", DashboardBody=widget) - client.put_dashboard(DashboardName="test3", DashboardBody=widget) - # Doesnt delete anything if all dashboards to be deleted do not exist - try: - client.delete_dashboards(DashboardNames=["test2", "test1", "test_no_match"]) - except ClientError as err: - err.response["Error"]["Code"].should.equal("ResourceNotFound") - else: - raise RuntimeError("Should of raised error") - - resp = client.list_dashboards() - len(resp["DashboardEntries"]).should.equal(3) - - -@mock_cloudwatch -def test_get_dashboard(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' - client.put_dashboard(DashboardName="test1", DashboardBody=widget) - - resp = client.get_dashboard(DashboardName="test1") - resp.should.contain("DashboardArn") - resp.should.contain("DashboardBody") - resp["DashboardName"].should.equal("test1") - - -@mock_cloudwatch -def test_get_dashboard_fail(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - - try: - client.get_dashboard(DashboardName="test1") - except ClientError as err: - err.response["Error"]["Code"].should.equal("ResourceNotFound") - else: - raise RuntimeError("Should have raised error") - - -@mock_cloudwatch -def test_create_alarm(): - region = "eu-west-1" - cloudwatch = boto3.client("cloudwatch", region) - - name = "tester" - cloudwatch.put_metric_alarm( - AlarmActions=["arn:alarm"], - AlarmDescription="A test", - AlarmName=name, - ComparisonOperator="GreaterThanOrEqualToThreshold", - Dimensions=[{"Name": "InstanceId", "Value": "i-0123457"}], - EvaluationPeriods=5, - InsufficientDataActions=["arn:insufficient"], - Namespace="{0}_namespace".format(name), - MetricName="{0}_metric".format(name), - OKActions=["arn:ok"], - Period=60, - Statistic="Average", - Threshold=2, - Unit="Seconds", - ) - - alarms = cloudwatch.describe_alarms()["MetricAlarms"] - alarms.should.have.length_of(1) - alarm = alarms[0] - alarm.should.have.key("AlarmName").equal("tester") - alarm.should.have.key("Namespace").equal("tester_namespace") - alarm.should.have.key("MetricName").equal("tester_metric") - alarm.should.have.key("ComparisonOperator").equal("GreaterThanOrEqualToThreshold") - alarm.should.have.key("Threshold").equal(2.0) - alarm.should.have.key("Period").equal(60) - alarm.should.have.key("EvaluationPeriods").equal(5) - alarm.should.have.key("Statistic").should.equal("Average") - alarm.should.have.key("AlarmDescription").equal("A test") - alarm.should.have.key("Dimensions").equal( - [{"Name": "InstanceId", "Value": "i-0123457"}] - ) - alarm.should.have.key("AlarmActions").equal(["arn:alarm"]) - alarm.should.have.key("OKActions").equal(["arn:ok"]) - alarm.should.have.key("InsufficientDataActions").equal(["arn:insufficient"]) - alarm.should.have.key("Unit").equal("Seconds") - alarm.should.have.key("AlarmArn").equal( - "arn:aws:cloudwatch:{}:{}:alarm:{}".format(region, ACCOUNT_ID, name) - ) - - -@mock_cloudwatch -def test_delete_alarm(): - cloudwatch = boto3.client("cloudwatch", region_name="eu-central-1") - - alarms = cloudwatch.describe_alarms()["MetricAlarms"] - alarms.should.have.length_of(0) - - name = "tester" - cloudwatch.put_metric_alarm( - AlarmActions=["arn:alarm"], - AlarmDescription="A test", - AlarmName=name, - ComparisonOperator="GreaterThanOrEqualToThreshold", - Dimensions=[{"Name": "InstanceId", "Value": "i-0123457"}], - EvaluationPeriods=5, - InsufficientDataActions=["arn:insufficient"], - Namespace="{0}_namespace".format(name), - MetricName="{0}_metric".format(name), - OKActions=["arn:ok"], - Period=60, - Statistic="Average", - Threshold=2, - Unit="Seconds", - ) - - alarms = cloudwatch.describe_alarms()["MetricAlarms"] - alarms.should.have.length_of(1) - - cloudwatch.delete_alarms(AlarmNames=[name]) - - alarms = cloudwatch.describe_alarms()["MetricAlarms"] - alarms.should.have.length_of(0) - - -@mock_cloudwatch -def test_delete_alarms_without_error(): - # given - cloudwatch = boto3.client("cloudwatch", "eu-west-1") - - # when/then - cloudwatch.delete_alarms(AlarmNames=["not-exists"]) - - -@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) - - assert "testalarm1" in alarms.get("MetricAlarms")[0].get("AlarmArn") - - -@mock_cloudwatch -def test_describe_alarms(): - 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, - ) - metric_data_queries = [ - { - "Id": "metricA", - "Expression": "metricB + metricC", - "Label": "metricA", - "ReturnData": True, - }, - { - "Id": "metricB", - "MetricStat": { - "Metric": { - "Namespace": "ns", - "MetricName": "metricB", - "Dimensions": [{"Name": "Name", "Value": "B"}], - }, - "Period": 60, - "Stat": "Sum", - }, - "ReturnData": False, - }, - { - "Id": "metricC", - "MetricStat": { - "Metric": { - "Namespace": "AWS/Lambda", - "MetricName": "metricC", - "Dimensions": [{"Name": "Name", "Value": "C"}], - }, - "Period": 60, - "Stat": "Sum", - "Unit": "Seconds", - }, - "ReturnData": False, - }, - ] - conn.put_metric_alarm( - AlarmName="testalarm2", - EvaluationPeriods=1, - DatapointsToAlarm=1, - Metrics=metric_data_queries, - ComparisonOperator="GreaterThanThreshold", - Threshold=1.0, - ) - alarms = conn.describe_alarms() - metric_alarms = alarms.get("MetricAlarms") - metric_alarms.should.have.length_of(2) - single_metric_alarm = [ - alarm for alarm in metric_alarms if alarm["AlarmName"] == "testalarm1" - ][0] - multiple_metric_alarm = [ - alarm for alarm in metric_alarms if alarm["AlarmName"] == "testalarm2" - ][0] - - single_metric_alarm["MetricName"].should.equal("cpu") - single_metric_alarm.shouldnt.have.property("Metrics") - single_metric_alarm["Namespace"].should.equal("blah") - single_metric_alarm["Period"].should.equal(10) - single_metric_alarm["EvaluationPeriods"].should.equal(5) - single_metric_alarm["Statistic"].should.equal("Average") - single_metric_alarm["ComparisonOperator"].should.equal("GreaterThanThreshold") - single_metric_alarm["Threshold"].should.equal(2) - - multiple_metric_alarm.shouldnt.have.property("MetricName") - multiple_metric_alarm["EvaluationPeriods"].should.equal(1) - multiple_metric_alarm["DatapointsToAlarm"].should.equal(1) - multiple_metric_alarm["Metrics"].should.equal(metric_data_queries) - multiple_metric_alarm["ComparisonOperator"].should.equal("GreaterThanThreshold") - multiple_metric_alarm["Threshold"].should.equal(1.0) - - -@mock_cloudwatch -def test_alarm_state(): - client = boto3.client("cloudwatch", region_name="eu-central-1") - - client.put_metric_alarm( - AlarmName="testalarm1", - MetricName="cpu", - Namespace="blah", - Period=10, - EvaluationPeriods=5, - Statistic="Average", - Threshold=2, - ComparisonOperator="GreaterThanThreshold", - ActionsEnabled=True, - ) - client.put_metric_alarm( - AlarmName="testalarm2", - MetricName="cpu", - Namespace="blah", - Period=10, - EvaluationPeriods=5, - Statistic="Average", - Threshold=2, - ComparisonOperator="GreaterThanThreshold", - ) - - # This is tested implicitly as if it doesnt work the rest will die - client.set_alarm_state( - AlarmName="testalarm1", - StateValue="ALARM", - StateReason="testreason", - StateReasonData='{"some": "json_data"}', - ) - - resp = client.describe_alarms(StateValue="ALARM") - len(resp["MetricAlarms"]).should.equal(1) - resp["MetricAlarms"][0]["AlarmName"].should.equal("testalarm1") - resp["MetricAlarms"][0]["StateValue"].should.equal("ALARM") - resp["MetricAlarms"][0]["ActionsEnabled"].should.equal(True) - - resp = client.describe_alarms(StateValue="OK") - len(resp["MetricAlarms"]).should.equal(1) - resp["MetricAlarms"][0]["AlarmName"].should.equal("testalarm2") - resp["MetricAlarms"][0]["StateValue"].should.equal("OK") - resp["MetricAlarms"][0]["ActionsEnabled"].should.equal(False) - - # Just for sanity - resp = client.describe_alarms() - len(resp["MetricAlarms"]).should.equal(2) - - @mock_cloudwatch def test_put_metric_data_no_dimensions(): conn = boto3.client("cloudwatch", region_name="us-east-1") @@ -355,6 +30,29 @@ def test_put_metric_data_no_dimensions(): metric["MetricName"].should.equal("metric") +@mock_cloudwatch +def test_put_metric_data_can_not_have_nan(): + client = boto3.client("cloudwatch", region_name="us-west-2") + utc_now = datetime.now(tz=pytz.utc) + with pytest.raises(ClientError) as exc: + client.put_metric_data( + Namespace="mynamespace", + MetricData=[ + { + "MetricName": "mymetric", + "Timestamp": utc_now, + "Value": Decimal("NaN"), + "Unit": "Count", + } + ], + ) + err = exc.value.response["Error"] + err["Code"].should.equal("InvalidParameterValue") + err["Message"].should.equal( + "The value NaN for parameter MetricData.member.1.Value is invalid." + ) + + @mock_cloudwatch def test_put_metric_data_with_statistics(): conn = boto3.client("cloudwatch", region_name="us-east-1") @@ -1257,178 +955,3 @@ def test_put_metric_alarm_error_evaluate_low_sample_count_percentile(): "Option unknown is not supported. " "Supported options for parameter EvaluateLowSampleCountPercentile are evaluate and ignore." ) - - -@mock_cloudwatch -def test_list_tags_for_resource(): - # given - client = boto3.client("cloudwatch", region_name="eu-central-1") - alarm_name = "test-alarm" - client.put_metric_alarm( - AlarmName=alarm_name, - AlarmDescription="test alarm", - ActionsEnabled=True, - MetricName="5XXError", - Namespace="AWS/ApiGateway", - Statistic="Sum", - Dimensions=[ - {"Name": "ApiName", "Value": "test-api"}, - {"Name": "Stage", "Value": "default"}, - ], - Period=60, - Unit="Seconds", - EvaluationPeriods=1, - DatapointsToAlarm=1, - Threshold=1.0, - ComparisonOperator="GreaterThanOrEqualToThreshold", - Tags=[{"Key": "key-1", "Value": "value-1"}], - ) - arn = client.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"][0]["AlarmArn"] - - # when - response = client.list_tags_for_resource(ResourceARN=arn) - - # then - response["Tags"].should.equal([{"Key": "key-1", "Value": "value-1"}]) - - -@mock_cloudwatch -def test_list_tags_for_resource_with_unknown_resource(): - # given - region_name = "eu-central-1" - client = boto3.client("cloudwatch", region_name=region_name) - - # when - response = client.list_tags_for_resource( - ResourceARN=make_arn_for_alarm( - region=region_name, account_id=ACCOUNT_ID, alarm_name="unknown" - ) - ) - - # then - response["Tags"].should.be.empty - - -@mock_cloudwatch -def test_tag_resource(): - # given - client = boto3.client("cloudwatch", region_name="eu-central-1") - alarm_name = "test-alarm" - client.put_metric_alarm( - AlarmName=alarm_name, - AlarmDescription="test alarm", - ActionsEnabled=True, - MetricName="5XXError", - Namespace="AWS/ApiGateway", - Statistic="Sum", - Dimensions=[ - {"Name": "ApiName", "Value": "test-api"}, - {"Name": "Stage", "Value": "default"}, - ], - Period=60, - Unit="Seconds", - EvaluationPeriods=1, - DatapointsToAlarm=1, - Threshold=1.0, - ComparisonOperator="GreaterThanOrEqualToThreshold", - Tags=[{"Key": "key-1", "Value": "value-1"}], - ) - arn = client.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"][0]["AlarmArn"] - - # when - client.tag_resource(ResourceARN=arn, Tags=[{"Key": "key-2", "Value": "value-2"}]) - - # then - response = client.list_tags_for_resource(ResourceARN=arn) - sorted(response["Tags"], key=itemgetter("Key")).should.equal( - sorted( - [ - {"Key": "key-1", "Value": "value-1"}, - {"Key": "key-2", "Value": "value-2"}, - ], - key=itemgetter("Key"), - ) - ) - - -@mock_cloudwatch -def test_tag_resource_error_not_exists(): - # given - region_name = "eu-central-1" - client = boto3.client("cloudwatch", region_name=region_name) - - # when - with pytest.raises(ClientError) as e: - client.tag_resource( - ResourceARN=make_arn_for_alarm( - region=region_name, account_id=ACCOUNT_ID, alarm_name="unknown" - ), - Tags=[{"Key": "key-1", "Value": "value-1"},], - ) - - # then - ex = e.value - ex.operation_name.should.equal("TagResource") - ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(404) - ex.response["Error"]["Code"].should.contain("ResourceNotFoundException") - ex.response["Error"]["Message"].should.equal("Unknown") - - -@mock_cloudwatch -def test_untag_resource(): - # given - client = boto3.client("cloudwatch", region_name="eu-central-1") - alarm_name = "test-alarm" - client.put_metric_alarm( - AlarmName=alarm_name, - AlarmDescription="test alarm", - ActionsEnabled=True, - MetricName="5XXError", - Namespace="AWS/ApiGateway", - Statistic="Sum", - Dimensions=[ - {"Name": "ApiName", "Value": "test-api"}, - {"Name": "Stage", "Value": "default"}, - ], - Period=60, - Unit="Seconds", - EvaluationPeriods=1, - DatapointsToAlarm=1, - Threshold=1.0, - ComparisonOperator="GreaterThanOrEqualToThreshold", - Tags=[ - {"Key": "key-1", "Value": "value-1"}, - {"Key": "key-2", "Value": "value-2"}, - ], - ) - arn = client.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"][0]["AlarmArn"] - - # when - client.untag_resource(ResourceARN=arn, TagKeys=["key-2"]) - - # then - response = client.list_tags_for_resource(ResourceARN=arn) - response["Tags"].should.equal([{"Key": "key-1", "Value": "value-1"}]) - - -@mock_cloudwatch -def test_untag_resource_error_not_exists(): - # given - region_name = "eu-central-1" - client = boto3.client("cloudwatch", region_name=region_name) - - # when - with pytest.raises(ClientError) as e: - client.untag_resource( - ResourceARN=make_arn_for_alarm( - region=region_name, account_id=ACCOUNT_ID, alarm_name="unknown" - ), - TagKeys=["key-1"], - ) - - # then - ex = e.value - ex.operation_name.should.equal("UntagResource") - ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(404) - ex.response["Error"]["Code"].should.contain("ResourceNotFoundException") - ex.response["Error"]["Message"].should.equal("Unknown") diff --git a/tests/test_cloudwatch/test_cloudwatch_dashboards.py b/tests/test_cloudwatch/test_cloudwatch_dashboards.py new file mode 100644 index 000000000..0d105e7bc --- /dev/null +++ b/tests/test_cloudwatch/test_cloudwatch_dashboards.py @@ -0,0 +1,85 @@ +import boto3 +from botocore.exceptions import ClientError +import sure # noqa + +from moto import mock_cloudwatch + + +@mock_cloudwatch +def test_put_list_dashboard(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' + + client.put_dashboard(DashboardName="test1", DashboardBody=widget) + resp = client.list_dashboards() + + len(resp["DashboardEntries"]).should.equal(1) + + +@mock_cloudwatch +def test_put_list_prefix_nomatch_dashboard(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' + + client.put_dashboard(DashboardName="test1", DashboardBody=widget) + resp = client.list_dashboards(DashboardNamePrefix="nomatch") + + len(resp["DashboardEntries"]).should.equal(0) + + +@mock_cloudwatch +def test_delete_dashboard(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' + + client.put_dashboard(DashboardName="test1", DashboardBody=widget) + client.put_dashboard(DashboardName="test2", DashboardBody=widget) + client.put_dashboard(DashboardName="test3", DashboardBody=widget) + client.delete_dashboards(DashboardNames=["test2", "test1"]) + + resp = client.list_dashboards(DashboardNamePrefix="test3") + len(resp["DashboardEntries"]).should.equal(1) + + +@mock_cloudwatch +def test_delete_dashboard_fail(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' + + client.put_dashboard(DashboardName="test1", DashboardBody=widget) + client.put_dashboard(DashboardName="test2", DashboardBody=widget) + client.put_dashboard(DashboardName="test3", DashboardBody=widget) + # Doesnt delete anything if all dashboards to be deleted do not exist + try: + client.delete_dashboards(DashboardNames=["test2", "test1", "test_no_match"]) + except ClientError as err: + err.response["Error"]["Code"].should.equal("ResourceNotFound") + else: + raise RuntimeError("Should of raised error") + + resp = client.list_dashboards() + len(resp["DashboardEntries"]).should.equal(3) + + +@mock_cloudwatch +def test_get_dashboard(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}' + client.put_dashboard(DashboardName="test1", DashboardBody=widget) + + resp = client.get_dashboard(DashboardName="test1") + resp.should.contain("DashboardArn") + resp.should.contain("DashboardBody") + resp["DashboardName"].should.equal("test1") + + +@mock_cloudwatch +def test_get_dashboard_fail(): + client = boto3.client("cloudwatch", region_name="eu-central-1") + + try: + client.get_dashboard(DashboardName="test1") + except ClientError as err: + err.response["Error"]["Code"].should.equal("ResourceNotFound") + else: + raise RuntimeError("Should have raised error") diff --git a/tests/test_cloudwatch/test_cloudwatch_tags.py b/tests/test_cloudwatch/test_cloudwatch_tags.py new file mode 100644 index 000000000..8a72f3b0c --- /dev/null +++ b/tests/test_cloudwatch/test_cloudwatch_tags.py @@ -0,0 +1,185 @@ +from operator import itemgetter + +import boto3 +from botocore.exceptions import ClientError +import pytest +import sure # noqa + +from moto import mock_cloudwatch +from moto.cloudwatch.utils import make_arn_for_alarm +from moto.core import ACCOUNT_ID + + +@mock_cloudwatch +def test_list_tags_for_resource(): + # given + client = boto3.client("cloudwatch", region_name="eu-central-1") + alarm_name = "test-alarm" + client.put_metric_alarm( + AlarmName=alarm_name, + AlarmDescription="test alarm", + ActionsEnabled=True, + MetricName="5XXError", + Namespace="AWS/ApiGateway", + Statistic="Sum", + Dimensions=[ + {"Name": "ApiName", "Value": "test-api"}, + {"Name": "Stage", "Value": "default"}, + ], + Period=60, + Unit="Seconds", + EvaluationPeriods=1, + DatapointsToAlarm=1, + Threshold=1.0, + ComparisonOperator="GreaterThanOrEqualToThreshold", + Tags=[{"Key": "key-1", "Value": "value-1"}], + ) + arn = client.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"][0]["AlarmArn"] + + # when + response = client.list_tags_for_resource(ResourceARN=arn) + + # then + response["Tags"].should.equal([{"Key": "key-1", "Value": "value-1"}]) + + +@mock_cloudwatch +def test_list_tags_for_resource_with_unknown_resource(): + # given + region_name = "eu-central-1" + client = boto3.client("cloudwatch", region_name=region_name) + + # when + response = client.list_tags_for_resource( + ResourceARN=make_arn_for_alarm( + region=region_name, account_id=ACCOUNT_ID, alarm_name="unknown" + ) + ) + + # then + response["Tags"].should.be.empty + + +@mock_cloudwatch +def test_tag_resource(): + # given + client = boto3.client("cloudwatch", region_name="eu-central-1") + alarm_name = "test-alarm" + client.put_metric_alarm( + AlarmName=alarm_name, + AlarmDescription="test alarm", + ActionsEnabled=True, + MetricName="5XXError", + Namespace="AWS/ApiGateway", + Statistic="Sum", + Dimensions=[ + {"Name": "ApiName", "Value": "test-api"}, + {"Name": "Stage", "Value": "default"}, + ], + Period=60, + Unit="Seconds", + EvaluationPeriods=1, + DatapointsToAlarm=1, + Threshold=1.0, + ComparisonOperator="GreaterThanOrEqualToThreshold", + Tags=[{"Key": "key-1", "Value": "value-1"}], + ) + arn = client.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"][0]["AlarmArn"] + + # when + client.tag_resource(ResourceARN=arn, Tags=[{"Key": "key-2", "Value": "value-2"}]) + + # then + response = client.list_tags_for_resource(ResourceARN=arn) + sorted(response["Tags"], key=itemgetter("Key")).should.equal( + sorted( + [ + {"Key": "key-1", "Value": "value-1"}, + {"Key": "key-2", "Value": "value-2"}, + ], + key=itemgetter("Key"), + ) + ) + + +@mock_cloudwatch +def test_tag_resource_error_not_exists(): + # given + region_name = "eu-central-1" + client = boto3.client("cloudwatch", region_name=region_name) + + # when + with pytest.raises(ClientError) as e: + client.tag_resource( + ResourceARN=make_arn_for_alarm( + region=region_name, account_id=ACCOUNT_ID, alarm_name="unknown" + ), + Tags=[{"Key": "key-1", "Value": "value-1"},], + ) + + # then + ex = e.value + ex.operation_name.should.equal("TagResource") + ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(404) + ex.response["Error"]["Code"].should.contain("ResourceNotFoundException") + ex.response["Error"]["Message"].should.equal("Unknown") + + +@mock_cloudwatch +def test_untag_resource(): + # given + client = boto3.client("cloudwatch", region_name="eu-central-1") + alarm_name = "test-alarm" + client.put_metric_alarm( + AlarmName=alarm_name, + AlarmDescription="test alarm", + ActionsEnabled=True, + MetricName="5XXError", + Namespace="AWS/ApiGateway", + Statistic="Sum", + Dimensions=[ + {"Name": "ApiName", "Value": "test-api"}, + {"Name": "Stage", "Value": "default"}, + ], + Period=60, + Unit="Seconds", + EvaluationPeriods=1, + DatapointsToAlarm=1, + Threshold=1.0, + ComparisonOperator="GreaterThanOrEqualToThreshold", + Tags=[ + {"Key": "key-1", "Value": "value-1"}, + {"Key": "key-2", "Value": "value-2"}, + ], + ) + arn = client.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"][0]["AlarmArn"] + + # when + client.untag_resource(ResourceARN=arn, TagKeys=["key-2"]) + + # then + response = client.list_tags_for_resource(ResourceARN=arn) + response["Tags"].should.equal([{"Key": "key-1", "Value": "value-1"}]) + + +@mock_cloudwatch +def test_untag_resource_error_not_exists(): + # given + region_name = "eu-central-1" + client = boto3.client("cloudwatch", region_name=region_name) + + # when + with pytest.raises(ClientError) as e: + client.untag_resource( + ResourceARN=make_arn_for_alarm( + region=region_name, account_id=ACCOUNT_ID, alarm_name="unknown" + ), + TagKeys=["key-1"], + ) + + # then + ex = e.value + ex.operation_name.should.equal("UntagResource") + ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(404) + ex.response["Error"]["Code"].should.contain("ResourceNotFoundException") + ex.response["Error"]["Message"].should.equal("Unknown")