Index: tests/ec2/cloudwatch/test_connection.py |
diff --git a/tests/ec2/cloudwatch/test_connection.py b/tests/ec2/cloudwatch/test_connection.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..c549c1d2f7e1c9fa97e326500cebb4a938ab8624 |
--- /dev/null |
+++ b/tests/ec2/cloudwatch/test_connection.py |
@@ -0,0 +1,133 @@ |
+# Copyright (c) 2010 Hunter Blanks http://artifex.org/~hblanks/ |
+# All rights reserved. |
+# |
+# Permission is hereby granted, free of charge, to any person obtaining a |
+# copy of this software and associated documentation files (the |
+# "Software"), to deal in the Software without restriction, including |
+# without limitation the rights to use, copy, modify, merge, publish, dis- |
+# tribute, sublicense, and/or sell copies of the Software, and to permit |
+# persons to whom the Software is furnished to do so, subject to the fol- |
+# lowing conditions: |
+# |
+# The above copyright notice and this permission notice shall be included |
+# in all copies or substantial portions of the Software. |
+# |
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
+# IN THE SOFTWARE. |
+ |
+""" |
+Initial, and very limited, unit tests for CloudWatchConnection. |
+""" |
+ |
+import datetime |
+import time |
+import unittest |
+ |
+from boto.ec2.cloudwatch import CloudWatchConnection |
+from boto.ec2.cloudwatch.metric import Metric |
+ |
+class CloudWatchConnectionTest(unittest.TestCase): |
+ |
+ def test_build_list_params(self): |
+ c = CloudWatchConnection() |
+ params = {} |
+ c.build_list_params( |
+ params, ['thing1', 'thing2', 'thing3'], 'ThingName%d') |
+ expected_params = { |
+ 'ThingName1': 'thing1', |
+ 'ThingName2': 'thing2', |
+ 'ThingName3': 'thing3' |
+ } |
+ self.assertEqual(params, expected_params) |
+ |
+ def test_build_put_params_one(self): |
+ c = CloudWatchConnection() |
+ params = {} |
+ c.build_put_params(params, name="N", value=1, dimensions={"D": "V"}) |
+ expected_params = { |
+ 'MetricData.member.1.MetricName': 'N', |
+ 'MetricData.member.1.Value': 1, |
+ 'MetricData.member.1.Dimensions.member.1.Name': 'D', |
+ 'MetricData.member.1.Dimensions.member.1.Value': 'V', |
+ } |
+ self.assertEqual(params, expected_params) |
+ |
+ def test_build_put_params_multiple_metrics(self): |
+ c = CloudWatchConnection() |
+ params = {} |
+ c.build_put_params(params, name=["N", "M"], value=[1, 2], dimensions={"D": "V"}) |
+ expected_params = { |
+ 'MetricData.member.1.MetricName': 'N', |
+ 'MetricData.member.1.Value': 1, |
+ 'MetricData.member.1.Dimensions.member.1.Name': 'D', |
+ 'MetricData.member.1.Dimensions.member.1.Value': 'V', |
+ 'MetricData.member.2.MetricName': 'M', |
+ 'MetricData.member.2.Value': 2, |
+ 'MetricData.member.2.Dimensions.member.1.Name': 'D', |
+ 'MetricData.member.2.Dimensions.member.1.Value': 'V', |
+ } |
+ self.assertEqual(params, expected_params) |
+ |
+ def test_build_put_params_multiple_dimensions(self): |
+ c = CloudWatchConnection() |
+ params = {} |
+ c.build_put_params(params, name="N", value=[1, 2], dimensions=[{"D": "V"}, {"D": "W"}]) |
+ expected_params = { |
+ 'MetricData.member.1.MetricName': 'N', |
+ 'MetricData.member.1.Value': 1, |
+ 'MetricData.member.1.Dimensions.member.1.Name': 'D', |
+ 'MetricData.member.1.Dimensions.member.1.Value': 'V', |
+ 'MetricData.member.2.MetricName': 'N', |
+ 'MetricData.member.2.Value': 2, |
+ 'MetricData.member.2.Dimensions.member.1.Name': 'D', |
+ 'MetricData.member.2.Dimensions.member.1.Value': 'W', |
+ } |
+ self.assertEqual(params, expected_params) |
+ |
+ def test_build_put_params_invalid(self): |
+ c = CloudWatchConnection() |
+ params = {} |
+ try: |
+ c.build_put_params(params, name=["N", "M"], value=[1, 2, 3]) |
+ except: |
+ pass |
+ else: |
+ self.fail("Should not accept lists of different lengths.") |
+ |
+ def test_get_metric_statistics(self): |
+ c = CloudWatchConnection() |
+ m = c.list_metrics()[0] |
+ end = datetime.datetime.now() |
+ start = end - datetime.timedelta(hours=24*14) |
+ c.get_metric_statistics( |
+ 3600*24, start, end, m.name, m.namespace, ['Average', 'Sum']) |
+ |
+ def test_put_metric_data(self): |
+ c = CloudWatchConnection() |
+ now = datetime.datetime.now() |
+ name, namespace = 'unit-test-metric', 'boto-unit-test' |
+ c.put_metric_data(namespace, name, 5, now, 'Bytes') |
+ |
+ # Uncomment the following lines for a slower but more thorough |
+ # test. (Hurrah for eventual consistency...) |
+ # |
+ # metric = Metric(connection=c) |
+ # metric.name = name |
+ # metric.namespace = namespace |
+ # time.sleep(60) |
+ # l = metric.query( |
+ # now - datetime.timedelta(seconds=60), |
+ # datetime.datetime.now(), |
+ # 'Average') |
+ # assert l |
+ # for row in l: |
+ # self.assertEqual(row['Unit'], 'Bytes') |
+ # self.assertEqual(row['Average'], 5.0) |
+ |
+if __name__ == '__main__': |
+ unittest.main() |