KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _set_alarm_actions method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...168    return json.dumps(response)169class ValidationError(CommonServiceException):170    def __init__(self, message: str):171        super().__init__("ValidationError", message, 400, True)172def _set_alarm_actions(context, alarm_names, enabled):173    backend = cloudwatch_backends[context.region]174    for name in alarm_names:175        alarm = backend.alarms.get(name)176        if alarm:177            alarm.actions_enabled = enabled178def _cleanup_describe_output(alarm):179    reason_data = alarm.get("StateReasonData")180    if reason_data is not None and reason_data in ("{}", ""):181        alarm.pop("StateReasonData")182    if (183        alarm.get("StateReason", "") == MOTO_INITIAL_UNCHECKED_REASON184        and alarm.get("StateValue") != StateValue.INSUFFICIENT_DATA185    ):186        alarm["StateValue"] = StateValue.INSUFFICIENT_DATA187class CloudwatchProvider(CloudwatchApi, ServiceLifecycleHook):188    """189    Cloudwatch provider.190    LIMITATIONS:191        - no alarm rule evaluation192    """193    def __init__(self):194        self.tags = TaggingService()195        self.alarm_scheduler = None196    def on_after_init(self):197        ROUTER.add(PATH_GET_RAW_METRICS, self.get_raw_metrics)198        self.alarm_scheduler = AlarmScheduler()199    def on_before_start(self):200        # re-schedule alarms for persistence use-case201        def restart_alarms(*args):202            poll_condition(lambda: SERVICE_PLUGINS.is_running("cloudwatch"))203            self.alarm_scheduler.restart_existing_alarms()204        start_worker_thread(restart_alarms)205    def on_before_stop(self):206        self.alarm_scheduler.shutdown_scheduler()207    def delete_alarms(self, context: RequestContext, alarm_names: AlarmNames) -> None:208        moto.call_moto(context)209        for alarm_name in alarm_names:210            arn = aws_stack.cloudwatch_alarm_arn(alarm_name)211            self.alarm_scheduler.delete_scheduler_for_alarm(arn)212    def get_raw_metrics(self, request: Request):213        region = aws_stack.extract_region_from_auth_header(request.headers)214        backend = cloudwatch_backends[region]215        if backend:216            result = backend.metric_data217        else:218            result = []219        result = [220            {221                "ns": r.namespace,222                "n": r.name,223                "v": r.value,224                "t": r.timestamp,225                "d": [{"n": d.name, "v": d.value} for d in r.dimensions],226            }227            for r in result228        ]229        return {"metrics": result}230    def list_tags_for_resource(231        self, context: RequestContext, resource_arn: AmazonResourceName232    ) -> ListTagsForResourceOutput:233        tags = self.tags.list_tags_for_resource(resource_arn)234        return ListTagsForResourceOutput(Tags=tags.get("Tags", []))235    def untag_resource(236        self, context: RequestContext, resource_arn: AmazonResourceName, tag_keys: TagKeyList237    ) -> UntagResourceOutput:238        self.tags.untag_resource(resource_arn, tag_keys)239        return UntagResourceOutput()240    def tag_resource(241        self, context: RequestContext, resource_arn: AmazonResourceName, tags: TagList242    ) -> TagResourceOutput:243        self.tags.tag_resource(resource_arn, tags)244        return TagResourceOutput()245    @handler("PutMetricAlarm", expand=False)246    def put_metric_alarm(247        self,248        context: RequestContext,249        request: PutMetricAlarmInput,250    ) -> None:251        # missing will be the default, when not set (but it will not explicitly be set)252        if not request.get("TreatMissingData", "missing") in [253            "breaching",254            "notBreaching",255            "ignore",256            "missing",257        ]:258            raise ValidationError(259                f"The value {request['TreatMissingData']} is not supported for TreatMissingData parameter. Supported values are [breaching, notBreaching, ignore, missing]."260            )261        # do some sanity checks:262        if request.get("Period"):263            # Valid values are 10, 30, and any multiple of 60.264            value = request.get("Period")265            if value not in (10, 30):266                if value % 60 != 0:267                    raise ValidationError("Period must be 10, 30 or a multiple of 60")268        if request.get("Statistic"):269            if not request.get("Statistic") in [270                "SampleCount",271                "Average",272                "Sum",273                "Minimum",274                "Maximum",275            ]:276                raise ValidationError(277                    f"Value '{request.get('Statistic')}' at 'statistic' failed to satisfy constraint: Member must satisfy enum value set: [Maximum, SampleCount, Sum, Minimum, Average]"278                )279        moto.call_moto(context)280        name = request.get("AlarmName")281        arn = aws_stack.cloudwatch_alarm_arn(name)282        self.tags.tag_resource(arn, request.get("Tags"))283        self.alarm_scheduler.schedule_metric_alarm(arn)284    @handler("PutCompositeAlarm", expand=False)285    def put_composite_alarm(286        self,287        context: RequestContext,288        request: PutCompositeAlarmInput,289    ) -> None:290        backend = cloudwatch_backends[context.region]291        backend.put_metric_alarm(292            name=request.get("AlarmName"),293            namespace=None,294            metric_name=None,295            metric_data_queries=None,296            comparison_operator=None,297            evaluation_periods=None,298            datapoints_to_alarm=None,299            period=None,300            threshold=None,301            statistic=None,302            extended_statistic=None,303            description=request.get("AlarmDescription"),304            dimensions=[],305            alarm_actions=request.get("AlarmActions", []),306            ok_actions=request.get("OKActions", []),307            insufficient_data_actions=request.get("InsufficientDataActions", []),308            unit=None,309            actions_enabled=request.get("ActionsEnabled"),310            treat_missing_data=None,311            evaluate_low_sample_count_percentile=None,312            threshold_metric_id=None,313            rule=request.get("AlarmRule"),314            tags=request.get("Tags", []),315        )316        LOG.warning(317            "Composite Alarms configuration is not yet supported, alarm state will not be evaluated"318        )319    @handler("EnableAlarmActions")320    def enable_alarm_actions(self, context: RequestContext, alarm_names: AlarmNames) -> None:321        _set_alarm_actions(context, alarm_names, enabled=True)322    @handler("DisableAlarmActions")323    def disable_alarm_actions(self, context: RequestContext, alarm_names: AlarmNames) -> None:324        _set_alarm_actions(context, alarm_names, enabled=False)325    @handler("DescribeAlarms", expand=False)326    def describe_alarms(327        self, context: RequestContext, request: DescribeAlarmsInput328    ) -> DescribeAlarmsOutput:329        response = moto.call_moto(context)330        for c in response["CompositeAlarms"]:331            _cleanup_describe_output(c)332        for m in response["MetricAlarms"]:333            _cleanup_describe_output(m)...

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.