KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use cloudwatch_api method in localstack
Best Python code snippet using localstack_python
cloudwatch_api_test.py
Source:cloudwatch_api_test.py 
1#!/usr/bin/env python2# ----------------------------------------------------------------------3# Numenta Platform for Intelligent Computing (NuPIC)4# Copyright (C) 2015, Numenta, Inc. Unless you have purchased from5# Numenta, Inc. a separate commercial license for this software code, the6# following terms and conditions apply:7#8# This program is free software: you can redistribute it and/or modify9# it under the terms of the GNU Affero Public License version 3 as10# published by the Free Software Foundation.11#12# This program is distributed in the hope that it will be useful,13# but WITHOUT ANY WARRANTY; without even the implied warranty of14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.15# See the GNU Affero Public License for more details.16#17# You should have received a copy of the GNU Affero Public License18# along with this program. If not, see http://www.gnu.org/licenses.19#20# http://numenta.org/licenses/21# ----------------------------------------------------------------------22"""23Unit tests for Cloudwatch API24"""25import unittest26from mock import patch27from paste.fixture import TestApp28import htm.it.app29from htmengine import utils as app_utils30import htm.it.app.adapters.datasource as datasource_adapter_factory31from htm.it.app.webservices import cloudwatch_api32from htm.it.test_utils.app.webservices import (33 getDefaultHTTPHeaders,34 getInvalidHTTPHeaders,35 webservices_assertions as assertions36)37class CWDefaultHandlerTest(unittest.TestCase):38 """39 Unit tests CWDefaultHandler40 """41 def setUp(self):42 self.headers = getDefaultHTTPHeaders(htm.it.app.config)43 self.app = TestApp(cloudwatch_api.app.wsgifunc())44 adapter = datasource_adapter_factory.createCloudwatchDatasourceAdapter()45 self.resources = adapter.describeSupportedMetrics()46 self.regions = adapter.describeRegions()47 def _getCloudWatchCommon(self, url, expectedResult):48 response = self.app.get(url, headers=self.headers)49 assertions.assertSuccess(self, response)50 result = app_utils.jsonDecode(response.body)51 self.assertIsInstance(result, dict)52 self.assertEqual(result, expectedResult)53 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."54 "createCloudwatchDatasourceAdapter")55 def testCWDefaultHandlerDefault(self, adapterMock):56 """57 Test for Get '/_metrics/cloudwatch'58 response is validated for appropriate headers, body and status59 """60 adapterMock.return_value.describeRegions.return_value = []61 adapterMock.return_value.describeSupportedMetrics.return_value = {}62 self._getCloudWatchCommon("", {"regions": {}, "namespaces": {}})63 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."64 "createCloudwatchDatasourceAdapter")65 def testCWDefaultHandlerDefaultWithSlash(self, adapterMock):66 """67 Test for Get '/_metrics/cloudwatch/'68 response is validated for appropriate headers, body and status69 """70 adapterMock.return_value.describeRegions.return_value = []71 adapterMock.return_value.describeSupportedMetrics.return_value = {}72 self._getCloudWatchCommon("/", {"regions": {}, "namespaces": {}})73class CWNamespaceHandlerTest(unittest.TestCase):74 """75 Unit tests CWNamespaceHandler76 """77 @classmethod78 def setUpClass(cls):79 adapter = datasource_adapter_factory.createCloudwatchDatasourceAdapter()80 cls.resources = adapter.describeSupportedMetrics()81 cls.regions = adapter.describeRegions()82 def setUp(self):83 self.headers = getDefaultHTTPHeaders(htm.it.app.config)84 self.app = TestApp(cloudwatch_api.app.wsgifunc())85 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."86 "createCloudwatchDatasourceAdapter")87 def testGetListNamespaceNoRegions(self, adapterMock):88 """89 Test for Get '/_metrics/cloudwatch/namespaces'90 response is validated for appropriate headers, body and status91 """92 adapterMock.return_value.describeRegions.return_value = []93 response = self.app.get("/namespaces", headers=self.headers)94 assertions.assertSuccess(self, response)95 result = app_utils.jsonDecode(response.body)96 # Added Autostacks namespaces to this list for now, to maintain API97 # backwards-compatibility during adapter refactor98 self.assertEqual(result, {'Autostacks': {'metrics': ['InstanceCount']}})99 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."100 "createCloudwatchDatasourceAdapter")101 def testGetNamespaceAWSType(self, adapterMock):102 """103 Test for Get '/_metrics/cloudwatch/AWS/<namespace>'104 response is validated for appropriate headers, body and status105 """106 adapterMock.return_value.describeRegions.return_value = self.regions107 adapterMock.return_value.describeSupportedMetrics.return_value = (108 self.resources)109 response = self.app.get("/AWS/EC2", headers=self.headers)110 #self.assertTrue(getMetrics.called)111 assertions.assertSuccess(self, response)112 result = app_utils.jsonDecode(response.body)113 self.assertEqual(result,114 {'AWS/EC2': {'dimensions': ['InstanceId',115 'AutoScalingGroupName'],116 'metrics': ['CPUUtilization',117 'NetworkIn',118 'DiskWriteBytes',119 'NetworkOut',120 'DiskReadBytes']}})121 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."122 "createCloudwatchDatasourceAdapter")123 def testGetNamespaceAWSInvalidType(self, adapterMock):124 """125 Test for Get '/_metrics/cloudwatch/AWS/<invalid_namespace>'126 response is validated for appropriate headers, body and status127 response is validated for error message128 """129 adapterMock.return_value.describeRegions.return_value = self.regions130 adapterMock.return_value.describeSupportedMetrics.return_value = (131 self.resources)132 response = self.app.get("/AWS/foo", headers=self.headers, status="*")133 assertions.assertNotFound(self, response)134 self.assertEqual("Namespace 'AWS/foo' was not found", response.body)135class TestCWRegionsHandler(unittest.TestCase):136 """137 Unit tests CWRegionsHandler138 """139 @classmethod140 def setUpClass(cls):141 adapter = datasource_adapter_factory.createCloudwatchDatasourceAdapter()142 cls.resources = adapter.describeSupportedMetrics()143 cls.regions = adapter.describeRegions()144 def setUp(self):145 self.headers = getDefaultHTTPHeaders(htm.it.app.config)146 self.app = TestApp(cloudwatch_api.app.wsgifunc())147 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."148 "createCloudwatchDatasourceAdapter")149 def testGetListRegionsEmptyResponse(self, adapterMock):150 """151 Test for Get '/_metrics/cloudwatch/regions'152 response is validated for appropriate headers, body and status153 """154 adapterMock.return_value.describeRegions.return_value = []155 response = self.app.get("/regions", headers=self.headers)156 assertions.assertSuccess(self, response)157 result = app_utils.jsonDecode(response.body)158 self.assertEqual(result, {})159 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."160 "createCloudwatchDatasourceAdapter")161 def testGetListRegionsNonEmptyResponse(self, adapterMock):162 """163 Test for Get '/_metrics/cloudwatch/regions'164 response is validated for appropriate headers, body and status165 and pre-defined values166 """167 adapterMock.return_value.describeRegions.return_value = self.regions168 response = self.app.get("/regions", headers=self.headers)169 assertions.assertSuccess(self, response)170 result = app_utils.jsonDecode(response.body)171 self.assertEqual(result, dict(self.regions))172 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."173 "createCloudwatchDatasourceAdapter")174 def testGetListRegionMetricsEmptyResponse(self, adapterMock):175 """176 Test for Get '/_metrics/cloudwatch/regions/<invalid-region-name>'177 response is validated for appropriate headers, body and status178 response is validated against expected error message179 """180 adapterMock.return_value.describeRegions.return_value = self.regions181 response = self.app.get('/regions/fake-region', status="*",182 headers=self.headers)183 assertions.assertNotFound(self, response)184 self.assertIn("Region 'fake-region' was not found", response.body)185class TestCWInstanceHandler(unittest.TestCase):186 """187 Units tests CWInstanceHandler188 """189 @classmethod190 def setUpClass(cls):191 adapter = datasource_adapter_factory.createCloudwatchDatasourceAdapter()192 cls.resources = adapter.describeSupportedMetrics()193 cls.regions = adapter.describeRegions()194 def setUp(self):195 self.headers = getDefaultHTTPHeaders(htm.it.app.config)196 self.app = TestApp(cloudwatch_api.app.wsgifunc())197 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."198 "createCloudwatchDatasourceAdapter")199 def testGetInstanceWithInvalidRegion(self, adapterMock):200 """201 Test for202 Get '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/instances/'203 with invalid region204 response is validated for appropriate headers, body and status205 and error message206 """207 adapterMock.return_value.describeRegions.return_value = self.regions208 adapterMock.return_value.describeSupportedMetrics.return_value = (209 self.resources)210 response = self.app.get("/fake-region/AWS/EC2/instances/i-832311",211 headers=self.headers, status="*")212 assertions.assertNotFound(self, response)213 self.assertEqual(response.body, "Region 'fake-region' was not found")214 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."215 "createCloudwatchDatasourceAdapter")216 def testGetInstanceWithInvalidNamespace(self, adapterMock):217 """218 Test for Get219 '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/instances/<InstancdId>'220 with invalid InstancceId221 response is validated for appropriate headers, body and status222 """223 adapterMock.return_value.describeRegions.return_value = self.regions224 adapterMock.return_value.describeSupportedMetrics.return_value = (225 self.resources)226 response = self.app.get("/us-east-1/AWS/foo/instances/i-832311",227 headers=self.headers, status="*")228 assertions.assertNotFound(self, response)229 self.assertEqual(response.body, "Namespace 'AWS/foo' was not found")230 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."231 "createCloudwatchDatasourceAdapter")232 def testGetInstanceWithDimension(self,adapterMock):233 """234 Test for Get235 '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/\236 instances/<InstancdId>'237 with valid InstancceId238 response is validated for appropriate headers, body and status239 """240 adapterMock.return_value.describeRegions.return_value = self.regions241 adapterMock.return_value.describeSupportedMetrics.return_value = (242 self.resources)243 adapterMock.return_value.describeResources.return_value = [244 {'grn': u'aws://us-west-2/Instance/i-548acc3a',245 'name': u'Bar',246 'resID': u'i-548acc3a'}]247 response = self.app.get("/us-east-1/AWS/EC2/instances/i-548acc3a",248 headers=self.headers)249 assertions.assertSuccess(self, response)250 result = app_utils.jsonDecode(response.body)251 self.assertTrue(all(metric["metric"]252 in self.resources['AWS::EC2::Instance']253 for metric in result))254 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."255 "createCloudwatchDatasourceAdapter")256 def testGetInstanceWithDimensionNotFoundEmpty(self, adapterMock):257 """258 Test for Get259 '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/\260 instances/<InstancdId>/Dimenion'261 with invalid InstancceId262 response is validated for appropriate headers, body and status263 Tests scenario when given instance name is not264 present in ouput from getDimensionsFromRegion265 """266 adapterMock.return_value.describeRegions.return_value = self.regions267 adapterMock.return_value.describeSupportedMetrics.return_value = (268 self.resources)269 adapterMock.return_value.describeResources.return_value = [270 {'grn': u'aws://us-west-2/Instance/i-548acc3a',271 'name': u'Bar',272 'resID': u'i-548acc3a'}]273 response = self.app.get("/us-east-1/AWS/EC2/instances/i-1234567a",274 headers=self.headers)275 assertions.assertSuccess(self, response)276 result = app_utils.jsonDecode(response.body)277 self.assertEqual(result, [])278class TestCWMetricHandler(unittest.TestCase):279 """280 Unit tests CWMetricHandler281 """282 @classmethod283 def setUpClass(cls):284 adapter = datasource_adapter_factory.createCloudwatchDatasourceAdapter()285 cls.resources = adapter.describeSupportedMetrics()286 cls.regions = adapter.describeRegions()287 def setUp(self):288 self.headers = getDefaultHTTPHeaders(htm.it.app.config)289 self.app = TestApp(cloudwatch_api.app.wsgifunc())290 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."291 "createCloudwatchDatasourceAdapter")292 def testGetMetricsWithInvalidRegion(self, adapterMock):293 """294 Test for295 Get '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/metricName'296 with invalid region297 response is validated for appropriate headers, body and status298 and reponse is validated for error message299 """300 adapterMock.return_value.describeRegions.return_value = self.regions301 adapterMock.return_value.describeSupportedMetrics.return_value = (302 self.resources)303 response = self.app.get("/fake-region/AWS/EC2/CPUUtilization",304 headers=self.headers, status="*")305 assertions.assertNotFound(self, response)306 self.assertEqual(response.body, "Region 'fake-region' was not found")307 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."308 "createCloudwatchDatasourceAdapter")309 def testGetMetricInvalidNamespace(self, adapterMock):310 """311 Test for312 Get '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/metricName'313 with invalid namespace314 response is validated for appropriate headers, body and status315 and reponse is validated for error message316 """317 adapterMock.return_value.describeRegions.return_value = self.regions318 adapterMock.return_value.describeSupportedMetrics.return_value = (319 self.resources)320 response = self.app.get("/us-east-1/AWS/foo/CPUUtilization",321 headers=self.headers, status="*")322 assertions.assertNotFound(self, response)323 self.assertEqual(response.body, "Namespace 'AWS/foo' was not found")324 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."325 "createCloudwatchDatasourceAdapter")326 def testGetMetricValidInputEmptyResponse(self, adapterMock):327 """328 Test for329 Get '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/metricName'330 response is validated for appropriate headers, body and status331 """332 adapterMock.return_value.describeRegions.return_value = self.regions333 adapterMock.return_value.describeSupportedMetrics.return_value = (334 self.resources)335 adapterMock.return_value.describeResources.return_value = []336 response = self.app.get("/us-east-1/AWS/EC2/CPUUtilization",337 headers=self.headers)338 assertions.assertSuccess(self, response)339 result = app_utils.jsonDecode(response.body)340 self.assertEqual(result, [])341 @patch("htm.it.app.webservices.cloudwatch_api.datasource_adapter_factory."342 "createCloudwatchDatasourceAdapter")343 def testGetMetricDimensionWithResponse(self, adapterMock):344 """345 Test for346 Get '/_metrics/cloudwatch/<region-name>/AWS/<namespace>/metricName'347 response is validated for appropriate headers, body and status348 and response is validated with pre-defined responses349 """350 adapterMock.return_value.describeRegions.return_value = self.regions351 adapterMock.return_value.describeSupportedMetrics.return_value = (352 self.resources)353 adapterMock.return_value.describeResources.return_value = [354 {'grn': u'aws://us-west-2/Instance/i-d48ccaba',355 'name': u'Foo',356 'resID': u'i-d48ccaba'},357 {'grn': u'aws://us-west-2/Instance/i-548acc3a',358 'name': u'Bar',359 'resID': u'i-548acc3a'}]360 response = self.app.get("/us-east-1/AWS/EC2/CPUUtilization",361 headers=self.headers)362 assertions.assertSuccess(self, response)363 result = app_utils.jsonDecode(response.body)364 self.assertItemsEqual(result,365 [366 {'datasource': 'cloudwatch',367 'dimensions': {'InstanceId': 'i-d48ccaba'},368 'metric': 'CPUUtilization',369 'namespace': 'AWS/EC2',370 'region': 'us-east-1'},371 {'datasource': 'cloudwatch',372 'dimensions': {'InstanceId': 'i-548acc3a'},373 'metric': 'CPUUtilization',374 'namespace': 'AWS/EC2',375 'region': 'us-east-1'},376 {'datasource': 'cloudwatch',377 'dimensions': {'AutoScalingGroupName': 'i-d48ccaba'},378 'metric': 'CPUUtilization',379 'namespace': 'AWS/EC2',380 'region': 'us-east-1'},381 {'datasource': 'cloudwatch',382 'dimensions': {'AutoScalingGroupName': 'i-548acc3a'},383 'metric': 'CPUUtilization',384 'namespace': 'AWS/EC2',385 'region': 'us-east-1'}386 ])387class CWApiUnhappyTest(unittest.TestCase):388 """389 Unhappy tests for Cloudwatch API390 """391 def setUp(self):392 self.app = TestApp(cloudwatch_api.app.wsgifunc())393 self.headers = getDefaultHTTPHeaders(htm.it.app.config)394 def testNoAuthHeaders(self):395 """396 negative test for authentication guarded route.397 invoke get request without passing authentication headers398 response is validated for appropriate headers and body399 """400 response = self.app.get("", status="*")401 assertions.assertInvalidAuthenticationResponse(self, response)402 def testInvalidAuthHeaders(self):403 """404 negative test for authentication guarded route.405 invoke get request with invalid authentication headers406 response is validated for appropriate headers and body407 """408 invalidHeaders = getInvalidHTTPHeaders()409 response = self.app.get("", status="*", headers=invalidHeaders)410 assertions.assertInvalidAuthenticationResponse(self, response)411 def testInvalidRoute(self):412 """413 Invoke non supported route414 resoponse is validated for appropriate headers and body415 """416 response = self.app.get("/foo", status="*", headers=self.headers)417 assertions.assertRouteNotFound(self, response)418 def testInvalidMethod(self):419 """420 Invoe non supported methods421 resoponse is validated for appropriate headers and body422 """423 response = self.app.post("", status="*", headers=self.headers)424 assertions.assertMethodNotAllowed(self, response)425 headers = dict(response.headers)426 self.assertEqual(headers["Allow"], "GET")427if __name__ == "__main__":...
aws-dashboard.py
Source:aws-dashboard.py
1#!/usr/bin/env python32# pylint: disable=C0301,C01113"""4A script for rendering AWS dashboards to be viewed without logging in.5Supply AWS credentials via an assigned role in IAM or environment variables.6Usage example: ./aws-dashboard.py all > foo.html7Logger runtime output will go to stderr.8"""9import base6410import datetime11import json12import logging13import sys14import boto315from botocore.exceptions import ClientError16__author__ = "Markus Koskinen"17__license__ = "BSD"18_LOG_FORMAT = "%(asctime)s\t%(levelname)s\t%(name)s\t%(message)s"19_DEBUG = False20logging.basicConfig(level=logging.INFO, format=_LOG_FORMAT)21def syntax(execname):22 print("Syntax: %s [dashboard_name]" % execname)23 sys.exit(1)24def auth():25 """ TODO: This not quite how it should be done. """26 try:27 api = boto3.client('cloudwatch')28 except ClientError as e:29 logging.error("Client error: %s", e)30 exit(1)31 return e32 else:33 return api34def get_dashboard_images(dashboard_name="all", encode_base64=True):35 """ Returns a list of base64 encoded PNGs of dashboard metrics. """36 cloudwatch_api = auth()37 dashboard_list = [d['DashboardName'] for d in cloudwatch_api.list_dashboards()['DashboardEntries']]38 logging.info("Dashboards available: %s Rendering: %s", dashboard_list, dashboard_name)39 if dashboard_name != "all" and dashboard_name in dashboard_list:40 dashboard_list = [dashboard_name]41 dashboard_widget_properties = {}42 dashboard_images = {}43 logging.debug("Dashboards available: %s ", dashboard_list)44 for dashboard in dashboard_list:45 dashboard_widget_properties[dashboard] = [dp['properties'] for dp in json.loads(cloudwatch_api.get_dashboard(DashboardName=dashboard)['DashboardBody'])['widgets'] if 'metrics' in dp['properties']]46 dashboard_images[dashboard] = []47 for dashboard_widget_property in dashboard_widget_properties[dashboard]:48 #logging.debug(json.dumps(dashboard_widget_property))49 dashboard_image = cloudwatch_api.get_metric_widget_image(MetricWidget=json.dumps(dashboard_widget_property))['MetricWidgetImage']50 if encode_base64:51 dashboard_image = base64.b64encode(dashboard_image).decode('utf-8')52 dashboard_images[dashboard].append(dashboard_image)53 result = [item for sublist in dashboard_images.values() for item in sublist]54 logging.info("Result size: %d ", len(result))55 return result56def main(argv):57 """ Outputs a HTML page to stdout with inline base64 encoded PNG dashboards. """58 dashboard_name = "all"59 if len(argv) == 2:60 dashboard_name = argv[1]61 result = "<html>\n"62 start_time = datetime.datetime.now()63 result += "<!-- Generation started: " + start_time.isoformat() + " -->\n"64 for image in get_dashboard_images(dashboard_name=dashboard_name):65 result += " <img src='data:image/png;base64," + str(image) + "' />\n"66 end_time = datetime.datetime.now()67 result += "<!-- Generation ended: " + end_time.isoformat() + " -->\n"68 result += "</html>\n"69 print("%s" % result)70 logging.info("Completed. Start time: %s Runtime: %s ", start_time, str(end_time-start_time))71 return 072if __name__ == "__main__":73 if len(sys.argv) not in (1, 2):74 syntax(sys.argv[0])...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
