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

How to use init_resource_status method in localstack

Best Python code snippet using localstack_python

template_deployer.py

Source:template_deployer.py

...1105        for resource_id, resource in resources.items():1106            add_default_resource_props(1107                resource, self.stack_name, resource_id=resource_id, existing_resources=resources1108            )1109    def init_resource_status(self, resources=None, stack=None, action="CREATE"):1110        resources = resources or self.resources1111        stack = stack or self.stack1112        for resource_id, resource in resources.items():1113            stack.set_resource_status(resource_id, "%s_IN_PROGRESS" % action)1114    def update_resource_details(self, resource_id, result, stack=None, action="CREATE"):1115        stack = stack or self.stack1116        # update resource state1117        update_resource_details(stack, resource_id, result, action)1118        # update physical resource id1119        resource = stack.resources[resource_id]1120        physical_id = resource.get("PhysicalResourceId")1121        physical_id = physical_id or determine_resource_physical_id(resource_id, stack=stack)1122        if not resource.get("PhysicalResourceId") or action == "UPDATE":1123            if physical_id:1124                resource["PhysicalResourceId"] = physical_id1125        # set resource status1126        stack.set_resource_status(resource_id, "%s_COMPLETE" % action, physical_res_id=physical_id)1127        return physical_id1128    def get_change_config(self, action, resource, change_set_id=None):1129        return {1130            "Type": "Resource",1131            "ResourceChange": {1132                "Action": action,1133                "LogicalResourceId": resource.get("LogicalResourceId"),1134                "PhysicalResourceId": resource.get("PhysicalResourceId"),1135                "ResourceType": resource.get("Type"),1136                "Replacement": "False",1137                "ChangeSetId": change_set_id,1138            },1139        }1140    def resource_config_differs(self, resource_new):1141        """Return whether the given resource properties differ from the existing config (for stack updates)."""1142        resource_id = resource_new["LogicalResourceId"]1143        resource_old = self.resources[resource_id]1144        props_old = resource_old["Properties"]1145        props_new = resource_new["Properties"]1146        ignored_keys = ["LogicalResourceId", "PhysicalResourceId"]1147        old_keys = set(props_old.keys()) - set(ignored_keys)1148        new_keys = set(props_new.keys()) - set(ignored_keys)1149        if old_keys != new_keys:1150            return True1151        for key in old_keys:1152            if props_old[key] != props_new[key]:1153                return True1154        old_status = self.stack.resource_states.get(resource_id) or {}1155        previous_state = (1156            old_status.get("PreviousResourceStatus") or old_status.get("ResourceStatus") or ""1157        )1158        if old_status and "DELETE" in previous_state:1159            return True1160    def merge_properties(self, resource_id, old_stack, new_stack):1161        old_resources = old_stack.template["Resources"]1162        new_resources = new_stack.template["Resources"]1163        new_resource = new_resources[resource_id]1164        old_resource = old_resources[resource_id] = old_resources.get(resource_id) or {}1165        for key, value in new_resource.items():1166            if key == "Properties":1167                continue1168            old_resource[key] = old_resource.get(key, value)1169        old_res_props = old_resource["Properties"] = old_resource.get("Properties", {})1170        for key, value in new_resource["Properties"].items():1171            old_res_props[key] = value1172        # overwrite original template entirely1173        old_stack.template_original["Resources"][resource_id] = new_stack.template_original[1174            "Resources"1175        ][resource_id]1176    def resolve_param(1177        self, logical_id: str, param_type: str, default_value: Optional[str] = None1178    ) -> Optional[str]:1179        if param_type == "AWS::SSM::Parameter::Value<String>":1180            ssm_client = aws_stack.connect_to_service("ssm")1181            param = ssm_client.get_parameter(Name=default_value)1182            return param["Parameter"]["Value"]1183        return None1184    def apply_parameter_changes(self, old_stack, new_stack) -> None:1185        parameters = {1186            p["ParameterKey"]: p1187            for p in old_stack.metadata["Parameters"]  # go through current parameter values1188        }1189        for logical_id, value in new_stack.template["Parameters"].items():1190            default = value.get("Default")1191            provided_param_value = parameters.get(logical_id)1192            param = {1193                "ParameterKey": logical_id,1194                "ParameterValue": provided_param_value if default is None else default,1195            }1196            if default is not None:1197                resolved_value = self.resolve_param(logical_id, value.get("Type"), default)1198                if resolved_value is not None:1199                    param["ResolvedValue"] = resolved_value1200            parameters[logical_id] = param1201        parameters.update({p["ParameterKey"]: p for p in new_stack.metadata["Parameters"]})1202        for change_set in new_stack.change_sets:1203            parameters.update({p["ParameterKey"]: p for p in change_set.metadata["Parameters"]})1204        # TODO: unclear/undocumented behavior in implicitly updating old_stack parameter here1205        old_stack.metadata["Parameters"] = [v for v in parameters.values() if v]1206    # TODO: fix circular import with cloudformation_api.py when importing Stack here1207    def construct_changes(1208        self,1209        existing_stack,1210        new_stack,1211        initialize=False,1212        change_set_id=None,1213        append_to_changeset=False,1214    ):1215        from localstack.services.cloudformation.cloudformation_api import StackChangeSet1216        old_resources = existing_stack.template["Resources"]1217        new_resources = new_stack.template["Resources"]1218        deletes = [val for key, val in old_resources.items() if key not in new_resources]1219        adds = [val for key, val in new_resources.items() if initialize or key not in old_resources]1220        modifies = [val for key, val in new_resources.items() if key in old_resources]1221        changes = []1222        for action, items in (("Remove", deletes), ("Add", adds), ("Modify", modifies)):1223            for item in items:1224                item["Properties"] = item.get("Properties", {})1225                change = self.get_change_config(action, item, change_set_id=change_set_id)1226                changes.append(change)1227        # append changes to change set1228        if append_to_changeset and isinstance(new_stack, StackChangeSet):1229            new_stack.changes.extend(changes)1230        return changes1231    def apply_changes(1232        self,1233        existing_stack,1234        new_stack,1235        stack_name,1236        change_set_id=None,1237        initialize=False,1238        action=None,1239    ):1240        old_resources = existing_stack.template["Resources"]1241        new_resources = new_stack.template["Resources"]1242        action = action or "CREATE"1243        self.init_resource_status(old_resources, action="UPDATE")1244        # apply parameter changes to existing stack1245        self.apply_parameter_changes(existing_stack, new_stack)1246        # construct changes1247        changes = self.construct_changes(1248            existing_stack,1249            new_stack,1250            initialize=initialize,1251            change_set_id=change_set_id,1252        )1253        # check if we have actual changes in the stack, and prepare properties1254        contains_changes = False1255        for change in changes:1256            res_action = change["ResourceChange"]["Action"]1257            resource = new_resources.get(change["ResourceChange"]["LogicalResourceId"])...

aix_init_function.py

Source:aix_init_function.py

...8    ip_locked=models.aix_ip_status(status_id=2,status='locked')9    ip_locked.save()10    ip_used=models.aix_ip_status(status_id=3,status='used')11    ip_used.save()12def init_resource_status():13    resource_available=models.resource_available(available=True)14    resource_available.save()15    resource_unavailable=models.resource_available(available=False)16    resource_unavailable.save()17def init_nim():18    nim=models.nim_server(ip=setting.NIMserver['ip'],19                          username=setting.NIMserver['username'],20                          password=setting.NIMserver['password'],21                          prompt=setting.NIMserver['prompt'])22    nim.save()23    24def init_vio_server():25    vio_1=models.vio_server(name='vio_1',26                            ip=setting.VIOServer1['ip'],27                            username=setting.VIOServer1['username'],28                            password=setting.VIOServer1['password'],29                            prompt=setting.VIOServer1['prompt'])30    31    vio_2=models.vio_server(name='vio_2',32                            ip=setting.VIOServer2['ip'],33                            username=setting.VIOServer2['username'],34                            password=setting.VIOServer2['password'],35                            prompt=setting.VIOServer2['prompt'])36    vio_1.save()37    vio_2.save()38def init_hmc():39    hmc=models.hmc(ip=setting.HMC['ip'],40                   username=setting.HMC['username'],41                   password=setting.HMC['password'],42                   prompt=setting.HMC['prompt'])43    hmc.save()44def init_aix_version():45    for ver in setting.AIX_Version:46        aix_ver=models.aix_version(version=ver['version'],47                                   spot=ver['spot'],48                                   lpp_source=ver['lpp_source'],49                                   mksysb=ver['mksysb'])50        aix_ver.save()51def init_vioclient_status():52    for status in setting.vioclient_status:53        vio_client_status=models.vioclient_status(status_id=status['status_id'],54                                                  status=status['status'])55        vio_client_status.save()56def init():57    init_ip_status()58    init_resource_status()59    init_vio_server()60    init_nim()61#################################################62#  clear table, reset all resource to available #63#################################################64def reset_ip():65    available_status=models.aix_ip_status.objects.get(status_id=1)66    for ip in models.aix_service_ip.objects.all():67        if ip.status!=available_status:68            ip.status=available_status69            ip.save()70    for ip in models.aix_manage_ip.objects.all():71        if ip.status!=available_status:72            ip.status=available_status...

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.