Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #2801 from bblommers/bugfix/cloudwatch_timestamp

Browse Source 
Cloudwatch - Fix timestamp format
This commit is contained in:
Steve Pulec 2020-03-15 16:43:03 -05:00 committed by GitHub
commit 091b6cdef4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 152 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
moto/cloudwatch/models.py
View File

@ -2,13 +2,14 @@ import json
from boto3 import Session from boto3 import Session
from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.core.utils import iso_8601_datetime_without_milliseconds
from moto.core import BaseBackend, BaseModel from moto.core import BaseBackend, BaseModel
from moto.core.exceptions import RESTError from moto.core.exceptions import RESTError
from datetime import datetime, timedelta from datetime import datetime, timedelta
from dateutil.tz import tzutc from dateutil.tz import tzutc
from uuid import uuid4 from uuid import uuid4
from .utils import make_arn_for_dashboard from .utils import make_arn_for_dashboard
from dateutil import parser
from moto.core import ACCOUNT_ID as DEFAULT_ACCOUNT_ID from moto.core import ACCOUNT_ID as DEFAULT_ACCOUNT_ID
@ -146,7 +147,7 @@ class Dashboard(BaseModel):
class Statistics: class Statistics:
def __init__(self, stats, dt): def __init__(self, stats, dt):
self.timestamp = iso_8601_datetime_with_milliseconds(dt) self.timestamp = iso_8601_datetime_without_milliseconds(dt)
self.values = [] self.values = []
self.stats = stats self.stats = stats
@ -278,8 +279,7 @@ class CloudWatchBackend(BaseBackend):
# Preserve "datetime" for get_metric_statistics comparisons # Preserve "datetime" for get_metric_statistics comparisons
timestamp = metric_member.get("Timestamp") timestamp = metric_member.get("Timestamp")
if timestamp is not None and type(timestamp) != datetime: if timestamp is not None and type(timestamp) != datetime:
timestamp = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ") timestamp = parser.parse(timestamp)
timestamp = timestamp.replace(tzinfo=tzutc())
self.metric_data.append( self.metric_data.append(
MetricDatum( MetricDatum(
namespace, namespace,

265
tests/test_cloudwatch/test_cloudwatch.py
View File

@ -1,117 +1,148 @@
import boto import boto
from boto.ec2.cloudwatch.alarm import MetricAlarm from boto.ec2.cloudwatch.alarm import MetricAlarm
import sure # noqa from datetime import datetime
import sure # noqa
from moto import mock_cloudwatch_deprecated
from moto import mock_cloudwatch_deprecated
def alarm_fixture(name="tester", action=None):
action = action or ["arn:alarm"] def alarm_fixture(name="tester", action=None):
return MetricAlarm( action = action or ["arn:alarm"]
name=name, return MetricAlarm(
namespace="{0}_namespace".format(name), name=name,
metric="{0}_metric".format(name), namespace="{0}_namespace".format(name),
comparison=">=", metric="{0}_metric".format(name),
threshold=2.0, comparison=">=",
period=60, threshold=2.0,
evaluation_periods=5, period=60,
statistic="Average", evaluation_periods=5,
description="A test", statistic="Average",
dimensions={"InstanceId": ["i-0123456,i-0123457"]}, description="A test",
alarm_actions=action, dimensions={"InstanceId": ["i-0123456,i-0123457"]},
ok_actions=["arn:ok"], alarm_actions=action,
insufficient_data_actions=["arn:insufficient"], ok_actions=["arn:ok"],
unit="Seconds", insufficient_data_actions=["arn:insufficient"],
) unit="Seconds",
)
@mock_cloudwatch_deprecated
def test_create_alarm(): @mock_cloudwatch_deprecated
conn = boto.connect_cloudwatch() def test_create_alarm():
conn = boto.connect_cloudwatch()
alarm = alarm_fixture()
conn.create_alarm(alarm) alarm = alarm_fixture()
conn.create_alarm(alarm)
alarms = conn.describe_alarms()
alarms.should.have.length_of(1) alarms = conn.describe_alarms()
alarm = alarms[0] alarms.should.have.length_of(1)
alarm.name.should.equal("tester") alarm = alarms[0]
alarm.namespace.should.equal("tester_namespace") alarm.name.should.equal("tester")
alarm.metric.should.equal("tester_metric") alarm.namespace.should.equal("tester_namespace")
alarm.comparison.should.equal(">=") alarm.metric.should.equal("tester_metric")
alarm.threshold.should.equal(2.0) alarm.comparison.should.equal(">=")
alarm.period.should.equal(60) alarm.threshold.should.equal(2.0)
alarm.evaluation_periods.should.equal(5) alarm.period.should.equal(60)
alarm.statistic.should.equal("Average") alarm.evaluation_periods.should.equal(5)
alarm.description.should.equal("A test") alarm.statistic.should.equal("Average")
dict(alarm.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]}) alarm.description.should.equal("A test")
list(alarm.alarm_actions).should.equal(["arn:alarm"]) dict(alarm.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
list(alarm.ok_actions).should.equal(["arn:ok"]) list(alarm.alarm_actions).should.equal(["arn:alarm"])
list(alarm.insufficient_data_actions).should.equal(["arn:insufficient"]) list(alarm.ok_actions).should.equal(["arn:ok"])
alarm.unit.should.equal("Seconds") list(alarm.insufficient_data_actions).should.equal(["arn:insufficient"])
alarm.unit.should.equal("Seconds")
@mock_cloudwatch_deprecated
def test_delete_alarm(): @mock_cloudwatch_deprecated
conn = boto.connect_cloudwatch() def test_delete_alarm():
conn = boto.connect_cloudwatch()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0) alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
alarm = alarm_fixture()
conn.create_alarm(alarm) alarm = alarm_fixture()
conn.create_alarm(alarm)
alarms = conn.describe_alarms()
alarms.should.have.length_of(1) alarms = conn.describe_alarms()
alarms.should.have.length_of(1)
alarms[0].delete()
alarms[0].delete()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0) alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
@mock_cloudwatch_deprecated
def test_put_metric_data(): @mock_cloudwatch_deprecated
conn = boto.connect_cloudwatch() def test_put_metric_data():
conn = boto.connect_cloudwatch()
conn.put_metric_data(
namespace="tester", conn.put_metric_data(
name="metric", namespace="tester",
value=1.5, name="metric",
dimensions={"InstanceId": ["i-0123456,i-0123457"]}, value=1.5,
) dimensions={"InstanceId": ["i-0123456,i-0123457"]},
)
metrics = conn.list_metrics()
metrics.should.have.length_of(1) metrics = conn.list_metrics()
metric = metrics[0] metrics.should.have.length_of(1)
metric.namespace.should.equal("tester") metric = metrics[0]
metric.name.should.equal("metric") metric.namespace.should.equal("tester")
dict(metric.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]}) metric.name.should.equal("metric")
dict(metric.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
@mock_cloudwatch_deprecated
def test_describe_alarms(): @mock_cloudwatch_deprecated
conn = boto.connect_cloudwatch() def test_describe_alarms():
conn = boto.connect_cloudwatch()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0) alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
conn.create_alarm(alarm_fixture(name="nfoobar", action="afoobar"))
conn.create_alarm(alarm_fixture(name="nfoobaz", action="afoobaz")) conn.create_alarm(alarm_fixture(name="nfoobar", action="afoobar"))
conn.create_alarm(alarm_fixture(name="nbarfoo", action="abarfoo")) conn.create_alarm(alarm_fixture(name="nfoobaz", action="afoobaz"))
conn.create_alarm(alarm_fixture(name="nbazfoo", action="abazfoo")) conn.create_alarm(alarm_fixture(name="nbarfoo", action="abarfoo"))
conn.create_alarm(alarm_fixture(name="nbazfoo", action="abazfoo"))
alarms = conn.describe_alarms()
alarms.should.have.length_of(4) alarms = conn.describe_alarms()
alarms = conn.describe_alarms(alarm_name_prefix="nfoo") alarms.should.have.length_of(4)
alarms.should.have.length_of(2) alarms = conn.describe_alarms(alarm_name_prefix="nfoo")
alarms = conn.describe_alarms(alarm_names=["nfoobar", "nbarfoo", "nbazfoo"]) alarms.should.have.length_of(2)
alarms.should.have.length_of(3) alarms = conn.describe_alarms(alarm_names=["nfoobar", "nbarfoo", "nbazfoo"])
alarms = conn.describe_alarms(action_prefix="afoo") alarms.should.have.length_of(3)
alarms.should.have.length_of(2) alarms = conn.describe_alarms(action_prefix="afoo")
alarms.should.have.length_of(2)
for alarm in conn.describe_alarms():
alarm.delete() for alarm in conn.describe_alarms():
alarm.delete()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0) alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
@mock_cloudwatch_deprecated
def test_get_metric_statistics():
conn = boto.connect_cloudwatch()
metric_timestamp = datetime(2018, 4, 9, 13, 0, 0, 0)
conn.put_metric_data(
namespace="tester",
name="metric",
value=1.5,
dimensions={"InstanceId": ["i-0123456,i-0123457"]},
timestamp=metric_timestamp,
)
metric_kwargs = dict(
namespace="tester",
metric_name="metric",
start_time=metric_timestamp,
end_time=datetime.now(),
period=3600,
statistics=["Minimum"],
)
datapoints = conn.get_metric_statistics(**metric_kwargs)
datapoints.should.have.length_of(1)
datapoint = datapoints[0]
datapoint.should.have.key("Minimum").which.should.equal(1.5)
datapoint.should.have.key("Timestamp").which.should.equal(metric_timestamp)