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

Copied to Clipboard!

How to use notification_configs method in localstack

Best Python code snippet using localstack_python

monitor.py

Source:monitor.py

1# -*- coding: UTF-8 â*-2"""3@Author: LennonChin4@Contact: i@coderap.com5@Date: 2021-10-196"""7import sys8import os9import random10import datetime11import requests12import json13import time14import hmac15import hashlib16import base6417import urllib.parse18class Utils:19    @staticmethod20    def time_title(message):21        return "[{}] {}".format(datetime.datetime.now().strftime('%H:%M:%S'), message)22    @staticmethod23    def log(message):24        print(Utils.time_title(message))25    @staticmethod26    def send_message(notification_configs, message, **kwargs):27        if len(message) == 0:28            return29        # Wrapper for exception caught30        def invoke(func, configs):31            try:32                func(configs, message, **kwargs)33            except Exception as err:34                Utils.log(err)35        # DingTalk message36        invoke(Utils.send_dingtalk_message, notification_configs["dingtalk"])37        # Bark message38        invoke(Utils.send_bark_message, notification_configs["bark"])39        # Telegram message40        invoke(Utils.send_telegram_message, notification_configs["telegram"])41    @staticmethod42    def send_dingtalk_message(dingtalk_configs, message, **kwargs):43        if len(dingtalk_configs["access_token"]) == 0 or len(dingtalk_configs["secret_key"]) == 0:44            return45        timestamp = str(round(time.time() * 1000))46        secret_enc = dingtalk_configs["secret_key"].encode('utf-8')47        string_to_sign = '{}\n{}'.format(timestamp, dingtalk_configs["secret_key"])48        string_to_sign_enc = string_to_sign.encode('utf-8')49        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()50        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))51        headers = {52            'Content-Type': 'application/json'53        }54        params = {55            "access_token": dingtalk_configs["access_token"],56            "timestamp": timestamp,57            "sign": sign58        }59        content = {60            "msgtype": "text" if "message_type" not in kwargs else kwargs["message_type"],61            "text": {62                "content": message63            }64        }65        response = requests.post("https://oapi.dingtalk.com/robot/send", headers=headers, params=params, json=content)66        Utils.log("Dingtalkéåºæ¶æ¯çæï¼{}".format(response.status_code))67    @staticmethod68    def send_telegram_message(telegram_configs, message, **kwargs):69        if len(telegram_configs["bot_token"]) == 0 or len(telegram_configs["chat_id"]) == 0:70            return71        headers = {72            'Content-Type': 'application/json'73        }74        proxies = {75            "https": telegram_configs["http_proxy"],76        }77        content = {78            "chat_id": telegram_configs["chat_id"],79            "text": message80        }81        url = "https://api.telegram.org/bot{}/sendMessage".format(telegram_configs["bot_token"])82        response = requests.post(url, headers=headers, proxies=proxies, json=content)83        Utils.log("Telegraméåºæ¶æ¯çæï¼{}".format(response.status_code))84    @staticmethod85    def send_bark_message(bark_configs, message, **kwargs):86        if len(bark_configs["url"]) == 0:87            return88        url = "{}/{}".format(bark_configs["url"], urllib.parse.quote(message, safe=""))89        response = requests.post(url, params=bark_configs["query_parameters"])90        Utils.log("Barkéåºæ¶æ¯çæï¼{}".format(response.status_code))91class AppleStoreMonitor:92    headers = {93        'sec-ch-ua': '"Google Chrome";v="93", " Not;A Brand";v="99", "Chromium";v="93"',94        'Referer': 'https://www.apple.com.cn/store',95        'DNT': '1',96        'sec-ch-ua-mobile': '?0',97        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.82 Safari/537.36',98        'sec-ch-ua-platform': '"macOS"',99    }100    def __init__(self):101        self.count = 1102    @staticmethod103    def config():104        """105        é²è¡åé¡æä½106        """107        products = json.load(open('products.json', encoding='utf-8'))108        configs = {109            "selected_products": {},110            "selected_area": "",111            "exclude_stores": [],112            "notification_configs": {113                "dingtalk": {114                    "access_token": "",115                    "secret_key": ""116                },117                "telegram": {118                    "bot_token": "",119                    "chat_id": "",120                    "http_proxy": ""121                },122                "bark": {123                    "url": "",124                    "query_parameters": {125                        "url": None,126                        "isArchive": None,127                        "group": None,128                        "icon": None,129                        "automaticallyCopy": None,130                        "copy": None131                    }132                }133            },134            "scan_interval": 30,135            "alert_exception": False136        }137        while True:138            # chose product type139            print('--------------------')140            for index, item in enumerate(products):141                print('[{}] {}'.format(index, item))142            product_type = list(products)[int(input('é¸æè¦æ¥çåèï¼'))]143            # chose product classification144            print('--------------------')145            for index, (key, value) in enumerate(products[product_type].items()):146                print('[{}] {}'.format(index, key))147            product_classification = list(products[product_type])[int(input('é¸æè¦æ¥çåèåé¡ï¼'))]148            # chose product classification149            print('--------------------')150            for index, (key, value) in enumerate(products[product_type][product_classification].items()):151                print('[{}] {}'.format(index, value))152            product_model = list(products[product_type][product_classification])[int(input('é¸æè¦æ¥çIPHONEåèï¼'))]153            configs["selected_products"][product_model] = (154                product_classification, products[product_type][product_classification][product_model])155            print('--------------------')156            if len(input('æ¯å¦å¢å æ´ç¢å[Enterç¹¼çºæ·»å ï¼éEnteréµéåº]ï¼')) != 0:157                break158        # config area159        print('é¸æé ç´å°åï¼')160        url_param = ['state', 'city', 'district']161        choice_params = {}162        param_dict = {}163        for step, param in enumerate(url_param):164            print('è«ç¨å¾...{}/{}'.format(step + 1, len(url_param)))165            response = requests.get("https://www.apple.com.cn/shop/address-lookup", headers=AppleStoreMonitor.headers,166                                    params=choice_params)167            result_param = json.loads(response.text)['body'][param]168            if type(result_param) is dict:169                result_data = result_param['data']170                print('--------------------')171                for index, item in enumerate(result_data):172                    print('[{}] {}'.format(index, item['value']))173                input_index = int(input('è«é¸æåèï¼'))174                choice_result = result_data[input_index]['value']175                param_dict[param] = choice_result176                choice_params[param] = param_dict[param]177            else:178                choice_params[param] = result_param179        print('æ£å¨è¼å¥ç¶²çµ¡è³æº...')180        response = requests.get("https://www.apple.com.cn/shop/address-lookup", headers=AppleStoreMonitor.headers,181                                params=choice_params)182        selected_area = json.loads(response.text)['body']['provinceCityDistrict']183        configs["selected_area"] = selected_area184        print('--------------------')185        print("é¸æé ç´å°åæ¯ï¼{}ï¼è¼å¥é ç´å°æ¢å¨åç´çåº...".format(selected_area))186        store_params = {187            "location": selected_area,188            "parts.0": list(configs["selected_products"].keys())[0]189        }190        response = requests.get("https://www.apple.com.cn/shop/fulfillment-messages",191                                headers=AppleStoreMonitor.headers, params=store_params)192        stores = json.loads(response.text)['body']["content"]["pickupMessage"]["stores"]193        for index, store in enumerate(stores):194            print("[{}] {}ï¼å°åï¼{}".format(index, store["storeName"], store["retailStore"]["address"]["street"]))195        exclude_stores_indexes = input('æé¤ç¡éç£æ¸¬çç´çåºï¼è¼¸å¥åºèè¾[ç´æ¥Enterä»£è¡¨å¨é¨ç£æ¸¬ï¼å¤ååºåºèä»¥ç©ºæ ¼åé]ï¼').strip().split()196        if len(exclude_stores_indexes) != 0:197            print("å·²é¸æçç¡éç£æ¸¬ç´åºï¼{}".format("ï¼".join(list(map(lambda i: stores[int(i)]["storeName"], exclude_stores_indexes)))))198            configs["exclude_stores"] = list(map(lambda i: stores[int(i)]["storeNumber"], exclude_stores_indexes))199        print('--------------------')200        # config notification configurations201        notification_configs = configs["notification_configs"]202        # config dingtalk notification203        dingtalk_access_token = input('è¾å¥ééæ©å¨äººAccess Token[å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼')204        dingtalk_secret_key = input('è¾å¥ééæ©å¨äººSecret Key[å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼')205        # write dingtalk configs206        notification_configs["dingtalk"]["access_token"] = dingtalk_access_token207        notification_configs["dingtalk"]["secret_key"] = dingtalk_secret_key208        # config telegram notification209        print('--------------------')210        telegram_chat_id = input('è¾å¥Telegramæºå¨äººChat ID[å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼')211        telegram_bot_token = input('è¾å¥Telegramæºå¨äººToken[å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼')212        telegram_http_proxy = input('è¾å¥Telegram HTTPä»£çå°å[å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼')213        # write telegram configs214        notification_configs["telegram"]["chat_id"] = telegram_chat_id215        notification_configs["telegram"]["bot_token"] = telegram_bot_token216        notification_configs["telegram"]["http_proxy"] = telegram_http_proxy217        # config bark notification218        print('--------------------')219        bark_url = input('è¾å¥Bark URL[å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼')220        # write dingtalk configs221        notification_configs["bark"]["url"] = bark_url222        # è¼¸å¥æçééæé223        print('--------------------')224        configs["scan_interval"] = int(input('è¾å¥æçééæé[ä»¥ç§çºå®ä½ï¼é»èªçº30ç§ï¼å¦ä¸éç½®ç´æ¥Enterå³å¯]ï¼') or 30)225        # æ¯å¦å°ç°å¸¸é²è¡è¦å226        print('--------------------')227        configs["alert_exception"] = (input('æ¯å¦å¨ç¨åºç°å¸¸æç¼åºéç¥[Y/nï¼é»èªçºn]ï¼').lower().strip() or "n") == "y"228        with open('apple_store_monitor_configs.json', 'w') as file:229            json.dump(configs, file, indent=2)230            print('--------------------')231            print("æçéç½®å·±çæï¼ä¸¦å·²å¯«å¥å°{}æä»¶ä¸\nè¯·ä½¿ç¨ python {} start å½ä»¤å§åç£æ¸¬".format(file.name, os.path.abspath(__file__)))232    def start(self):233        """234        éå§ç£æ¸¬235        """236        configs = json.load(open('apple_store_monitor_configs.json', encoding='utf-8'))237        selected_products = configs["selected_products"]238        selected_area = configs["selected_area"]239        exclude_stores = configs["exclude_stores"]240        notification_configs = configs["notification_configs"]241        scan_interval = configs["scan_interval"]242        alert_exception = configs["alert_exception"]243        products_info = []244        for index, product_info in enumerate(selected_products.items()):245            products_info.append("ã{}ã{}".format(index, " ".join(product_info[1])))246        message = "æºåéå§ç£æ¸¬ï¼ååä¿¡æ¯å¦ä¸ï¼\n{}\nåè²¨ååï¼{}\næçé »æ¬¡ï¼{}ç§/æ¬¡".format("\n".join(products_info), selected_area,247                                                                   scan_interval)248        Utils.log(message)249        Utils.send_message(notification_configs, message)250        params = {251            "location": selected_area,252            "mt": "regular",253        }254        code_index = 0255        product_codes = selected_products.keys()256        for product_code in product_codes:257            params["parts.{}".format(code_index)] = product_code258            code_index += 1259        # ä¸æ¬¡æ´é»éç¥æé260        last_exactly_time = -1261        while True:262            available_list = []263            tm_hour = time.localtime(time.time()).tm_hour264            try:265                # æ´æ°è«æ±æé266                params["_"] = int(time.time() * 1000)267                response = requests.get("https://www.apple.com.cn/shop/fulfillment-messages",268                                        headers=AppleStoreMonitor.headers,269                                        params=params)270                json_result = json.loads(response.text)271                stores = json_result['body']['content']['pickupMessage']['stores']272                Utils.log(273                    '-------------------- ç¬¬{}æ¬¡æç --------------------'.format(274                        self.count + 1))275                for item in stores:276                    store_name = item['storeName']277                    if item["storeNumber"] in exclude_stores:278                        print("ã{}ï¼å·²æé¤ã".format(store_name))279                        continue280                    print("{:-<100}".format("ã{}ã".format(store_name)))281                    for product_code in product_codes:282                        pickup_search_quote = item['partsAvailability'][product_code]['pickupSearchQuote']283                        pickup_display = item['partsAvailability'][product_code]['pickupDisplay']284                        store_pickup_product_title = item['partsAvailability'][product_code]['storePickupProductTitle']285                        print('\tã{}ã{}'.format(pickup_search_quote, store_pickup_product_title))286                        if pickup_search_quote == 'ä»å¤©å¯åè²¨' or pickup_display != 'unavailable':287                            available_list.append((store_name, product_code, store_pickup_product_title))288                if len(available_list) > 0:289                    messages = []290                    print("å½ä¸è²¨æºï¼è«æ³¨æ >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")291                    Utils.log("ä»¥ä¸ç´çåºé ç´å¯ç¨ï¼")292                    for item in available_list:293                        messages.append("ã{}ã {}".format(item[0], item[2]))294                        print("ã{}ã{}".format(item[0], item[2]))295                    print("å½ä¸è²¨æºï¼è«æ³¨æ >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")296                    Utils.send_message(notification_configs,297                                       Utils.time_title(298                                           "ç¬¬{}æ¬¡æçå°ç´çåºæè²¨ï¼ä¿¡æ¯å¦ä¸ï¼\n{}".format(self.count, "\n".join(messages))))299            except Exception as err:300                Utils.log(err)301                # 6:00 ~ 23:00æåéå¼å¸¸æ¶æ¯302                if alert_exception and 6 <= tm_hour <= 23:303                    Utils.send_message(notification_configs,304                                       Utils.time_title("ç¬¬{}æ¬¡æçå°åºç¾ç°å¸¸ï¼{}".format(self.count, repr(err))))305            if len(available_list) == 0:306                interval = max(random.randint(int(scan_interval / 2), scan_interval * 2), 5)307                Utils.log('{}ç§å¾é²è¡ç¬¬{}æ¬¡åè©¦...'.format(interval, self.count))308                # æ´ç¹éç¥ï¼ç¨äºé¶æ®µæ§æ£æµåºç¨æ¯å¦æ£å¸¸309                if last_exactly_time != tm_hour and (6 <= tm_hour <= 23):310                    Utils.send_message(notification_configs,311                                       Utils.time_title("å·²æç{}æ¬¡ï¼æçç¨åºéä½æ£å¸¸".format(self.count)))312                    last_exactly_time = tm_hour313                time.sleep(interval)314            else:315                time.sleep(5)316            # æ¬¡æ°èªå¢317            self.count += 1318if __name__ == '__main__':319    args = sys.argv320    if len(args) != 2:321        print("""322        Usage: python {} <option>323        option can be:324        \tconfig: pre config of products or notification325        \tstart: start to monitor326        """.format(args[0]))327        exit(1)328    if args[1] == "config":329        AppleStoreMonitor.config()330    if args[1] == "start":...

notification_manager.py

Source:notification_manager.py

1from abc import ABC, abstractmethod2from collections import namedtuple3import asyncio4from threading import Thread5from pathlib import Path6import time7from typing import List, Any, Optional, Callable8import yaml9from sensor_manager import SensorManager10from email_manager import EmailManager11class NotificationConfig(ABC):12    def __init__(self, get_time: Callable[[], float], sender_email: str, receiver_email: str, message_subject: str, message: Optional[str], check_interval: float):13        self.get_time = get_time14        self.sender_email = sender_email15        self.receiver_email = receiver_email16        self.message_subject = message_subject17        self.message = message18        self.check_interval = check_interval19    @classmethod20    @abstractmethod21    def get_type_name(cls):22        pass23    @classmethod24    def from_dict(cls, raw_dict, get_time):25        """get_time: function to get current time in seconds from epoch"""26        if raw_dict['type'] != cls.get_type_name():27            raise Exception('raw dictionary does not seem to encode for this message type')28        cls_params = cls.params_from_dict(raw_dict)29        if cls_params == False:30            raise Exception('raw dictionary does not seem to encode for this message type')31        32        result = cls(33            get_time=get_time,34            sender_email=raw_dict['sender'],35            receiver_email=raw_dict['receiver'],36            message_subject=raw_dict['message_subject'],37            message=raw_dict['message'] if 'message' in raw_dict else None,38            check_interval=float(raw_dict['check_interval']),39            **cls_params)40        return result41    @classmethod42    @abstractmethod43    def params_from_dict(cls, raw_dict):44        pass45    @abstractmethod46    def check_signal(self, current_sensor_values) -> bool:47        pass48class BaseThresholdNotificationConfig(NotificationConfig):49    def __init__(self, sensor_id: str, threshold: float, min_breach_duration: float, **kwargs):50        super().__init__(**kwargs)51        self.sensor_id = sensor_id52        self.threshold = threshold53        self.min_breach_duration = min_breach_duration54        self.previous_check_value = None55        self.tracking_breach = False56        self.current_breach_start_timestamp = None57        self.current_breach_notified = False58    @classmethod59    @abstractmethod60    def get_type_name(cls):61        pass62    @classmethod63    def params_from_dict(cls, raw_dict):64        return {65            'sensor_id': raw_dict['sensor'],66            'threshold': float(raw_dict['threshold']),67            'min_breach_duration': float(raw_dict['min_breach_duration'])68        }69    def check_signal(self, current_sensor_values) -> bool:70        breached = False71        72        try:73            breached = self.check_threshold_breached(current_sensor_values)74        except Exception as e:75            breached = self.tracking_breach76        if breached and not self.tracking_breach:77            self.tracking_breach = True78            self.current_breach_start_timestamp = self.get_time()79            self.current_breach_notified = False80        elif not breached and self.tracking_breach:81            self.tracking_breach = False82        if breached and self.tracking_breach:83            current_timestamp = self.get_time()84            duration = current_timestamp - self.current_breach_start_timestamp85            duration_reached = duration >= self.min_breach_duration86            result = duration_reached and not self.current_breach_notified 87            if duration_reached:88                self.current_breach_notified = True89            90            return result91        return False92    @abstractmethod93    def check_threshold_breached(self, current_sensor_values):94        pass95    def __str__(self):96        return f'Notification {{ {self.get_type_name()}: {str(self.threshold)}, sender: {self.sender_email}, receiver: {self.receiver_email} }}'97class FallBelowNotificationConfig(BaseThresholdNotificationConfig):98	@classmethod99	def get_type_name(cls):100		return 'fall_below'101	102	def check_threshold_breached(self, current_sensor_values) -> bool:103		current_value = current_sensor_values[self.sensor_id]104		result = False105		if self.previous_check_value is None:106			result = self.threshold > current_value107		else:108			result = self.threshold > current_value and self.previous_check_value > self.threshold109		self.previous_check_value = current_value110		return result111class RiseAboveNotificationConfig(BaseThresholdNotificationConfig):112	@classmethod113	def get_type_name(cls):114		return 'rise_above'115	def check_threshold_breached(self, current_sensor_values) -> bool:116		current_value = current_sensor_values[self.sensor_id]117		if current_value is None:118			print('delaying check because sensor read value is none')119			raise Exception('no valid value, could not check threshold breached')120		else:121			return current_value > self.threshold122class SystemStartNotificationConfig(NotificationConfig):123	def __init__(self, *args, **kwargs):124		super().__init__(*args, **kwargs)125		self.checked_once = False126	@classmethod127	def params_from_dict(cls, raw_dict):128		return {}129	@classmethod130	def get_type_name(cls):131		return 'system_start'132	def check_signal(self, current_sensor_values) -> bool:133		result = not self.checked_once134		self.checked_once = True135		return result136class NotificationManager:137    def __init__(self, notification_configs, sensor_manager: SensorManager, email_manager: EmailManager, get_time = time.time):138        self.notification_configs = notification_configs139        self.sensor_manager = sensor_manager140        self.email_manager = email_manager141        self.get_time = get_time142        self.last_check_timestamps = {}143        self.current_sensor_values = None144        self.master_check_interval = 1 if len(notification_configs) == 0 else min([x.check_interval for x in notification_configs])145        self.watch_thread = None146        print('Notification Manager master_check_interval', self.master_check_interval)147    def start_watching(self):148        if self.watch_thread is not None:149            raise Exception('can only start one watching thread per notification manager')150        self.watch_thread = Thread(target=self.watch_loop)151        self.watch_thread.start()152    def watch_loop(self):153        while True:154            self.single_watch_step()155            time.sleep(self.master_check_interval)156    def single_watch_step(self):157        current_timestamp = self.get_time()158        self.current_sensor_values = asyncio.run(self.sensor_manager.get_latest_values_filled_with_previous())159        try:160            for index, notification_config in enumerate(self.notification_configs):161                if not index in self.last_check_timestamps or\162                    self.last_check_timestamps[index] + notification_config.check_interval <= current_timestamp:163                    164                    self.last_check_timestamps[index] = current_timestamp165                    try:166                        if notification_config.check_signal(self.current_sensor_values):167                            self.send_notification(notification_config)168                    except Exception as e:169                        print(f'an error occurred while attempting to check or send notification {notification_config}', e)170        except Exception as e:171            print('an error occurred while checking for possible notifications', e)172    def send_notification(self, notification_config: NotificationConfig):173        print('send notification for config:', notification_config)174        175        text = notification_config.message176        if text is None:177            text = 'notification was sent automatically'178        self.email_manager.send_email(179            sender_address=notification_config.sender_email,180            receiver_address=notification_config.receiver_email,181            subject=notification_config.message_subject,182            text=text)183    @staticmethod184    def parse_notifications_config(config_file_path: Path, get_time=time.time) -> List[Any]:185        with open(config_file_path, 'rb') as file:186            raw_notification_configs = yaml.load(file, Loader=yaml.Loader)187            potential_config_types = [FallBelowNotificationConfig, RiseAboveNotificationConfig, SystemStartNotificationConfig]188            notification_configs = []189            for index, raw_notification_config in enumerate(raw_notification_configs):190                found_matching_type = False191                for potential_type in potential_config_types:192                    try:193                        parsed = potential_type.from_dict(raw_notification_config, get_time=get_time)194                        if parsed is not None:195                            notification_configs.append(parsed)196                            found_matching_type = True197                            break	198                    except Exception as e:199                        print(f'error when parsing config {index}: {e}')200                        pass201                if not found_matching_type:202                    raise Exception(f'found no matching type for notification config at index: {index}, raw: {raw_notification_config}')203                    ...

asg_event_notifications_util.py

Source:asg_event_notifications_util.py

1from __future__ import absolute_import2from __future__ import print_function3import boto34import click5@click.group()6def cli():7    pass8def get_asg_infos():9    response = client.describe_auto_scaling_groups(MaxRecords=100)10    auto_scaling_groups = response['AutoScalingGroups']11    return auto_scaling_groups12def get_asg_names():13    asg_names = list()14    for asg in get_asg_infos():15        asg_names.append(asg['AutoScalingGroupName'])16    return asg_names17def get_asg_event_notifications(asg):18    event_notifications = list()19    response = \20        client.describe_notification_configurations(AutoScalingGroupNames=[asg],21            MaxRecords=100)22    notification_configs = response['NotificationConfigurations']23    for notification in notification_configs:24        event_notifications.append(notification['NotificationType'])25    return event_notifications26@click.command()27def show_asg_event_notifications():28    try:29        for asg in get_asg_names():30            event_notifications = get_asg_event_notifications(asg)31            if event_notifications:32                print(("Event notifications: {0} are set for ASG: {1}".format(event_notifications,33                        asg)))34            else:35                print(("No Event Notifications found for ASG {}".format(asg)))36    except Exception as e:37        print(e)38@click.command()39@click.option('--topic_arn', help='The ARN of Amazon SNS topic',40              required=True)41@click.option('--event',42              help='The type of event that causes the notification to be sent'43              , default='autoscaling:EC2_INSTANCE_LAUNCH_ERROR')44@click.option('--confirm', default=False, required=False, is_flag=True,45              help='Set this to create event notification for asg')46def create_asg_event_notifications(47    topic_arn,48    event,49    confirm,50    ):51    asg_names = get_asg_names()52    asg_to_create_event_notifications = list()53    for asg_name in asg_names:54        event_notifications = get_asg_event_notifications(asg_name)55        if event in event_notifications:56             continue57        else:58             asg_to_create_event_notifications.append(asg_name)59    if confirm is False:60        print(("Would have created the event notification for asgs {}".format(asg_to_create_event_notifications)))61    else:62        try:63            for asg in asg_to_create_event_notifications:64                response = \65                    client.put_notification_configuration(AutoScalingGroupName=asg,66                        TopicARN=topic_arn, NotificationTypes=[event])67                print(("Created {0} event notifications for auto scaling group {1}").format(event,68                       asg))69        except Exception as e:70            print(e)71cli.add_command(show_asg_event_notifications)72cli.add_command(create_asg_event_notifications)73if __name__ == '__main__':74    client = boto3.client('autoscaling')...

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.