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

How to use detect_stack_resource_drift method in localstack

Best Python code snippet using localstack_python

client.py

Source:client.py

...51    def describe_stacks(self, StackName: str = None, NextToken: str = None) -> Dict:52        pass53    def detect_stack_drift(self, StackName: str, LogicalResourceIds: List = None) -> Dict:54        pass55    def detect_stack_resource_drift(self, StackName: str, LogicalResourceId: str) -> Dict:56        pass57    def estimate_template_cost(self, TemplateBody: str = None, TemplateURL: str = None, Parameters: List = None) -> Dict:58        pass59    def execute_change_set(self, ChangeSetName: str, StackName: str = None, ClientRequestToken: str = None) -> Dict:60        pass61    def generate_presigned_url(self, ClientMethod: str = None, Params: Dict = None, ExpiresIn: int = None, HttpMethod: str = None):62        pass63    def get_paginator(self, operation_name: str = None) -> Paginator:64        pass65    def get_stack_policy(self, StackName: str) -> Dict:66        pass67    def get_template(self, StackName: str = None, ChangeSetName: str = None, TemplateStage: str = None) -> Dict:68        pass69    def get_template_summary(self, TemplateBody: str = None, TemplateURL: str = None, StackName: str = None, StackSetName: str = None) -> Dict:...

test_drift_stack.py

Source:test_drift_stack.py

1import os2import io3import sys4import unittest5from unittest.mock import patch6from fzfaws.cloudformation.drift_stack import drift_stack, wait_drift_result7from fzfaws.cloudformation import Cloudformation8from fzfaws.utils import FileLoader9class TestCloudformationDriftStack(unittest.TestCase):10    def setUp(self):11        self.capturedOutput = io.StringIO()12        sys.stdout = self.capturedOutput13        self.cloudformation_details = {14            "StackId": "arn:aws:cloudformation:ap-southeast-2:1111111:stack/dotbare-cicd/0ae5ef60-9651-11ea-b6d0-0223bf2782f0",15            "StackName": "dotbare-cicd",16            "Description": "CodeBuild template for dotbare, webhook trigger from Github only on Master push",17            "RollbackConfiguration": {"RollbackTriggers": []},18            "StackStatus": "UPDATE_COMPLETE",19            "DisableRollback": False,20            "NotificationARNs": [],21            "Capabilities": ["CAPABILITY_NAMED_IAM"],22            "Tags": [],23            "DriftInformation": {"StackDriftStatus": "IN_SYNC"},24        }25    def tearDown(self):26        sys.stdout = sys.__stdout__27    @patch("fzfaws.cloudformation.drift_stack.Cloudformation")28    def test_info(self, MockedCloudformation):29        self.capturedOutput.truncate(0)30        self.capturedOutput.seek(0)31        cloudformation = MockedCloudformation()32        cloudformation.stack_details = self.cloudformation_details33        cloudformation.stack_name = "testing1"34        cloudformation.client.describe_stack_resource_drifts.return_value = {}35        drift_stack(profile="root", region="us-east-1", info=True)36        MockedCloudformation.assert_called_with("root", "us-east-1")37        cloudformation.set_stack.assert_called_once()38        cloudformation.client.describe_stack_resource_drifts.assert_called_once_with(39            StackName="testing1"40        )41        self.assertRegex(42            self.capturedOutput.getvalue(), r'"StackDriftStatus": "IN_SYNC"'43        )44    @patch("fzfaws.cloudformation.drift_stack.wait_drift_result")45    @patch("fzfaws.cloudformation.drift_stack.Cloudformation")46    def test_drift_entire_stack(self, MockedCloudformation, mocked_wait):47        self.capturedOutput.truncate(0)48        self.capturedOutput.seek(0)49        cloudformation = MockedCloudformation()50        cloudformation.stack_name = "testing1"51        cloudformation.stack_details = self.cloudformation_details52        cloudformation.client.detect_stack_drift.return_value = {53            "StackDriftDetectionId": "1111111"54        }55        drift_stack(wait=True)56        MockedCloudformation.assert_called_with(False, False)57        mocked_wait.assert_called_once_with(cloudformation, "1111111")58        cloudformation.client.detect_stack_drift.assert_called_once_with(59            StackName="testing1"60        )61        self.assertRegex(self.capturedOutput.getvalue(), r"Drift detection initiated")62        self.assertRegex(self.capturedOutput.getvalue(), r"DriftDetectionId: 1111111")63        self.assertRegex(64            self.capturedOutput.getvalue(), r'"StackDriftStatus": "IN_SYNC"'65        )66    @patch("fzfaws.cloudformation.drift_stack.Cloudformation")67    def test_single_selected_resource(self, MockedCloudformation):68        self.capturedOutput.truncate(0)69        self.capturedOutput.seek(0)70        cloudformation = MockedCloudformation()71        cloudformation.stack_name = "testing1"72        cloudformation.stack_details = self.cloudformation_details73        cloudformation.get_stack_resources.return_value = ["asg1"]74        cloudformation.client.detect_stack_resource_drift.return_value = {75            "StackResourceDrift": {76                "LogicalResourceId": "asg1",77                "StackResourceDriftStatus": "IN_SYNC",78            }79        }80        drift_stack(select=True)81        MockedCloudformation.assert_called_with(False, False)82        cloudformation.get_stack_resources.assert_called_once()83        cloudformation.client.detect_stack_resource_drift.assert_called_once_with(84            StackName="testing1", LogicalResourceId="asg1"85        )86        self.assertRegex(self.capturedOutput.getvalue(), r"LogicalResourceId: asg1")87        self.assertRegex(88            self.capturedOutput.getvalue(), r"StackResourceDriftStatus: IN_SYNC"89        )90    @patch("fzfaws.cloudformation.drift_stack.wait_drift_result")91    @patch("fzfaws.cloudformation.drift_stack.Cloudformation")92    def test_multiple_selected_resource(self, MockedCloudformation, mocked_wait):93        self.capturedOutput.truncate(0)94        self.capturedOutput.seek(0)95        cloudformation = MockedCloudformation()96        cloudformation.stack_name = "testing1"97        cloudformation.stack_details = self.cloudformation_details98        cloudformation.get_stack_resources.return_value = ["asg1", "sg1"]99        cloudformation.client.detect_stack_drift.return_value = {100            "StackDriftDetectionId": "1111111"101        }102        drift_stack(profile=True, select=True)103        MockedCloudformation.assert_called_with(True, False)104        cloudformation.get_stack_resources.assert_called_once()105        cloudformation.set_stack.assert_called_once()106        cloudformation.client.detect_stack_drift.assert_called_once_with(107            StackName="testing1", LogicalResourceIds=["asg1", "sg1"]108        )109        mocked_wait.assert_not_called()110        self.assertRegex(self.capturedOutput.getvalue(), r"Drift detection initiated")111        self.assertRegex(self.capturedOutput.getvalue(), r"DriftDetectionId: 1111111")112        drift_stack(profile=True, select=True, wait=True)113        mocked_wait.assert_called_once_with(cloudformation, "1111111")114    @patch("fzfaws.cloudformation.drift_stack.Cloudformation")115    def test_wait(self, MockedCloudformation):116        self.capturedOutput.truncate(0)117        self.capturedOutput.seek(0)118        cloudformation = MockedCloudformation()119        cloudformation._get_waiter_config.return_value = (1, 120)120        cloudformation.client.describe_stack_drift_detection_status.return_value = {121            "DetectionStatus": "DETECTION_COMPLETE",122            "StackResourceDriftStatus": "IN_SYNC",123            "DriftedStackResourceCount": 0,124        }125        wait_drift_result(cloudformation, "1111111")126        cloudformation.client.describe_stack_drift_detection_status.assert_called_once_with(127            StackDriftDetectionId="1111111"128        )129        self.assertRegex(self.capturedOutput.getvalue(), r"StackDriftStatus: IN_SYNC")130        self.assertRegex(131            self.capturedOutput.getvalue(), r"DriftedStackResourceCount: 0"...

drift_stack.py

Source:drift_stack.py

...52    else:53        logical_id_list: List[str] = cloudformation.get_stack_resources()54        if len(logical_id_list) == 1:55            # get individual resource drift status56            response = cloudformation.client.detect_stack_resource_drift(57                StackName=cloudformation.stack_name,58                LogicalResourceId=logical_id_list[0],59            )60            print(json.dumps(response["StackResourceDrift"], indent=4, default=str))61            print(80 * "-")62            print(63                "LogicalResourceId: %s"64                % response["StackResourceDrift"]["LogicalResourceId"]65            )66            print(67                "StackResourceDriftStatus: %s"68                % response["StackResourceDrift"]["StackResourceDriftStatus"]69            )70        else:...

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.