OLD | NEW |
(Empty) | |
| 1 # Copyright (c) 2010 Hunter Blanks http://artifex.org/~hblanks/ |
| 2 # All rights reserved. |
| 3 # |
| 4 # Permission is hereby granted, free of charge, to any person obtaining a |
| 5 # copy of this software and associated documentation files (the |
| 6 # "Software"), to deal in the Software without restriction, including |
| 7 # without limitation the rights to use, copy, modify, merge, publish, dis- |
| 8 # tribute, sublicense, and/or sell copies of the Software, and to permit |
| 9 # persons to whom the Software is furnished to do so, subject to the fol- |
| 10 # lowing conditions: |
| 11 # |
| 12 # The above copyright notice and this permission notice shall be included |
| 13 # in all copies or substantial portions of the Software. |
| 14 # |
| 15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
| 16 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
| 17 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
| 18 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
| 19 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
| 20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
| 21 # IN THE SOFTWARE. |
| 22 |
| 23 """ |
| 24 Initial, and very limited, unit tests for CloudWatchConnection. |
| 25 """ |
| 26 |
| 27 import datetime |
| 28 import time |
| 29 import unittest |
| 30 |
| 31 from boto.ec2.cloudwatch import CloudWatchConnection |
| 32 from boto.ec2.cloudwatch.metric import Metric |
| 33 |
| 34 class CloudWatchConnectionTest(unittest.TestCase): |
| 35 |
| 36 def test_build_list_params(self): |
| 37 c = CloudWatchConnection() |
| 38 params = {} |
| 39 c.build_list_params( |
| 40 params, ['thing1', 'thing2', 'thing3'], 'ThingName%d') |
| 41 expected_params = { |
| 42 'ThingName1': 'thing1', |
| 43 'ThingName2': 'thing2', |
| 44 'ThingName3': 'thing3' |
| 45 } |
| 46 self.assertEqual(params, expected_params) |
| 47 |
| 48 def test_build_put_params_one(self): |
| 49 c = CloudWatchConnection() |
| 50 params = {} |
| 51 c.build_put_params(params, name="N", value=1, dimensions={"D": "V"}) |
| 52 expected_params = { |
| 53 'MetricData.member.1.MetricName': 'N', |
| 54 'MetricData.member.1.Value': 1, |
| 55 'MetricData.member.1.Dimensions.member.1.Name': 'D', |
| 56 'MetricData.member.1.Dimensions.member.1.Value': 'V', |
| 57 } |
| 58 self.assertEqual(params, expected_params) |
| 59 |
| 60 def test_build_put_params_multiple_metrics(self): |
| 61 c = CloudWatchConnection() |
| 62 params = {} |
| 63 c.build_put_params(params, name=["N", "M"], value=[1, 2], dimensions={"D
": "V"}) |
| 64 expected_params = { |
| 65 'MetricData.member.1.MetricName': 'N', |
| 66 'MetricData.member.1.Value': 1, |
| 67 'MetricData.member.1.Dimensions.member.1.Name': 'D', |
| 68 'MetricData.member.1.Dimensions.member.1.Value': 'V', |
| 69 'MetricData.member.2.MetricName': 'M', |
| 70 'MetricData.member.2.Value': 2, |
| 71 'MetricData.member.2.Dimensions.member.1.Name': 'D', |
| 72 'MetricData.member.2.Dimensions.member.1.Value': 'V', |
| 73 } |
| 74 self.assertEqual(params, expected_params) |
| 75 |
| 76 def test_build_put_params_multiple_dimensions(self): |
| 77 c = CloudWatchConnection() |
| 78 params = {} |
| 79 c.build_put_params(params, name="N", value=[1, 2], dimensions=[{"D": "V"
}, {"D": "W"}]) |
| 80 expected_params = { |
| 81 'MetricData.member.1.MetricName': 'N', |
| 82 'MetricData.member.1.Value': 1, |
| 83 'MetricData.member.1.Dimensions.member.1.Name': 'D', |
| 84 'MetricData.member.1.Dimensions.member.1.Value': 'V', |
| 85 'MetricData.member.2.MetricName': 'N', |
| 86 'MetricData.member.2.Value': 2, |
| 87 'MetricData.member.2.Dimensions.member.1.Name': 'D', |
| 88 'MetricData.member.2.Dimensions.member.1.Value': 'W', |
| 89 } |
| 90 self.assertEqual(params, expected_params) |
| 91 |
| 92 def test_build_put_params_invalid(self): |
| 93 c = CloudWatchConnection() |
| 94 params = {} |
| 95 try: |
| 96 c.build_put_params(params, name=["N", "M"], value=[1, 2, 3]) |
| 97 except: |
| 98 pass |
| 99 else: |
| 100 self.fail("Should not accept lists of different lengths.") |
| 101 |
| 102 def test_get_metric_statistics(self): |
| 103 c = CloudWatchConnection() |
| 104 m = c.list_metrics()[0] |
| 105 end = datetime.datetime.now() |
| 106 start = end - datetime.timedelta(hours=24*14) |
| 107 c.get_metric_statistics( |
| 108 3600*24, start, end, m.name, m.namespace, ['Average', 'Sum']) |
| 109 |
| 110 def test_put_metric_data(self): |
| 111 c = CloudWatchConnection() |
| 112 now = datetime.datetime.now() |
| 113 name, namespace = 'unit-test-metric', 'boto-unit-test' |
| 114 c.put_metric_data(namespace, name, 5, now, 'Bytes') |
| 115 |
| 116 # Uncomment the following lines for a slower but more thorough |
| 117 # test. (Hurrah for eventual consistency...) |
| 118 # |
| 119 # metric = Metric(connection=c) |
| 120 # metric.name = name |
| 121 # metric.namespace = namespace |
| 122 # time.sleep(60) |
| 123 # l = metric.query( |
| 124 # now - datetime.timedelta(seconds=60), |
| 125 # datetime.datetime.now(), |
| 126 # 'Average') |
| 127 # assert l |
| 128 # for row in l: |
| 129 # self.assertEqual(row['Unit'], 'Bytes') |
| 130 # self.assertEqual(row['Average'], 5.0) |
| 131 |
| 132 if __name__ == '__main__': |
| 133 unittest.main() |
OLD | NEW |