Merge pull request #2599 from bblommers/feature/#2010

CloudWatch - Implement list_metrics pagination

tests/test_cloudwatch/test_cloudwatch.py

@@ -1,8 +1,5 @@
 import boto
 from boto.ec2.cloudwatch.alarm import MetricAlarm
-import boto3
-from datetime import datetime, timedelta
-import pytz
 import sure  # noqa
 
 from moto import mock_cloudwatch_deprecated

tests/test_cloudwatch/test_cloudwatch_boto3.py

@@ -1,8 +1,10 @@
-from __future__ import unicode_literals
+# from __future__ import unicode_literals
 
 import boto3
 from botocore.exceptions import ClientError
 from datetime import datetime, timedelta
+from nose.tools import assert_raises
+from uuid import uuid4
 import pytz
 import sure  # noqa
 
@@ -155,13 +157,14 @@ def test_put_metric_data_no_dimensions():
 @mock_cloudwatch
 def test_put_metric_data_with_statistics():
     conn = boto3.client("cloudwatch", region_name="us-east-1")
+    utc_now = datetime.now(tz=pytz.utc)
 
     conn.put_metric_data(
         Namespace="tester",
         MetricData=[
             dict(
                 MetricName="statmetric",
-                Timestamp=datetime(2015, 1, 1),
+                Timestamp=utc_now,
                 # no Value to test  https://github.com/spulec/moto/issues/1615
                 StatisticValues=dict(
                     SampleCount=123.0, Sum=123.0, Minimum=123.0, Maximum=123.0
@@ -203,3 +206,86 @@ def test_get_metric_statistics():
     datapoint = stats["Datapoints"][0]
     datapoint["SampleCount"].should.equal(1.0)
     datapoint["Sum"].should.equal(1.5)
+
+
+@mock_cloudwatch
+def test_list_metrics():
+    cloudwatch = boto3.client("cloudwatch", "eu-west-1")
+    # Verify namespace has to exist
+    res = cloudwatch.list_metrics(Namespace="unknown/")["Metrics"]
+    res.should.be.empty
+    # Create some metrics to filter on
+    create_metrics(cloudwatch, namespace="list_test_1/", metrics=4, data_points=2)
+    create_metrics(cloudwatch, namespace="list_test_2/", metrics=4, data_points=2)
+    # Verify we can retrieve everything
+    res = cloudwatch.list_metrics()["Metrics"]
+    len(res).should.equal(16)  # 2 namespaces * 4 metrics * 2 data points
+    # Verify we can filter by namespace/metric name
+    res = cloudwatch.list_metrics(Namespace="list_test_1/")["Metrics"]
+    len(res).should.equal(8)  # 1 namespace * 4 metrics * 2 data points
+    res = cloudwatch.list_metrics(Namespace="list_test_1/", MetricName="metric1")[
+        "Metrics"
+    ]
+    len(res).should.equal(2)  # 1 namespace * 1 metrics * 2 data points
+    # Verify format
+    res.should.equal(
+        [
+            {u"Namespace": "list_test_1/", u"Dimensions": [], u"MetricName": "metric1"},
+            {u"Namespace": "list_test_1/", u"Dimensions": [], u"MetricName": "metric1"},
+        ]
+    )
+    # Verify unknown namespace still has no results
+    res = cloudwatch.list_metrics(Namespace="unknown/")["Metrics"]
+    res.should.be.empty
+
+
+@mock_cloudwatch
+def test_list_metrics_paginated():
+    cloudwatch = boto3.client("cloudwatch", "eu-west-1")
+    # Verify that only a single page of metrics is returned
+    cloudwatch.list_metrics()["Metrics"].should.be.empty
+    # Verify we can't pass a random NextToken
+    with assert_raises(ClientError) as e:
+        cloudwatch.list_metrics(NextToken=str(uuid4()))
+    e.exception.response["Error"]["Message"].should.equal(
+        "Request parameter NextToken is invalid"
+    )
+    # Add a boatload of metrics
+    create_metrics(cloudwatch, namespace="test", metrics=100, data_points=1)
+    # Verify that a single page is returned until we've reached 500
+    first_page = cloudwatch.list_metrics()
+    first_page["Metrics"].shouldnt.be.empty
+    len(first_page["Metrics"]).should.equal(100)
+    create_metrics(cloudwatch, namespace="test", metrics=200, data_points=2)
+    first_page = cloudwatch.list_metrics()
+    len(first_page["Metrics"]).should.equal(500)
+    first_page.shouldnt.contain("NextToken")
+    # Verify that adding more data points results in pagination
+    create_metrics(cloudwatch, namespace="test", metrics=60, data_points=10)
+    first_page = cloudwatch.list_metrics()
+    len(first_page["Metrics"]).should.equal(500)
+    first_page["NextToken"].shouldnt.be.empty
+    # Retrieve second page - and verify there's more where that came from
+    second_page = cloudwatch.list_metrics(NextToken=first_page["NextToken"])
+    len(second_page["Metrics"]).should.equal(500)
+    second_page.should.contain("NextToken")
+    # Last page should only have the last 100 results, and no NextToken (indicating that pagination is finished)
+    third_page = cloudwatch.list_metrics(NextToken=second_page["NextToken"])
+    len(third_page["Metrics"]).should.equal(100)
+    third_page.shouldnt.contain("NextToken")
+    # Verify that we can't reuse an existing token
+    with assert_raises(ClientError) as e:
+        cloudwatch.list_metrics(NextToken=first_page["NextToken"])
+    e.exception.response["Error"]["Message"].should.equal(
+        "Request parameter NextToken is invalid"
+    )
+
+
+def create_metrics(cloudwatch, namespace, metrics=5, data_points=5):
+    for i in range(0, metrics):
+        metric_name = "metric" + str(i)
+        for j in range(0, data_points):
+            cloudwatch.put_metric_data(
+                Namespace=namespace,
+                MetricData=[{"MetricName": metric_name, "Value": j, "Unit": "Seconds"}],
+            )