Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use describe_stack_resource_drifts method in localstack

Best Python code snippet using localstack_python

test_detect_drift.py

Source:test_detect_drift.py

1import unittest2import sys3import json4sys.path.insert(0, './drift_detector')5from drift_detector.drift_detector import detect_drift6from unittest.mock import MagicMock7class MockCFClient: pass8mock_cf_client = MockCFClient()9class TestDetectDrift(unittest.TestCase):10    def setUp(self):11        mock_cf_client.detect_stack_drift = MagicMock(return_value={12            'StackDriftDetectionId': 4213        })14        mock_cf_client.describe_stack_drift_detection_status = MagicMock(return_value={15            'StackId': 'stack_id',16            'DetectionStatus': 'DETECTION_COMPLETE'17        })18        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={19            'StackResourceDrifts': [20                {21                    'StackResourceDriftStatus': 'MODIFIED',22                    'PhysicalResourceId': 'physical_resource_id',23                    'ResourceType': 'resource_type'24                }25            ]26        })27    def test_detect_drift(self):28        """29        Test that drift is correctly detected30        """31        mock_stacks = [32            {33                'StackName': 'stack_name',34                'StackId': 'stack_id'35            }36        ]37        stacks, _ = detect_drift(mock_cf_client, json.dumps(mock_stacks, default=str), 5)38        self.assertEqual(stacks, [39            {40                'StackName': 'stack_name',41                'StackId': 'stack_id',42                'drift': [43                    {44                        'PhysicalResourceId': 'physical_resource_id',45                        'ResourceType': 'resource_type',46                        'StackResourceDriftStatus': 'MODIFIED',47                    },48                ],49                'no_of_drifted_resources': 1,50                'no_of_resources': 1,51            }52        ])53    def test_detect_drift_with_no_drift(self):54        """55        Test that no drift is correctly detected56        """57        mock_stacks = [58            {59                'StackName': 'stack_name',60                'StackId': 'stack_id'61            },62        ]63        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={64            'StackResourceDrifts': [65                {66                    'StackResourceDriftStatus': 'IN_SYNC',67                    'PhysicalResourceId': 'physical_resource_id_one',68                    'ResourceType': 'resource_type',69                },70                {71                    'StackResourceDriftStatus': 'IN_SYNC',72                    'PhysicalResourceId': 'physical_resource_id_two',73                    'ResourceType': 'resource_type',74                },75            ],76        })77        stacks, _ = detect_drift(mock_cf_client, json.dumps(mock_stacks, default=str), 5)78        self.assertEqual(stacks, [79            {80                'StackName': 'stack_name',81                'StackId': 'stack_id',82                'drift': [83                    {84                        'PhysicalResourceId': 'physical_resource_id_one',85                        'ResourceType': 'resource_type',86                        'StackResourceDriftStatus': 'IN_SYNC',87                    },88                    {89                        'PhysicalResourceId': 'physical_resource_id_two',90                        'ResourceType': 'resource_type',91                        'StackResourceDriftStatus': 'IN_SYNC',92                    },93                ],94                'no_of_drifted_resources': 0,95                'no_of_resources': 2,96            },97        ])98    def test_detect_drift_sorting(self):99        """100        Test that drift is correctly detected and sorted101        """102        mock_stacks = [103            {104                'StackName': 'stack_name',105                'StackId': 'stack_id'106            },107        ]108        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={109            'StackResourceDrifts': [110                {111                    'StackResourceDriftStatus': 'IN_SYNC',112                    'PhysicalResourceId': 'aaaaaaaa',113                    'ResourceType': 'resource_type',114                },115                {116                    'StackResourceDriftStatus': 'IN_SYNC',117                    'PhysicalResourceId': 'cccccccc',118                    'ResourceType': 'resource_type',119                },120                {121                    'StackResourceDriftStatus': 'IN_SYNC',122                    'PhysicalResourceId': 'bbbbbbbb',123                    'ResourceType': 'resource_type',124                },125            ],126        })127        stacks, _ = detect_drift(mock_cf_client, json.dumps(mock_stacks, default=str), 5)128        self.assertEqual(stacks, [129            {130                'StackName': 'stack_name',131                'StackId': 'stack_id',132                'drift': [133                    {134                        'PhysicalResourceId': 'aaaaaaaa',135                        'ResourceType': 'resource_type',136                        'StackResourceDriftStatus': 'IN_SYNC',137                    },138                    {139                        'PhysicalResourceId': 'bbbbbbbb',140                        'ResourceType': 'resource_type',141                        'StackResourceDriftStatus': 'IN_SYNC',142                    },143                    {144                        'PhysicalResourceId': 'cccccccc',145                        'ResourceType': 'resource_type',146                        'StackResourceDriftStatus': 'IN_SYNC',147                    },148                ],149                'no_of_drifted_resources': 0,150                'no_of_resources': 3,151            },152        ])153    def test_detect_drift_for_resource_with_arn_as_id(self):154        """155        Test that drift is correctly detected for resources with arn id156        """157        mock_stacks = [158            {159                'StackName': 'stack_name',160                'StackId': 'stack_id'161            }162        ]163        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={164            'StackResourceDrifts': [165                {166                    'StackResourceDriftStatus': 'MODIFIED',167                    'PhysicalResourceId': 'arn:aws:lambda:us-east-1:450349639042:function:serverless-housekeeping-gdrive-prod-stuff:4',168                    'ResourceType': 'resource_type',169                },170            ],171        })172        stacks, _ = detect_drift(mock_cf_client, json.dumps(mock_stacks, default=str), 5)173        self.assertEqual(stacks, [174            {175                'StackName': 'stack_name',176                'StackId': 'stack_id',177                'drift': [178                    {179                        'PhysicalResourceId': 'serverless-housekeeping-gdrive-prod-stuff:4',180                        'ResourceType': 'resource_type',181                        'StackResourceDriftStatus': 'MODIFIED',182                    },183                ],184                'no_of_drifted_resources': 1,185                'no_of_resources': 1,186            },187        ])188    def test_detect_drift_with_failed_detection(self):189        """190        Test that failed detection are correctly handled191        """192        mock_stacks = [193            {194                'StackName': 'stack_name',195                'StackId': 'stack_id'196            },197        ]198        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={199            'StackResourceDrifts': [200                {201                    'StackResourceDriftStatus': 'IN_SYNC',202                    'PhysicalResourceId': 'physical_resource_id_one',203                    'ResourceType': 'resource_type',204                },205                {206                    'StackResourceDriftStatus': 'IN_SYNC',207                    'PhysicalResourceId': 'physical_resource_id_two',208                    'ResourceType': 'resource_type',209                },210            ],211        })212        stacks, _ = detect_drift(mock_cf_client, json.dumps(mock_stacks, default=str), 5)213        self.assertEqual(stacks, [214            {215                'StackName': 'stack_name',216                'StackId': 'stack_id',217                'drift': [218                    {219                        'PhysicalResourceId': 'physical_resource_id_one',220                        'ResourceType': 'resource_type',221                        'StackResourceDriftStatus': 'IN_SYNC',222                    },223                    {224                        'PhysicalResourceId': 'physical_resource_id_two',225                        'ResourceType': 'resource_type',226                        'StackResourceDriftStatus': 'IN_SYNC',227                    },228                ],229                'no_of_drifted_resources': 0,230                'no_of_resources': 2,231            },232        ])233if __name__ == '__main__':...

test_append_drift_info.py

Source:test_append_drift_info.py

1import unittest2import sys3sys.path.insert(0, './drift_detector')4from unittest.mock import MagicMock5from drift_detector.drift_detector import append_drift_info6class MockCFClient: pass7mock_cf_client = MockCFClient()8class TestAppendDriftInfo(unittest.TestCase):9    def test_append_drift_info_with_detected_drift(self):10        """11        Test that drift info is correctly appended for detected drift12        """13        mock_stacks = [14            {15                'StackName': 'stack_name',16                'StackId': 'stack_id'17            }18        ]19        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={20            'StackResourceDrifts': [21                {22                    'StackResourceDriftStatus': 'MODIFIED',23                    'PhysicalResourceId': 'physical_resource_id',24                    'ResourceType': 'resource_type'25                }26            ]27        })28        result = append_drift_info(mock_cf_client, mock_stacks)29        self.assertEqual([30            {31                'StackName': 'stack_name',32                'StackId': 'stack_id',33                'drift': [34                    {35                        'PhysicalResourceId': 'physical_resource_id',36                        'StackResourceDriftStatus': 'MODIFIED',37                        'ResourceType': 'resource_type'38                    }39                ],40                'no_of_drifted_resources': 1,41                'no_of_resources': 1,42            }43        ], result)44    def test_append_drift_info_with_no_detected_drift(self):45        """46        Test that no drift is correctly appended47        """48        mock_stacks = [49            {50                'StackName': 'stack_name',51                'StackId': 'stack_id'52            }53        ]54        mock_cf_client.describe_stack_resource_drifts = MagicMock(return_value={55            'StackResourceDrifts': [56                {57                    'StackResourceDriftStatus': 'IN_SYNC',58                    'PhysicalResourceId': 'physical_resource_id_one',59                    'ResourceType': 'resource_type'60                },61                {62                    'StackResourceDriftStatus': 'IN_SYNC',63                    'PhysicalResourceId': 'physical_resource_id_two',64                    'ResourceType': 'resource_type'65                }66            ]67        })68        stacks = append_drift_info(mock_cf_client, mock_stacks)69        self.assertEqual([70            {71                'StackName': 'stack_name',72                'StackId': 'stack_id',73                'drift': [74                    {75                        'PhysicalResourceId': 'physical_resource_id_one',76                        'ResourceType': 'resource_type',77                        'StackResourceDriftStatus': 'IN_SYNC'78                    },79                    {80                        'PhysicalResourceId': 'physical_resource_id_two',81                        'ResourceType': 'resource_type',82                        'StackResourceDriftStatus': 'IN_SYNC'83                    }84                ],85                'no_of_drifted_resources': 0,86                'no_of_resources': 287            }88        ], stacks)89if __name__ == '__main__':...

cli.py

Source:cli.py

...38        print("WARNING: some resources were skipped", file=sys.stderr)39        exit(2)40def get_desired_state(session: boto3.Session, stack: str) -> Iterable[DesiredState]:41    cfn = session.client("cloudformation")42    response = cfn.describe_stack_resource_drifts(43        StackName=stack,44        StackResourceDriftStatusFilters=["MODIFIED"],45    )46    next_token = response.get("NextToken", False)47    yield from get_desired_states(response["StackResourceDrifts"])48    while next_token:49        response = cfn.describe_stack_resource_drifts(50            StackName=stack,51            StackResourceDriftStatusFilters=["MODIFIED"],52            NextToken=next_token,53        )54        next_token = response.get("NextToken", False)55        yield from get_desired_states(response["StackResourceDrifts"])56def set_state(session: boto3.Session, state: "DesiredState") -> None:57    cc = session.client("cloudcontrol")58    patch = json.dumps(create_patch(state.differences))59    try:60        request_token = cc.update_resource(TypeName=state.type, Identifier=state.id, PatchDocument=patch,)[61            "ProgressEvent"62        ]["RequestToken"]63    except cc.exceptions.UnsupportedActionException as e:...

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.