How to use set_secret method in localstack

Best Python code snippet using localstack_python

mes_request.py

Source:mes_request.py

...68                                    headers=self.refresh_header())69            response_content = json.loads(response.text)70            if 'access_token' in response_content:71                self.access_token = response_content['access_token']72                self.az_kv.set_secret("AccessToken", self.access_token)73                self.az_kv.set_secret("IsValid", "YES")74                self.az_kv.set_secret("AuthRetryCounter", 1)75            else:76                # if the refresh failed, request brand new API credentials77                response = requests.post(self.api_domain + "/oauth/token",78                                        data="grant_type=client_credentials",79                                        headers=self.header(self.api_key))80                response_content = json.loads(response.text)81                if 'access_token' in response_content:82                    self.access_token = response_content['access_token']83                    self.refresh_token = response_content['refresh_token']84                    self.az_kv.set_secret("AccessToken", self.access_token) 85                    self.az_kv.set_secret("RefreshToken", self.refresh_token)86                    self.az_kv.set_secret("IsValid", "YES")87                    self.az_kv.set_secret("AuthRetryCounter", 1)88                else:89                    if response_content['error'] and response_content['error'] == 'invalid_client':90                        # Set flag to avoid unwanted retries in case of invalid key/client91                        self.retry_counter = int(self.retry_counter) + 192                        self.az_kv.set_secret("AuthRetryCounter", self.retry_counter)93                        if int(self.retry_counter) >= 10:94                            self.az_kv.set_secret("IsValid", "NO")95                    logging.error("Your Lookout application key has expired. " +96                                "Please get a new key and set up this connector app again.\n" +97                                "Go to https://mtp.lookout.com and generate a new key by " +98                                "navigating to System => Application Keys.")99                    logging.error("Exiting...")100                    sys.exit(1)101        except requests.exceptions.ProxyError as e:102            logging.error("Cannot connect to proxy. Remote end closed connection without response")            103        except requests.exceptions.RequestException as e:104            logging.error(e)105            106    def get_oauth(self):107        '''108        Retrieve OAuth tokens from Lookout API109        - Returns the access_token and the refresh_token110        - If the access token is already stored, returns the111          variables stored locally112        '''113        token_json = {}114        if self.access_token:115            logging.info("The access token has been found locally")116            return self.access_token, self.refresh_token117        logging.info("Could not find an access token, getting one now")118        119        try:120            response = requests.post(self.api_domain + "/oauth/token",121                                    data="grant_type=client_credentials",122                                    headers=self.header(self.api_key))123            try:124                token_json = json.loads(response.text)                125            except (AttributeError, ValueError) as e:126                logging.info("Exception when requesting new access token: " + str(e))127                logging.info("Refreshing access token...")128                self.refresh_oauth()129            if 'access_token' in token_json and 'error' not in token_json:130                logging.info("Storing creds in Azure Vault")131                self.access_token = token_json['access_token']132                self.refresh_token = token_json['refresh_token']133                self.az_kv.set_secret("AccessToken", self.access_token)134                self.az_kv.set_secret("RefreshToken", self.refresh_token)135                self.az_kv.set_secret("IsValid", "YES")136                self.az_kv.set_secret("AuthRetryCounter", 1)137                logging.info("Got authenticated")138                return self.access_token, self.refresh_token139            else: 140                if token_json['error'] and token_json['error'] == 'invalid_client':141                    # Set flag to avoid unwanted retries in case of invalid key/client142                    self.retry_counter = int(self.retry_counter) + 1143                    self.az_kv.set_secret("AuthRetryCounter", self.retry_counter)144                    if int(self.retry_counter) >= 10:145                        self.az_kv.set_secret("IsValid", "NO")146                logging.info("Auth API retry count :  " + str(self.retry_counter))147                logging.info("Error in oauth")148                logging.info(str(token_json))149                return False150                151        except requests.exceptions.ProxyError as e:152            logging.error("Cannot connect to proxy. Remote end closed connection without response")            153        except requests.exceptions.RequestException as e:154            logging.error(e)155            156    def get_events(self):157        '''158        Method to collect events from Metis API159        - Gets access token and stream position from Azure Vault160        - Requests events (retries if error HTTP code)161        - Collect events lists from Metis API, returns full list of events162        '''163        164        events = []165        retry_count = 0166        more_events = True167        168        if self.is_valid == "NO" or int(self.retry_counter) >= 10:169            logging.info("Please check API key, Auth API responds with Invalid Client error after 10 retries")170            return events171        if not self.access_token:172            self.get_oauth()173        if self.access_token:174            # Added cycle count to avoid long data polling175            cycle_count = 0176            while more_events and retry_count < 10 and cycle_count < 10 :177                logging.info("Fetching Events from Position {}".format(self.stream_position))178                try:179                    response = requests.get(self.api_domain + "/events?eventType=DEVICE,THREAT,AUDIT",180                                            headers=self.header(self.access_token),181                                            params={"streamPosition": self.stream_position})182                    if response.status_code == 400 and response.json()['errorCode'] in self.stale_token_errors:183                        self.refresh_oauth()184                        continue185                    elif response.status_code != requests.codes.ok:186                        logging.info("Received error code {}, trying again to get events".format(response.status_code))187                        retry_count = retry_count + 1188                        continue189                except requests.exceptions.ProxyError as e:190                    logging.error("Cannot connect to proxy. Remote end closed connection without response")                191                except requests.exceptions.RequestException as e:192                    logging.error(e)                193                    194                cycle_count = cycle_count + 1195                events = events + response.json()['events']196                self.stream_position = response.json()['streamPosition']197                198                #update stream position in Azure Vault199                self.az_kv.set_secret("StreamPosition", self.stream_position)200                more_events = response.json()['moreEvents']201                logging.info("Fetched Event Count {}".format(len(events)))202                logging.info("More Events to Fetch : {}".format(more_events))203            if retry_count >= 10:204                logging.error("Too many failed attempts to retrieve events, shutting down.")205                sys.exit(2)...

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.