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.