import copy
import json
import os

import pytest
from botocore.exceptions import WaiterError
from localstack_snapshot.snapshots.transformer import RegexTransformer
from tests.aws.services.cloudformation.conftest import skip_if_legacy_engine

from localstack.aws.connect import ServiceLevelClientFactory
from localstack.testing.pytest import markers
from localstack.utils.strings import short_uid


@skip_if_legacy_engine()
@markers.snapshot.skip_snapshot_verify(
    paths=[
        "delete-describe..*",
        #
        # Before/After Context
        "$..Capabilities",
        "$..IncludeNestedStacks",
        "$..Scope",
        "$..Details",
        "$..Parameters",
        "$..PolicyAction",
        "$..PhysicalResourceId",
    ]
)
class TestCaptureUpdateProcess:
    @markers.aws.validated
    def test_direct_update(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        Update a stack with a static change (i.e. in the text of the template).

        Conclusions:
        - A static change in the template that's not invoking an intrinsic function
            (`Ref`, `Fn::GetAtt` etc.) is resolved by the deployment engine synchronously
            during the `create_change_set` invocation
        """
        name1 = f"topic-1-{short_uid()}"
        name2 = f"topic-2-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(name1, "topic-1"))
        snapshot.add_transformer(RegexTransformer(name2, "topic-2"))
        t1 = {
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": name1,
                    },
                },
            },
        }
        t2 = {
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": name2,
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t2)

    @markers.aws.validated
    def test_dynamic_update(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        Update a stack with two resources:
        - A is changed statically
        - B refers to the changed value of A via an intrinsic function

        Conclusions:
        - The value of B on creation is "known after apply" even though the resolved
          property value is known statically
        - The nature of the change to B is "known after apply"
        - The CloudFormation engine does not resolve intrinsic function calls when determining the
            nature of the update
        """
        name1 = f"topic-1-{short_uid()}"
        name2 = f"topic-2-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(name1, "topic-1"))
        snapshot.add_transformer(RegexTransformer(name2, "topic-2"))
        t1 = {
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": name1,
                    },
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {
                            "Fn::GetAtt": ["Foo", "TopicName"],
                        },
                    },
                },
            },
        }
        t2 = {
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": name2,
                    },
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {
                            "Fn::GetAtt": ["Foo", "TopicName"],
                        },
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t2)

    @markers.aws.validated
    def test_parameter_changes(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        Update a stack with two resources:
        - A is changed via a template parameter
        - B refers to the changed value of A via an intrinsic function

        Conclusions:
        - The value of B on creation is "known after apply" even though the resolved
          property value is known statically
        - The nature of the change to B is "known after apply"
        - The CloudFormation engine does not resolve intrinsic function calls when determining the
            nature of the update
        """
        name1 = f"topic-1-{short_uid()}"
        name2 = f"topic-2-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(name1, "topic-1"))
        snapshot.add_transformer(RegexTransformer(name2, "topic-2"))
        t1 = {
            "Parameters": {
                "TopicName": {
                    "Type": "String",
                },
            },
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": {"Ref": "TopicName"},
                    },
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {
                            "Fn::GetAtt": ["Foo", "TopicName"],
                        },
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t1, p1={"TopicName": name1}, p2={"TopicName": name2})

    @markers.aws.validated
    def test_mappings_with_static_fields(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        Update a stack with two resources:
        - A is changed via looking up a static value in a mapping
        - B refers to the changed value of A via an intrinsic function

        Conclusions:
        - On first deploy the contents of the map is resolved completely
        - The nature of the change to B is "known after apply"
        - The CloudFormation engine does not resolve intrinsic function calls when determining the
            nature of the update
        """
        name1 = f"topic-1-{short_uid()}"
        name2 = f"topic-2-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(name1, "topic-name-1"))
        snapshot.add_transformer(RegexTransformer(name2, "topic-name-2"))
        t1 = {
            "Mappings": {
                "MyMap": {
                    "MyKey": {"key1": name1, "key2": name2},
                },
            },
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": {
                            "Fn::FindInMap": [
                                "MyMap",
                                "MyKey",
                                "key1",
                            ],
                        },
                    },
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {
                            "Fn::GetAtt": ["Foo", "TopicName"],
                        },
                    },
                },
            },
        }
        t2 = {
            "Mappings": {
                "MyMap": {
                    "MyKey": {
                        "key1": name1,
                        "key2": name2,
                    },
                },
            },
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": {
                            "Fn::FindInMap": [
                                "MyMap",
                                "MyKey",
                                "key2",
                            ],
                        },
                    },
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {
                            "Fn::GetAtt": ["Foo", "TopicName"],
                        },
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t2)

    @markers.aws.validated
    def test_mappings_with_parameter_lookup(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        Update a stack with two resources:
        - A is changed via looking up a static value in a mapping but the key comes from
          a template parameter
        - B refers to the changed value of A via an intrinsic function

        Conclusions:
        - The same conclusions as `test_mappings_with_static_fields`
        """
        name1 = f"topic-1-{short_uid()}"
        name2 = f"topic-2-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(name1, "topic-name-1"))
        snapshot.add_transformer(RegexTransformer(name2, "topic-name-2"))
        t1 = {
            "Parameters": {
                "TopicName": {
                    "Type": "String",
                },
            },
            "Mappings": {
                "MyMap": {
                    "MyKey": {"key1": name1, "key2": name2},
                },
            },
            "Resources": {
                "Foo": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": {
                        "TopicName": {
                            "Fn::FindInMap": [
                                "MyMap",
                                "MyKey",
                                {
                                    "Ref": "TopicName",
                                },
                            ],
                        },
                    },
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {
                            "Fn::GetAtt": ["Foo", "TopicName"],
                        },
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t1, p1={"TopicName": "key1"}, p2={"TopicName": "key2"})

    @markers.aws.validated
    def test_conditions(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        Toggle a resource from present to not present via a condition

        Conclusions:
        - Adding the second resource creates an `Add` resource change
        """
        t1 = {
            "Parameters": {
                "EnvironmentType": {
                    "Type": "String",
                }
            },
            "Conditions": {
                "IsProduction": {
                    "Fn::Equals": [
                        {"Ref": "EnvironmentType"},
                        "prod",
                    ],
                }
            },
            "Resources": {
                "Bucket": {
                    "Type": "AWS::S3::Bucket",
                },
                "Parameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": "test",
                    },
                    "Condition": "IsProduction",
                },
            },
        }

        capture_update_process(
            snapshot, t1, t1, p1={"EnvironmentType": "not-prod"}, p2={"EnvironmentType": "prod"}
        )

    @markers.aws.validated
    @pytest.mark.skip(
        "Unlike AWS CFN, the update graph understands the dependent resource does not "
        "need modification also when the IncludePropertyValues flag is off."
        # TODO: we may achieve the same limitation by pruning the resolution of traversals.
    )
    def test_unrelated_changes_update_propagation(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        - Resource B depends on resource A which is updated, but the referenced parameter does not
          change

        Conclusions:
        - No update to resource B
        """
        topic_name = f"MyTopic{short_uid()}"
        snapshot.add_transformer(RegexTransformer(topic_name, "topic-name"))
        t1 = {
            "Resources": {
                "Parameter1": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": topic_name,
                        "Description": "original",
                    },
                },
                "Parameter2": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {"Fn::GetAtt": ["Parameter1", "Value"]},
                    },
                },
            },
        }
        t2 = {
            "Resources": {
                "Parameter1": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": topic_name,
                        "Description": "changed",
                    },
                },
                "Parameter2": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {"Fn::GetAtt": ["Parameter1", "Value"]},
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t2)

    @markers.aws.validated
    @pytest.mark.skip(
        "Deployment now succeeds but our describer incorrectly does not assign a change for Parameter2"
    )
    def test_unrelated_changes_requires_replacement(
        self,
        snapshot,
        capture_update_process,
    ):
        """
        - Resource B depends on resource A which is updated, but the referenced parameter does not
          change, however resource A requires replacement

        Conclusions:
        - Resource B is updated
        """
        parameter_name_1 = f"MyParameter{short_uid()}"
        parameter_name_2 = f"MyParameter{short_uid()}"
        snapshot.add_transformer(RegexTransformer(parameter_name_1, "parameter-1-name"))
        snapshot.add_transformer(RegexTransformer(parameter_name_2, "parameter-2-name"))
        t1 = {
            "Resources": {
                "Parameter1": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Name": parameter_name_1,
                        "Type": "String",
                        "Value": "value",
                    },
                },
                "Parameter2": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {"Fn::GetAtt": ["Parameter1", "Value"]},
                    },
                },
            },
        }
        t2 = {
            "Resources": {
                "Parameter1": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Name": parameter_name_2,
                        "Type": "String",
                        "Value": "value",
                    },
                },
                "Parameter2": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": {"Fn::GetAtt": ["Parameter1", "Value"]},
                    },
                },
            },
        }
        capture_update_process(snapshot, t1, t2)

    @markers.aws.validated
    @pytest.mark.parametrize(
        "template",
        [
            pytest.param(
                {
                    "Parameters": {
                        "ParameterValue": {
                            "Type": "String",
                        },
                    },
                    "Resources": {
                        "Parameter": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": {"Ref": "ParameterValue"},
                            },
                        }
                    },
                },
                id="change_dynamic",
            ),
            pytest.param(
                {
                    "Parameters": {
                        "ParameterValue": {
                            "Type": "String",
                        },
                    },
                    "Resources": {
                        "Parameter1": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Name": "param-name",
                                "Type": "String",
                                "Value": {"Ref": "ParameterValue"},
                            },
                        },
                        "Parameter2": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": {"Fn::GetAtt": ["Parameter1", "Name"]},
                            },
                        },
                    },
                },
                id="change_unrelated_property",
            ),
            pytest.param(
                {
                    "Parameters": {
                        "ParameterValue": {
                            "Type": "String",
                        },
                    },
                    "Resources": {
                        "Parameter1": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": {"Ref": "ParameterValue"},
                            },
                        },
                        "Parameter2": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": {"Fn::GetAtt": ["Parameter1", "Type"]},
                            },
                        },
                    },
                },
                id="change_unrelated_property_not_create_only",
            ),
            pytest.param(
                {
                    "Parameters": {
                        "ParameterValue": {
                            "Type": "String",
                            "Default": "value-1",
                            "AllowedValues": ["value-1", "value-2"],
                        }
                    },
                    "Conditions": {
                        "ShouldCreateParameter": {
                            "Fn::Equals": [{"Ref": "ParameterValue"}, "value-2"]
                        }
                    },
                    "Resources": {
                        "SSMParameter1": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": "first",
                            },
                        },
                        "SSMParameter2": {
                            "Type": "AWS::SSM::Parameter",
                            "Condition": "ShouldCreateParameter",
                            "Properties": {
                                "Type": "String",
                                "Value": "first",
                            },
                        },
                    },
                },
                id="change_parameter_for_condition_create_resource",
            ),
        ],
    )
    def test_base_dynamic_parameter_scenarios(
        self, snapshot, capture_update_process, template, request
    ):
        if request.node.callspec.id in {
            "change_unrelated_property",
            "change_unrelated_property_not_create_only",
        }:
            pytest.skip(
                reason="AWS appears to incorrectly mark the dependent resource as needing update when describe "
                "changeset is invoked without the inclusion of property values."
            )
        capture_update_process(
            snapshot,
            template,
            template,
            {"ParameterValue": "value-1"},
            {"ParameterValue": "value-2"},
        )

    @markers.aws.validated
    def test_execute_with_ref(self, snapshot, aws_client, deploy_cfn_template):
        name1 = f"param-1-{short_uid()}"
        snapshot.add_transformer(snapshot.transform.regex(name1, "<name-1>"))
        name2 = f"param-2-{short_uid()}"
        snapshot.add_transformer(snapshot.transform.regex(name2, "<name-2>"))
        value = "my-value"
        param2_name = f"output-param-{short_uid()}"
        snapshot.add_transformer(snapshot.transform.regex(param2_name, "<output-parameter>"))

        t1 = {
            "Resources": {
                "Parameter1": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Name": name1,
                        "Type": "String",
                        "Value": value,
                    },
                },
                "Parameter2": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Name": param2_name,
                        "Type": "String",
                        "Value": {"Ref": "Parameter1"},
                    },
                },
            }
        }
        t2 = copy.deepcopy(t1)
        t2["Resources"]["Parameter1"]["Properties"]["Name"] = name2

        stack = deploy_cfn_template(template=json.dumps(t1))
        stack_id = stack.stack_id

        before_value = aws_client.ssm.get_parameter(Name=param2_name)["Parameter"]["Value"]
        snapshot.match("before-value", before_value)

        deploy_cfn_template(stack_name=stack_id, template=json.dumps(t2), is_update=True)

        after_value = aws_client.ssm.get_parameter(Name=param2_name)["Parameter"]["Value"]
        snapshot.match("after-value", after_value)

    @markers.aws.validated
    @pytest.mark.parametrize(
        "template_1, template_2",
        [
            (
                {
                    "Mappings": {"GenericMapping": {"EnvironmentA": {"ParameterValue": "value-1"}}},
                    "Resources": {
                        "MySSMParameter": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": {
                                    "Fn::FindInMap": [
                                        "GenericMapping",
                                        "EnvironmentA",
                                        "ParameterValue",
                                    ]
                                },
                            },
                        }
                    },
                },
                {
                    "Mappings": {"GenericMapping": {"EnvironmentA": {"ParameterValue": "value-2"}}},
                    "Resources": {
                        "MySSMParameter": {
                            "Type": "AWS::SSM::Parameter",
                            "Properties": {
                                "Type": "String",
                                "Value": {
                                    "Fn::FindInMap": [
                                        "GenericMapping",
                                        "EnvironmentA",
                                        "ParameterValue",
                                    ]
                                },
                            },
                        }
                    },
                },
            )
        ],
        ids=["update_string_referencing_resource"],
    )
    def test_base_mapping_scenarios(
        self,
        snapshot,
        capture_update_process,
        template_1,
        template_2,
    ):
        capture_update_process(snapshot, template_1, template_2)

    @markers.aws.validated
    def test_single_resource_static_update(
        self, aws_client: ServiceLevelClientFactory, snapshot, cleanups
    ):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())
        parameter_name = f"parameter-{short_uid()}"
        value1 = "foo"
        value2 = "bar"

        t1 = {
            "Resources": {
                "MyParameter": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Name": parameter_name,
                        "Type": "String",
                        "Value": value1,
                    },
                },
            },
        }

        stack_name = f"stack-{short_uid()}"
        change_set_name = f"cs-{short_uid()}"
        cs_result = aws_client.cloudformation.create_change_set(
            StackName=stack_name,
            ChangeSetName=change_set_name,
            TemplateBody=json.dumps(t1),
            ChangeSetType="CREATE",
        )
        cs_id = cs_result["Id"]
        stack_id = cs_result["StackId"]
        aws_client.cloudformation.get_waiter("change_set_create_complete").wait(ChangeSetName=cs_id)
        cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=stack_id))

        describe_result = aws_client.cloudformation.describe_change_set(ChangeSetName=cs_id)
        snapshot.match("describe-1", describe_result)

        aws_client.cloudformation.execute_change_set(ChangeSetName=cs_id)
        aws_client.cloudformation.get_waiter("stack_create_complete").wait(StackName=stack_id)

        parameter = aws_client.ssm.get_parameter(Name=parameter_name)["Parameter"]
        snapshot.match("parameter-1", parameter)

        t2 = copy.deepcopy(t1)
        t2["Resources"]["MyParameter"]["Properties"]["Value"] = value2

        change_set_name = f"cs-{short_uid()}"
        cs_result = aws_client.cloudformation.create_change_set(
            StackName=stack_name,
            ChangeSetName=change_set_name,
            TemplateBody=json.dumps(t2),
        )
        cs_id = cs_result["Id"]
        aws_client.cloudformation.get_waiter("change_set_create_complete").wait(ChangeSetName=cs_id)

        describe_result = aws_client.cloudformation.describe_change_set(ChangeSetName=cs_id)
        snapshot.match("describe-2", describe_result)

        aws_client.cloudformation.execute_change_set(ChangeSetName=cs_id)
        aws_client.cloudformation.get_waiter("stack_update_complete").wait(StackName=stack_id)

        parameter = aws_client.ssm.get_parameter(Name=parameter_name)["Parameter"]
        snapshot.match("parameter-2", parameter)

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(paths=["$..PhysicalResourceId"])
    def test_queue_update(
        self, aws_client, deploy_cfn_template, capture_per_resource_events, snapshot
    ):
        """
        A test where one of the templates creates a resource without any specified properties. We
        make sure the second operation on that resource is an UPDATE not CREATE.
        """
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        template1 = os.path.join(
            os.path.dirname(__file__), "../../templates/sqs_queue_update_1.yaml"
        )
        template2 = os.path.join(
            os.path.dirname(__file__), "../../templates/sqs_queue_update_2.yaml"
        )

        # Deploy order:
        # - 1
        # - 2
        # - 1

        stack = deploy_cfn_template(template_path=template1)
        outputs = stack.outputs

        queue_url = outputs["QueueUrl"]

        queue_attributes = aws_client.sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["All"]
        )
        snapshot.match("queue-attributes-1", queue_attributes)

        deploy_cfn_template(template_path=template2, is_update=True, stack_name=stack.stack_id)
        queue_attributes = aws_client.sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["All"]
        )
        snapshot.match("queue-attributes-2", queue_attributes)

        deploy_cfn_template(template_path=template1, is_update=True, stack_name=stack.stack_id)
        queue_attributes = aws_client.sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["All"]
        )
        snapshot.match("queue-attributes-3", queue_attributes)

        per_resource_events = capture_per_resource_events(stack.stack_id)

        snapshot.match("events", per_resource_events["StandardQueue"])


@skip_if_legacy_engine()
@markers.aws.validated
@markers.snapshot.skip_snapshot_verify(
    paths=[
        "delete-describe..*",
        #
        # Before/After Context
        "$..Capabilities",
        "$..IncludeNestedStacks",
        "$..Scope",
        "$..Details",
        "$..Parameters",
        "$..Replacement",
        "$..PolicyAction",
        "$..PhysicalResourceId",
    ]
)
def test_dynamic_ssm_parameter_lookup(
    snapshot,
    aws_client: ServiceLevelClientFactory,
    aws_client_no_retry: ServiceLevelClientFactory,
    cleanups,
    create_parameter,
    capture_update_process,
):
    """
    Test reading parameter values from SSM works correctly"""
    parameter_name = f"param-{short_uid()}"
    value1 = f"value-1-{short_uid()}"
    value2 = f"value-2-{short_uid()}"

    snapshot.add_transformers_list(
        [
            snapshot.transform.key_value("OutputValue"),
            snapshot.transform.regex(parameter_name, "<parameter-name>"),
            snapshot.transform.regex(value1, "<parameter-value-1>"),
            snapshot.transform.regex(value2, "<parameter-value-2>"),
        ]
        + snapshot.transform.cloudformation_api(),
    )

    create_parameter(Name=parameter_name, Value=value1, Type="String")

    template = {
        "Parameters": {
            "InputValue": {
                "Type": "AWS::SSM::Parameter::Value<String>",
            },
        },
        "Resources": {
            "DerivedParameter": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": {"Ref": "InputValue"},
                },
            },
        },
        "Outputs": {
            "DerivedParameterName": {
                "Value": {"Ref": "DerivedParameter"},
            },
        },
    }

    def update_parameter_value():
        aws_client.ssm.put_parameter(Name=parameter_name, Value=value2, Overwrite=True)

    capture_update_process(
        snapshot,
        template,
        template,
        p1={"InputValue": parameter_name},
        p2={"InputValue": parameter_name},
        custom_update_step=update_parameter_value,
    )


@skip_if_legacy_engine()
@markers.aws.validated
@markers.snapshot.skip_snapshot_verify(
    paths=[
        "delete-describe..*",
        #
        # Before/After Context
        "$..Capabilities",
        "$..IncludeNestedStacks",
        "$..Scope",
        "$..Details",
        "$..Parameters",
        "$..Replacement",
        "$..PolicyAction",
        "$..PhysicalResourceId",
    ]
)
def test_dynamic_ssm_parameter_lookup_no_change(
    snapshot,
    aws_client: ServiceLevelClientFactory,
    aws_client_no_retry: ServiceLevelClientFactory,
    cleanups,
    create_parameter,
    capture_update_process,
):
    """
    Test reading parameter values from SSM works correctly"""
    parameter_name = f"param-{short_uid()}"
    parameter_value = f"value-{short_uid()}"

    snapshot.add_transformers_list(
        [
            snapshot.transform.key_value("OutputValue"),
            snapshot.transform.regex(parameter_name, "<parameter-name>"),
            snapshot.transform.regex(parameter_value, "<parameter-value>"),
        ]
        + snapshot.transform.cloudformation_api(),
    )

    create_parameter(Name=parameter_name, Value=parameter_value, Type="String")

    t1 = {
        "Parameters": {
            "InputValue": {
                "Type": "AWS::SSM::Parameter::Value<String>",
            },
        },
        "Resources": {
            "DerivedParameter": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": {"Ref": "InputValue"},
                },
            },
        },
        "Outputs": {
            "DerivedParameterName": {
                "Value": {"Ref": "DerivedParameter"},
            },
        },
    }
    t2 = copy.deepcopy(t1)
    t2["Resources"]["AnotherParameter"] = {
        "Type": "AWS::SSM::Parameter",
        "Properties": {
            "Type": "String",
            "Value": "new parameter",
        },
    }

    capture_update_process(
        snapshot,
        t1,
        t2,
        p1={"InputValue": parameter_name},
        p2={"InputValue": parameter_name},
    )


@skip_if_legacy_engine()
@markers.aws.validated
@markers.snapshot.skip_snapshot_verify(paths=["$..StatusReason"])
def test_describe_failed_change_set(aws_client: ServiceLevelClientFactory, snapshot, cleanups):
    snapshot.add_transformer(snapshot.transform.cloudformation_api())

    template = {
        "Resources": {
            "MyFoo": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Fn::Transform": {
                        "Name": "AWS::Include",
                        "Parameters": {
                            "Location": "s3://doesnotexist/key.yml",
                        },
                    },
                },
            },
        },
    }

    stack_name = f"cs-{short_uid()}"
    change_set_name = f"cs-{short_uid()}"
    res = aws_client.cloudformation.create_change_set(
        StackName=stack_name,
        ChangeSetName=change_set_name,
        ChangeSetType="CREATE",
        TemplateBody=json.dumps(template),
    )

    cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=res["StackId"]))

    with pytest.raises(WaiterError):
        aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
            ChangeSetName=res["Id"]
        )

    describe = aws_client.cloudformation.describe_change_set(ChangeSetName=res["Id"])
    snapshot.match("describe", describe)


@skip_if_legacy_engine()
@markers.aws.validated
def test_list_change_sets(deploy_cfn_template, aws_client, snapshot):
    snapshot.add_transformer(snapshot.transform.cloudformation_api())
    template = {
        "Resources": {
            "MyParameter": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": short_uid(),
                },
            },
        },
    }

    # first create an executed change set
    stack = deploy_cfn_template(template=json.dumps(template))
    stack_id = stack.stack_id

    # now create a non-executed change set
    template2 = copy.deepcopy(template)
    template2["Resources"]["MyParameter"]["Properties"]["Value"] = short_uid()

    non_executed_change_set_name = f"cs-{short_uid()}"
    non_executed_change_set_id = aws_client.cloudformation.create_change_set(
        ChangeSetName=non_executed_change_set_name,
        StackName=stack.stack_id,
        ChangeSetType="UPDATE",
        TemplateBody=json.dumps(template2),
    )["Id"]
    aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
        ChangeSetName=non_executed_change_set_id
    )

    # now create and delete a change set
    template3 = copy.deepcopy(template)
    template3["Resources"]["MyParameter"]["Properties"]["Value"] = short_uid()

    deleted_change_set_name = f"cs-{short_uid()}"
    deleted_change_set_id = aws_client.cloudformation.create_change_set(
        ChangeSetName=deleted_change_set_name,
        StackName=stack.stack_id,
        ChangeSetType="UPDATE",
        TemplateBody=json.dumps(template3),
    )["Id"]
    aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
        ChangeSetName=deleted_change_set_id
    )

    aws_client.cloudformation.delete_change_set(ChangeSetName=deleted_change_set_id)

    change_sets = aws_client.cloudformation.list_change_sets(StackName=stack_id)
    snapshot.match("change-sets", change_sets)
