Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use schedule_metric_alarm method in localstack

Best Python code snippet using localstack_python

alarm_scheduler.py

Source:alarm_scheduler.py

...41        Shutsdown the scheduler, must be called before application stops42        """43        self.scheduler.close()44        self.thread.join(5)45    def schedule_metric_alarm(self, alarm_arn: str) -> None:46        """(Re-)schedules the alarm, if the alarm is re-scheduled, the running alarm scheduler will be cancelled before47        starting a new one"""48        alarm_details = get_metric_alarm_details_for_alarm_arn(alarm_arn)49        self.delete_scheduler_for_alarm(alarm_arn)50        if not self._is_alarm_supported(alarm_details):51            LOG.warning(52                "Given alarm configuration not yet supported, alarm state will not be evaluated."53            )54            return55        period = alarm_details["Period"]56        evaluation_periods = alarm_details["EvaluationPeriods"]57        schedule_period = evaluation_periods * period58        def on_error(e):59            LOG.exception("Error executing scheduled alarm", exc_info=e)60        task = self.scheduler.schedule(61            func=calculate_alarm_state,62            period=schedule_period,63            fixed_rate=True,64            args=[alarm_arn],65            on_error=on_error,66        )67        self.scheduled_alarms[alarm_arn] = task68    def delete_scheduler_for_alarm(self, alarm_arn: str) -> None:69        """70        Deletes the recurring scheduler for an alarm71        :param alarm_arn: the arn of the alarm to be removed72        """73        task = self.scheduled_alarms.pop(alarm_arn, None)74        if task:75            task.cancel()76    def restart_existing_alarms(self) -> None:77        """78        Only used re-create persistent state. Reschedules alarms that already exist79        """80        service = "cloudwatch"81        for region in aws_stack.get_valid_regions_for_service(service):82            client = aws_stack.connect_to_service(service, region_name=region)83            result = client.describe_alarms()84            for metric_alarm in result["MetricAlarms"]:85                arn = metric_alarm["AlarmArn"]86                self.schedule_metric_alarm(alarm_arn=arn)87    def _is_alarm_supported(self, alarm_details: MetricAlarm) -> bool:88        required_parameters = ["Period", "Statistic", "MetricName", "Threshold"]89        for param in required_parameters:90            if param not in alarm_details.keys():91                LOG.debug(92                    f"Currently only simple MetricAlarm are supported. Alarm is missing '{param}'. ExtendedStatistic is not yet supported."93                )94                return False95        if alarm_details["ComparisonOperator"] not in COMPARISON_OPS.keys():96            LOG.debug(97                f"ComparisonOperator '{alarm_details['ComparisonOperator']}' not yet supported."98            )99            return False100        return 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.