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

Added dashboard methods + tests

Browse Source
This commit is contained in:
Terry Cain 2017-09-22 16:38:20 +01:00
parent 4029afeb5b
commit c965fdd47f
4 changed files with 269 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
moto/cloudwatch/models.py
View File

@ -2,6 +2,11 @@ from moto.core import BaseBackend, BaseModel
import boto.ec2.cloudwatch
import datetime
from .utils import make_arn_for_dashboard
DEFAULT_ACCOUNT_ID = 123456789012
class Dimension(object):
@ -44,10 +49,34 @@ class MetricDatum(BaseModel):
'value']) for dimension in dimensions]
class Dashboard(BaseModel):
def __init__(self, name, body):
# Guaranteed to be unique for now as the name is also the key of a dictionary where they are stored
self.arn = make_arn_for_dashboard(DEFAULT_ACCOUNT_ID, name)
self.name = name
self.body = body
self.last_modified = datetime.datetime.now()
@property
def last_modified_iso(self):
return self.last_modified.isoformat()
@property
def size(self):
return len(self)
def __len__(self):
return len(self.body)
def __repr__(self):
return '<CloudWatchDashboard {0}>'.format(self.name)
class CloudWatchBackend(BaseBackend):
def __init__(self):
self.alarms = {}
self.dashboards = {}
self.metric_data = []
def put_metric_alarm(self, name, namespace, metric_name, comparison_operator, evaluation_periods,
@ -110,6 +139,31 @@ class CloudWatchBackend(BaseBackend):
def get_all_metrics(self):
return self.metric_data
def put_dashboard(self, name, body):
self.dashboards[name] = Dashboard(name, body)
def list_dashboards(self, prefix=''):
for key, value in self.dashboards.items():
if key.startswith(prefix):
yield value
def delete_dashboards(self, dashboards):
to_delete = set(dashboards)
all_dashboards = set(self.dashboards.keys())
left_over = to_delete - all_dashboards
if len(left_over) > 0:
# Some dashboards are not found
return False, 'The specified dashboard does not exist. [{0}]'.format(', '.join(left_over))
for dashboard in to_delete:
del self.dashboards[dashboard]
return True, None
def get_dashboard(self, dashboard):
return self.dashboards.get(dashboard)
class LogGroup(BaseModel):

141
moto/cloudwatch/responses.py
View File

@ -1,9 +1,18 @@
import json
from moto.core.responses import BaseResponse
from .models import cloudwatch_backends
class CloudWatchResponse(BaseResponse):
@property
def cloudwatch_backend(self):
return cloudwatch_backends[self.region]
def _error(self, code, message, status=400):
template = self.response_template(ERROR_RESPONSE_TEMPLATE)
return template.render(code=code, message=message), dict(status=status)
def put_metric_alarm(self):
name = self._get_param('AlarmName')
namespace = self._get_param('Namespace')
@ -20,15 +29,14 @@ class CloudWatchResponse(BaseResponse):
insufficient_data_actions = self._get_multi_param(
"InsufficientDataActions.member")
unit = self._get_param('Unit')
cloudwatch_backend = cloudwatch_backends[self.region]
alarm = cloudwatch_backend.put_metric_alarm(name, namespace, metric_name,
comparison_operator,
evaluation_periods, period,
threshold, statistic,
description, dimensions,
alarm_actions, ok_actions,
insufficient_data_actions,
unit)
alarm = self.cloudwatch_backend.put_metric_alarm(name, namespace, metric_name,
comparison_operator,
evaluation_periods, period,
threshold, statistic,
description, dimensions,
alarm_actions, ok_actions,
insufficient_data_actions,
unit)
template = self.response_template(PUT_METRIC_ALARM_TEMPLATE)
return template.render(alarm=alarm)
@ -37,28 +45,26 @@ class CloudWatchResponse(BaseResponse):
alarm_name_prefix = self._get_param('AlarmNamePrefix')
alarm_names = self._get_multi_param('AlarmNames.member')
state_value = self._get_param('StateValue')
cloudwatch_backend = cloudwatch_backends[self.region]
if action_prefix:
alarms = cloudwatch_backend.get_alarms_by_action_prefix(
alarms = self.cloudwatch_backend.get_alarms_by_action_prefix(
action_prefix)
elif alarm_name_prefix:
alarms = cloudwatch_backend.get_alarms_by_alarm_name_prefix(
alarms = self.cloudwatch_backend.get_alarms_by_alarm_name_prefix(
alarm_name_prefix)
elif alarm_names:
alarms = cloudwatch_backend.get_alarms_by_alarm_names(alarm_names)
alarms = self.cloudwatch_backend.get_alarms_by_alarm_names(alarm_names)
elif state_value:
alarms = cloudwatch_backend.get_alarms_by_state_value(state_value)
alarms = self.cloudwatch_backend.get_alarms_by_state_value(state_value)
else:
alarms = cloudwatch_backend.get_all_alarms()
alarms = self.cloudwatch_backend.get_all_alarms()
template = self.response_template(DESCRIBE_ALARMS_TEMPLATE)
return template.render(alarms=alarms)
def delete_alarms(self):
alarm_names = self._get_multi_param('AlarmNames.member')
cloudwatch_backend = cloudwatch_backends[self.region]
cloudwatch_backend.delete_alarms(alarm_names)
self.cloudwatch_backend.delete_alarms(alarm_names)
template = self.response_template(DELETE_METRIC_ALARMS_TEMPLATE)
return template.render()
@ -89,19 +95,26 @@ class CloudWatchResponse(BaseResponse):
dimension_index += 1
metric_data.append([metric_name, value, dimensions])
metric_index += 1
cloudwatch_backend = cloudwatch_backends[self.region]
cloudwatch_backend.put_metric_data(namespace, metric_data)
self.cloudwatch_backend.put_metric_data(namespace, metric_data)
template = self.response_template(PUT_METRIC_DATA_TEMPLATE)
return template.render()
def list_metrics(self):
cloudwatch_backend = cloudwatch_backends[self.region]
metrics = cloudwatch_backend.get_all_metrics()
metrics = self.cloudwatch_backend.get_all_metrics()
template = self.response_template(LIST_METRICS_TEMPLATE)
return template.render(metrics=metrics)
def delete_dashboards(self):
raise NotImplementedError()
dashboards = self._get_multi_param('DashboardNames.member')
if dashboards is None:
return self._error('InvalidParameterValue', 'Need at least 1 dashboard')
status, error = self.cloudwatch_backend.delete_dashboards(dashboards)
if not status:
return self._error('ResourceNotFound', error)
template = self.response_template(DELETE_DASHBOARD_TEMPLATE)
return template.render()
def describe_alarm_history(self):
raise NotImplementedError()
@ -116,16 +129,39 @@ class CloudWatchResponse(BaseResponse):
raise NotImplementedError()
def get_dashboard(self):
raise NotImplementedError()
dashboard_name = self._get_param('DashboardName')
dashboard = self.cloudwatch_backend.get_dashboard(dashboard_name)
if dashboard is None:
return self._error('ResourceNotFound', 'Dashboard does not exist')
template = self.response_template(GET_DASHBOARD_TEMPLATE)
return template.render(dashboard=dashboard)
def get_metric_statistics(self):
raise NotImplementedError()
def list_dashboards(self):
raise NotImplementedError()
prefix = self._get_param('DashboardNamePrefix', '')
dashboards = self.cloudwatch_backend.list_dashboards(prefix)
template = self.response_template(LIST_DASHBOARD_RESPONSE)
return template.render(dashboards=dashboards)
def put_dashboard(self):
raise NotImplementedError()
name = self._get_param('DashboardName')
body = self._get_param('DashboardBody')
try:
json.loads(body)
except ValueError:
return self._error('InvalidParameterInput', 'Body is invalid JSON')
self.cloudwatch_backend.put_dashboard(name, body)
template = self.response_template(PUT_DASHBOARD_RESPONSE)
return template.render()
def set_alarm_state(self):
raise NotImplementedError()
@ -229,3 +265,58 @@ LIST_METRICS_TEMPLATE = """<ListMetricsResponse xmlns="http://monitoring.amazona
</NextToken>
</ListMetricsResult>
</ListMetricsResponse>"""
PUT_DASHBOARD_RESPONSE = """<PutDashboardResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<PutDashboardResult>
<DashboardValidationMessages/>
</PutDashboardResult>
<ResponseMetadata>
<RequestId>44b1d4d8-9fa3-11e7-8ad3-41b86ac5e49e</RequestId>
</ResponseMetadata>
</PutDashboardResponse>"""
LIST_DASHBOARD_RESPONSE = """<ListDashboardsResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<ListDashboardsResult>
<DashboardEntries>
{% for dashboard in dashboards %}
<member>
<DashboardArn>{{ dashboard.arn }}</DashboardArn>
<LastModified>{{ dashboard.last_modified_iso }}</LastModified>
<Size>{{ dashboard.size }}</Size>
<DashboardName>{{ dashboard.name }}</DashboardName>
</member>
{% endfor %}
</DashboardEntries>
</ListDashboardsResult>
<ResponseMetadata>
<RequestId>c3773873-9fa5-11e7-b315-31fcc9275d62</RequestId>
</ResponseMetadata>
</ListDashboardsResponse>"""
DELETE_DASHBOARD_TEMPLATE = """<DeleteDashboardsResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<DeleteDashboardsResult/>
<ResponseMetadata>
<RequestId>68d1dc8c-9faa-11e7-a694-df2715690df2</RequestId>
</ResponseMetadata>
</DeleteDashboardsResponse>"""
GET_DASHBOARD_TEMPLATE = """<GetDashboardResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<GetDashboardResult>
<DashboardArn>{{ dashboard.arn }}</DashboardArn>
<DashboardBody>{{ dashboard.body }}</DashboardBody>
<DashboardName>{{ dashboard.name }}</DashboardName>
</GetDashboardResult>
<ResponseMetadata>
<RequestId>e3c16bb0-9faa-11e7-b315-31fcc9275d62</RequestId>
</ResponseMetadata>
</GetDashboardResponse>
"""
ERROR_RESPONSE_TEMPLATE = """<ErrorResponse xmlns="http://monitoring.amazonaws.com/doc/2010-08-01/">
<Error>
<Type>Sender</Type>
<Code>{{ code }}</Code>
<Message>{{ message }}</Message>
</Error>
<RequestId>5e45fd1e-9fa3-11e7-b720-89e8821d38c4</RequestId>
</ErrorResponse>"""

5
moto/cloudwatch/utils.py Normal file
View File

@ -0,0 +1,5 @@
from __future__ import unicode_literals
def make_arn_for_dashboard(account_id, name):
return "arn:aws:cloudwatch::{0}dashboard/{1}".format(account_id, name)

94
tests/test_cloudwatch/test_cloudwatch_boto3.py Normal file
View File

@ -0,0 +1,94 @@
from __future__ import unicode_literals
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 of raised error')