Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use cloudwatch method in localstack
Best Python code snippet using localstack_python
cloudwatch.py
Source:cloudwatch.py 
1import re2from datetime import datetime, timedelta3from typing import ClassVar, Dict, List, Optional, Type, Tuple4from attr import define, field5from resoto_plugin_aws.aws_client import AwsClient6from resoto_plugin_aws.resource.base import AwsApiSpec, AwsResource, GraphBuilder7from resoto_plugin_aws.utils import ToDict8from resotolib.baseresources import BaseAccount, ModelReference # noqa: F4019from resotolib.json import from_json10from resotolib.json_bender import S, Bend, Bender, ForallBend, bend11from resotolib.types import Json12from resotolib.utils import chunks13# noinspection PyUnresolvedReferences14class CloudwatchTaggable:15 def update_resource_tag(self, client: AwsClient, key: str, value: str) -> bool:16 if isinstance(self, AwsResource):17 if spec := self.api_spec:18 client.call(19 aws_service=spec.service,20 action="tag_resource",21 result_name=None,22 ResourceARN=self.arn,23 Tags=[{"Key": key, "Value": value}],24 )25 return True26 return False27 return False28 def delete_resource_tag(self, client: AwsClient, key: str) -> bool:29 if isinstance(self, AwsResource):30 if spec := self.api_spec:31 client.call(32 aws_service=spec.service,33 action="untag_resource",34 result_name=None,35 ResourceARN=self.arn,36 TagKeys=[key],37 )38 return True39 return False40 return False41@define(eq=False, slots=False)42class AwsCloudwatchDimension:43 kind: ClassVar[str] = "aws_cloudwatch_dimension"44 mapping: ClassVar[Dict[str, Bender]] = {"name": S("Name"), "value": S("Value")}45 name: Optional[str] = field(default=None)46 value: Optional[str] = field(default=None)47@define(eq=False, slots=False)48class AwsCloudwatchMetric:49 kind: ClassVar[str] = "aws_cloudwatch_metric"50 mapping: ClassVar[Dict[str, Bender]] = {51 "namespace": S("Namespace"),52 "metric_name": S("MetricName"),53 "dimensions": S("Dimensions", default=[]) >> ForallBend(AwsCloudwatchDimension.mapping),54 }55 namespace: Optional[str] = field(default=None)56 metric_name: Optional[str] = field(default=None)57 dimensions: List[AwsCloudwatchDimension] = field(factory=list)58@define(eq=False, slots=False)59class AwsCloudwatchMetricStat:60 kind: ClassVar[str] = "aws_cloudwatch_metric_stat"61 mapping: ClassVar[Dict[str, Bender]] = {62 "metric": S("Metric") >> Bend(AwsCloudwatchMetric.mapping),63 "period": S("Period"),64 "stat": S("Stat"),65 "unit": S("Unit"),66 }67 metric: Optional[AwsCloudwatchMetric] = field(default=None)68 period: Optional[int] = field(default=None)69 stat: Optional[str] = field(default=None)70 unit: Optional[str] = field(default=None)71@define(eq=False, slots=False)72class AwsCloudwatchMetricDataQuery:73 kind: ClassVar[str] = "aws_cloudwatch_metric_data_query"74 mapping: ClassVar[Dict[str, Bender]] = {75 "id": S("Id"),76 "metric_stat": S("MetricStat") >> Bend(AwsCloudwatchMetricStat.mapping),77 "expression": S("Expression"),78 "label": S("Label"),79 "return_data": S("ReturnData"),80 "period": S("Period"),81 "account_id": S("AccountId"),82 }83 id: Optional[str] = field(default=None)84 metric_stat: Optional[AwsCloudwatchMetricStat] = field(default=None)85 expression: Optional[str] = field(default=None)86 label: Optional[str] = field(default=None)87 return_data: Optional[bool] = field(default=None)88 period: Optional[int] = field(default=None)89 account_id: Optional[str] = field(default=None)90@define(eq=False, slots=False)91class AwsCloudwatchAlarm(CloudwatchTaggable, AwsResource):92 kind: ClassVar[str] = "aws_cloudwatch_alarm"93 api_spec: ClassVar[AwsApiSpec] = AwsApiSpec("cloudwatch", "describe-alarms", "MetricAlarms")94 reference_kinds: ClassVar[ModelReference] = {95 "predecessors": {"default": ["aws_ec2_instance"], "delete": ["aws_ec2_instance"]},96 }97 mapping: ClassVar[Dict[str, Bender]] = {98 "id": S("AlarmName"),99 "name": S("AlarmName"),100 "mtime": S("AlarmConfigurationUpdatedTimestamp"),101 "arn": S("AlarmArn"),102 "cloudwatch_alarm_description": S("AlarmDescription"),103 "cloudwatch_actions_enabled": S("ActionsEnabled"),104 "cloudwatch_ok_actions": S("OKActions", default=[]),105 "cloudwatch_alarm_actions": S("AlarmActions", default=[]),106 "cloudwatch_insufficient_data_actions": S("InsufficientDataActions", default=[]),107 "cloudwatch_state_value": S("StateValue"),108 "cloudwatch_state_reason": S("StateReason"),109 "cloudwatch_state_reason_data": S("StateReasonData"),110 "cloudwatch_state_updated_timestamp": S("StateUpdatedTimestamp"),111 "cloudwatch_metric_name": S("MetricName"),112 "cloudwatch_namespace": S("Namespace"),113 "cloudwatch_statistic": S("Statistic"),114 "cloudwatch_extended_statistic": S("ExtendedStatistic"),115 "cloudwatch_dimensions": S("Dimensions", default=[]) >> ForallBend(AwsCloudwatchDimension.mapping),116 "cloudwatch_period": S("Period"),117 "cloudwatch_unit": S("Unit"),118 "cloudwatch_evaluation_periods": S("EvaluationPeriods"),119 "cloudwatch_datapoints_to_alarm": S("DatapointsToAlarm"),120 "cloudwatch_threshold": S("Threshold"),121 "cloudwatch_comparison_operator": S("ComparisonOperator"),122 "cloudwatch_treat_missing_data": S("TreatMissingData"),123 "cloudwatch_evaluate_low_sample_count_percentile": S("EvaluateLowSampleCountPercentile"),124 "cloudwatch_metrics": S("Metrics", default=[]) >> ForallBend(AwsCloudwatchMetricDataQuery.mapping),125 "cloudwatch_threshold_metric_id": S("ThresholdMetricId"),126 }127 arn: Optional[str] = field(default=None)128 cloudwatch_alarm_description: Optional[str] = field(default=None)129 cloudwatch_actions_enabled: Optional[bool] = field(default=None)130 cloudwatch_ok_actions: List[str] = field(factory=list)131 cloudwatch_alarm_actions: List[str] = field(factory=list)132 cloudwatch_insufficient_data_actions: List[str] = field(factory=list)133 cloudwatch_state_value: Optional[str] = field(default=None)134 cloudwatch_state_reason: Optional[str] = field(default=None)135 cloudwatch_state_reason_data: Optional[str] = field(default=None)136 cloudwatch_state_updated_timestamp: Optional[datetime] = field(default=None)137 cloudwatch_metric_name: Optional[str] = field(default=None)138 cloudwatch_namespace: Optional[str] = field(default=None)139 cloudwatch_statistic: Optional[str] = field(default=None)140 cloudwatch_extended_statistic: Optional[str] = field(default=None)141 cloudwatch_dimensions: List[AwsCloudwatchDimension] = field(factory=list)142 cloudwatch_period: Optional[int] = field(default=None)143 cloudwatch_unit: Optional[str] = field(default=None)144 cloudwatch_evaluation_periods: Optional[int] = field(default=None)145 cloudwatch_datapoints_to_alarm: Optional[int] = field(default=None)146 cloudwatch_threshold: Optional[float] = field(default=None)147 cloudwatch_comparison_operator: Optional[str] = field(default=None)148 cloudwatch_treat_missing_data: Optional[str] = field(default=None)149 cloudwatch_evaluate_low_sample_count_percentile: Optional[str] = field(default=None)150 cloudwatch_metrics: List[AwsCloudwatchMetricDataQuery] = field(factory=list)151 cloudwatch_threshold_metric_id: Optional[str] = field(default=None)152 @classmethod153 def collect(cls: Type[AwsResource], json: List[Json], builder: GraphBuilder) -> None:154 def add_tags(alarm: AwsCloudwatchAlarm) -> None:155 tags = builder.client.list("cloudwatch", "list-tags-for-resource", "Tags", ResourceARN=alarm.arn)156 if tags:157 alarm.tags = bend(ToDict(), tags)158 for js in json:159 instance = cls.from_api(js)160 builder.add_node(instance, js)161 builder.submit_work(add_tags, instance)162 def connect_in_graph(self, builder: GraphBuilder, source: Json) -> None:163 super().connect_in_graph(builder, source)164 for dimension in self.cloudwatch_dimensions:165 builder.dependant_node(166 self, reverse=True, delete_same_as_default=True, kind="aws_ec2_instance", id=dimension.value167 )168 def delete_resource(self, client: AwsClient) -> bool:169 client.call(aws_service=self.api_spec.service, action="delete_alarms", result_name=None, AlarmNames=[self.name])170 return True171@define(hash=True, frozen=True)172class AwsCloudwatchQuery:173 metric_name: str174 namespace: str175 dimensions: Tuple[Tuple[str, str], ...]176 period: timedelta177 ref_id: str178 metric_id: str179 def to_json(self) -> Json:180 return {181 "Id": self.metric_id,182 "MetricStat": {183 "Metric": {184 "Namespace": self.namespace,185 "MetricName": self.metric_name,186 "Dimensions": [{"Name": k, "Value": v} for k, v in self.dimensions],187 },188 "Period": int((self.period.total_seconds() / 60) * 60), # round to the next 60 seconds189 "Stat": "Sum",190 "Unit": "Count",191 },192 "ReturnData": True,193 }194 @staticmethod195 def create(196 metric_name: str,197 namespace: str,198 period: timedelta,199 ref_id: str,200 metric_id: Optional[str] = None,201 **dimensions: str,202 ) -> "AwsCloudwatchQuery":203 dims = "_".join(f"{k}+{v}" for k, v in dimensions.items())204 rid = metric_id or re.sub("\\W", "_", f"{metric_name}-{namespace}-{dims}".lower())205 return AwsCloudwatchQuery(206 metric_name=metric_name,207 namespace=namespace,208 period=period,209 dimensions=tuple(dimensions.items()),210 ref_id=ref_id,211 metric_id=rid,212 )213@define(eq=False, slots=False)214class AwsCloudwatchMessageData:215 mapping: ClassVar[Dict[str, Bender]] = {"code": S("Code"), "value": S("Value")}216 code: Optional[str] = field(default=None)217 value: Optional[str] = field(default=None)218@define(eq=False, slots=False)219class AwsCloudwatchMetricData:220 mapping: ClassVar[Dict[str, Bender]] = {221 "id": S("Id"),222 "label": S("Label"),223 "metric_timestamps": S("Timestamps", default=[]),224 "metric_values": S("Values", default=[]),225 "metric_status_code": S("StatusCode"),226 "metric_messages": S("Messages", default=[]) >> ForallBend(AwsCloudwatchMessageData.mapping),227 }228 id: str = field(default=None)229 label: Optional[str] = field(default=None)230 metric_timestamps: List[datetime] = field(factory=list)231 metric_values: List[float] = field(factory=list)232 metric_status_code: Optional[str] = field(default=None)233 metric_messages: List[AwsCloudwatchMessageData] = field(factory=list)234 def first_non_zero(self) -> Optional[Tuple[datetime, float]]:235 for timestamp, value in zip(self.metric_timestamps, self.metric_values):236 if value != 0:237 return timestamp, value238 return None239 @classmethod240 def called_apis(cls) -> List[AwsApiSpec]:241 return [AwsApiSpec("cloudwatch", "get_metric_data")]242 @staticmethod243 def query_for(244 client: AwsClient,245 queries: List[AwsCloudwatchQuery],246 start_time: datetime,247 end_time: datetime,248 scan_desc: bool = True,249 ) -> "Dict[AwsCloudwatchQuery, AwsCloudwatchMetricData]":250 lookup = {q.metric_id: q for q in queries}251 result: Dict[AwsCloudwatchQuery, AwsCloudwatchMetricData] = {}252 # the api only allows for up to 500 metrics at once253 for chunk in chunks(queries, 499):254 part = client.list(255 "cloudwatch",256 "get-metric-data",257 "MetricDataResults",258 MetricDataQueries=[a.to_json() for a in chunk],259 StartTime=start_time,260 EndTime=end_time,261 ScanBy="TimestampDescending" if scan_desc else "TimestampAscending",262 )263 for single in part:264 metric = from_json(bend(AwsCloudwatchMetricData.mapping, single), AwsCloudwatchMetricData)265 result[lookup[metric.id]] = metric266 return result...
test_agent_aws_cloudwatch.py
Source:test_agent_aws_cloudwatch.py
1#!/usr/bin/env python32# Copyright (C) 2019 tribe29 GmbH - License: GNU General Public License v23# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and4# conditions defined in the file COPYING, which is part of this source code package.5# pylint: disable=redefined-outer-name6from typing import List, Optional, Tuple7import pytest8from cmk.special_agents.agent_aws import (9 AWSConfig,10 CloudwatchAlarms,11 CloudwatchAlarmsLimits,12 ResultDistributor,13)14from .agent_aws_fake_clients import FakeCloudwatchClient15@pytest.fixture()16def get_cloudwatch_alarms_sections():17 def _create_cloudwatch_alarms_sections(alarm_names):18 region = "region"19 config = AWSConfig("hostname", [], (None, None))20 config.add_single_service_config("cloudwatch_alarms", alarm_names)21 fake_cloudwatch_client = FakeCloudwatchClient()22 cloudwatch_alarms_limits_distributor = ResultDistributor()23 cloudwatch_alarms_limits = CloudwatchAlarmsLimits(24 fake_cloudwatch_client, region, config, cloudwatch_alarms_limits_distributor25 )26 cloudwatch_alarms = CloudwatchAlarms(fake_cloudwatch_client, region, config)27 cloudwatch_alarms_limits_distributor.add(cloudwatch_alarms)28 return cloudwatch_alarms_limits, cloudwatch_alarms29 return _create_cloudwatch_alarms_sections30cloudwatch_params: List[Tuple[Optional[List[str]], int]] = [31 (None, 2),32 ([], 2),33 (["AlarmName-0"], 1),34 (["not found"], 1),35 (["AlarmName-0", "too many"], 1),36 (["AlarmName-0", "AlarmName-1"], 2),37 (["AlarmName-0", "AlarmName-1", "too many"], 2),38]39@pytest.mark.parametrize("alarm_names,amount_alarms", cloudwatch_params)40def test_agent_aws_cloudwatch_alarms_limits(41 get_cloudwatch_alarms_sections, alarm_names, amount_alarms42):43 cloudwatch_alarms_limits, _cloudwatch_alarms = get_cloudwatch_alarms_sections(alarm_names)44 cloudwatch_alarms_limits_results = cloudwatch_alarms_limits.run().results45 assert cloudwatch_alarms_limits.cache_interval == 30046 assert cloudwatch_alarms_limits.period == 60047 assert cloudwatch_alarms_limits.name == "cloudwatch_alarms_limits"48 assert len(cloudwatch_alarms_limits_results) == 149 cloudwatch_alarms_limits_result = cloudwatch_alarms_limits_results[0]50 assert cloudwatch_alarms_limits_result.piggyback_hostname == ""51 assert len(cloudwatch_alarms_limits_result.content) == 152 cloudwatch_alarms_limits_content = cloudwatch_alarms_limits_result.content[0]53 assert cloudwatch_alarms_limits_content.key == "cloudwatch_alarms"54 assert cloudwatch_alarms_limits_content.title == "CloudWatch Alarms"55 assert cloudwatch_alarms_limits_content.limit == 500056 assert cloudwatch_alarms_limits_content.amount == 257@pytest.mark.parametrize("alarm_names,amount_alarms", cloudwatch_params)58def test_agent_aws_cloudwatch_alarms( # type:ignore[no-untyped-def]59 get_cloudwatch_alarms_sections, alarm_names, amount_alarms60) -> None:61 cloudwatch_alarms_limits, cloudwatch_alarms = get_cloudwatch_alarms_sections(alarm_names)62 _cloudwatch_alarms_limits_results = cloudwatch_alarms_limits.run().results # noqa: F84163 cloudwatch_alarms_results = cloudwatch_alarms.run().results64 assert cloudwatch_alarms.cache_interval == 30065 assert cloudwatch_alarms.period == 60066 assert cloudwatch_alarms.name == "cloudwatch_alarms"67 assert len(cloudwatch_alarms_results) == 168 cloudwatch_alarms_result = cloudwatch_alarms_results[0]69 assert cloudwatch_alarms_result.piggyback_hostname == ""70 assert len(cloudwatch_alarms_result.content) == amount_alarms71@pytest.mark.parametrize("alarm_names,amount_alarms", cloudwatch_params)72def test_agent_aws_cloudwatch_alarms_without_limits(73 get_cloudwatch_alarms_sections, alarm_names, amount_alarms74):75 _cloudwatch_alarms_limits, cloudwatch_alarms = get_cloudwatch_alarms_sections(alarm_names)76 cloudwatch_alarms_results = cloudwatch_alarms.run().results77 assert cloudwatch_alarms.cache_interval == 30078 assert cloudwatch_alarms.period == 60079 assert cloudwatch_alarms.name == "cloudwatch_alarms"80 assert len(cloudwatch_alarms_results) == 181 cloudwatch_alarms_result = cloudwatch_alarms_results[0]82 assert cloudwatch_alarms_result.piggyback_hostname == ""...
test-utils_cloudwatch.py
Source:test-utils_cloudwatch.py
1from AWSScout2.utils_cloudwatch import *2#3# Test for Scout2 CloudWatch functions4#5class TestScout2CloudWatchUtilsClass:6 configPrintException(True)7 #8 # Test get_cloudwatch_region in us-east-19 #10 def test_get_cloudwatch_region(self):11 # TODO: change to us-east-112 credentials = read_creds('default')13 service_config = {'regions': {'us-east-1': {}}}14 get_cloudwatch_region(params = {'region': 'us-east-1', 'creds': credentials, 'cloudwatch_config': service_config})15 # 16 # Test get_cloudwatch_info (multiple thread of get_cloudwatch_region)17 # 1. in us-east-1 and us-west-118 # 2. in empty region intersection 19 #20 def test_get_cloudwatch_info(self):21 credentials = read_creds('default')22 service_config = {'regions': {'us-east-1': {}, 'us-west-1': {}}} #, 'cn-north-1': {}}}23 get_cloudwatch_info(credentials, service_config, ['us-east-1', 'us-west-1'], 'aws')24 get_cloudwatch_info(credentials, service_config, ['us-east-1', 'us-west-1'], 'aws-us-gov')25# get_cloudwatch_info(credentials, service_config, ['us-gov-west-1'], 'aws-us-gov')26 #27 # Smoke tests for status display functions28 #29 def test_cloudwatch_status_init(self):30 cloudwatch_status_init()31 32 def test_cloudwatch_status(self):33 cloudwatch_status(True)34 cloudwatch_status(False)35 cloudwatch_status()36 def test_formatted_status(self):37 formatted_status(1, 42, True)...
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?
