Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use cloudwatch_client method in localstack

Best Python code snippet using localstack_python

test_cloudwatch.py

Source:test_cloudwatch.py

1import gzip2import json3from datetime import datetime, timedelta, timezone4from urllib.request import Request, urlopen5import pytest6import requests7from dateutil.tz import tzutc8from localstack import config9from localstack.services.cloudwatch.provider import PATH_GET_RAW_METRICS10from localstack.utils.aws import aws_stack11from localstack.utils.common import retry, short_uid, to_str12PUBLICATION_RETRIES = 513class TestCloudwatch:14    def test_put_metric_data(self, cloudwatch_client):15        metric_name = "metric-%s" % short_uid()16        namespace = "namespace-%s" % short_uid()17        # Put metric data without value18        data = [19            {20                "MetricName": metric_name,21                "Dimensions": [{"Name": "foo", "Value": "bar"}],22                "Timestamp": datetime(2019, 1, 3, tzinfo=tzutc()),23                "Unit": "Seconds",24            }25        ]26        rs = cloudwatch_client.put_metric_data(Namespace=namespace, MetricData=data)27        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]28        # Get metric statistics29        rs = cloudwatch_client.get_metric_statistics(30            Namespace=namespace,31            MetricName=metric_name,32            StartTime=datetime(2019, 1, 1),33            EndTime=datetime(2019, 1, 10),34            Period=120,35            Statistics=["Average"],36        )37        assert metric_name == rs["Label"]38        assert 1 == len(rs["Datapoints"])39        assert data[0]["Timestamp"] == rs["Datapoints"][0]["Timestamp"]40        rs = cloudwatch_client.list_metrics(Namespace=namespace, MetricName=metric_name)41        assert 1 == len(rs["Metrics"])42        assert namespace == rs["Metrics"][0]["Namespace"]43    def test_put_metric_data_gzip(self, cloudwatch_client):44        metric_name = "test-metric"45        namespace = "namespace"46        data = (47            "Action=PutMetricData&MetricData.member.1."48            "MetricName=%s&MetricData.member.1.Value=1&"49            "Namespace=%s&Version=2010-08-01" % (metric_name, namespace)50        )51        bytes_data = bytes(data, encoding="utf-8")52        encoded_data = gzip.compress(bytes_data)53        url = config.get_edge_url()54        headers = aws_stack.mock_aws_request_headers("cloudwatch", internal=True)55        authorization = aws_stack.mock_aws_request_headers("monitoring")["Authorization"]56        headers.update(57            {58                "Content-Type": "application/x-www-form-urlencoded; charset=utf-8",59                "Content-Length": len(encoded_data),60                "Content-Encoding": "GZIP",61                "User-Agent": "aws-sdk-nodejs/2.819.0 linux/v12.18.2 callback",62                "Authorization": authorization,63            }64        )65        request = Request(url, encoded_data, headers, method="POST")66        urlopen(request)67        rs = cloudwatch_client.list_metrics(Namespace=namespace, MetricName=metric_name)68        assert 1 == len(rs["Metrics"])69        assert namespace == rs["Metrics"][0]["Namespace"]70    def test_get_metric_data(self, cloudwatch_client):71        cloudwatch_client.put_metric_data(72            Namespace="some/thing", MetricData=[dict(MetricName="someMetric", Value=23)]73        )74        cloudwatch_client.put_metric_data(75            Namespace="some/thing", MetricData=[dict(MetricName="someMetric", Value=18)]76        )77        cloudwatch_client.put_metric_data(78            Namespace="ug/thing", MetricData=[dict(MetricName="ug", Value=23)]79        )80        # filtering metric data with current time interval81        response = cloudwatch_client.get_metric_data(82            MetricDataQueries=[83                {84                    "Id": "some",85                    "MetricStat": {86                        "Metric": {87                            "Namespace": "some/thing",88                            "MetricName": "someMetric",89                        },90                        "Period": 60,91                        "Stat": "Sum",92                    },93                },94                {95                    "Id": "part",96                    "MetricStat": {97                        "Metric": {"Namespace": "ug/thing", "MetricName": "ug"},98                        "Period": 60,99                        "Stat": "Sum",100                    },101                },102            ],103            StartTime=datetime.utcnow() - timedelta(hours=1),104            EndTime=datetime.utcnow(),105        )106        assert 2 == len(response["MetricDataResults"])107        for data_metric in response["MetricDataResults"]:108            if data_metric["Id"] == "some":109                assert 41.0 == data_metric["Values"][0]110            if data_metric["Id"] == "part":111                assert 23.0 == data_metric["Values"][0]112        # filtering metric data with current time interval113        response = cloudwatch_client.get_metric_data(114            MetricDataQueries=[115                {116                    "Id": "some",117                    "MetricStat": {118                        "Metric": {119                            "Namespace": "some/thing",120                            "MetricName": "someMetric",121                        },122                        "Period": 60,123                        "Stat": "Sum",124                    },125                },126                {127                    "Id": "part",128                    "MetricStat": {129                        "Metric": {"Namespace": "ug/thing", "MetricName": "ug"},130                        "Period": 60,131                        "Stat": "Sum",132                    },133                },134            ],135            StartTime=datetime.utcnow() + timedelta(hours=1),136            EndTime=datetime.utcnow() + timedelta(hours=2),137        )138        for data_metric in response["MetricDataResults"]:139            if data_metric["Id"] == "some":140                assert len(data_metric["Values"]) == 0141            if data_metric["Id"] == "part":142                assert len(data_metric["Values"]) == 0143        # get raw metric data144        url = "%s%s" % (config.get_edge_url(), PATH_GET_RAW_METRICS)145        result = requests.get(url)146        assert 200 == result.status_code147        result = json.loads(to_str(result.content))148        assert len(result["metrics"]) >= 3149    def test_multiple_dimensions(self, cloudwatch_client):150        namespaces = [151            "ns1-%s" % short_uid(),152            "ns2-%s" % short_uid(),153            "ns3-%s" % short_uid(),154        ]155        num_dimensions = 2156        for ns in namespaces:157            for i in range(3):158                rs = cloudwatch_client.put_metric_data(159                    Namespace=ns,160                    MetricData=[161                        {162                            "MetricName": "someMetric",163                            "Value": 123,164                            "Dimensions": [165                                {166                                    "Name": "foo",167                                    "Value": "bar-%s" % (i % num_dimensions),168                                }169                            ],170                        }171                    ],172                )173                assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]174        rs = cloudwatch_client.list_metrics()175        metrics = [m for m in rs["Metrics"] if m.get("Namespace") in namespaces]176        assert metrics177        assert len(metrics) == len(namespaces) * num_dimensions178    def test_describe_alarms_converts_date_format_correctly(self, cloudwatch_client):179        alarm_name = "a-%s" % short_uid()180        cloudwatch_client.put_metric_alarm(181            AlarmName=alarm_name,182            EvaluationPeriods=1,183            ComparisonOperator="GreaterThanThreshold",184        )185        try:186            result = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])187            alarm = result["MetricAlarms"][0]188            assert isinstance(alarm["AlarmConfigurationUpdatedTimestamp"], datetime)189            assert isinstance(alarm["StateUpdatedTimestamp"], datetime)190        finally:191            cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])192    def test_put_composite_alarm_describe_alarms_converts_date_format_correctly(193        self, cloudwatch_client194    ):195        alarm_name = "a-%s" % short_uid()196        alarm_rule = 'ALARM("my_other_alarm")'197        cloudwatch_client.put_composite_alarm(198            AlarmName=alarm_name,199            AlarmRule=alarm_rule,200        )201        try:202            result = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])203            alarm = result["CompositeAlarms"][0]204            assert alarm["AlarmName"] == alarm_name205            assert alarm["AlarmRule"] == alarm_rule206        finally:207            cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])208    def test_store_tags(self, cloudwatch_client):209        alarm_name = "a-%s" % short_uid()210        response = cloudwatch_client.put_metric_alarm(211            AlarmName=alarm_name,212            EvaluationPeriods=1,213            ComparisonOperator="GreaterThanThreshold",214        )215        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]216        alarm_arn = aws_stack.cloudwatch_alarm_arn(alarm_name)217        tags = [{"Key": "tag1", "Value": "foo"}, {"Key": "tag2", "Value": "bar"}]218        response = cloudwatch_client.tag_resource(ResourceARN=alarm_arn, Tags=tags)219        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]220        response = cloudwatch_client.list_tags_for_resource(ResourceARN=alarm_arn)221        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]222        assert tags, response["Tags"]223        response = cloudwatch_client.untag_resource(ResourceARN=alarm_arn, TagKeys=["tag1"])224        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]225        response = cloudwatch_client.list_tags_for_resource(ResourceARN=alarm_arn)226        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]227        assert [{"Key": "tag2", "Value": "bar"}] == response["Tags"]228        # clean up229        cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])230    def test_list_metrics_uniqueness(self, cloudwatch_client):231        # create metrics with same namespace and dimensions but different metric names232        cloudwatch_client.put_metric_data(233            Namespace="AWS/EC2",234            MetricData=[235                {236                    "MetricName": "CPUUtilization",237                    "Dimensions": [{"Name": "InstanceId", "Value": "i-46cdcd06a11207ab3"}],238                    "Value": 15,239                }240            ],241        )242        cloudwatch_client.put_metric_data(243            Namespace="AWS/EC2",244            MetricData=[245                {246                    "MetricName": "Memory",247                    "Dimensions": [{"Name": "InstanceId", "Value": "i-46cdcd06a11207ab3"}],248                    "Value": 30,249                }250            ],251        )252        results = cloudwatch_client.list_metrics(Namespace="AWS/EC2")["Metrics"]253        assert 2 == len(results)254        # duplicating existing metric255        cloudwatch_client.put_metric_data(256            Namespace="AWS/EC2",257            MetricData=[258                {259                    "MetricName": "CPUUtilization",260                    "Dimensions": [{"Name": "InstanceId", "Value": "i-46cdcd06a11207ab3"}],261                    "Value": 15,262                }263            ],264        )265        # asserting only unique values are returned266        results = cloudwatch_client.list_metrics(Namespace="AWS/EC2")["Metrics"]267        assert 2 == len(results)268    def test_put_metric_alarm_escape_character(self, cloudwatch_client):269        cloudwatch_client.put_metric_alarm(270            AlarmName="cpu-mon",271            AlarmDescription="<",272            MetricName="CPUUtilization-2",273            Namespace="AWS/EC2",274            Statistic="Sum",275            Period=600,276            Threshold=1,277            ComparisonOperator="GreaterThanThreshold",278            EvaluationPeriods=1,279            AlarmActions=["arn:aws:sns:us-east-1:111122223333:MyTopic"],280        )281        result = cloudwatch_client.describe_alarms()282        assert result.get("MetricAlarms")[0]["AlarmDescription"] == "<"283        cloudwatch_client.delete_alarms(AlarmNames=["cpu-mon"])284    def test_set_alarm(285        self, sns_client, cloudwatch_client, sqs_client, sns_create_topic, sqs_create_queue286    ):287        # create topics for state 'ALARM' and 'OK'288        sns_topic_alarm = sns_create_topic()289        topic_arn_alarm = sns_topic_alarm["TopicArn"]290        sns_topic_ok = sns_create_topic()291        topic_arn_ok = sns_topic_ok["TopicArn"]292        # create queues for 'ALARM' and 'OK' (will receive sns messages)293        uid = short_uid()294        queue_url_alarm = sqs_create_queue(QueueName=f"AlarmQueue-{uid}")295        queue_url_ok = sqs_create_queue(QueueName=f"OKQueue-{uid}")296        arn_queue_alarm = sqs_client.get_queue_attributes(297            QueueUrl=queue_url_alarm, AttributeNames=["QueueArn"]298        )["Attributes"]["QueueArn"]299        arn_queue_ok = sqs_client.get_queue_attributes(300            QueueUrl=queue_url_ok, AttributeNames=["QueueArn"]301        )["Attributes"]["QueueArn"]302        sqs_client.set_queue_attributes(303            QueueUrl=queue_url_alarm,304            Attributes={"Policy": get_sqs_policy(arn_queue_alarm, topic_arn_alarm)},305        )306        sqs_client.set_queue_attributes(307            QueueUrl=queue_url_ok, Attributes={"Policy": get_sqs_policy(arn_queue_ok, topic_arn_ok)}308        )309        alarm_name = "test-alarm"310        alarm_description = "Test Alarm when CPU exceeds 50 percent"311        expected_trigger = {312            "MetricName": "CPUUtilization-3",313            "Namespace": "AWS/EC2",314            "Unit": "Percent",315            "Period": 300,316            "EvaluationPeriods": 2,317            "ComparisonOperator": "GreaterThanThreshold",318            "Threshold": 50.0,319            "TreatMissingData": "ignore",320            "EvaluateLowSampleCountPercentile": "",321            "Dimensions": [{"value": "i-0317828c84edbe100", "name": "InstanceId"}],322            "StatisticType": "Statistic",323            "Statistic": "AVERAGE",324        }325        try:326            # subscribe to SQS327            subscription_alarm = sns_client.subscribe(328                TopicArn=topic_arn_alarm, Protocol="sqs", Endpoint=arn_queue_alarm329            )330            subscription_ok = sns_client.subscribe(331                TopicArn=topic_arn_ok, Protocol="sqs", Endpoint=arn_queue_ok332            )333            # create alarm with actions for "OK" and "ALARM"334            cloudwatch_client.put_metric_alarm(335                AlarmName=alarm_name,336                AlarmDescription=alarm_description,337                MetricName=expected_trigger["MetricName"],338                Namespace=expected_trigger["Namespace"],339                ActionsEnabled=True,340                Period=expected_trigger["Period"],341                Threshold=expected_trigger["Threshold"],342                Dimensions=[{"Name": "InstanceId", "Value": "i-0317828c84edbe100"}],343                Unit=expected_trigger["Unit"],344                Statistic=expected_trigger["Statistic"].capitalize(),345                OKActions=[topic_arn_ok],346                AlarmActions=[topic_arn_alarm],347                EvaluationPeriods=expected_trigger["EvaluationPeriods"],348                ComparisonOperator=expected_trigger["ComparisonOperator"],349                TreatMissingData=expected_trigger["TreatMissingData"],350            )351            # trigger alarm352            state_value = "ALARM"353            state_reason = "testing alarm"354            cloudwatch_client.set_alarm_state(355                AlarmName=alarm_name, StateReason=state_reason, StateValue=state_value356            )357            retry(358                check_message,359                retries=PUBLICATION_RETRIES,360                sleep_before=1,361                sqs_client=sqs_client,362                expected_queue_url=queue_url_alarm,363                expected_topic_arn=topic_arn_alarm,364                expected_new=state_value,365                expected_reason=state_reason,366                alarm_name=alarm_name,367                alarm_description=alarm_description,368                expected_trigger=expected_trigger,369            )370            # trigger OK371            state_value = "OK"372            state_reason = "resetting alarm"373            cloudwatch_client.set_alarm_state(374                AlarmName=alarm_name, StateReason=state_reason, StateValue=state_value375            )376            retry(377                check_message,378                retries=PUBLICATION_RETRIES,379                sleep_before=1,380                sqs_client=sqs_client,381                expected_queue_url=queue_url_ok,382                expected_topic_arn=topic_arn_ok,383                expected_new=state_value,384                expected_reason=state_reason,385                alarm_name=alarm_name,386                alarm_description=alarm_description,387                expected_trigger=expected_trigger,388            )389        finally:390            # cleanup391            sns_client.unsubscribe(SubscriptionArn=subscription_alarm["SubscriptionArn"])392            sns_client.unsubscribe(SubscriptionArn=subscription_ok["SubscriptionArn"])393            cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])394    @pytest.mark.aws_validated395    @pytest.mark.skip_snapshot_verify(396        paths=[397            "$..evaluatedDatapoints",398            "$..startDate",  # only sometimes visible? part of StateReasonData399            "$..alarm-triggered-describe.MetricAlarms[0].StateReason",  # reason contains datapoint + date400            "$..alarm-triggered-sqs-msg.NewStateReason",401        ]402    )403    def test_put_metric_alarm(404        self,405        sns_client,406        cloudwatch_client,407        sqs_client,408        sns_create_topic,409        sqs_create_queue,410        snapshot,411    ):412        sns_topic_alarm = sns_create_topic()413        topic_arn_alarm = sns_topic_alarm["TopicArn"]414        snapshot.add_transformer(snapshot.transform.cloudwatch_api())415        snapshot.add_transformer(416            snapshot.transform.regex(topic_arn_alarm.split(":")[-1], "<topic_arn>"), priority=2417        )418        # as we add metrics, we use a unique namespace to ensure the test runs on AWS419        namespace = f"test-nsp-{short_uid()}"420        snapshot.add_transformer(snapshot.transform.regex(namespace, "<metric-namespace>"))421        sqs_queue = sqs_create_queue()422        arn_queue = sqs_client.get_queue_attributes(423            QueueUrl=sqs_queue, AttributeNames=["QueueArn"]424        )["Attributes"]["QueueArn"]425        # required for AWS:426        sqs_client.set_queue_attributes(427            QueueUrl=sqs_queue,428            Attributes={"Policy": get_sqs_policy(arn_queue, topic_arn_alarm)},429        )430        metric_name = "my-metric1"431        dimension = [{"Name": "InstanceId", "Value": "abc"}]432        alarm_name = f"test-alarm-{short_uid()}"433        try:434            subscription = sns_client.subscribe(435                TopicArn=topic_arn_alarm, Protocol="sqs", Endpoint=arn_queue436            )437            data = [438                {439                    "MetricName": metric_name,440                    "Dimensions": dimension,441                    "Value": 21,442                    "Timestamp": datetime.utcnow().replace(tzinfo=timezone.utc),443                    "Unit": "Seconds",444                },445                {446                    "MetricName": metric_name,447                    "Dimensions": dimension,448                    "Value": 22,449                    "Timestamp": datetime.utcnow().replace(tzinfo=timezone.utc),450                    "Unit": "Seconds",451                },452            ]453            cloudwatch_client.put_metric_data(Namespace=namespace, MetricData=data)454            # create alarm with action for "ALARM"455            cloudwatch_client.put_metric_alarm(456                AlarmName=alarm_name,457                AlarmDescription="testing cloudwatch alarms",458                MetricName=metric_name,459                Namespace=namespace,460                ActionsEnabled=True,461                Period=30,462                Threshold=2,463                Dimensions=dimension,464                Unit="Seconds",465                Statistic="Average",466                OKActions=[topic_arn_alarm],467                AlarmActions=[topic_arn_alarm],468                EvaluationPeriods=1,469                ComparisonOperator="GreaterThanThreshold",470                TreatMissingData="notBreaching",471            )472            response = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])473            snapshot.match("describe-alarm", response)474            retry(475                _check_alarm_triggered,476                retries=60,477                sleep=3.0,478                sleep_before=5,479                expected_state="ALARM",480                sqs_client=sqs_client,481                sqs_queue=sqs_queue,482                alarm_name=alarm_name,483                cloudwatch_client=cloudwatch_client,484                snapshot=snapshot,485                identifier="alarm-triggered",486            )487            # missing are treated as not breaching, so we should reach OK state again488            retry(489                _check_alarm_triggered,490                retries=60,491                sleep=3.0,492                sleep_before=5,493                expected_state="OK",494                sqs_client=sqs_client,495                sqs_queue=sqs_queue,496                alarm_name=alarm_name,497                cloudwatch_client=cloudwatch_client,498                snapshot=snapshot,499                identifier="ok-triggered",500            )501        finally:502            sns_client.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])503            cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])504    @pytest.mark.aws_validated505    @pytest.mark.skip_snapshot_verify(paths=["$..evaluatedDatapoints"])506    def test_breaching_alarm_actions(507        self,508        sns_client,509        cloudwatch_client,510        sqs_client,511        sns_create_topic,512        sqs_create_queue,513        snapshot,514    ):515        sns_topic_alarm = sns_create_topic()516        topic_arn_alarm = sns_topic_alarm["TopicArn"]517        snapshot.add_transformer(snapshot.transform.cloudwatch_api())518        snapshot.add_transformer(519            snapshot.transform.regex(topic_arn_alarm.split(":")[-1], "<topic_arn>"), priority=2520        )521        sqs_queue = sqs_create_queue()522        arn_queue = sqs_client.get_queue_attributes(523            QueueUrl=sqs_queue, AttributeNames=["QueueArn"]524        )["Attributes"]["QueueArn"]525        # required for AWS:526        sqs_client.set_queue_attributes(527            QueueUrl=sqs_queue,528            Attributes={"Policy": get_sqs_policy(arn_queue, topic_arn_alarm)},529        )530        metric_name = "my-metric101"531        dimension = [{"Name": "InstanceId", "Value": "abc"}]532        namespace = "test/breaching-alarm"533        alarm_name = f"test-alarm-{short_uid()}"534        try:535            subscription = sns_client.subscribe(536                TopicArn=topic_arn_alarm, Protocol="sqs", Endpoint=arn_queue537            )538            snapshot.match("cloudwatch_sns_subscription", subscription)539            cloudwatch_client.put_metric_alarm(540                AlarmName=alarm_name,541                AlarmDescription="testing cloudwatch alarms",542                MetricName=metric_name,543                Namespace=namespace,544                Period=10,545                Threshold=2,546                Dimensions=dimension,547                Unit="Seconds",548                Statistic="Average",549                OKActions=[topic_arn_alarm],550                AlarmActions=[topic_arn_alarm],551                EvaluationPeriods=2,552                ComparisonOperator="GreaterThanThreshold",553                TreatMissingData="breaching",554            )555            response = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])556            assert response["MetricAlarms"][0]["ActionsEnabled"]557            retry(558                _check_alarm_triggered,559                retries=80,560                sleep=3.0,561                sleep_before=5,562                expected_state="ALARM",563                sqs_client=sqs_client,564                sqs_queue=sqs_queue,565                alarm_name=alarm_name,566                cloudwatch_client=cloudwatch_client,567                snapshot=snapshot,568                identifier="alarm-1",569            )570        finally:571            sns_client.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])572            cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])573    @pytest.mark.aws_validated574    def test_enable_disable_alarm_actions(575        self,576        sns_client,577        cloudwatch_client,578        sqs_client,579        sns_create_topic,580        sqs_create_queue,581        snapshot,582    ):583        sns_topic_alarm = sns_create_topic()584        topic_arn_alarm = sns_topic_alarm["TopicArn"]585        snapshot.add_transformer(snapshot.transform.cloudwatch_api())586        snapshot.add_transformer(587            snapshot.transform.regex(topic_arn_alarm.split(":")[-1], "<topic_arn>"), priority=2588        )589        sqs_queue = sqs_create_queue()590        arn_queue = sqs_client.get_queue_attributes(591            QueueUrl=sqs_queue, AttributeNames=["QueueArn"]592        )["Attributes"]["QueueArn"]593        # required for AWS:594        sqs_client.set_queue_attributes(595            QueueUrl=sqs_queue,596            Attributes={"Policy": get_sqs_policy(arn_queue, topic_arn_alarm)},597        )598        metric_name = "my-metric101"599        dimension = [{"Name": "InstanceId", "Value": "abc"}]600        namespace = "test/enable"601        alarm_name = f"test-alarm-{short_uid()}"602        try:603            subscription = sns_client.subscribe(604                TopicArn=topic_arn_alarm, Protocol="sqs", Endpoint=arn_queue605            )606            snapshot.match("cloudwatch_sns_subscription", subscription)607            cloudwatch_client.put_metric_alarm(608                AlarmName=alarm_name,609                AlarmDescription="testing cloudwatch alarms",610                MetricName=metric_name,611                Namespace=namespace,612                Period=10,613                Threshold=2,614                Dimensions=dimension,615                Unit="Seconds",616                Statistic="Average",617                OKActions=[topic_arn_alarm],618                AlarmActions=[topic_arn_alarm],619                EvaluationPeriods=2,620                ComparisonOperator="GreaterThanThreshold",621                TreatMissingData="ignore",622            )623            response = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])624            assert response["MetricAlarms"][0]["ActionsEnabled"]625            snapshot.match("describe_alarm", response)626            cloudwatch_client.set_alarm_state(627                AlarmName=alarm_name, StateValue="ALARM", StateReason="testing alarm"628            )629            retry(630                _check_alarm_triggered,631                retries=80,632                sleep=3.0,633                sleep_before=5,634                expected_state="ALARM",635                sqs_client=sqs_client,636                sqs_queue=sqs_queue,637                alarm_name=alarm_name,638                cloudwatch_client=cloudwatch_client,639                snapshot=snapshot,640                identifier="alarm-state",641            )642            # disable alarm action643            cloudwatch_client.disable_alarm_actions(AlarmNames=[alarm_name])644            cloudwatch_client.set_alarm_state(645                AlarmName=alarm_name, StateValue="OK", StateReason="testing OK state"646            )647            response = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])648            snapshot.match("describe_alarm_disabled", response)649            assert response["MetricAlarms"][0]["StateValue"] == "OK"650            assert not response["MetricAlarms"][0]["ActionsEnabled"]651            retry(652                _check_alarm_triggered,653                retries=80,654                sleep=3.0,655                sleep_before=5,656                expected_state="OK",657                alarm_name=alarm_name,658                cloudwatch_client=cloudwatch_client,659                snapshot=snapshot,660                identifier="ok-state-action-disabled",661            )662            # enable alarm action663            cloudwatch_client.enable_alarm_actions(AlarmNames=[alarm_name])664            response = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])665            snapshot.match("describe_alarm_enabled", response)666            assert response["MetricAlarms"][0]["ActionsEnabled"]667        finally:668            sns_client.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])669            cloudwatch_client.delete_alarms(AlarmNames=[alarm_name])670def _check_alarm_triggered(671    expected_state,672    alarm_name,673    cloudwatch_client,674    sqs_client=None,675    sqs_queue=None,676    snapshot=None,677    identifier=None,678):679    response = cloudwatch_client.describe_alarms(AlarmNames=[alarm_name])680    assert response["MetricAlarms"][0]["StateValue"] == expected_state681    if snapshot:682        snapshot.match(f"{identifier}-describe", response)683    if not sqs_queue or not sqs_client:684        return685    result = sqs_client.receive_message(QueueUrl=sqs_queue, VisibilityTimeout=0)686    msg = result["Messages"][0]687    body = json.loads(msg["Body"])688    message = json.loads(body["Message"])689    if snapshot:690        snapshot.match(f"{identifier}-sqs-msg", message)691    receipt_handle = msg["ReceiptHandle"]692    sqs_client.delete_message(QueueUrl=sqs_queue, ReceiptHandle=receipt_handle)693    assert message["NewStateValue"] == expected_state694def check_message(695    sqs_client,696    expected_queue_url,697    expected_topic_arn,698    expected_new,699    expected_reason,700    alarm_name,701    alarm_description,702    expected_trigger,703):704    receive_result = sqs_client.receive_message(QueueUrl=expected_queue_url)705    message = None706    for msg in receive_result["Messages"]:707        body = json.loads(msg["Body"])708        if body["TopicArn"] == expected_topic_arn:709            message = json.loads(body["Message"])710            receipt_handle = msg["ReceiptHandle"]711            sqs_client.delete_message(QueueUrl=expected_queue_url, ReceiptHandle=receipt_handle)712            break713    assert message["NewStateValue"] == expected_new714    assert message["NewStateReason"] == expected_reason715    assert message["AlarmName"] == alarm_name716    assert message["AlarmDescription"] == alarm_description717    assert message["Trigger"] == expected_trigger718    return message719def get_sqs_policy(sqs_queue_arn, sns_topic_arn):720    return f"""721{{722  "Version":"2012-10-17",723  "Statement":[724    {{725      "Effect": "Allow",726      "Principal": {{ "AWS": "*" }},727      "Action": "sqs:SendMessage",728      "Resource": "{sqs_queue_arn}",729      "Condition":{{730        "ArnEquals":{{731        "aws:SourceArn":"{sns_topic_arn}"732        }}733      }}734    }}735  ]736}}...

test_simple_server.py

Source:test_simple_server.py

1#!/usr/bin/python2"""3Licensed under the Amazon Software License (the "License"). You may not use this file4except in compliance with the License. A copy of the License is located at5http://aws.amazon.com/asl/6or in the "license" file accompanying this file. This file is distributed on an "AS IS"7BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the8License for the specific language governing permissions and limitations under the License.'9Sends GET requests to the specified target host and posts metrics10to CloudWatch based on the result. This will start one testing11thread per CPU.12Usage:13  ./test_simple_server.py <aws_region> <target_hostname>14"""15import boto316import multiprocessing17import requests18import sys19from threading import Thread20import time21METRIC_BATCH_SIZE = 1022def make_call(hostname):23    try:24        r = requests.get("http://"+hostname, timeout=5.0)25        r.raise_for_status()26        return True27    except:28        return False29def emit_metrics(cloudwatch_client, successes, errors):30    metric_data = []31    for i in range(successes):32        metric_data.append({"MetricName": "Success", "Value": 1})33        metric_data.append({"MetricName": "Error", "Value": 0})34    for i in range(errors):35        metric_data.append({"MetricName": "Success", "Value": 0})36        metric_data.append({"MetricName": "Error", "Value": 1})37    cloudwatch_client.put_metric_data(Namespace="ReInvent-EC2SpotTestDev-Metrics", MetricData=metric_data)38def rate_limit(start):39    elapsed = time.time() - start40    if (elapsed < 1.0):41        time.sleep(1.0 - elapsed)42def start_test_loop(cloudwatch_client, hostname):43    while True:44        successes = 045        errors = 046        for i in range(METRIC_BATCH_SIZE):47            start = time.time()48            is_success = make_call(hostname)49            if is_success:50                successes += 151            else:52                errors += 153            rate_limit(start)54        emit_metrics(cloudwatch_client, successes, errors)55aws_region = sys.argv[1]56cloudwatch_client = boto3.client('cloudwatch', region_name=aws_region)57hostname = sys.argv[2]58cpu_count = multiprocessing.cpu_count()59for i in range(cpu_count):60    t = Thread(target=start_test_loop, args=(cloudwatch_client, hostname))...

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.