Cloudwatch - Do not allow NaN values (#4399)
This commit is contained in:
parent
58df83f39f
commit
198f2696b9
@ -438,6 +438,12 @@ class CloudWatchBackend(BaseBackend):
|
|||||||
self.alarms.pop(alarm_name, None)
|
self.alarms.pop(alarm_name, None)
|
||||||
|
|
||||||
def put_metric_data(self, namespace, metric_data):
|
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:
|
for metric_member in metric_data:
|
||||||
# Preserve "datetime" for get_metric_statistics comparisons
|
# Preserve "datetime" for get_metric_statistics comparisons
|
||||||
timestamp = metric_member.get("Timestamp")
|
timestamp = metric_member.get("Timestamp")
|
||||||
|
250
tests/test_cloudwatch/test_cloudwatch_alarms.py
Normal file
250
tests/test_cloudwatch/test_cloudwatch_alarms.py
Normal file
@ -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)
|
@ -1,345 +1,20 @@
|
|||||||
from datetime import datetime, timedelta
|
|
||||||
from operator import itemgetter
|
|
||||||
|
|
||||||
import boto3
|
import boto3
|
||||||
from botocore.exceptions import ClientError
|
|
||||||
from dateutil.tz import tzutc
|
|
||||||
from freezegun import freeze_time
|
|
||||||
import pytest
|
import pytest
|
||||||
from uuid import uuid4
|
|
||||||
import pytz
|
import pytz
|
||||||
import sure # noqa
|
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 import mock_cloudwatch
|
||||||
from moto.cloudwatch.utils import make_arn_for_alarm
|
|
||||||
from moto.core import ACCOUNT_ID
|
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
|
@mock_cloudwatch
|
||||||
def test_put_metric_data_no_dimensions():
|
def test_put_metric_data_no_dimensions():
|
||||||
conn = boto3.client("cloudwatch", region_name="us-east-1")
|
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")
|
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
|
@mock_cloudwatch
|
||||||
def test_put_metric_data_with_statistics():
|
def test_put_metric_data_with_statistics():
|
||||||
conn = boto3.client("cloudwatch", region_name="us-east-1")
|
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. "
|
"Option unknown is not supported. "
|
||||||
"Supported options for parameter EvaluateLowSampleCountPercentile are evaluate and ignore."
|
"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")
|
|
||||||
|
85
tests/test_cloudwatch/test_cloudwatch_dashboards.py
Normal file
85
tests/test_cloudwatch/test_cloudwatch_dashboards.py
Normal file
@ -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")
|
185
tests/test_cloudwatch/test_cloudwatch_tags.py
Normal file
185
tests/test_cloudwatch/test_cloudwatch_tags.py
Normal file
@ -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")
|
Loading…
Reference in New Issue
Block a user