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

How to use resource_config_differs method in localstack

Best Python code snippet using localstack_python

template_deployer.py

Source:template_deployer.py

...1157                "Replacement": "False",1158                "ChangeSetId": change_set_id,1159            },1160        }1161    def resource_config_differs(self, resource_new):1162        """Return whether the given resource properties differ from the existing config (for stack updates)."""1163        resource_id = resource_new["LogicalResourceId"]1164        resource_old = self.resources[resource_id]1165        props_old = resource_old["Properties"]1166        props_new = resource_new["Properties"]1167        ignored_keys = ["LogicalResourceId", "PhysicalResourceId"]1168        old_keys = set(props_old.keys()) - set(ignored_keys)1169        new_keys = set(props_new.keys()) - set(ignored_keys)1170        if old_keys != new_keys:1171            return True1172        for key in old_keys:1173            if props_old[key] != props_new[key]:1174                return True1175        old_status = self.stack.resource_states.get(resource_id) or {}1176        previous_state = (1177            old_status.get("PreviousResourceStatus") or old_status.get("ResourceStatus") or ""1178        )1179        if old_status and "DELETE" in previous_state:1180            return True1181    def merge_properties(self, resource_id, old_stack, new_stack):1182        old_resources = old_stack.template["Resources"]1183        new_resources = new_stack.template["Resources"]1184        new_resource = new_resources[resource_id]1185        old_resource = old_resources[resource_id] = old_resources.get(resource_id) or {}1186        for key, value in new_resource.items():1187            if key == "Properties":1188                continue1189            old_resource[key] = old_resource.get(key, value)1190        old_res_props = old_resource["Properties"] = old_resource.get("Properties", {})1191        for key, value in new_resource["Properties"].items():1192            old_res_props[key] = value1193        # overwrite original template entirely1194        old_stack.template_original["Resources"][resource_id] = new_stack.template_original[1195            "Resources"1196        ][resource_id]1197    def resolve_param(1198        self, logical_id: str, param_type: str, default_value: Optional[str] = None1199    ) -> Optional[str]:1200        if param_type == "AWS::SSM::Parameter::Value<String>":1201            ssm_client = aws_stack.connect_to_service("ssm")1202            param = ssm_client.get_parameter(Name=default_value)1203            return param["Parameter"]["Value"]1204        return None1205    def apply_parameter_changes(self, old_stack, new_stack) -> None:1206        parameters = {1207            p["ParameterKey"]: p1208            for p in old_stack.metadata["Parameters"]  # go through current parameter values1209        }1210        for logical_id, value in new_stack.template["Parameters"].items():1211            default = value.get("Default")1212            provided_param_value = parameters.get(logical_id)1213            param = {1214                "ParameterKey": logical_id,1215                "ParameterValue": provided_param_value if default is None else default,1216            }1217            if default is not None:1218                resolved_value = self.resolve_param(logical_id, value.get("Type"), default)1219                if resolved_value is not None:1220                    param["ResolvedValue"] = resolved_value1221            parameters[logical_id] = param1222        def _update_params(params_list: List[Dict]):1223            for param in params_list:1224                # make sure we preserve parameter values if UsePreviousValue=true1225                if not param.get("UsePreviousValue"):1226                    parameters.update({param["ParameterKey"]: param})1227        _update_params(new_stack.metadata["Parameters"])1228        for change_set in new_stack.change_sets:1229            _update_params(change_set.metadata["Parameters"])1230        # TODO: unclear/undocumented behavior in implicitly updating old_stack parameter here1231        # Note: Indeed it seems that parameters from Change Sets are applied to a stack1232        #   itself, and are preserved even after a change set has been deleted. However,1233        #   a proper implementation would distinguish between (1) Change Sets and (2) Change1234        #   Set Executions - the former are only a template for the changes to be applied,1235        #   whereas the latter actually perform changes (including parameter updates).1236        #   Also, (1) can be deleted, and (2) can only be rolled back (in case of errors).1237        #   Once we have the distinction between (1) and (2) in place, this logic (updating1238        #   the parameters of the stack itself) will become obsolete, then the parameter1239        #   values can be determined by replaying the values of the sequence of (immutable)1240        #   Change Set Executions.1241        old_stack.metadata["Parameters"] = [v for v in parameters.values() if v]1242    # TODO: fix circular import with cloudformation_api.py when importing Stack here1243    def construct_changes(1244        self,1245        existing_stack,1246        new_stack,1247        initialize=False,1248        change_set_id=None,1249        append_to_changeset=False,1250        filter_unchanged_resources=False,1251    ):1252        from localstack.services.cloudformation.provider import StackChangeSet1253        old_resources = existing_stack.template["Resources"]1254        new_resources = new_stack.template["Resources"]1255        deletes = [val for key, val in old_resources.items() if key not in new_resources]1256        adds = [val for key, val in new_resources.items() if initialize or key not in old_resources]1257        modifies = [val for key, val in new_resources.items() if key in old_resources]1258        changes = []1259        for action, items in (("Remove", deletes), ("Add", adds), ("Modify", modifies)):1260            for item in items:1261                item["Properties"] = item.get("Properties", {})1262                if (1263                    not filter_unchanged_resources1264                    or action != "Modify"1265                    or self.resource_config_differs(item)1266                ):1267                    change = self.get_change_config(action, item, change_set_id=change_set_id)1268                    changes.append(change)1269        # append changes to change set1270        if append_to_changeset and isinstance(new_stack, StackChangeSet):1271            new_stack.changes.extend(changes)1272        return changes1273    def apply_changes(1274        self,1275        existing_stack,1276        new_stack,1277        stack_name,1278        change_set_id=None,1279        initialize=False,1280        action=None,1281    ):1282        old_resources = existing_stack.template["Resources"]1283        new_resources = new_stack.template["Resources"]1284        action = action or "CREATE"1285        self.init_resource_status(old_resources, action="UPDATE")1286        # apply parameter changes to existing stack1287        self.apply_parameter_changes(existing_stack, new_stack)1288        # construct changes1289        changes = self.construct_changes(1290            existing_stack,1291            new_stack,1292            initialize=initialize,1293            change_set_id=change_set_id,1294        )1295        # check if we have actual changes in the stack, and prepare properties1296        contains_changes = False1297        for change in changes:1298            res_action = change["ResourceChange"]["Action"]1299            resource = new_resources.get(change["ResourceChange"]["LogicalResourceId"])1300            if res_action != "Modify" or self.resource_config_differs(resource):1301                contains_changes = True1302            if res_action in ["Modify", "Add"]:1303                self.merge_properties(resource["LogicalResourceId"], existing_stack, new_stack)1304        if not contains_changes:1305            raise NoStackUpdates("No updates are to be performed.")1306        # merge stack outputs and conditions1307        existing_stack.outputs.update(new_stack.outputs)1308        existing_stack.conditions.update(new_stack.conditions)1309        # start deployment loop1310        return self.apply_changes_in_loop(1311            changes, existing_stack, stack_name, action=action, new_stack=new_stack1312        )1313    def apply_changes_in_loop(self, changes, stack, stack_name, action=None, new_stack=None):1314        from localstack.services.cloudformation.provider import StackChangeSet...

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.