Repository: getmoto/moto
Pull request #9053 (2025-07-10 02:40:44)

ssm: Delete ssm parameter based on resource name
Use the `resource_name` to delete the SSM Parameter when using Cloudformation.

Will fallback to using the Cloudformation JSON if `resource_name` is not specified.

The JSON `cloudformation_json` would not match the resource name if Parameter values are being used in the Cloudformation template to represent the SSM ParameterStore name.

In the `cloudformation_json` case below, the SSM parameter that is passed to `ssm_backend.delete_parameter` is the literal value, not the parsed value.

```
{
  "Properties": {
    "Name": {
      "Fn::Sub": "${Name}"
    },
    "Type": "String",
    "Value": {
      "Ref": "Value"
    }
  },
  "Type": "AWS::SSM::Parameter"
}

```

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/9053?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
Attention: Patch coverage is `66.66667%` with `2 lines` in your changes missing coverage. Please review.
> Project coverage is 92.87%. Comparing base [(`e26ff6d`)](https://app.codecov.io/gh/getmoto/moto/commit/e26ff6de856d4d47ea2f4878c9e8e2636342e639?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`e54c169`)](https://app.codecov.io/gh/getmoto/moto/commit/e54c169fb2c9ee707ed5eac0dc79e60a97ec519e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 2 commits behind head on master.

| [Files with missing lines](https://app.codecov.io/gh/getmoto/moto/pull/9053?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Patch % | Lines |
|---|---|---|
| [moto/ssm/models.py](https://app.codecov.io/gh/getmoto/moto/pull/9053?src=pr&el=tree&filepath=moto%2Fssm%2Fmodels.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9zc20vbW9kZWxzLnB5) | 66.66% | [2 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/9053?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |

:x: Your patch check has failed because the patch coverage (66.66%) is below the target coverage (90.00%). You can increase the patch coverage or adjust the [target](https://docs.codecov.com/docs/commit-status#target) coverage.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #9053      +/-   ##
==========================================
- Coverage   92.87%   92.87%   -0.01%     
==========================================
  Files        1287     1287              
  Lines      113475   113485      +10     
==========================================
+ Hits       105386   105394       +8     
- Misses       8089     8091       +2     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/9053/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/9053/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.62% <0.00%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/9053/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `92.84% <66.66%> (-0.01%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/9053?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>

Patch:
diff --git a/moto/core/base_backend.py b/moto/core/base_backend.py
index 75ca40e6f..095df2057 100644
--- a/moto/core/base_backend.py
+++ b/moto/core/base_backend.py
@@ -38,7 +38,7 @@ class InstanceTrackerMeta(type):
         service = cls.__module__.split(".")[1]
         if name not in model_data[service]:
             model_data[service][name] = cls
-        cls.instances: ClassVar[List["BaseModel"]] = []  # type: ignore
+        cls.instances_tracked: ClassVar[List["BaseModel"]] = []  # type: ignore
         return cls
 
 
diff --git a/moto/core/common_models.py b/moto/core/common_models.py
index 6fcdd35d5..973b2eaf2 100644
--- a/moto/core/common_models.py
+++ b/moto/core/common_models.py
@@ -11,7 +11,7 @@ class BaseModel(metaclass=InstanceTrackerMeta):
         **kwargs: Any,
     ) -> "BaseModel":
         instance = super(BaseModel, cls).__new__(cls)
-        cls.instances.append(instance)  # type: ignore[attr-defined]
+        cls.instances_tracked.append(instance)  # type: ignore[attr-defined]
         return instance
 
 
diff --git a/moto/core/model_instances.py b/moto/core/model_instances.py
index a73f333dd..26ae5ad3d 100644
--- a/moto/core/model_instances.py
+++ b/moto/core/model_instances.py
@@ -12,4 +12,4 @@ def reset_model_data() -> None:
     # Remove all references to the models stored
     for models in model_data.values():
         for model in models.values():
-            model.instances = []  # type: ignore[attr-defined]
+            model.instances_tracked = []  # type: ignore[attr-defined]
diff --git a/moto/moto_api/_internal/responses.py b/moto/moto_api/_internal/responses.py
index 00df8b134..0cc970be6 100644
--- a/moto/moto_api/_internal/responses.py
+++ b/moto/moto_api/_internal/responses.py
@@ -64,7 +64,7 @@ class MotoAPIResponse(BaseResponse):
             for name in sorted(models):
                 model = models[name]
                 results[service][name] = []
-                for instance in model.instances:  # type: ignore[attr-defined]
+                for instance in model.instances_tracked:  # type: ignore[attr-defined]
                     inst_result = {}
                     for attr in dir(instance):
                         if not attr.startswith("_"):
diff --git a/moto/moto_server/threaded_moto_server.py b/moto/moto_server/threaded_moto_server.py
index 73b3f8935..e0e83b17f 100644
--- a/moto/moto_server/threaded_moto_server.py
+++ b/moto/moto_server/threaded_moto_server.py
@@ -36,7 +36,7 @@ class ThreadedMotoServer:
 
     def get_host_and_port(self) -> Tuple[str, int]:
         assert self._server is not None, "Make sure to call start() first"
-        host, port = self._server.server_address
+        host, port = self._server.server_address[:2]
         return (str(host), port)
 
     def stop(self) -> None:
diff --git a/moto/redshift/exceptions.py b/moto/redshift/exceptions.py
index b087457bb..2181e26da 100644
--- a/moto/redshift/exceptions.py
+++ b/moto/redshift/exceptions.py
@@ -1,18 +1,13 @@
-import json
 from typing import List, Optional
 
-from moto.core.exceptions import JsonRESTError
+from moto.core.exceptions import ServiceException
 
 
-class RedshiftClientError(JsonRESTError):
+class RedshiftClientError(ServiceException):
     def __init__(self, code: str, message: str):
-        super().__init__(error_type=code, message=message)
-        self.description = json.dumps(
-            {
-                "Error": {"Code": code, "Message": message, "Type": "Sender"},
-                "RequestId": "6876f774-7273-11e4-85dc-39e55ca848d1",
-            }
-        )
+        super().__init__(message)
+        self.code = code
+        self.message = message
 
 
 class ClusterNotFoundError(RedshiftClientError):
@@ -87,8 +82,6 @@ class InvalidParameterValueError(RedshiftClientError):
 
 
 class ResourceNotFoundFaultError(RedshiftClientError):
-    code = 404
-
     def __init__(
         self,
         resource_type: Optional[str] = None,
diff --git a/moto/redshift/models.py b/moto/redshift/models.py
index 03e542b90..1618f74f2 100644
--- a/moto/redshift/models.py
+++ b/moto/redshift/models.py
@@ -1,4 +1,5 @@
-import copy
+from __future__ import annotations
+
 import datetime
 from collections import OrderedDict
 from typing import Any, Dict, Iterable, List, Optional
@@ -7,7 +8,7 @@ from dateutil.tz import tzutc
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel, CloudFormationModel
-from moto.core.utils import iso_8601_datetime_with_milliseconds
+from moto.core.utils import utcnow
 from moto.ec2 import ec2_backends
 from moto.ec2.models.security_groups import SecurityGroup as EC2SecurityGroup
 from moto.moto_api._internal import mock_random
@@ -70,7 +71,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
 
     def __init__(
         self,
-        redshift_backend: "RedshiftBackend",
+        redshift_backend: RedshiftBackend,
         cluster_identifier: str,
         node_type: str,
         master_username: str,
@@ -86,37 +87,33 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
         automated_snapshot_retention_period: str,
         port: str,
         cluster_version: str,
-        allow_version_upgrade: str,
+        allow_version_upgrade: bool,
         number_of_nodes: str,
-        publicly_accessible: str,
-        encrypted: str,
+        publicly_accessible: bool,
+        encrypted: bool,
         region_name: str,
         tags: Optional[List[Dict[str, str]]] = None,
         iam_roles_arn: Optional[List[str]] = None,
-        enhanced_vpc_routing: Optional[str] = None,
+        enhanced_vpc_routing: Optional[bool] = False,
         restored_from_snapshot: bool = False,
         kms_key_id: Optional[str] = None,
     ):
         super().__init__(redshift_backend.account_id, region_name, tags)
         self.redshift_backend = redshift_backend
         self.cluster_identifier = cluster_identifier
-        self.create_time = iso_8601_datetime_with_milliseconds()
+        self.create_time = utcnow()
         self.status = "available"
         self.node_type = node_type
         self.master_username = master_username
         self.master_user_password = master_user_password
         self.db_name = db_name if db_name else "dev"
         self.vpc_security_group_ids = vpc_security_group_ids
-        self.enhanced_vpc_routing = (
-            enhanced_vpc_routing if enhanced_vpc_routing is not None else False
-        )
+        self.enhanced_vpc_routing = enhanced_vpc_routing
         self.cluster_subnet_group_name = cluster_subnet_group_name
         self.publicly_accessible = publicly_accessible
         self.encrypted = encrypted
 
-        self.allow_version_upgrade = (
-            allow_version_upgrade if allow_version_upgrade is not None else True
-        )
+        self.allow_version_upgrade = allow_version_upgrade
         self.cluster_version = cluster_version if cluster_version else "1.0"
         self.port = int(port) if port else 5439
         self.automated_snapshot_retention_period = (
@@ -180,14 +177,14 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
         return "AWS::Redshift::Cluster"
 
     @classmethod
-    def create_from_cloudformation_json(  # type: ignore[misc]
+    def create_from_cloudformation_json(
         cls,
         resource_name: str,
         cloudformation_json: Any,
         account_id: str,
         region_name: str,
         **kwargs: Any,
-    ) -> "Cluster":
+    ) -> Cluster:
         redshift_backend = redshift_backends[account_id][region_name]
         properties = cloudformation_json["Properties"]
 
@@ -240,11 +237,11 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
         raise UnformattedGetAttTemplateException()
 
     @property
-    def endpoint(self) -> str:
+    def address(self) -> str:
         return f"{self.cluster_identifier}.cg034hpkmmjt.{self.region}.redshift.amazonaws.com"
 
     @property
-    def security_groups(self) -> List["SecurityGroup"]:
+    def security_groups(self) -> List[SecurityGroup]:
         return [
             security_group
             for security_group in self.redshift_backend.describe_cluster_security_groups()
@@ -253,7 +250,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
         ]
 
     @property
-    def vpc_security_groups(self) -> List["EC2SecurityGroup"]:
+    def vpc_security_groups(self) -> List[EC2SecurityGroup]:
         return [
             security_group
             for security_group in self.redshift_backend.ec2_backend.describe_security_groups()
@@ -261,12 +258,11 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
         ]
 
     @property
-    def parameter_groups(self) -> List["ParameterGroup"]:
+    def parameter_groups(self) -> List[ParameterGroup]:
         return [
             parameter_group
             for parameter_group in self.redshift_backend.describe_cluster_parameter_groups()
-            if parameter_group.cluster_parameter_group_name
-            in self.cluster_parameter_group_name
+            if parameter_group.name in self.cluster_parameter_group_name
         ]
 
     @property
@@ -279,68 +275,68 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
     def resume(self) -> None:
         self.status = "available"
 
-    def to_json(self) -> Dict[str, Any]:
-        json_response = {
-            "MasterUsername": self.master_username,
-            "MasterUserPassword": "****",
-            "ClusterVersion": self.cluster_version,
-            "VpcSecurityGroups": [
-                {"Status": "active", "VpcSecurityGroupId": group.id}
-                for group in self.vpc_security_groups
-            ],
-            "ClusterSubnetGroupName": self.cluster_subnet_group_name,
-            "AvailabilityZone": self.availability_zone,
-            "ClusterStatus": self.status,
-            "NumberOfNodes": self.number_of_nodes,
-            "AutomatedSnapshotRetentionPeriod": self.automated_snapshot_retention_period,
-            "PubliclyAccessible": self.publicly_accessible,
-            "Encrypted": self.encrypted,
-            "DBName": self.db_name,
-            "PreferredMaintenanceWindow": self.preferred_maintenance_window,
-            "ClusterParameterGroups": [
-                {
-                    "ParameterApplyStatus": "in-sync",
-                    "ParameterGroupName": group.cluster_parameter_group_name,
-                }
-                for group in self.parameter_groups
-            ],
-            "ClusterSecurityGroups": [
-                {
-                    "Status": "active",
-                    "ClusterSecurityGroupName": group.cluster_security_group_name,
-                }
-                for group in self.security_groups
-            ],
+    @property
+    def vpc_security_group_membership_list(self) -> List[Dict[str, str]]:
+        return [
+            {"VpcSecurityGroupId": group.id, "Status": "active"}
+            for group in self.vpc_security_groups
+        ]
+
+    @property
+    def cluster_parameter_group_status_list(self) -> List[Dict[str, str]]:
+        return [
+            {
+                "ParameterGroupName": group.name,
+                "ParameterApplyStatus": "in-sync",
+            }
+            for group in self.parameter_groups
+        ]
+
+    @property
+    def cluster_security_group_membership_list(self) -> List[Dict[str, str]]:
+        return [
+            {
+                "ClusterSecurityGroupName": group.cluster_security_group_name,
+                "Status": "active",
+            }
+            for group in self.security_groups
+        ]
+
+    @property
+    def endpoint(self) -> Dict[str, str | int]:
+        return {
+            "Address": self.address,
             "Port": self.port,
-            "NodeType": self.node_type,
-            "ClusterIdentifier": self.cluster_identifier,
-            "AllowVersionUpgrade": self.allow_version_upgrade,
-            "Endpoint": {"Address": self.endpoint, "Port": self.port},
-            "ClusterCreateTime": self.create_time,
-            "PendingModifiedValues": [],
-            "Tags": self.tags,
-            "EnhancedVpcRouting": self.enhanced_vpc_routing,
-            "IamRoles": [
-                {"ApplyStatus": "in-sync", "IamRoleArn": iam_role_arn}
-                for iam_role_arn in self.iam_roles_arn
-            ],
-            "KmsKeyId": self.kms_key_id,
-            "TotalStorageCapacityInMegaBytes": self.total_storage_capacity,
         }
-        if self.restored_from_snapshot:
-            json_response["RestoreStatus"] = {
-                "Status": "completed",
-                "CurrentRestoreRateInMegaBytesPerSecond": 123.0,
-                "SnapshotSizeInMegaBytes": 123,
-                "ProgressInMegaBytes": 123,
-                "ElapsedTimeInSeconds": 123,
-                "EstimatedTimeToCompletionInSeconds": 123,
-            }
-        if self.cluster_snapshot_copy_status is not None:
-            json_response["ClusterSnapshotCopyStatus"] = (
-                self.cluster_snapshot_copy_status
-            )
-        return json_response
+
+    @property
+    def pending_modified_values(self) -> list[str]:
+        return []
+
+    @property
+    def iam_roles(self) -> List[Dict[str, str]]:
+        return [
+            {"ApplyStatus": "in-sync", "IamRoleArn": iam_role_arn}
+            for iam_role_arn in self.iam_roles_arn
+        ]
+
+    @property
+    def total_storage_capacity_in_mega_bytes(self) -> int:
+        return self.total_storage_capacity
+
+    @property
+    def restore_status(self) -> Optional[Dict[str, Any]]:
+        if not self.restored_from_snapshot:
+            return None
+        status = {
+            "Status": "completed",
+            "CurrentRestoreRateInMegaBytesPerSecond": 123.0,
+            "SnapshotSizeInMegaBytes": 123,
+            "ProgressInMegaBytes": 123,
+            "ElapsedTimeInSeconds": 123,
+            "EstimatedTimeToCompletionInSeconds": 123,
+        }
+        return status
 
 
 class SnapshotCopyGrant(TaggableResourceMixin, BaseModel):
@@ -350,12 +346,6 @@ class SnapshotCopyGrant(TaggableResourceMixin, BaseModel):
         self.snapshot_copy_grant_name = snapshot_copy_grant_name
         self.kms_key_id = kms_key_id
 
-    def to_json(self) -> Dict[str, Any]:
-        return {
-            "SnapshotCopyGrantName": self.snapshot_copy_grant_name,
-            "KmsKeyId": self.kms_key_id,
-        }
-
 
 class SubnetGroup(TaggableResourceMixin, CloudFormationModel):
     resource_type = "subnetgroup"
@@ -374,6 +364,7 @@ class SubnetGroup(TaggableResourceMixin, CloudFormationModel):
         self.cluster_subnet_group_name = cluster_subnet_group_name
         self.description = description
         self.subnet_ids = subnet_ids
+        self.status = "Complete"
         if not self.subnets:
             raise InvalidSubnetError(subnet_ids)
 
@@ -387,14 +378,14 @@ class SubnetGroup(TaggableResourceMixin, CloudFormationModel):
         return "AWS::Redshift::ClusterSubnetGroup"
 
     @classmethod
-    def create_from_cloudformation_json(  # type: ignore[misc]
+    def create_from_cloudformation_json(
         cls,
         resource_name: str,
         cloudformation_json: Any,
         account_id: str,
         region_name: str,
         **kwargs: Any,
-    ) -> "SubnetGroup":
+    ) -> SubnetGroup:
         redshift_backend = redshift_backends[account_id][region_name]
         properties = cloudformation_json["Properties"]
 
@@ -407,7 +398,7 @@ class SubnetGroup(TaggableResourceMixin, CloudFormationModel):
         return subnet_group
 
     @property
-    def subnets(self) -> Any:  # type: ignore[misc]
+    def subnets(self) -> Any:
         return self.ec2_backend.describe_subnets(filters={"subnet-id": self.subnet_ids})
 
     @property
@@ -418,22 +409,16 @@ class SubnetGroup(TaggableResourceMixin, CloudFormationModel):
     def resource_id(self) -> str:
         return self.cluster_subnet_group_name
 
-    def to_json(self) -> Dict[str, Any]:
-        return {
-            "VpcId": self.vpc_id,
-            "Description": self.description,
-            "ClusterSubnetGroupName": self.cluster_subnet_group_name,
-            "SubnetGroupStatus": "Complete",
-            "Subnets": [
-                {
-                    "SubnetStatus": "Active",
-                    "SubnetIdentifier": subnet.id,
-                    "SubnetAvailabilityZone": {"Name": subnet.availability_zone},
-                }
-                for subnet in self.subnets
-            ],
-            "Tags": self.tags,
-        }
+    @property
+    def subnet_list(self) -> List[Dict[str, Any]]:
+        return [
+            {
+                "SubnetStatus": "Active",
+                "SubnetIdentifier": subnet.id,
+                "SubnetAvailabilityZone": {"Name": subnet.availability_zone},
+            }
+            for subnet in self.subnets
+        ]
 
 
 class SecurityGroup(TaggableResourceMixin, BaseModel):
@@ -451,20 +436,13 @@ class SecurityGroup(TaggableResourceMixin, BaseModel):
         self.cluster_security_group_name = cluster_security_group_name
         self.description = description
         self.ingress_rules: List[str] = []
+        self.ec2_security_groups: list[str] = []
+        self.ip_ranges: list[str] = []
 
     @property
     def resource_id(self) -> str:
         return self.cluster_security_group_name
 
-    def to_json(self) -> Dict[str, Any]:
-        return {
-            "EC2SecurityGroups": [],
-            "IPRanges": [],
-            "Description": self.description,
-            "ClusterSecurityGroupName": self.cluster_security_group_name,
-            "Tags": self.tags,
-        }
-
 
 class ParameterGroup(TaggableResourceMixin, CloudFormationModel):
     resource_type = "parametergroup"
@@ -479,8 +457,8 @@ class ParameterGroup(TaggableResourceMixin, CloudFormationModel):
         tags: Optional[List[Dict[str, str]]] = None,
     ):
         super().__init__(account_id, region_name, tags)
-        self.cluster_parameter_group_name = cluster_parameter_group_name
-        self.group_family = group_family
+        self.name = cluster_parameter_group_name
+        self.family = group_family
         self.description = description
 
     @staticmethod
@@ -493,14 +471,14 @@ class ParameterGroup(TaggableResourceMixin, CloudFormationModel):
         return "AWS::Redshift::ClusterParameterGroup"
 
     @classmethod
-    def create_from_cloudformation_json(  # type: ignore[misc]
+    def create_from_cloudformation_json(
         cls,
         resource_name: str,
         cloudformation_json: Any,
         account_id: str,
         region_name: str,
         **kwargs: Any,
-    ) -> "ParameterGroup":
+    ) -> ParameterGroup:
         redshift_backend = redshift_backends[account_id][region_name]
         properties = cloudformation_json["Properties"]
 
@@ -514,15 +492,7 @@ class ParameterGroup(TaggableResourceMixin, CloudFormationModel):
 
     @property
     def resource_id(self) -> str:
-        return self.cluster_parameter_group_name
-
-    def to_json(self) -> Dict[str, Any]:
-        return {
-            "ParameterGroupFamily": self.group_family,
-            "Description": self.description,
-            "ParameterGroupName": self.cluster_parameter_group_name,
-            "Tags": self.tags,
-        }
+        return self.name
 
 
 class Snapshot(TaggableResourceMixin, BaseModel):
@@ -539,38 +509,33 @@ class Snapshot(TaggableResourceMixin, BaseModel):
         snapshot_type: str = "manual",
     ):
         super().__init__(account_id, region_name, tags)
-        self.cluster = copy.copy(cluster)
         self.snapshot_identifier = snapshot_identifier
         self.snapshot_type = snapshot_type
         self.status = "available"
-        self.create_time = iso_8601_datetime_with_milliseconds()
+        self.create_time = utcnow()
         self.iam_roles_arn = iam_roles_arn or []
+        self.cluster_identifier = cluster.cluster_identifier
+        self.port = cluster.port
+        self.availability_zone = cluster.availability_zone
+        self.master_username = cluster.master_username
+        self.master_user_password = cluster.master_user_password
+        self.cluster_version = cluster.cluster_version
+        self.node_type = cluster.node_type
+        self.number_of_nodes = cluster.number_of_nodes
+        self.db_name = cluster.db_name
+        self.enhanced_vpc_routing = cluster.enhanced_vpc_routing
+        self.encrypted = cluster.encrypted
 
     @property
     def resource_id(self) -> str:
-        return f"{self.cluster.cluster_identifier}/{self.snapshot_identifier}"
+        return f"{self.cluster_identifier}/{self.snapshot_identifier}"
 
-    def to_json(self) -> Dict[str, Any]:
-        return {
-            "SnapshotIdentifier": self.snapshot_identifier,
-            "ClusterIdentifier": self.cluster.cluster_identifier,
-            "SnapshotCreateTime": self.create_time,
-            "Status": self.status,
-            "Port": self.cluster.port,
-            "AvailabilityZone": self.cluster.availability_zone,
-            "MasterUsername": self.cluster.master_username,
-            "ClusterVersion": self.cluster.cluster_version,
-            "SnapshotType": self.snapshot_type,
-            "NodeType": self.cluster.node_type,
-            "NumberOfNodes": self.cluster.number_of_nodes,
-            "DBName": self.cluster.db_name,
-            "Tags": self.tags,
-            "EnhancedVpcRouting": self.cluster.enhanced_vpc_routing,
-            "IamRoles": [
-                {"ApplyStatus": "in-sync", "IamRoleArn": iam_role_arn}
-                for iam_role_arn in self.iam_roles_arn
-            ],
-        }
+    @property
+    def iam_roles(self) -> List[Dict[str, str]]:
+        return [
+            {"ApplyStatus": "in-sync", "IamRoleArn": iam_role_arn}
+            for iam_role_arn in self.iam_roles_arn
+        ]
 
 
 class RedshiftBackend(BaseBackend):
@@ -607,10 +572,7 @@ class RedshiftBackend(BaseBackend):
         cluster_identifier = kwargs["cluster_identifier"]
         cluster = self.clusters[cluster_identifier]
         if cluster.cluster_snapshot_copy_status is None:
-            if (
-                cluster.encrypted == "true"
-                and kwargs["snapshot_copy_grant_name"] is None
-            ):
+            if cluster.encrypted and kwargs["snapshot_copy_grant_name"] is None:
                 raise InvalidParameterValueError(
                     "SnapshotCopyGrantName is required for Snapshot Copy on KMS encrypted clusters."
                 )
@@ -909,7 +871,7 @@ class RedshiftBackend(BaseBackend):
         if cluster_identifier:
             cluster_snapshots = []
             for snapshot in self.snapshots.values():
-                if snapshot.cluster.cluster_identifier == cluster_identifier:
+                if snapshot.cluster_identifier == cluster_identifier:
                     if snapshot.snapshot_type in snapshot_types:
                         cluster_snapshots.append(snapshot)
             if cluster_snapshots:
@@ -942,21 +904,21 @@ class RedshiftBackend(BaseBackend):
             snapshot_identifier=snapshot_identifier
         )[0]
         create_kwargs = {
-            "node_type": snapshot.cluster.node_type,
-            "master_username": snapshot.cluster.master_username,
-            "master_user_password": snapshot.cluster.master_user_password,
-            "db_name": snapshot.cluster.db_name,
+            "node_type": snapshot.node_type,
+            "master_username": snapshot.master_username,
+            "master_user_password": snapshot.master_user_password,
+            "db_name": snapshot.db_name,
             "cluster_type": "multi-node"
-            if snapshot.cluster.number_of_nodes > 1
+            if snapshot.number_of_nodes > 1
             else "single-node",
-            "availability_zone": snapshot.cluster.availability_zone,
-            "port": snapshot.cluster.port,
-            "cluster_version": snapshot.cluster.cluster_version,
-            "number_of_nodes": snapshot.cluster.number_of_nodes,
-            "encrypted": snapshot.cluster.encrypted,
-            "tags": snapshot.cluster.tags,
+            "availability_zone": snapshot.availability_zone,
+            "port": snapshot.port,
+            "cluster_version": snapshot.cluster_version,
+            "number_of_nodes": snapshot.number_of_nodes,
+            "encrypted": snapshot.encrypted,
+            "tags": snapshot.tags,
             "restored_from_snapshot": True,
-            "enhanced_vpc_routing": snapshot.cluster.enhanced_vpc_routing,
+            "enhanced_vpc_routing": snapshot.enhanced_vpc_routing,
         }
         create_kwargs.update(kwargs)
         return self.create_cluster(**create_kwargs)
@@ -1012,7 +974,7 @@ class RedshiftBackend(BaseBackend):
         return resource
 
     @staticmethod
-    def _describe_tags_for_resources(resources: Iterable[Any]) -> List[Dict[str, Any]]:  # type: ignore[misc]
+    def _describe_tags_for_resources(resources: Iterable[Any]) -> List[Dict[str, Any]]:
         tagged_resources = []
         for resource in resources:
             for tag in resource.tags:
diff --git a/moto/redshift/responses.py b/moto/redshift/responses.py
index 0fd111fb0..3495a6d08 100644
--- a/moto/redshift/responses.py
+++ b/moto/redshift/responses.py
@@ -1,52 +1,16 @@
-import json
-from typing import Any, Dict, List
+from typing import Any, List
 
-import xmltodict
-from jinja2 import Template
-
-from moto.core.common_types import TYPE_RESPONSE
-from moto.core.responses import BaseResponse
+from moto.core.responses import ActionResult, BaseResponse, EmptyResult
 
 from .models import RedshiftBackend, redshift_backends
 
 
-def convert_json_error_to_xml(json_error: Any) -> str:
-    error = json.loads(json_error)
-    code = error["Error"]["Code"]
-    message = error["Error"]["Message"]
-    template = Template(
-        """
-        <RedshiftClientError>
-            <Error>
-              <Code>{{ code }}</Code>
-              <Message>{{ message }}</Message>
-              <Type>Sender</Type>
-            </Error>
-            <RequestId>6876f774-7273-11e4-85dc-39e55ca848d1</RequestId>
-        </RedshiftClientError>"""
-    )
-    return template.render(code=code, message=message)
-
-
-def itemize(data: Any) -> Dict[str, Any]:
-    """
-    The xmltodict.unparse requires we modify the shape of the input dictionary slightly. Instead of a dict of the form:
-        {'key': ['value1', 'value2']}
-    We must provide:
-        {'key': {'item': ['value1', 'value2']}}
-    """
-    if isinstance(data, dict):
-        ret = {}
-        for key in data:
-            ret[key] = itemize(data[key])
-        return ret
-    elif isinstance(data, list):
-        return {"item": [itemize(value) for value in data]}
-    else:
-        return data
-
-
 class RedshiftResponse(BaseResponse):
+    RESPONSE_KEY_PATH_TO_TRANSFORMER = {
+        "CreateClusterResult.Cluster.ClusterStatus": lambda _: "creating",
+        "RestoreFromClusterSnapshotResult.Cluster.ClusterStatus": lambda _: "creating",
+    }
+
     def __init__(self) -> None:
         super().__init__(service_name="redshift")
 
@@ -54,21 +18,6 @@ class RedshiftResponse(BaseResponse):
     def redshift_backend(self) -> RedshiftBackend:
         return redshift_backends[self.current_account][self.region]
 
-    def get_response(self, response: Any) -> str:
-        if self.request_json:
-            return json.dumps(response)
-        else:
-            xml = xmltodict.unparse(itemize(response), full_document=False)
-            if hasattr(xml, "decode"):
-                xml = xml.decode("utf-8")
-            return xml
-
-    def call_action(self) -> TYPE_RESPONSE:
-        status, headers, body = super().call_action()
-        if status >= 400 and not self.request_json:
-            body = convert_json_error_to_xml(body)
-        return status, headers, body
-
     def unpack_list_params(self, label: str, child_label: str) -> Any:
         root = self._get_multi_param_dict(label) or {}
         return root.get(child_label, [])
@@ -101,7 +50,7 @@ class RedshiftResponse(BaseResponse):
             subnet_ids = self._get_multi_param("SubnetIds.SubnetIdentifier")
         return subnet_ids
 
-    def create_cluster(self) -> str:
+    def create_cluster(self) -> ActionResult:
         cluster_kwargs = {
             "cluster_identifier": self._get_param("ClusterIdentifier"),
             "node_type": self._get_param("NodeType"),
@@ -124,58 +73,30 @@ class RedshiftResponse(BaseResponse):
             ),
             "port": self._get_int_param("Port"),
             "cluster_version": self._get_param("ClusterVersion"),
-            "allow_version_upgrade": self._get_bool_param("AllowVersionUpgrade"),
+            "allow_version_upgrade": self._get_bool_param("AllowVersionUpgrade", True),
             "number_of_nodes": self._get_int_param("NumberOfNodes"),
-            "publicly_accessible": self._get_param("PubliclyAccessible"),
-            "encrypted": self._get_param("Encrypted"),
+            "publicly_accessible": self._get_bool_param("PubliclyAccessible", False),
+            "encrypted": self._get_bool_param("Encrypted", False),
             "region_name": self.region,
             "tags": self.unpack_list_params("Tags", "Tag"),
             "iam_roles_arn": self._get_iam_roles(),
-            "enhanced_vpc_routing": self._get_param("EnhancedVpcRouting"),
+            "enhanced_vpc_routing": self._get_bool_param("EnhancedVpcRouting", False),
             "kms_key_id": self._get_param("KmsKeyId"),
         }
-        cluster = self.redshift_backend.create_cluster(**cluster_kwargs).to_json()
-        cluster["ClusterStatus"] = "creating"
-        return self.get_response(
-            {
-                "CreateClusterResponse": {
-                    "CreateClusterResult": {"Cluster": cluster},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        cluster = self.redshift_backend.create_cluster(**cluster_kwargs)
+        return ActionResult({"Cluster": cluster})
 
-    def pause_cluster(self) -> str:
+    def pause_cluster(self) -> ActionResult:
         cluster_id = self._get_param("ClusterIdentifier")
-        cluster = self.redshift_backend.pause_cluster(cluster_id).to_json()
-        return self.get_response(
-            {
-                "PauseClusterResponse": {
-                    "PauseClusterResult": {"Cluster": cluster},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        cluster = self.redshift_backend.pause_cluster(cluster_id)
+        return ActionResult({"Cluster": cluster})
 
-    def resume_cluster(self) -> str:
+    def resume_cluster(self) -> ActionResult:
         cluster_id = self._get_param("ClusterIdentifier")
-        cluster = self.redshift_backend.resume_cluster(cluster_id).to_json()
-        return self.get_response(
-            {
-                "ResumeClusterResponse": {
-                    "ResumeClusterResult": {"Cluster": cluster},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        cluster = self.redshift_backend.resume_cluster(cluster_id)
+        return ActionResult({"Cluster": cluster})
 
-    def restore_from_cluster_snapshot(self) -> str:
+    def restore_from_cluster_snapshot(self) -> ActionResult:
         enhanced_vpc_routing = self._get_bool_param("EnhancedVpcRouting")
         node_type = self._get_param("NodeType")
         number_of_nodes = self._get_int_param("NumberOfNodes")
@@ -207,39 +128,16 @@ class RedshiftResponse(BaseResponse):
             restore_kwargs["node_type"] = node_type
         if number_of_nodes is not None:
             restore_kwargs["number_of_nodes"] = number_of_nodes
-        cluster = self.redshift_backend.restore_from_cluster_snapshot(
-            **restore_kwargs
-        ).to_json()
-        cluster["ClusterStatus"] = "creating"
-        return self.get_response(
-            {
-                "RestoreFromClusterSnapshotResponse": {
-                    "RestoreFromClusterSnapshotResult": {"Cluster": cluster},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        cluster = self.redshift_backend.restore_from_cluster_snapshot(**restore_kwargs)
+        return ActionResult({"Cluster": cluster})
 
-    def describe_clusters(self) -> str:
+    def describe_clusters(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         clusters = self.redshift_backend.describe_clusters(cluster_identifier)
 
-        return self.get_response(
-            {
-                "DescribeClustersResponse": {
-                    "DescribeClustersResult": {
-                        "Clusters": [cluster.to_json() for cluster in clusters]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Clusters": clusters})
 
-    def modify_cluster(self) -> str:
+    def modify_cluster(self) -> ActionResult:
         request_kwargs = {
             "cluster_identifier": self._get_param("ClusterIdentifier"),
             "new_cluster_identifier": self._get_param("NewClusterIdentifier"),
@@ -275,18 +173,9 @@ class RedshiftResponse(BaseResponse):
 
         cluster = self.redshift_backend.modify_cluster(**cluster_kwargs)
 
-        return self.get_response(
-            {
-                "ModifyClusterResponse": {
-                    "ModifyClusterResult": {"Cluster": cluster.to_json()},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Cluster": cluster})
 
-    def delete_cluster(self) -> str:
+    def delete_cluster(self) -> ActionResult:
         request_kwargs = {
             "cluster_identifier": self._get_param("ClusterIdentifier"),
             "final_cluster_snapshot_identifier": self._get_param(
@@ -297,18 +186,9 @@ class RedshiftResponse(BaseResponse):
 
         cluster = self.redshift_backend.delete_cluster(**request_kwargs)
 
-        return self.get_response(
-            {
-                "DeleteClusterResponse": {
-                    "DeleteClusterResult": {"Cluster": cluster.to_json()},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Cluster": cluster})
 
-    def create_cluster_subnet_group(self) -> str:
+    def create_cluster_subnet_group(self) -> ActionResult:
         cluster_subnet_group_name = self._get_param("ClusterSubnetGroupName")
         description = self._get_param("Description")
         subnet_ids = self._get_subnet_ids()
@@ -322,55 +202,23 @@ class RedshiftResponse(BaseResponse):
             tags=tags,
         )
 
-        return self.get_response(
-            {
-                "CreateClusterSubnetGroupResponse": {
-                    "CreateClusterSubnetGroupResult": {
-                        "ClusterSubnetGroup": subnet_group.to_json()
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"ClusterSubnetGroup": subnet_group})
 
-    def describe_cluster_subnet_groups(self) -> str:
+    def describe_cluster_subnet_groups(self) -> ActionResult:
         subnet_identifier = self._get_param("ClusterSubnetGroupName")
         subnet_groups = self.redshift_backend.describe_cluster_subnet_groups(
             subnet_identifier
         )
 
-        return self.get_response(
-            {
-                "DescribeClusterSubnetGroupsResponse": {
-                    "DescribeClusterSubnetGroupsResult": {
-                        "ClusterSubnetGroups": [
-                            subnet_group.to_json() for subnet_group in subnet_groups
-                        ]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"ClusterSubnetGroups": subnet_groups})
 
-    def delete_cluster_subnet_group(self) -> str:
+    def delete_cluster_subnet_group(self) -> ActionResult:
         subnet_identifier = self._get_param("ClusterSubnetGroupName")
         self.redshift_backend.delete_cluster_subnet_group(subnet_identifier)
 
-        return self.get_response(
-            {
-                "DeleteClusterSubnetGroupResponse": {
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    }
-                }
-            }
-        )
+        return EmptyResult()
 
-    def create_cluster_security_group(self) -> str:
+    def create_cluster_security_group(self) -> ActionResult:
         cluster_security_group_name = self._get_param("ClusterSecurityGroupName")
         description = self._get_param("Description")
         tags = self.unpack_list_params("Tags", "Tag")
@@ -381,84 +229,45 @@ class RedshiftResponse(BaseResponse):
             tags=tags,
         )
 
-        return self.get_response(
-            {
-                "CreateClusterSecurityGroupResponse": {
-                    "CreateClusterSecurityGroupResult": {
-                        "ClusterSecurityGroup": security_group.to_json()
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"ClusterSecurityGroup": security_group})
 
-    def describe_cluster_security_groups(self) -> str:
+    def describe_cluster_security_groups(self) -> ActionResult:
         cluster_security_group_name = self._get_param("ClusterSecurityGroupName")
         security_groups = self.redshift_backend.describe_cluster_security_groups(
             cluster_security_group_name
         )
 
-        return self.get_response(
-            {
-                "DescribeClusterSecurityGroupsResponse": {
-                    "DescribeClusterSecurityGroupsResult": {
-                        "ClusterSecurityGroups": [
-                            security_group.to_json()
-                            for security_group in security_groups
-                        ]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"ClusterSecurityGroups": security_groups})
 
-    def delete_cluster_security_group(self) -> str:
+    def delete_cluster_security_group(self) -> ActionResult:
         security_group_identifier = self._get_param("ClusterSecurityGroupName")
         self.redshift_backend.delete_cluster_security_group(security_group_identifier)
 
-        return self.get_response(
-            {
-                "DeleteClusterSecurityGroupResponse": {
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    }
-                }
-            }
-        )
+        return EmptyResult()
 
-    def authorize_cluster_security_group_ingress(self) -> str:
+    def authorize_cluster_security_group_ingress(self) -> ActionResult:
         cluster_security_group_name = self._get_param("ClusterSecurityGroupName")
         cidr_ip = self._get_param("CIDRIP")
 
         security_group = self.redshift_backend.authorize_cluster_security_group_ingress(
             cluster_security_group_name, cidr_ip
         )
-
-        return self.get_response(
-            {
-                "AuthorizeClusterSecurityGroupIngressResponse": {
-                    "AuthorizeClusterSecurityGroupIngressResult": {
-                        "ClusterSecurityGroup": {
-                            "ClusterSecurityGroupName": cluster_security_group_name,
-                            "Description": security_group.description,
-                            "IPRanges": [
-                                {
-                                    "Status": "authorized",
-                                    "CIDRIP": cidr_ip,
-                                    "Tags": security_group.tags,
-                                },
-                            ],
-                        }
-                    }
-                }
+        result = {
+            "ClusterSecurityGroup": {
+                "ClusterSecurityGroupName": cluster_security_group_name,
+                "Description": security_group.description,
+                "IPRanges": [
+                    {
+                        "Status": "authorized",
+                        "CIDRIP": cidr_ip,
+                        "Tags": security_group.tags,
+                    },
+                ],
             }
-        )
+        }
+        return ActionResult(result)
 
-    def create_cluster_parameter_group(self) -> str:
+    def create_cluster_parameter_group(self) -> ActionResult:
         cluster_parameter_group_name = self._get_param("ParameterGroupName")
         group_family = self._get_param("ParameterGroupFamily")
         description = self._get_param("Description")
@@ -468,58 +277,25 @@ class RedshiftResponse(BaseResponse):
             cluster_parameter_group_name, group_family, description, self.region, tags
         )
 
-        return self.get_response(
-            {
-                "CreateClusterParameterGroupResponse": {
-                    "CreateClusterParameterGroupResult": {
-                        "ClusterParameterGroup": parameter_group.to_json()
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"ClusterParameterGroup": parameter_group})
 
-    def describe_cluster_parameter_groups(self) -> str:
+    def describe_cluster_parameter_groups(self) -> ActionResult:
         cluster_parameter_group_name = self._get_param("ParameterGroupName")
         parameter_groups = self.redshift_backend.describe_cluster_parameter_groups(
             cluster_parameter_group_name
         )
 
-        return self.get_response(
-            {
-                "DescribeClusterParameterGroupsResponse": {
-                    "DescribeClusterParameterGroupsResult": {
-                        "ParameterGroups": [
-                            parameter_group.to_json()
-                            for parameter_group in parameter_groups
-                        ]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"ParameterGroups": parameter_groups})
 
-    def delete_cluster_parameter_group(self) -> str:
+    def delete_cluster_parameter_group(self) -> ActionResult:
         cluster_parameter_group_name = self._get_param("ParameterGroupName")
         self.redshift_backend.delete_cluster_parameter_group(
             cluster_parameter_group_name
         )
 
-        return self.get_response(
-            {
-                "DeleteClusterParameterGroupResponse": {
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    }
-                }
-            }
-        )
+        return EmptyResult()
 
-    def create_cluster_snapshot(self) -> str:
+    def create_cluster_snapshot(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         snapshot_identifier = self._get_param("SnapshotIdentifier")
         tags = self.unpack_list_params("Tags", "Tag")
@@ -527,53 +303,24 @@ class RedshiftResponse(BaseResponse):
         snapshot = self.redshift_backend.create_cluster_snapshot(
             cluster_identifier, snapshot_identifier, self.region, tags
         )
-        return self.get_response(
-            {
-                "CreateClusterSnapshotResponse": {
-                    "CreateClusterSnapshotResult": {"Snapshot": snapshot.to_json()},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Snapshot": snapshot})
 
-    def describe_cluster_snapshots(self) -> str:
+    def describe_cluster_snapshots(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         snapshot_identifier = self._get_param("SnapshotIdentifier")
         snapshot_type = self._get_param("SnapshotType")
         snapshots = self.redshift_backend.describe_cluster_snapshots(
             cluster_identifier, snapshot_identifier, snapshot_type
         )
-        return self.get_response(
-            {
-                "DescribeClusterSnapshotsResponse": {
-                    "DescribeClusterSnapshotsResult": {
-                        "Snapshots": [snapshot.to_json() for snapshot in snapshots]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Snapshots": snapshots})
 
-    def delete_cluster_snapshot(self) -> str:
+    def delete_cluster_snapshot(self) -> ActionResult:
         snapshot_identifier = self._get_param("SnapshotIdentifier")
         snapshot = self.redshift_backend.delete_cluster_snapshot(snapshot_identifier)
 
-        return self.get_response(
-            {
-                "DeleteClusterSnapshotResponse": {
-                    "DeleteClusterSnapshotResult": {"Snapshot": snapshot.to_json()},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Snapshot": snapshot})
 
-    def create_snapshot_copy_grant(self) -> str:
+    def create_snapshot_copy_grant(self) -> ActionResult:
         copy_grant_kwargs = {
             "snapshot_copy_grant_name": self._get_param("SnapshotCopyGrantName"),
             "kms_key_id": self._get_param("KmsKeyId"),
@@ -583,35 +330,16 @@ class RedshiftResponse(BaseResponse):
         copy_grant = self.redshift_backend.create_snapshot_copy_grant(
             **copy_grant_kwargs
         )
-        return self.get_response(
-            {
-                "CreateSnapshotCopyGrantResponse": {
-                    "CreateSnapshotCopyGrantResult": {
-                        "SnapshotCopyGrant": copy_grant.to_json()
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"SnapshotCopyGrant": copy_grant})
 
-    def delete_snapshot_copy_grant(self) -> str:
+    def delete_snapshot_copy_grant(self) -> ActionResult:
         copy_grant_kwargs = {
             "snapshot_copy_grant_name": self._get_param("SnapshotCopyGrantName")
         }
         self.redshift_backend.delete_snapshot_copy_grant(**copy_grant_kwargs)
-        return self.get_response(
-            {
-                "DeleteSnapshotCopyGrantResponse": {
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    }
-                }
-            }
-        )
+        return EmptyResult()
 
-    def describe_snapshot_copy_grants(self) -> str:
+    def describe_snapshot_copy_grants(self) -> ActionResult:
         copy_grant_kwargs = {
             "snapshot_copy_grant_name": self._get_param("SnapshotCopyGrantName")
         }
@@ -619,72 +347,34 @@ class RedshiftResponse(BaseResponse):
         copy_grants = self.redshift_backend.describe_snapshot_copy_grants(
             **copy_grant_kwargs
         )
-        return self.get_response(
-            {
-                "DescribeSnapshotCopyGrantsResponse": {
-                    "DescribeSnapshotCopyGrantsResult": {
-                        "SnapshotCopyGrants": [
-                            copy_grant.to_json() for copy_grant in copy_grants
-                        ]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"SnapshotCopyGrants": copy_grants})
 
-    def create_tags(self) -> str:
+    def create_tags(self) -> ActionResult:
         resource_name = self._get_param("ResourceName")
         tags = self.unpack_list_params("Tags", "Tag")
 
         self.redshift_backend.create_tags(resource_name, tags)
 
-        return self.get_response(
-            {
-                "CreateTagsResponse": {
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    }
-                }
-            }
-        )
+        return EmptyResult()
 
-    def describe_tags(self) -> str:
+    def describe_tags(self) -> ActionResult:
         resource_name = self._get_param("ResourceName")
         resource_type = self._get_param("ResourceType")
 
         tagged_resources = self.redshift_backend.describe_tags(
             resource_name, resource_type
         )
-        return self.get_response(
-            {
-                "DescribeTagsResponse": {
-                    "DescribeTagsResult": {"TaggedResources": tagged_resources},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"TaggedResources": tagged_resources})
 
-    def delete_tags(self) -> str:
+    def delete_tags(self) -> ActionResult:
         resource_name = self._get_param("ResourceName")
         tag_keys = self.unpack_list_params("TagKeys", "TagKey")
 
         self.redshift_backend.delete_tags(resource_name, tag_keys)
 
-        return self.get_response(
-            {
-                "DeleteTagsResponse": {
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    }
-                }
-            }
-        )
+        return EmptyResult()
 
-    def enable_snapshot_copy(self) -> str:
+    def enable_snapshot_copy(self) -> ActionResult:
         snapshot_copy_kwargs = {
             "cluster_identifier": self._get_param("ClusterIdentifier"),
             "destination_region": self._get_param("DestinationRegion"),
@@ -693,35 +383,17 @@ class RedshiftResponse(BaseResponse):
         }
         cluster = self.redshift_backend.enable_snapshot_copy(**snapshot_copy_kwargs)
 
-        return self.get_response(
-            {
-                "EnableSnapshotCopyResponse": {
-                    "EnableSnapshotCopyResult": {"Cluster": cluster.to_json()},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Cluster": cluster})
 
-    def disable_snapshot_copy(self) -> str:
+    def disable_snapshot_copy(self) -> ActionResult:
         snapshot_copy_kwargs = {
             "cluster_identifier": self._get_param("ClusterIdentifier")
         }
         cluster = self.redshift_backend.disable_snapshot_copy(**snapshot_copy_kwargs)
 
-        return self.get_response(
-            {
-                "DisableSnapshotCopyResponse": {
-                    "DisableSnapshotCopyResult": {"Cluster": cluster.to_json()},
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Cluster": cluster})
 
-    def modify_snapshot_copy_retention_period(self) -> str:
+    def modify_snapshot_copy_retention_period(self) -> ActionResult:
         snapshot_copy_kwargs = {
             "cluster_identifier": self._get_param("ClusterIdentifier"),
             "retention_period": self._get_param("RetentionPeriod"),
@@ -730,20 +402,9 @@ class RedshiftResponse(BaseResponse):
             **snapshot_copy_kwargs
         )
 
-        return self.get_response(
-            {
-                "ModifySnapshotCopyRetentionPeriodResponse": {
-                    "ModifySnapshotCopyRetentionPeriodResult": {
-                        "Clusters": [cluster.to_json()]
-                    },
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult({"Clusters": [cluster]})
 
-    def get_cluster_credentials(self) -> str:
+    def get_cluster_credentials(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         db_user = self._get_param("DbUser")
         auto_create = self._get_bool_param("AutoCreate", False)
@@ -753,18 +414,9 @@ class RedshiftResponse(BaseResponse):
             cluster_identifier, db_user, auto_create, duration_seconds
         )
 
-        return self.get_response(
-            {
-                "GetClusterCredentialsResponse": {
-                    "GetClusterCredentialsResult": cluster_credentials,
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult(cluster_credentials)
 
-    def enable_logging(self) -> str:
+    def enable_logging(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         bucket_name = self._get_param("BucketName")
         s3_key_prefix = self._get_param("S3KeyPrefix")
@@ -777,45 +429,18 @@ class RedshiftResponse(BaseResponse):
             log_destination_type=log_destination_type,
             log_exports=log_exports,
         )
-        return self.get_response(
-            {
-                "EnableLoggingResponse": {
-                    "EnableLoggingResult": config,
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult(config)
 
-    def disable_logging(self) -> str:
+    def disable_logging(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         config = self.redshift_backend.disable_logging(
             cluster_identifier=cluster_identifier,
         )
-        return self.get_response(
-            {
-                "DisableLoggingResponse": {
-                    "DisableLoggingResult": config,
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult(config)
 
-    def describe_logging_status(self) -> str:
+    def describe_logging_status(self) -> ActionResult:
         cluster_identifier = self._get_param("ClusterIdentifier")
         config = self.redshift_backend.describe_logging_status(
             cluster_identifier=cluster_identifier,
         )
-        return self.get_response(
-            {
-                "DescribeLoggingStatusResponse": {
-                    "DescribeLoggingStatusResult": config,
-                    "ResponseMetadata": {
-                        "RequestId": "384ac68d-3775-11df-8963-01868b7c937a"
-                    },
-                }
-            }
-        )
+        return ActionResult(config)
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 2e5c10655..adbcd47d6 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -1767,9 +1767,9 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         #
         # Second, go through the list of instances
         # It may contain FakeKeys created earlier, which are no longer tracked
-        for mp in FakeMultipart.instances:  # type: ignore
+        for mp in FakeMultipart.instances_tracked:  # type: ignore
             mp.dispose()
-        for key in FakeKey.instances:  # type: ignore
+        for key in FakeKey.instances_tracked:  # type: ignore
             key.dispose()
         super().reset()
 
diff --git a/moto/ssm/models.py b/moto/ssm/models.py
index ba9d22f54..ffba8d7b6 100644
--- a/moto/ssm/models.py
+++ b/moto/ssm/models.py
@@ -359,10 +359,13 @@ class Parameter(CloudFormationModel):
         account_id: str,
         region_name: str,
     ) -> None:
-        ssm_backend = ssm_backends[account_id][region_name]
-        properties = cloudformation_json["Properties"]
+        resource = resource_name
 
-        ssm_backend.delete_parameter(properties.get("Name"))
+        ssm_backend = ssm_backends[account_id][region_name]
+        if not resource:
+            properties = cloudformation_json["Properties"]
+            resource = properties.get("Name")
+        ssm_backend.delete_parameter(resource)
 
     @property
     def physical_resource_id(self) -> str:


Test patch:
diff --git a/tests/test_core/test_moto_api.py b/tests/test_core/test_moto_api.py
index 6b2b3c101..93f553956 100644
--- a/tests/test_core/test_moto_api.py
+++ b/tests/test_core/test_moto_api.py
@@ -97,7 +97,7 @@ def test_model_data_is_emptied_as_necessary() -> None:
     # No instances exist, because we have just reset it
     for classes_per_service in model_data.values():
         for _class in classes_per_service.values():
-            assert _class.instances == []  # type: ignore[attr-defined]
+            assert _class.instances_tracked == []  # type: ignore[attr-defined]
 
     # TODO: ensure that iam is not loaded, and IAM policies are not created
     # with mock_aws(load_static_data=False) ?
@@ -105,18 +105,18 @@ def test_model_data_is_emptied_as_necessary() -> None:
         # When just starting a mock, it is empty
         for classes_per_service in model_data.values():
             for _class in classes_per_service.values():
-                assert _class.instances == []  # type: ignore[attr-defined]
+                assert _class.instances_tracked == []  # type: ignore[attr-defined]
 
         # After creating a queue, some data will be present
         conn = boto3.client("sqs", region_name="us-west-1")
         conn.create_queue(QueueName="queue1")
 
-        assert len(model_data["sqs"]["Queue"].instances) == 1  # type: ignore[attr-defined]
+        assert len(model_data["sqs"]["Queue"].instances_tracked) == 1  # type: ignore[attr-defined]
 
     # But after the mock ends, it is empty again
     for classes_per_service in model_data.values():
         for _class in classes_per_service.values():
-            assert _class.instances == []  # type: ignore[attr-defined]
+            assert _class.instances_tracked == []  # type: ignore[attr-defined]
 
     # When we have multiple/nested mocks, the data should still be present after the first mock ends
     with mock_aws():
@@ -124,9 +124,9 @@ def test_model_data_is_emptied_as_necessary() -> None:
         conn.create_queue(QueueName="queue1")
         with mock_aws():
             # The data should still be here - instances should not reset if another mock is still active
-            assert len(model_data["sqs"]["Queue"].instances) == 1  # type: ignore[attr-defined]
+            assert len(model_data["sqs"]["Queue"].instances_tracked) == 1  # type: ignore[attr-defined]
         # The data should still be here - the inner mock has exited, but the outer mock is still active
-        assert len(model_data["sqs"]["Queue"].instances) == 1  # type: ignore[attr-defined]
+        assert len(model_data["sqs"]["Queue"].instances_tracked) == 1  # type: ignore[attr-defined]
 
 
 @mock_aws
@@ -138,10 +138,10 @@ class TestModelDataResetForClassDecorator(TestCase):
         # No data is present at the beginning
         for classes_per_service in model_data.values():
             for _class in classes_per_service.values():
-                assert _class.instances == []  # type: ignore[attr-defined]
+                assert _class.instances_tracked == []  # type: ignore[attr-defined]
 
         conn = boto3.client("sqs", region_name="us-west-1")
         conn.create_queue(QueueName="queue1")
 
     def test_should_find_bucket(self) -> None:
-        assert len(model_data["sqs"]["Queue"].instances) == 1  # type: ignore[attr-defined]
+        assert len(model_data["sqs"]["Queue"].instances_tracked) == 1  # type: ignore[attr-defined]
diff --git a/tests/test_core/utilities.py b/tests/test_core/utilities.py
index 0e9382c0a..c95db185a 100644
--- a/tests/test_core/utilities.py
+++ b/tests/test_core/utilities.py
@@ -25,7 +25,7 @@ class SimpleServer:
 
     def get_host_and_port(self) -> Tuple[str, int]:
         assert self._server
-        host, port = self._server.server_address
+        host, port = self._server.server_address[:2]
         return str(host), port
 
     def stop(self) -> None:
diff --git a/tests/test_elasticache/test_elasticache.py b/tests/test_elasticache/test_elasticache.py
index f97168e9c..b1e7e7944 100644
--- a/tests/test_elasticache/test_elasticache.py
+++ b/tests/test_elasticache/test_elasticache.py
@@ -1035,10 +1035,7 @@ def test_create_replication_group_cluster_disabled():
     assert replication_group["SnapshottingClusterId"] == f"{replication_group_id}-002"
     assert replication_group["MultiAZ"] == "enabled"
     assert replication_group["SnapshotRetentionLimit"] == 1
-    if not replication_group["ClusterEnabled"]:
-        assert True
-    else:
-        assert False
+    assert not replication_group["ClusterEnabled"]
     assert replication_group["CacheNodeType"] == "cache.t4g.micro"
     assert (
         replication_group["ARN"]
@@ -1315,10 +1312,7 @@ def test_describe_replication_groups_cluster_enabled():
     replication_group = describe_resp["ReplicationGroups"][0]
     assert replication_group["ReplicationGroupId"] == replication_group_id
     assert len(replication_group["MemberClusters"]) == 9
-    if replication_group["ClusterEnabled"]:
-        assert True
-    else:
-        assert False
+    assert replication_group["ClusterEnabled"]
 
 
 @mock_aws
diff --git a/tests/test_redshift/test_redshift.py b/tests/test_redshift/test_redshift.py
index 976faff35..c231f14d8 100644
--- a/tests/test_redshift/test_redshift.py
+++ b/tests/test_redshift/test_redshift.py
@@ -44,14 +44,12 @@ def test_create_cluster_boto3():
             "ParameterApplyStatus": "in-sync",
         }
     ]
-    assert cluster["ClusterSubnetGroupName"] == ""
     assert cluster["AvailabilityZone"] == "us-east-1a"
     assert cluster["PreferredMaintenanceWindow"] == "Mon:03:00-Mon:03:30"
     assert cluster["ClusterVersion"] == "1.0"
     assert cluster["AllowVersionUpgrade"] is True
     assert cluster["NumberOfNodes"] == 1
     assert cluster["EnhancedVpcRouting"] is False
-    assert cluster["KmsKeyId"] == ""
     assert cluster["Endpoint"]["Port"] == 5439
 
 
@@ -591,7 +589,7 @@ def test_authorize_security_group_ingress():
         client.authorize_cluster_security_group_ingress(
             ClusterSecurityGroupName="invalid_security_group", CIDRIP="192.168.10.0/28"
         )
-    assert ex.value.response["Error"]["Code"] == "ClusterSecurityGroupNotFoundFault"
+    assert ex.value.response["Error"]["Code"] == "ClusterSecurityGroupNotFound"
 
     assert (
         ex.value.response["Error"]["Message"]
diff --git a/tests/test_redshift/test_server.py b/tests/test_redshift/test_server.py
index a55baa4bd..7d49802cf 100644
--- a/tests/test_redshift/test_server.py
+++ b/tests/test_redshift/test_server.py
@@ -25,7 +25,7 @@ def test_describe_clusters():
     res = test_client.get("/?Action=DescribeClusters")
 
     result = res.data.decode("utf-8")
-    assert "<Clusters></Clusters>" in result
+    assert "<Clusters></Clusters>" in result or "<Clusters/>" in result
 
 
 @mock_aws
@@ -73,14 +73,12 @@ def test_create_cluster(is_json):
     result = result["CreateClusterResponse"]["CreateClusterResult"]["Cluster"]
 
     assert result["MasterUsername"] == "masteruser"
-    assert result["MasterUserPassword"] == "****"
     assert result["ClusterVersion"] == "1.0"
-    assert result["ClusterSubnetGroupName"] is None
     assert result["AvailabilityZone"] == "us-east-1a"
     assert result["ClusterStatus"] == "creating"
     assert result["NumberOfNodes"] == (1 if is_json else "1")
-    assert result["PubliclyAccessible"] is None
-    assert result["Encrypted"] is None
+    assert result["PubliclyAccessible"] == (False if is_json else "false")
+    assert result["Encrypted"] == (False if is_json else "false")
     assert result["DBName"] == "dev"
     assert result["NodeType"] == "ds2.xlarge"
     assert result["ClusterIdentifier"] == "examplecluster"
@@ -133,14 +131,12 @@ def test_create_cluster_multiple_params(is_json):
     result = result["CreateClusterResponse"]["CreateClusterResult"]["Cluster"]
 
     assert result["MasterUsername"] == "masteruser"
-    assert result["MasterUserPassword"] == "****"
     assert result["ClusterVersion"] == "2.0"
-    assert result["ClusterSubnetGroupName"] is None
     assert result["AvailabilityZone"] == "us-east-1a"
     assert result["ClusterStatus"] == "creating"
     assert result["NumberOfNodes"] == (3 if is_json else "3")
-    assert result["PubliclyAccessible"] is None
-    assert result["Encrypted"] == "True"
+    assert result["PubliclyAccessible"] == (False if is_json else "false")
+    assert result["Encrypted"] == (True if is_json else "true")
     assert result["DBName"] == "testdb"
     assert result["NodeType"] == "ds2.xlarge"
     assert result["ClusterIdentifier"] == "examplecluster"
@@ -153,7 +149,7 @@ def test_create_cluster_multiple_params(is_json):
     assert "Tags" in result
     tags = result["Tags"]
     if not is_json:
-        tags = tags["item"]
+        tags = tags["Tag"]
     assert tags == [{"Key": "key1", "Value": "val1"}, {"Key": "key2", "Value": "val2"}]
 
 
@@ -192,7 +188,7 @@ def test_create_and_describe_clusters(is_json):
     assert "Clusters" in result["DescribeClustersResponse"]["DescribeClustersResult"]
     result = result["DescribeClustersResponse"]["DescribeClustersResult"]["Clusters"]
     if not is_json:
-        result = result["item"]
+        result = result["Cluster"]
 
     assert len(result) == 2
     for cluster in result:
@@ -230,7 +226,7 @@ def test_create_and_describe_cluster_security_group(is_json):
         "DescribeClusterSecurityGroupsResult"
     ]["ClusterSecurityGroups"]
     if not is_json:
-        groups = groups["item"]
+        groups = groups["ClusterSecurityGroup"]
 
     descriptions = [g["Description"] for g in groups]
     assert "desc for csg1" in descriptions
@@ -258,7 +254,7 @@ def test_create_and_describe_cluster_security_group(is_json):
         assert len(groups) == 1
         assert groups[0]["ClusterSecurityGroupName"] == "csg1"
     else:
-        assert groups["item"]["ClusterSecurityGroupName"] == "csg1"
+        assert groups["ClusterSecurityGroup"]["ClusterSecurityGroupName"] == "csg1"
 
 
 @pytest.mark.parametrize("is_json", [True, False], ids=["JSON", "XML"])
@@ -280,7 +276,7 @@ def test_describe_unknown_cluster_security_group(is_json):
         response = json.loads(res.data.decode("utf-8"))
     else:
         response = xmltodict.parse(res.data.decode("utf-8"), dict_constructor=dict)[
-            "RedshiftClientError"
+            "ErrorResponse"
         ]
     error = response["Error"]
 
@@ -351,7 +347,7 @@ def test_create_cluster_with_security_group(is_json):
 
     security_groups = result["ClusterSecurityGroups"]
     if not is_json:
-        security_groups = security_groups["item"]
+        security_groups = security_groups["ClusterSecurityGroup"]
 
     assert len(security_groups) == 2
     for csg in security_group_names:
diff --git a/tests/test_ssm/test_ssm_parameterstore_cloudformation.py b/tests/test_ssm/test_ssm_parameterstore_cloudformation.py
new file mode 100644
index 000000000..0594ad7bf
--- /dev/null
+++ b/tests/test_ssm/test_ssm_parameterstore_cloudformation.py
@@ -0,0 +1,148 @@
+import json
+
+import boto3
+import botocore.exceptions
+import pytest
+
+from moto import mock_aws
+
+
+@mock_aws
+def test_cloudformation_lifecycle():
+    ssm_param_name = "test"
+    ssm_param_value = "initial value"
+    stack_template = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Description": "Test Stack",
+        "Resources": {
+            "SSMParameter": {
+                "Type": "AWS::SSM::Parameter",
+                "Properties": {
+                    "Name": ssm_param_name,
+                    "Type": "String",
+                    "Value": ssm_param_value,
+                },
+            }
+        },
+    }
+
+    cloudformation_client = boto3.client("cloudformation", region_name="us-east-1")
+
+    stack_template_str = json.dumps(stack_template)
+
+    cloudformation_client.create_stack(
+        StackName="test_stack",
+        TemplateBody=stack_template_str,
+        Capabilities=("CAPABILITY_IAM",),
+    )
+
+    client = boto3.client("ssm", region_name="us-east-1")
+    resp = client.get_parameter(Name=ssm_param_name)["Parameter"]
+    resp_param_name = resp["Name"]
+    resp_param_value = resp["Value"]
+
+    assert resp_param_name == ssm_param_name
+    assert resp_param_value == ssm_param_value
+
+    # Update the stack template with new value
+
+    new_ssm_param_value = "updated value"
+
+    stack_template["Resources"]["SSMParameter"]["Properties"]["Value"] = (
+        new_ssm_param_value
+    )
+    stack_template_str = json.dumps(stack_template)
+    cloudformation_client.update_stack(
+        StackName="test_stack",
+        TemplateBody=stack_template_str,
+        Capabilities=("CAPABILITY_IAM",),
+    )
+
+    resp = client.get_parameter(Name=ssm_param_name)["Parameter"]
+    resp_param_name = resp["Name"]
+    resp_param_value = resp["Value"]
+
+    assert resp_param_name == ssm_param_name
+    assert resp_param_value == new_ssm_param_value
+
+    # Stack deletion
+
+    cloudformation_client.delete_stack(StackName="test_stack")
+    with pytest.raises(botocore.exceptions.ClientError) as exc:
+        resp = client.get_parameter(Name=ssm_param_name)["Parameter"]
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ParameterNotFound"
+    assert err["Message"] == f"Parameter {ssm_param_name} not found."
+
+
+@mock_aws
+def test_cloudformation_lifecycle_with_parsing():
+    ssm_param_name = "test"
+    ssm_param_value = "initial value"
+    stack_template = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Description": "Test Stack",
+        "Parameters": {
+            "Name": {"Type": "String", "Default": ssm_param_name},
+            "Value": {"Type": "String", "Default": ssm_param_value},
+        },
+        "Resources": {
+            "SSMParameter": {
+                "Type": "AWS::SSM::Parameter",
+                "Properties": {
+                    "Name": {"Fn::Sub": "${Name}"},
+                    "Type": "String",
+                    "Value": {"Ref": "Value"},
+                },
+            }
+        },
+    }
+
+    cloudformation_client = boto3.client("cloudformation", region_name="us-east-1")
+
+    stack_template_str = json.dumps(stack_template)
+
+    cloudformation_client.create_stack(
+        StackName="test_stack",
+        TemplateBody=stack_template_str,
+        Capabilities=("CAPABILITY_IAM",),
+    )
+
+    cloudformation_client.describe_stack_resources(StackName="test_stack")
+
+    client = boto3.client("ssm", region_name="us-east-1")
+    resp = client.get_parameter(Name=ssm_param_name)
+    resp_param = resp.get("Parameter")
+    resp_param_name = resp_param["Name"]
+    resp_param_value = resp_param["Value"]
+
+    assert resp_param_name == ssm_param_name
+    assert resp_param_value == ssm_param_value
+
+    # Update the stack with new value
+
+    new_ssm_param_value = "updated value"
+
+    stack_template_str = json.dumps(stack_template)
+    cloudformation_client.update_stack(
+        StackName="test_stack",
+        TemplateBody=stack_template_str,
+        Capabilities=("CAPABILITY_IAM",),
+        Parameters=[{"ParameterKey": "Value", "ParameterValue": new_ssm_param_value}],
+    )
+
+    resp = client.get_parameter(Name=ssm_param_name)["Parameter"]
+    resp_param_name = resp["Name"]
+    resp_param_value = resp["Value"]
+
+    assert resp_param_name == ssm_param_name
+    assert resp_param_value == new_ssm_param_value
+
+    # Stack deletion
+
+    cloudformation_client.delete_stack(StackName="test_stack")
+    with pytest.raises(botocore.exceptions.ClientError) as exc:
+        resp = client.get_parameter(Name=ssm_param_name)["Parameter"]
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ParameterNotFound"
+    assert err["Message"] == f"Parameter {ssm_param_name} not found."