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

How to use delete_stack_instances method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

...164                raise OperationFailedError(f"Operation {operation_name} failed to complete! "165                                           f"Check 'cloudformation' StackSet '{self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME}' for more information")166            logger.info(f"Going to sleep for {self.STACK_OPERATION_WAIT_SLEEP} seconds")167            time.sleep(self.STACK_OPERATION_WAIT_SLEEP)168    def delete_stack_instances(self) -> None:169        """170        Delete the stack instance.171        :return: Response of the aws deletion command.172        """173        logger.info(f"Deleting stack instance for StackSet: {self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME}, "174                    f"region: {self.region_name}, AccountId: {self.customer_account_id}")175        try:176            response = self.cloudformation_client.delete_stack_instances(177                StackSetName=self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME,178                Regions=[self.region_name], RetainStacks=False,179                Accounts=[self.customer_account_id])180        except Exception as e:181            logger.warning(f"Stack instance deletion failed with error: {repr(e)}. "182                           f"Possibly this stack instance already exists.")183            return184        self.wait_for_stack_operation(response.get("OperationId"), "delete_stack_instances")185    def delete_stack_set(self) -> Dict:186        """187        Delete the stack set188        :return: Response of the delete command.189        """190        logger.info(f"Deleting StackSet: {self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME}")191        return self.cloudformation_client.delete_stack_set(StackSetName=self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME)192    def register_to_dome9(self) -> Dict:193        """194        Make API request to dome9 to lunch the onboarding.195        :return: Response from Dome9 API196        """197        dome9_api_keys = self.get_secret()198        logger.info("Initiating Dome9 Client")199        dome9_client = Client(access_id=dome9_api_keys['AccessId'], secret_key=dome9_api_keys['Secret'],200                              base_url=self.dome9_region_url)201        logger.info("Initiating Dome9 CloudAccountCredentials")202        credentials = CloudAccountCredentials(arn=self.customer_account_new_role_arn,203                                              secret=self.customer_account_external_id)204        logger.info("Initiating Dome9 CloudAccount")205        payload = CloudAccount(name=self.customer_account_name, credentials=credentials)206        logger.info("Sending API request to Dome9")207        response = dome9_client.aws_cloud_account.create(body=payload)208        logger.info(f"Received reply from dome9 API: {response}")209        return response210    def create_stack_set_flow(self) -> None:211        """212        Handle stack set creation213        :return:214        """215        try:216            self.create_stack_set()217            logger.info(f"StackSet '{self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME}' created")218        except Exception as e:219            repr_e_lower = repr(e).lower()220            if "already" in repr_e_lower and "exists" in repr_e_lower:221                logger.warning(222                    f"StackSet '{self.MASTER_ACCOUNT_PERMISSIONS_STACK_SET_NAME}' already exists. Skipping this step.")223                self.delete_stack_instances()224            else:225                raise226    def execute_onboarding_flow(self) -> Dict:227        """228        Execute the whole onboarding process.229        :return:230        """231        self.create_stack_set_flow()232        self.create_stack_instances()233        try:234            register_result = self.register_to_dome9()235            return {"Status": f"Onboarding finished. Received reply from Dome9: {str(register_result)}"}236        except Exception as e:237            logger.error(f"An error '{repr(e)}' occurred in the onboarding process. Aborting the flow. "238                         f"Going to delete the StackInstance")239            self.delete_stack_instances()240            raise241def lambda_handler(event: Dict, context: Dict) -> Dict:242    """243    Makes the magic happen.244    Creates Dome9 protected account:245    1) Creates needed infrastructure on the New Account's side.246    2) Triggers automatic onboardig.247    :param event: AWS lambda event248    :param context: AWS lambda context249    :return: Status250    """251    logger.info(f"Triggered by ControlTower event. Event details: {event}")252    event = event["detail"]253    event_state = event["serviceEventDetails"]["createManagedAccountStatus"]["state"]...

stack_sets.py

Source:stack_sets.py

...112        self.update_stack() if self.stack_status == 'ACTIVE' else self.create_stack()113    def delete_stack(self):114        # Delete existing stack instances115        if len(self.current_accounts) > 0:116            self.stack_set_waiter(self.delete_stack_instances(list(self.current_accounts), list(self.current_regions)), "Deletion")117        # Delete stack set118        args = {'StackSetName': self.stack_name}119        self.client.delete_stack_set(**args)120        logger.info("Delete complete!")121    def stack_set_waiter(self, operation_id, verb="Update"):122        logger.info("Stack Set " + verb + " Started")123        if self.print_events:124            args = {125                "StackSetName": self.stack_name,126                "OperationId": operation_id127            }128            operation = self.client.describe_stack_set_operation(**args)129            time.sleep(5)130            while operation['StackSetOperation']['Status'] not in ['SUCCEEDED', 'FAILED', 'STOPPED']:131                time.sleep(5)132                operation = self.client.describe_stack_set_operation(**args)133            # Print result134            results = self.client.list_stack_set_operation_results(**args)135            headers = ['Account', 'Region', 'Status', 'Reason']136            table = [[x['Account'], x['Region'], x['Status'], x.get('StatusReason', '')] for x in results['Summaries']]137        138            print(tabulate.tabulate(table, headers, tablefmt='simple'))139    def update_stack(self):140        args = {141            "Capabilities": [142                'CAPABILITY_IAM',143                'CAPABILITY_NAMED_IAM',144                'CAPABILITY_AUTO_EXPAND'145            ],146            "Parameters": self.config.build_params(self.session, self.stack, self.release, self.params, self.template_file),147            'StackSetName': self.stack_name,148            "Tags": self.construct_tags(),149        }150        args.update({'AdministrationRoleARN': self.administration_role} if self.administration_role else {})151        args.update({'ExecutionRoleName': self.execution_role} if self.execution_role else {})152        args.update({'TemplateBody': self.template_body} if self.template_body else {"TemplateURL": self.template_url})153        if self.template_body:154            logger.info("Using local template due to null template bucket")155        # Generate API calls based upon what is currently deployed156        api_calls = self.generate_instance_calls()157        # Run update on existing stacks158        result = self.client.update_stack_set(**args)159        operation_id = result.get('OperationId')160        self.stack_set_waiter(operation_id)161        # Delete or create as needed162        for call in api_calls:163            if call['type'] == "create":164                self.stack_set_waiter(self.create_stack_instances(call['accounts'], call['regions']))165            if call['type'] == "delete":166                self.stack_set_waiter(self.delete_stack_instances(call['accounts'], call['regions']))167    def generate_instance_calls(self):168        api_calls = []169        # Create x-y graph where x is accounts and y is regions170        graph = []171        for account in range(0, len(self.accounts)):172            graph.append([])173            for region in range(0, len(self.regions)):174                graph[account].append([])175                if self.regions[region] not in self.current_regions or self.accounts[account] not in self.current_accounts:176                    graph[account][region] = "create"177                    logger.debug(self.accounts[account] + ", " + self.regions[region] + " marked for creation")178                elif (self.regions[region] not in self.regions and self.regions[region] in self.current_regions) or (self.accounts[account] not in self.accounts and self.accounts[account] in self.current_accounts):179                    graph[account][region] = "delete"180                    logger.debug(self.accounts[account] + ", " + self.regions[region] + " marked for deletion")181                else:182                    graph[account][region] = "update"183                    logger.debug(self.accounts[account] + ", " + self.regions[region] + " marked for update")184        for call_type in ["create", "update", "delete"]:185            type_calls = []186            # Get all account based calls187            for account in range(0, len(self.accounts)):188                api_call = {189                    'type': call_type,190                    'accounts': [self.accounts[account]],191                    'regions': []192                }193                for region in range(0, len(self.regions)):194                    if graph[account][region] == call_type:195                        api_call['regions'].append(self.regions[region])196                        graph[account][region] = "done"197                if len(api_call['regions']) > 0:198                    type_calls.append(api_call)199            200            # Get all region based calls201            for region in range(0, len(self.regions)):202                api_call = {203                    'type': call_type,204                    'accounts': [],205                    'regions': [self.regions[region]]206                }207                for account in range(0, len(self.accounts)):208                    if graph[account][region] == call_type:209                        api_call['accounts'].append(self.accounts[account])210                        graph[account][region] = "done"211                if len(api_call['accounts']) > 0:212                    type_calls.append(api_call)213            # Merge account based calls214            for call in type_calls:215                for other_call in type_calls:216                    if call != other_call:217                        if call['regions'] == other_call['regions']:218                            call['accounts'] = call['accounts'] + other_call['accounts']219                            type_calls.remove(other_call)220            221            # Merge region based calls222            for call in type_calls:223                for other_call in type_calls:224                    if call != other_call:225                        if call['accounts'] == other_call['accounts']:226                            call['regions'] = call['regions'] + other_call['regions']227                            type_calls.remove(other_call)228            api_calls = api_calls + type_calls229        return api_calls230    def create_stack_instances(self, accounts, regions):231        logger.info("Creating " + str(len(accounts) * len(regions)) + " stack instances...")232        result = self.client.create_stack_instances(StackSetName=self.stack_name, Accounts=accounts, Regions=regions)233        return result['OperationId']234    def delete_stack_instances(self, accounts, regions):235        logger.info("Deleting " + str(len(accounts) * len(regions)) + " stack instances...")236        result = self.client.delete_stack_instances(StackSetName=self.stack_name, Accounts=accounts, Regions=regions, RetainStacks=False)237        return result['OperationId']238    def construct_tags(self): 239        tags = self.config.get_config_att('tags')240        if tags:241            tags = [ { 'Key': key, 'Value': value } for key, value in tags.items() ] 242            if len(tags) > 47:243                raise ValueError('Resources tag limit is 50, you have provided more than 47 tags. Please limit your tagging, save room for name and deployer tags.')244        else:245            tags = []246        tags.append({'Key': 'deployer:stack', 'Value': self.stack})247        tags.append({'Key': 'deployer:caller', 'Value': self.identity_arn})248        tags.append({'Key': 'deployer:git:commit', 'Value': self.commit})249        tags.append({'Key': 'deployer:git:origin', 'Value': self.origin})250        tags.append({'Key': 'deployer:config', 'Value': self.config.file_name.replace('\\', '/')})...

cloudformation.py

Source:cloudformation.py

...68                "MaxConcurrentPercentage": 10069            }70        )["OperationId"]71        return operation_id72    def delete_stack_instances(self, stack_set_name, accounts, regions, region_order=None):73        if region_order is None:74            region_order = []75        operation_id = self.client.delete_stack_instances(76            StackSetName=stack_set_name,77            Accounts=accounts,78            Regions=regions,79            OperationPreferences={80                "RegionOrder": region_order,81                "FailureTolerancePercentage": 0,82                "MaxConcurrentPercentage": 10083            },84            RetainStacks=False85        )["OperationId"]86        return operation_id87    def describe_stack_instance(self, stack_set_name, stack_instance_account, stack_instance_region):88        stack_instance_info = self.client.describe_stack_instance(89            StackSetName=stack_set_name,...

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.