TestMu 2025 - The Largest Virtual Software Testing Conference is Back | 50,000+ Attendees

How to use set_secret method in localstack

Best Python code snippet using localstack_python

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)...

Blogs

Check out the latest blogs from LambdaTest on this topic:

What is coaching leadership

Coaching is a term that is now being mentioned a lot more in the leadership space. Having grown successful teams I thought that I was well acquainted with this subject.

Options for Manual Test Case Development & Management

The purpose of developing test cases is to ensure the application functions as expected for the customer. Test cases provide basic application documentation for every function, feature, and integrated connection. Test case development often detects defects in the design or missing requirements early in the development process. Additionally, well-written test cases provide internal documentation for all application processing. Test case development is an important part of determining software quality and keeping defects away from customers.

How To Automate Desktops Apps Using WinAppDriver

QA’s and Unit Testing – Can QA Create Effective Unit Tests

Unit testing is typically software testing within the developer domain. As the QA role expands in DevOps, QAOps, DesignOps, or within an Agile team, QA testers often find themselves creating unit tests. QA testers may create unit tests within the code using a specified unit testing tool, or independently using a variety of methods.

Project Goal Prioritization in Context of Your Organization’s Strategic Objectives

One of the most important skills for leaders to have is the ability to prioritize. To understand how we can organize all of the tasks that must be completed in order to complete a project, we must first understand the business we are in, particularly the project goals. There might be several project drivers that stimulate project execution and motivate a company to allocate the appropriate funding.

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.