251 lines
8.0 KiB
Python
251 lines
8.0 KiB
Python
import boto3
|
|
|
|
from moto import mock_cloudwatch
|
|
from moto.core import DEFAULT_ACCOUNT_ID as 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=f"{name}_namespace",
|
|
MetricName=f"{name}_metric",
|
|
OKActions=["arn:ok"],
|
|
Period=60,
|
|
Statistic="Average",
|
|
Threshold=2,
|
|
Unit="Seconds",
|
|
)
|
|
|
|
alarms = cloudwatch.describe_alarms()["MetricAlarms"]
|
|
assert len(alarms) == 1
|
|
alarm = alarms[0]
|
|
assert alarm["AlarmName"] == "tester"
|
|
assert alarm["Namespace"] == "tester_namespace"
|
|
assert alarm["MetricName"] == "tester_metric"
|
|
assert alarm["ComparisonOperator"] == "GreaterThanOrEqualToThreshold"
|
|
assert alarm["Threshold"] == 2.0
|
|
assert alarm["Period"] == 60
|
|
assert alarm["EvaluationPeriods"] == 5
|
|
assert alarm["Statistic"] == "Average"
|
|
assert alarm["AlarmDescription"] == "A test"
|
|
assert alarm["Dimensions"] == [{"Name": "InstanceId", "Value": "i-0123457"}]
|
|
assert alarm["AlarmActions"] == ["arn:alarm"]
|
|
assert alarm["OKActions"] == ["arn:ok"]
|
|
assert alarm["InsufficientDataActions"] == ["arn:insufficient"]
|
|
assert alarm["Unit"] == "Seconds"
|
|
assert alarm["AlarmArn"] == f"arn:aws:cloudwatch:{region}:{ACCOUNT_ID}:alarm:{name}"
|
|
# default value should be True
|
|
assert alarm["ActionsEnabled"] is True
|
|
|
|
|
|
@mock_cloudwatch
|
|
def test_delete_alarm():
|
|
cloudwatch = boto3.client("cloudwatch", region_name="eu-central-1")
|
|
|
|
alarms = cloudwatch.describe_alarms()["MetricAlarms"]
|
|
assert len(alarms) == 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=f"{name}_namespace",
|
|
MetricName=f"{name}_metric",
|
|
OKActions=["arn:ok"],
|
|
Period=60,
|
|
Statistic="Average",
|
|
Threshold=2,
|
|
Unit="Seconds",
|
|
)
|
|
|
|
alarms = cloudwatch.describe_alarms()["MetricAlarms"]
|
|
assert len(alarms) == 1
|
|
|
|
cloudwatch.delete_alarms(AlarmNames=[name])
|
|
|
|
alarms = cloudwatch.describe_alarms()["MetricAlarms"]
|
|
assert len(alarms) == 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")
|
|
assert len(alarms.get("MetricAlarms")) == 1
|
|
alarm = alarms.get("MetricAlarms")[0]
|
|
assert "testalarm1" in alarm.get("AlarmArn")
|
|
assert alarm["ActionsEnabled"] is True
|
|
|
|
|
|
@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=False,
|
|
)
|
|
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")
|
|
assert len(metric_alarms) == 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]
|
|
|
|
assert single_metric_alarm["MetricName"] == "cpu"
|
|
assert "Metrics" not in single_metric_alarm
|
|
assert single_metric_alarm["Namespace"] == "blah"
|
|
assert single_metric_alarm["Period"] == 10
|
|
assert single_metric_alarm["EvaluationPeriods"] == 5
|
|
assert single_metric_alarm["Statistic"] == "Average"
|
|
assert single_metric_alarm["ComparisonOperator"] == "GreaterThanThreshold"
|
|
assert single_metric_alarm["Threshold"] == 2
|
|
assert single_metric_alarm["ActionsEnabled"] is False
|
|
|
|
assert "MetricName" not in multiple_metric_alarm
|
|
assert multiple_metric_alarm["EvaluationPeriods"] == 1
|
|
assert multiple_metric_alarm["DatapointsToAlarm"] == 1
|
|
assert multiple_metric_alarm["Metrics"] == metric_data_queries
|
|
assert multiple_metric_alarm["ComparisonOperator"] == "GreaterThanThreshold"
|
|
assert multiple_metric_alarm["Threshold"] == 1.0
|
|
assert multiple_metric_alarm["ActionsEnabled"] is True
|
|
|
|
|
|
@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")
|
|
assert len(resp["MetricAlarms"]) == 1
|
|
assert resp["MetricAlarms"][0]["AlarmName"] == "testalarm1"
|
|
assert resp["MetricAlarms"][0]["StateValue"] == "ALARM"
|
|
assert resp["MetricAlarms"][0]["ActionsEnabled"] is True
|
|
|
|
resp = client.describe_alarms(StateValue="OK")
|
|
assert len(resp["MetricAlarms"]) == 1
|
|
assert resp["MetricAlarms"][0]["AlarmName"] == "testalarm2"
|
|
assert resp["MetricAlarms"][0]["StateValue"] == "OK"
|
|
assert resp["MetricAlarms"][0]["ActionsEnabled"] is True
|
|
|
|
# Just for sanity
|
|
resp = client.describe_alarms()
|
|
assert len(resp["MetricAlarms"]) == 2
|