import copy
import json
import os
from collections import OrderedDict
from itertools import permutations

import botocore.exceptions
import pytest
import yaml
from botocore.exceptions import ClientError, WaiterError
from localstack_snapshot.snapshots.transformer import SortingTransformer
from tests.aws.services.cloudformation.conftest import (
    skip_if_legacy_engine,
    skipped_v2_items,
)

from localstack.aws.api.cloudformation import Capability
from localstack.aws.connect import ServiceLevelClientFactory
from localstack.services.cloudformation.engine.entities import StackIdentifier
from localstack.services.cloudformation.engine.yaml_parser import parse_yaml
from localstack.testing.aws.util import is_aws_cloud
from localstack.testing.pytest import markers
from localstack.utils.files import load_file
from localstack.utils.strings import short_uid
from localstack.utils.sync import retry, wait_until


class TestStacksApi:
    @markers.snapshot.skip_snapshot_verify(
        paths=["$..ChangeSetId", "$..EnableTerminationProtection"]
        + skipped_v2_items(
            "$..Parameters",
        ),
    )
    @markers.aws.validated
    def test_stack_lifecycle(self, deploy_cfn_template, snapshot, aws_client):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())
        snapshot.add_transformer(snapshot.transform.key_value("ParameterValue", "parameter-value"))
        api_name = f"test_{short_uid()}"
        template_path = os.path.join(
            os.path.dirname(__file__), "../../../templates/simple_api.yaml"
        )

        deployed = deploy_cfn_template(
            template_path=template_path,
            parameters={"ApiName": api_name},
        )
        stack_name = deployed.stack_name
        creation_description = aws_client.cloudformation.describe_stacks(StackName=stack_name)[
            "Stacks"
        ][0]
        snapshot.match("creation", creation_description)

        api_name = f"test_{short_uid()}"
        deploy_cfn_template(
            is_update=True,
            stack_name=deployed.stack_name,
            template_path=template_path,
            parameters={"ApiName": api_name},
        )
        update_description = aws_client.cloudformation.describe_stacks(StackName=stack_name)[
            "Stacks"
        ][0]
        snapshot.match("update", update_description)

        aws_client.cloudformation.delete_stack(
            StackName=stack_name,
        )
        aws_client.cloudformation.get_waiter("stack_delete_complete").wait(StackName=stack_name)

        with pytest.raises(aws_client.cloudformation.exceptions.ClientError) as e:
            aws_client.cloudformation.describe_stacks(StackName=stack_name)
        snapshot.match("describe_deleted_by_name_exc", e.value.response)

        deleted = aws_client.cloudformation.describe_stacks(StackName=deployed.stack_id)["Stacks"][
            0
        ]
        assert "DeletionTime" in deleted
        snapshot.match("deleted", deleted)

    @markers.aws.validated
    def test_stack_description_special_chars(self, deploy_cfn_template, snapshot, aws_client):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        template = {
            "AWSTemplateFormatVersion": "2010-09-09",
            "Description": "test <env>.test.net",
            "Resources": {
                "TestResource": {
                    "Type": "AWS::EC2::VPC",
                    "Properties": {"CidrBlock": "100.30.20.0/20"},
                }
            },
        }
        deployed = deploy_cfn_template(template=json.dumps(template))
        response = aws_client.cloudformation.describe_stacks(StackName=deployed.stack_id)["Stacks"][
            0
        ]
        snapshot.match("describe_stack", response)

    @skip_if_legacy_engine()
    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$..ResourceChange.Details",
            "$..ResourceChange.Scope",
            "$..StackStatusReason",
        ]
    )
    @pytest.mark.parametrize(
        "tags", [None, [{"Key": "foo", "Value": "bar"}]], ids=["no-tags", "with-tags"]
    )
    def test_stack_description_lifecycle(self, snapshot, aws_client, cleanups, tags):
        """
        Test when and how the stack metadata gets set:
        * tags
        * description
        """
        snapshot.add_transformer(snapshot.transform.cloudformation_api())
        template = {
            "AWSTemplateFormatVersion": "2010-09-09",
            "Description": "test <env>.test.net",
            "Resources": {
                "TestResource": {
                    "Type": "AWS::SSM::Parameter",
                    "Properties": {
                        "Type": "String",
                        "Value": "DummyValue",
                    },
                }
            },
        }
        stack_name = f"stack-{short_uid()}"
        change_set_name = f"cs-{short_uid()}"

        kwargs = {
            "StackName": stack_name,
            "ChangeSetName": change_set_name,
            "ChangeSetType": "CREATE",
            "TemplateBody": json.dumps(template),
        }
        if tags is not None:
            kwargs["Tags"] = tags

        change_set = aws_client.cloudformation.create_change_set(**kwargs)
        change_set_id = change_set["Id"]
        stack_id = change_set["StackId"]

        aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
            ChangeSetName=change_set_id, StackName=stack_id
        )
        change_set_description = aws_client.cloudformation.describe_change_set(
            ChangeSetName=change_set_id
        )
        snapshot.match("change-set-pre-execute", change_set_description)
        stack_description = aws_client.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][
            0
        ]
        snapshot.match("stack-pre-execute", stack_description)

        aws_client.cloudformation.execute_change_set(ChangeSetName=change_set_id)
        cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=stack_id))

        aws_client.cloudformation.get_waiter("stack_create_complete").wait(StackName=stack_id)
        stack_description = aws_client.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][
            0
        ]
        snapshot.match("stack-post-execute", stack_description)

    @markers.aws.validated
    def test_stack_name_creation(self, deploy_cfn_template, snapshot, aws_client):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        stack_name = f"*@{short_uid()}_$"

        with pytest.raises(Exception) as e:
            deploy_cfn_template(
                template_path=os.path.join(
                    os.path.dirname(__file__), "../../../templates/sns_topic_template.yaml"
                ),
                stack_name=stack_name,
            )

            snapshot.match("stack_response", e.value.response)

    @markers.aws.validated
    def test_create_stack_url_as_template(self, snapshot, aws_client):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        stack_name = f"stack-{short_uid()}"

        template_url = "https://raw.githubusercontent.com/aws-cloudformation/aws-cloudformation-templates/refs/heads/main/EC2/InstanceWithCfnInit.yaml"

        with pytest.raises(ClientError) as e:
            aws_client.cloudformation.create_stack(StackName=stack_name, TemplateBody=template_url)
        snapshot.match("stack_create_ec2", e.value)

    @markers.aws.validated
    @pytest.mark.parametrize("fileformat", ["yaml", "json"])
    def test_get_template_using_create_stack(self, snapshot, fileformat, aws_client):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        stack_name = f"stack-{short_uid()}"
        aws_client.cloudformation.create_stack(
            StackName=stack_name,
            TemplateBody=load_file(
                os.path.join(
                    os.path.dirname(__file__), f"../../../templates/sns_topic_template.{fileformat}"
                )
            ),
        )
        aws_client.cloudformation.get_waiter("stack_create_complete").wait(StackName=stack_name)

        template_original = aws_client.cloudformation.get_template(
            StackName=stack_name, TemplateStage="Original"
        )
        snapshot.match("template_original", template_original)

        template_processed = aws_client.cloudformation.get_template(
            StackName=stack_name, TemplateStage="Processed"
        )
        snapshot.match("template_processed", template_processed)

    @markers.aws.validated
    @pytest.mark.parametrize("fileformat", ["yaml", "json"])
    def test_get_template_using_changesets(
        self, deploy_cfn_template, snapshot, fileformat, aws_client
    ):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        stack = deploy_cfn_template(
            template_path=os.path.join(
                os.path.dirname(__file__), f"../../../templates/sns_topic_template.{fileformat}"
            )
        )

        template_original = aws_client.cloudformation.get_template(
            StackName=stack.stack_id, TemplateStage="Original"
        )
        snapshot.match("template_original", template_original)

        template_processed = aws_client.cloudformation.get_template(
            StackName=stack.stack_id, TemplateStage="Processed"
        )
        snapshot.match("template_processed", template_processed)

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=["$..ParameterValue", "$..PhysicalResourceId", "$..Capabilities"]
    )
    def test_stack_update_resources(
        self,
        deploy_cfn_template,
        is_change_set_finished,
        is_change_set_created_and_available,
        snapshot,
        aws_client,
    ):
        snapshot.add_transformer(snapshot.transform.cloudformation_api())
        snapshot.add_transformer(snapshot.transform.key_value("PhysicalResourceId"))

        api_name = f"test_{short_uid()}"

        # create stack
        deployed = deploy_cfn_template(
            template_path=os.path.join(
                os.path.dirname(__file__), "../../../templates/simple_api.yaml"
            ),
            parameters={"ApiName": api_name},
        )
        stack_name = deployed.stack_name
        stack_id = deployed.stack_id

        # assert snapshot of created stack
        snapshot.match(
            "stack_created",
            aws_client.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][0],
        )

        # update stack, with one additional resource
        api_name = f"test_{short_uid()}"
        deploy_cfn_template(
            is_update=True,
            stack_name=deployed.stack_name,
            template_path=os.path.join(
                os.path.dirname(__file__), "../../../templates/simple_api.update.yaml"
            ),
            parameters={"ApiName": api_name},
        )

        # assert snapshot of updated stack
        snapshot.match(
            "stack_updated",
            aws_client.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][0],
        )

        # describe stack resources
        resources = aws_client.cloudformation.describe_stack_resources(StackName=stack_name)
        snapshot.match("stack_resources", resources)

    @markers.aws.validated
    def test_update_stack_with_same_template_withoutchange(
        self, deploy_cfn_template, aws_client, snapshot
    ):
        template = load_file(
            os.path.join(os.path.dirname(__file__), "../../../templates/simple_no_change.yaml")
        )
        stack = deploy_cfn_template(template=template)

        with pytest.raises(Exception) as ctx:  # TODO: capture proper exception
            aws_client.cloudformation.update_stack(
                StackName=stack.stack_name, TemplateBody=template
            )
            aws_client.cloudformation.get_waiter("stack_update_complete").wait(
                StackName=stack.stack_name
            )

        snapshot.match("no_change_exception", ctx.value.response)

    @markers.aws.validated
    def test_update_stack_with_same_template_withoutchange_transformation(
        self, deploy_cfn_template, aws_client
    ):
        template = load_file(
            os.path.join(
                os.path.dirname(__file__),
                "../../../templates/simple_no_change_with_transformation.yaml",
            )
        )
        stack = deploy_cfn_template(template=template)

        # transformations will always work even if there's no change in the template!
        aws_client.cloudformation.update_stack(
            StackName=stack.stack_name,
            TemplateBody=template,
            Capabilities=["CAPABILITY_AUTO_EXPAND"],
        )
        aws_client.cloudformation.get_waiter("stack_update_complete").wait(
            StackName=stack.stack_name
        )

    @markers.aws.validated
    @skip_if_legacy_engine()
    def test_update_stack_actual_update(self, deploy_cfn_template, aws_client):
        template = load_file(
            os.path.join(os.path.dirname(__file__), "../../../templates/sqs_queue_update.yml")
        )
        queue_name = f"test-queue-{short_uid()}"
        stack = deploy_cfn_template(
            template=template, parameters={"QueueName": queue_name}, max_wait=360
        )

        queue_arn_1 = aws_client.sqs.get_queue_attributes(
            QueueUrl=stack.outputs["QueueUrl"], AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]
        assert queue_arn_1

        stack2 = deploy_cfn_template(
            template=template,
            stack_name=stack.stack_name,
            parameters={"QueueName": f"{queue_name}-new"},
            is_update=True,
            max_wait=360,
        )

        queue_arn_2 = aws_client.sqs.get_queue_attributes(
            QueueUrl=stack2.outputs["QueueUrl"], AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]
        assert queue_arn_2

        assert queue_arn_1 != queue_arn_2

    @markers.snapshot.skip_snapshot_verify(paths=["$..StackEvents"])
    @markers.aws.validated
    def test_list_events_after_deployment(self, deploy_cfn_template, snapshot, aws_client):
        snapshot.add_transformer(SortingTransformer("StackEvents", lambda x: x["Timestamp"]))
        snapshot.add_transformer(snapshot.transform.cloudformation_api())
        stack = deploy_cfn_template(
            template_path=os.path.join(
                os.path.dirname(__file__), "../../../templates/sns_topic_simple.yaml"
            )
        )
        response = aws_client.cloudformation.describe_stack_events(StackName=stack.stack_name)
        snapshot.match("events", response)

    @markers.aws.validated
    @pytest.mark.skip(reason="disable rollback not supported")
    @pytest.mark.parametrize("rollback_disabled, length_expected", [(False, 0), (True, 1)])
    def test_failure_options_for_stack_creation(
        self, rollback_disabled, length_expected, aws_client
    ):
        template_with_error = open(
            os.path.join(os.path.dirname(__file__), "../../../templates/multiple_bucket.yaml")
        ).read()

        stack_name = f"stack-{short_uid()}"
        bucket_1_name = f"bucket-{short_uid()}"
        bucket_2_name = f"bucket!#${short_uid()}"

        aws_client.cloudformation.create_stack(
            StackName=stack_name,
            TemplateBody=template_with_error,
            DisableRollback=rollback_disabled,
            Parameters=[
                {"ParameterKey": "BucketName1", "ParameterValue": bucket_1_name},
                {"ParameterKey": "BucketName2", "ParameterValue": bucket_2_name},
            ],
        )

        assert wait_until(
            lambda _: stack_process_is_finished(aws_client.cloudformation, stack_name),
            wait=10,
            strategy="exponential",
        )

        resources = aws_client.cloudformation.describe_stack_resources(StackName=stack_name)[
            "StackResources"
        ]
        created_resources = [
            resource for resource in resources if "CREATE_COMPLETE" in resource["ResourceStatus"]
        ]
        assert len(created_resources) == length_expected

        aws_client.cloudformation.delete_stack(StackName=stack_name)

    @markers.aws.validated
    @pytest.mark.skipif(reason="disable rollback not enabled", condition=not is_aws_cloud())
    @pytest.mark.parametrize("rollback_disabled, length_expected", [(False, 2), (True, 1)])
    def test_failure_options_for_stack_update(
        self, rollback_disabled, length_expected, aws_client, cleanups
    ):
        stack_name = f"stack-{short_uid()}"
        template = open(
            os.path.join(
                os.path.dirname(__file__), "../../../templates/multiple_bucket_update.yaml"
            ),
        ).read()

        aws_client.cloudformation.create_stack(
            StackName=stack_name,
            TemplateBody=template,
        )
        cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=stack_name))

        def _assert_stack_process_finished():
            return stack_process_is_finished(aws_client.cloudformation, stack_name)

        assert wait_until(_assert_stack_process_finished)
        resources = aws_client.cloudformation.describe_stack_resources(StackName=stack_name)[
            "StackResources"
        ]
        created_resources = [
            resource for resource in resources if "CREATE_COMPLETE" in resource["ResourceStatus"]
        ]
        assert len(created_resources) == 2

        aws_client.cloudformation.update_stack(
            StackName=stack_name,
            TemplateBody=template,
            DisableRollback=rollback_disabled,
            Parameters=[
                {"ParameterKey": "Days", "ParameterValue": "-1"},
            ],
        )

        assert wait_until(_assert_stack_process_finished)

        resources = aws_client.cloudformation.describe_stack_resources(StackName=stack_name)[
            "StackResources"
        ]
        updated_resources = [
            resource
            for resource in resources
            if resource["ResourceStatus"] in ["CREATE_COMPLETE", "UPDATE_COMPLETE"]
        ]
        assert len(updated_resources) == length_expected

    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_create_stack_with_custom_id(
        self, aws_client, cleanups, account_id, region_name, set_resource_custom_id
    ):
        stack_name = f"stack-{short_uid()}"
        custom_id = short_uid()

        set_resource_custom_id(
            StackIdentifier(account_id, region_name, stack_name), custom_id=custom_id
        )
        template = open(
            os.path.join(os.path.dirname(__file__), "../../../templates/sns_topic_simple.yaml"),
        ).read()

        stack = aws_client.cloudformation.create_stack(
            StackName=stack_name,
            TemplateBody=template,
        )
        cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=stack_name))

        assert stack["StackId"].split("/")[-1] == custom_id

        # We need to wait until the stack is created otherwise we can end up in a scenario
        # where we try to delete the stack before creating its resources, failing the test
        aws_client.cloudformation.get_waiter("stack_create_complete").wait(StackName=stack_name)


def stack_process_is_finished(cfn_client, stack_name):
    return (
        "PROGRESS"
        not in cfn_client.describe_stacks(StackName=stack_name)["Stacks"][0]["StackStatus"]
    )


@markers.aws.validated
@pytest.mark.skip(reason="Not Implemented")
def test_linting_error_during_creation(snapshot, aws_client):
    stack_name = f"stack-{short_uid()}"
    bad_template = {"Resources": "", "Outputs": ""}

    with pytest.raises(botocore.exceptions.ClientError) as ex:
        aws_client.cloudformation.create_stack(
            StackName=stack_name, TemplateBody=json.dumps(bad_template)
        )

    error_response = ex.value.response
    snapshot.match("error", error_response)


@markers.aws.validated
@pytest.mark.skip(reason="feature not implemented")
def test_notifications(
    deploy_cfn_template,
    sns_create_topic,
    is_stack_created,
    is_stack_updated,
    sqs_create_queue,
    sns_create_sqs_subscription,
    cleanup_stacks,
    aws_client,
):
    stack_name = f"stack-{short_uid()}"
    topic_arn = sns_create_topic()["TopicArn"]
    sqs_url = sqs_create_queue()
    sns_create_sqs_subscription(topic_arn, sqs_url)

    template = load_file(
        os.path.join(os.path.dirname(__file__), "../../../templates/sns_topic_parameter.yml")
    )
    aws_client.cloudformation.create_stack(
        StackName=stack_name,
        NotificationARNs=[topic_arn],
        TemplateBody=template,
        Parameters=[{"ParameterKey": "TopicName", "ParameterValue": f"topic-{short_uid()}"}],
    )
    cleanup_stacks([stack_name])

    assert wait_until(is_stack_created(stack_name))

    template = load_file(
        os.path.join(os.path.dirname(__file__), "../../../templates/sns_topic_parameter.yml")
    )
    aws_client.cloudformation.update_stack(
        StackName=stack_name,
        TemplateBody=template,
        Parameters=[
            {"ParameterKey": "TopicName", "ParameterValue": f"topic-{short_uid()}"},
        ],
    )
    assert wait_until(is_stack_updated(stack_name))

    messages = {}

    def _assert_messages():
        sqs_messages = aws_client.sqs.receive_message(QueueUrl=sqs_url)["Messages"]
        for sqs_message in sqs_messages:
            sns_message = json.loads(sqs_message["Body"])
            messages.update({sns_message["MessageId"]: sns_message})

        # Assert notifications of resources created
        assert [message for message in messages.values() if "CREATE_" in message["Message"]]

        # Assert notifications of resources deleted
        assert [message for message in messages.values() if "UPDATE_" in message["Message"]]

        # Assert notifications of resources deleted
        assert [message for message in messages.values() if "DELETE_" in message["Message"]]

    retry(_assert_messages, retries=10, sleep=2)


@markers.aws.validated
@markers.snapshot.skip_snapshot_verify(
    paths=[
        # parameters may be out of order
        "$..Stacks..Parameters",
    ]
)
def test_updating_an_updated_stack_sets_status(deploy_cfn_template, snapshot, aws_client):
    """
    The status of a stack that has been updated twice should be "UPDATE_COMPLETE"
    """
    snapshot.add_transformer(snapshot.transform.cloudformation_api())

    # need multiple templates to support updates to the stack
    template_1 = load_file(
        os.path.join(os.path.dirname(__file__), "../../../templates/stack_update_1.yaml")
    )
    template_2 = load_file(
        os.path.join(os.path.dirname(__file__), "../../../templates/stack_update_2.yaml")
    )
    template_3 = load_file(
        os.path.join(os.path.dirname(__file__), "../../../templates/stack_update_3.yaml")
    )

    topic_1_name = f"topic-1-{short_uid()}"
    topic_2_name = f"topic-2-{short_uid()}"
    topic_3_name = f"topic-3-{short_uid()}"
    snapshot.add_transformers_list(
        [
            snapshot.transform.regex(topic_1_name, "topic-1"),
            snapshot.transform.regex(topic_2_name, "topic-2"),
            snapshot.transform.regex(topic_3_name, "topic-3"),
        ]
    )

    parameters = {
        "Topic1Name": topic_1_name,
        "Topic2Name": topic_2_name,
        "Topic3Name": topic_3_name,
    }

    def wait_for(waiter_type: str) -> None:
        aws_client.cloudformation.get_waiter(waiter_type).wait(
            StackName=stack.stack_name,
            WaiterConfig={
                "Delay": 5,
                "MaxAttempts": 5,
            },
        )

    stack = deploy_cfn_template(template=template_1, parameters=parameters)
    wait_for("stack_create_complete")

    # update the stack
    deploy_cfn_template(
        template=template_2,
        is_update=True,
        stack_name=stack.stack_name,
        parameters=parameters,
    )
    wait_for("stack_update_complete")

    # update the stack again
    deploy_cfn_template(
        template=template_3,
        is_update=True,
        stack_name=stack.stack_name,
        parameters=parameters,
    )
    wait_for("stack_update_complete")

    res = aws_client.cloudformation.describe_stacks(StackName=stack.stack_name)
    snapshot.match("describe-result", res)


@markers.aws.validated
def test_update_termination_protection(deploy_cfn_template, snapshot, aws_client):
    snapshot.add_transformer(snapshot.transform.cloudformation_api())
    snapshot.add_transformer(snapshot.transform.key_value("ParameterValue", "parameter-value"))

    # create stack
    api_name = f"test_{short_uid()}"
    template_path = os.path.join(os.path.dirname(__file__), "../../../templates/simple_api.yaml")
    stack = deploy_cfn_template(template_path=template_path, parameters={"ApiName": api_name})

    # update termination protection (true)
    aws_client.cloudformation.update_termination_protection(
        EnableTerminationProtection=True, StackName=stack.stack_name
    )
    res = aws_client.cloudformation.describe_stacks(StackName=stack.stack_name)
    snapshot.match("describe-stack-1", res)

    # update termination protection (false)
    aws_client.cloudformation.update_termination_protection(
        EnableTerminationProtection=False, StackName=stack.stack_name
    )
    res = aws_client.cloudformation.describe_stacks(StackName=stack.stack_name)
    snapshot.match("describe-stack-2", res)


@markers.aws.validated
def test_events_resource_types(deploy_cfn_template, snapshot, aws_client):
    template_path = os.path.join(
        os.path.dirname(__file__), "../../../templates/cfn_cdk_sample_app.yaml"
    )
    stack = deploy_cfn_template(template_path=template_path, max_wait=500)
    events = aws_client.cloudformation.describe_stack_events(StackName=stack.stack_name)[
        "StackEvents"
    ]

    resource_types = list({event["ResourceType"] for event in events})
    resource_types.sort()
    snapshot.match("resource_types", resource_types)


@markers.aws.validated
def test_list_parameter_type(aws_client, deploy_cfn_template, cleanups):
    stack = deploy_cfn_template(
        template_path=os.path.join(
            os.path.dirname(__file__), "../../../templates/cfn_parameter_list_type.yaml"
        ),
        parameters={
            "ParamsList": "foo,bar",
        },
    )

    assert stack.outputs["ParamValue"] == "foo|bar"


@markers.aws.validated
def test_subnet_id_parameter_type(aws_client, deploy_cfn_template, cleanups, snapshot):
    vpc_id = aws_client.ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
    cleanups.append(lambda: aws_client.ec2.delete_vpc(VpcId=vpc_id))
    aws_client.ec2.get_waiter("vpc_available").wait(VpcIds=[vpc_id])
    subnet_id_1 = aws_client.ec2.create_subnet(VpcId=vpc_id, CidrBlock="10.0.0.0/24")["Subnet"][
        "SubnetId"
    ]
    cleanups.append(lambda: aws_client.ec2.delete_subnet(SubnetId=subnet_id_1))
    subnet_id_2 = aws_client.ec2.create_subnet(VpcId=vpc_id, CidrBlock="10.0.1.0/24")["Subnet"][
        "SubnetId"
    ]
    cleanups.append(lambda: aws_client.ec2.delete_subnet(SubnetId=subnet_id_2))

    subnets_list = ",".join([subnet_id_1, subnet_id_2])
    stack = deploy_cfn_template(
        template_path=os.path.join(
            os.path.dirname(__file__), "../../../templates/cfn_parameter_list_subnet_id_type.yaml"
        ),
        parameters={
            "ParamsList": subnets_list,
        },
    )

    snapshot.add_transformer(snapshot.transform.regex(subnet_id_1, "subnet-id-1"))
    snapshot.add_transformer(snapshot.transform.regex(subnet_id_2, "subnet-id-2"))
    snapshot.match("outputs", stack.outputs)


@markers.aws.validated
@pytest.mark.skipif(condition=not is_aws_cloud(), reason="rollback not implemented")
def test_blocked_stack_deletion(aws_client, cleanups, snapshot):
    """
    uses AWS::IAM::Policy for demonstrating this behavior

    1. create fails
    2. rollback fails even though create didn't even provision anything
    3. trying to delete the stack afterwards also doesn't work
    4. deleting the stack with retain resources works
    """
    cfn = aws_client.cloudformation
    stack_name = f"test-stacks-blocked-{short_uid()}"
    policy_name = f"test-broken-policy-{short_uid()}"
    snapshot.add_transformer(snapshot.transform.cloudformation_api())
    snapshot.add_transformer(snapshot.transform.regex(policy_name, "<policy-name>"))
    template_body = load_file(
        os.path.join(os.path.dirname(__file__), "../../../templates/iam_policy_invalid.yaml")
    )
    waiter_config = {"Delay": 1, "MaxAttempts": 20}

    snapshot.add_transformer(snapshot.transform.key_value("PhysicalResourceId"))
    snapshot.add_transformer(
        snapshot.transform.key_value("ResourceStatusReason", reference_replacement=False)
    )

    stack = cfn.create_stack(
        StackName=stack_name,
        TemplateBody=template_body,
        Parameters=[{"ParameterKey": "Name", "ParameterValue": policy_name}],
        Capabilities=[Capability.CAPABILITY_NAMED_IAM],
    )
    stack_id = stack["StackId"]
    cleanups.append(lambda: cfn.delete_stack(StackName=stack_id, RetainResources=["BrokenPolicy"]))
    with pytest.raises(WaiterError):
        cfn.get_waiter("stack_create_complete").wait(StackName=stack_id, WaiterConfig=waiter_config)
    stack_post_create = cfn.describe_stacks(StackName=stack_id)
    snapshot.match("stack_post_create", stack_post_create)

    cfn.delete_stack(StackName=stack_id)
    with pytest.raises(WaiterError):
        cfn.get_waiter("stack_delete_complete").wait(StackName=stack_id, WaiterConfig=waiter_config)
    stack_post_fail_delete = cfn.describe_stacks(StackName=stack_id)
    snapshot.match("stack_post_fail_delete", stack_post_fail_delete)

    cfn.delete_stack(StackName=stack_id, RetainResources=["BrokenPolicy"])
    cfn.get_waiter("stack_delete_complete").wait(StackName=stack_id, WaiterConfig=waiter_config)
    stack_post_success_delete = cfn.describe_stacks(StackName=stack_id)
    snapshot.match("stack_post_success_delete", stack_post_success_delete)
    stack_events = cfn.describe_stack_events(StackName=stack_id)
    snapshot.match("stack_events", stack_events)


MINIMAL_TEMPLATE = """
Resources:
    SimpleParam:
        Type: AWS::SSM::Parameter
        Properties:
            Value: test
            Type: String
"""


@markers.snapshot.skip_snapshot_verify(
    paths=["$..EnableTerminationProtection", "$..LastUpdatedTime"]
    + skipped_v2_items("$..Capabilities")
)
@markers.aws.validated
def test_name_conflicts(aws_client, snapshot, cleanups):
    """
    Tests behavior of creating a stack with the same name of one that was previously deleted

    1. Create Stack
    2. Delete Stack
    3. Create Stack with same name as in 1.

    Step 3 should be successful because you can re-use StackNames,
    but only one stack for a given stack name can be `ACTIVE` at one time.

    We didn't exhaustively test yet what is considered as Active by CloudFormation
    For now the assumption is that anything != "DELETE_COMPLETED" is considered "ACTIVE"
    """
    snapshot.add_transformer(snapshot.transform.cloudformation_api())

    stack_name = f"repeated-stack-{short_uid()}"
    cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=stack_name))
    stack = aws_client.cloudformation.create_stack(
        StackName=stack_name, TemplateBody=MINIMAL_TEMPLATE
    )
    stack_id = stack["StackId"]
    aws_client.cloudformation.get_waiter("stack_create_complete").wait(StackName=stack_name)

    # only one can be active at a time
    with pytest.raises(aws_client.cloudformation.exceptions.AlreadyExistsException) as e:
        aws_client.cloudformation.create_stack(StackName=stack_name, TemplateBody=MINIMAL_TEMPLATE)
    snapshot.match("create_stack_already_exists_exc", e.value.response)

    created_stack_desc = aws_client.cloudformation.describe_stacks(StackName=stack_name)["Stacks"][
        0
    ]["StackStatus"]
    snapshot.match("created_stack_desc", created_stack_desc)

    aws_client.cloudformation.delete_stack(StackName=stack_name)
    aws_client.cloudformation.get_waiter("stack_delete_complete").wait(StackName=stack_name)

    # describe with name fails
    with pytest.raises(aws_client.cloudformation.exceptions.ClientError) as e:
        aws_client.cloudformation.describe_stacks(StackName=stack_name)
    snapshot.match("deleted_stack_not_found_exc", e.value.response)

    # describe events with name fails
    with pytest.raises(aws_client.cloudformation.exceptions.ClientError) as e:
        aws_client.cloudformation.describe_stack_events(StackName=stack_name)
    snapshot.match("deleted_stack_events_not_found_by_name", e.value.response)

    # describe with stack id (ARN) succeeds
    deleted_stack_desc = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("deleted_stack_desc", deleted_stack_desc)

    # creating a new stack with the same name as the previously deleted one should work
    stack = aws_client.cloudformation.create_stack(
        StackName=stack_name, TemplateBody=MINIMAL_TEMPLATE
    )
    # should issue a new unique stack ID/ARN
    new_stack_id = stack["StackId"]
    assert stack_id != new_stack_id
    aws_client.cloudformation.get_waiter("stack_create_complete").wait(StackName=stack_name)

    new_stack_desc = aws_client.cloudformation.describe_stacks(StackName=stack_name)
    snapshot.match("new_stack_desc", new_stack_desc)
    assert len(new_stack_desc["Stacks"]) == 1
    assert new_stack_desc["Stacks"][0]["StackId"] == new_stack_id

    # can still access both by using the ARN (stack id)
    # and they should be different from each other
    stack_id_desc = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    new_stack_id_desc = aws_client.cloudformation.describe_stacks(StackName=new_stack_id)
    snapshot.match("stack_id_desc", stack_id_desc)
    snapshot.match("new_stack_id_desc", new_stack_id_desc)

    # check if the describing the stack events return the right stack
    stack_events = aws_client.cloudformation.describe_stack_events(StackName=stack_name)[
        "StackEvents"
    ]
    assert all(stack_event["StackId"] == new_stack_id for stack_event in stack_events)
    # describing events by the old stack id should still yield the old events
    stack_events = aws_client.cloudformation.describe_stack_events(StackName=stack_id)[
        "StackEvents"
    ]
    assert all(stack_event["StackId"] == stack_id for stack_event in stack_events)

    # deleting the stack by name should delete the new, not already deleted stack
    aws_client.cloudformation.delete_stack(StackName=stack_name)
    aws_client.cloudformation.get_waiter("stack_delete_complete").wait(StackName=stack_name)
    # describe with stack id returns stack deleted
    deleted_stack_desc = aws_client.cloudformation.describe_stacks(StackName=new_stack_id)
    snapshot.match("deleted_second_stack_desc", deleted_stack_desc)


@markers.aws.validated
def test_describe_stack_events_errors(aws_client, snapshot):
    with pytest.raises(botocore.exceptions.ParamValidationError) as e:
        aws_client.cloudformation.describe_stack_events()
    with pytest.raises(aws_client.cloudformation.exceptions.ClientError) as e:
        aws_client.cloudformation.describe_stack_events(StackName="does-not-exist")
    snapshot.match("describe_stack_events_stack_not_found", e.value.response)


TEMPLATE_ORDER_CASES = list(permutations(["A", "B", "C"]))


@markers.aws.validated
@markers.snapshot.skip_snapshot_verify(
    paths=[
        "$..StackId",
        # TODO
        "$..PhysicalResourceId",
        # TODO
        "$..ResourceProperties",
    ]
)
@pytest.mark.parametrize(
    "deploy_order", TEMPLATE_ORDER_CASES, ids=["-".join(vals) for vals in TEMPLATE_ORDER_CASES]
)
def test_stack_deploy_order(deploy_cfn_template, aws_client, snapshot, deploy_order: tuple[str]):
    snapshot.add_transformer(snapshot.transform.cloudformation_api())
    snapshot.add_transformer(snapshot.transform.key_value("EventId"))
    resources = {
        "A": {
            "Type": "AWS::SSM::Parameter",
            "Properties": {
                "Type": "String",
                "Value": "root",
            },
        },
        "B": {
            "Type": "AWS::SSM::Parameter",
            "Properties": {
                "Type": "String",
                "Value": {
                    "Ref": "A",
                },
            },
        },
        "C": {
            "Type": "AWS::SSM::Parameter",
            "Properties": {
                "Type": "String",
                "Value": {
                    "Ref": "B",
                },
            },
        },
    }

    resources = OrderedDict(
        [
            (logical_resource_id, resources[logical_resource_id])
            for logical_resource_id in deploy_order
        ]
    )
    assert len(resources) == 3

    stack = deploy_cfn_template(
        template=json.dumps(
            {
                "Resources": resources,
            }
        )
    )

    stack.destroy()

    events = aws_client.cloudformation.describe_stack_events(
        StackName=stack.stack_id,
    )["StackEvents"]

    filtered_events = []
    for event in events:
        # only the resources we care about
        if event["LogicalResourceId"] not in deploy_order:
            continue

        # only _COMPLETE events
        if not event["ResourceStatus"].endswith("_COMPLETE"):
            continue

        filtered_events.append(event)

    # sort by event time
    filtered_events.sort(key=lambda e: e["Timestamp"])

    snapshot.match("events", filtered_events)


@skip_if_legacy_engine()
@markers.aws.validated
@pytest.mark.parametrize(
    "deletions",
    [
        pytest.param(["C"], id="C"),
        pytest.param(["B", "C"], id="B-C"),
        pytest.param(["A", "B", "C"], id="A-B-C"),
    ],
)
@markers.snapshot.skip_snapshot_verify(
    paths=[
        "delete-describe.ChangeSetId",
        "$..EnableTerminationProtection",
        #
        # Before/After Context
        "$..Capabilities",
        "$..IncludeNestedStacks",
        "$..Scope",
        "$..Details",
        "$..Parameters",
        "$..PolicyAction",
        "$..PhysicalResourceId",
        "$..Changes..ResourceChange.BeforeContext.Properties.Value",
        "$..StackEvents..EventId",
        "$..StackEvents..ResourceStatusReason",
        "$..StackEvents..ResourceProperties.Value",
        "all-events..EventId",
    ]
)
def test_stack_deletion_order(
    aws_client, capture_update_process, capture_resource_state_changes, snapshot, deletions
):
    t1 = {
        "Resources": {
            "Dummy": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": "dummy",
                },
            },
            "A": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": "root",
                },
                "DependsOn": ["Dummy"],
            },
            "B": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": {
                        "Ref": "A",
                    },
                },
            },
            "C": {
                "Type": "AWS::SSM::Parameter",
                "Properties": {
                    "Type": "String",
                    "Value": {
                        "Ref": "B",
                    },
                },
            },
        }
    }
    t2 = copy.deepcopy(t1)
    for deletion in deletions:
        del t2["Resources"][deletion]

    stack_id = capture_update_process(snapshot, t1, t2)

    # since resource deployments are serializable, we can capture events and check parity with them
    events = list(capture_resource_state_changes(stack_id))
    to_snapshot = [(every["LogicalResourceId"], every["ResourceStatus"]) for every in events[::-1]]
    snapshot.match("all-events", to_snapshot)


@markers.snapshot.skip_snapshot_verify(
    paths=[
        # TODO: this property is present in the response from LocalStack when
        # there is an active changeset, however it is not present on AWS
        # because the change set has not been executed.
        "$..Stacks..ChangeSetId",
        # FIXME: tackle this when fixing API parity of CloudFormation
        "$..Capabilities",
        "$..IncludeNestedStacks",
        "$..LastUpdatedTime",
        "$..NotificationARNs",
        "$..ResourceChange",
        "$..StackResourceDetail.Metadata",
    ]
    + skipped_v2_items(
        "$..Stacks..Outputs..Description",
        "$..StackResourceDetail.DriftInformation",
    )
)
@markers.aws.validated
def test_no_echo_parameter(snapshot, aws_client, deploy_cfn_template):
    snapshot.add_transformer(snapshot.transform.cloudformation_api())
    snapshot.add_transformer(SortingTransformer("Parameters", lambda x: x.get("ParameterKey", "")))

    template_path = os.path.join(os.path.dirname(__file__), "../../../templates/cfn_no_echo.yml")
    template = open(template_path).read()

    deployment = deploy_cfn_template(
        template=template,
        parameters={"SecretParameter": "SecretValue"},
    )
    stack_id = deployment.stack_id
    stack_name = deployment.stack_name

    describe_stacks = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("describe_stacks", describe_stacks)

    # Check Resource Metadata.
    describe_stack_resources = aws_client.cloudformation.describe_stack_resources(
        StackName=stack_id
    )
    for resource in describe_stack_resources["StackResources"]:
        resource_logical_id = resource["LogicalResourceId"]

        # Get detailed information about the resource
        describe_stack_resource_details = aws_client.cloudformation.describe_stack_resource(
            StackName=stack_name, LogicalResourceId=resource_logical_id
        )
        snapshot.match(
            f"describe_stack_resource_details_{resource_logical_id}",
            describe_stack_resource_details,
        )

    # create a change set
    change_set_name = f"CreateChangeSetSecretParameterValue-{short_uid()}"
    aws_client.cloudformation.create_change_set(
        StackName=stack_name,
        TemplateBody=template,
        ChangeSetName=change_set_name,
        Parameters=[
            {"ParameterKey": "SecretParameter", "ParameterValue": "NewSecretValue1"},
        ],
    )
    aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
        StackName=stack_name,
        ChangeSetName=change_set_name,
    )
    change_sets = aws_client.cloudformation.describe_change_set(
        StackName=stack_id,
        ChangeSetName=change_set_name,
    )
    snapshot.match("describe_change_set_on_create_complete", change_sets)
    describe_stacks = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("describe_stack_on_create_complete", describe_stacks)

    # Update stack via update_stack (and change the value of SecretParameter)
    aws_client.cloudformation.update_stack(
        StackName=stack_name,
        TemplateBody=template,
        Parameters=[
            {"ParameterKey": "SecretParameter", "ParameterValue": "NewSecretValue1"},
        ],
    )
    aws_client.cloudformation.get_waiter("stack_update_complete").wait(StackName=stack_name)
    update_stacks = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("describe_updated_stacks", update_stacks)

    # Update stack via create_change_set (and change the value of SecretParameter)
    change_set_name = f"UpdateSecretParameterValue-{short_uid()}"
    aws_client.cloudformation.create_change_set(
        StackName=stack_name,
        TemplateBody=template,
        ChangeSetName=change_set_name,
        Parameters=[
            {"ParameterKey": "SecretParameter", "ParameterValue": "NewSecretValue2"},
        ],
    )
    aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
        StackName=stack_name,
        ChangeSetName=change_set_name,
    )
    change_sets = aws_client.cloudformation.describe_change_set(
        StackName=stack_id,
        ChangeSetName=change_set_name,
    )
    snapshot.match("describe_updated_change_set", change_sets)
    describe_stacks = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("describe_updated_stacks_change_set", describe_stacks)

    # Change `NoEcho` of a parameter from true to false and update stack via create_change_set.
    change_set_name = f"UpdateSecretParameterNoEchoToFalse-{short_uid()}"
    template_dict = parse_yaml(load_file(template_path))
    template_dict["Parameters"]["SecretParameter"]["NoEcho"] = False
    template_no_echo_false = yaml.dump(template_dict)
    aws_client.cloudformation.create_change_set(
        StackName=stack_name,
        TemplateBody=template_no_echo_false,
        ChangeSetName=change_set_name,
        Parameters=[
            {"ParameterKey": "SecretParameter", "ParameterValue": "NewSecretValue2"},
        ],
    )
    aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
        StackName=stack_name,
        ChangeSetName=change_set_name,
    )
    change_sets = aws_client.cloudformation.describe_change_set(
        StackName=stack_id,
        ChangeSetName=change_set_name,
    )
    snapshot.match("describe_updated_change_set_no_echo_true", change_sets)
    describe_stacks = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("describe_updated_stacks_no_echo_true", describe_stacks)

    # Change `NoEcho` of a parameter back from false to true and update stack via create_change_set.
    change_set_name = f"UpdateSecretParameterNoEchoToTrue-{short_uid()}"
    aws_client.cloudformation.create_change_set(
        StackName=stack_name,
        TemplateBody=template,
        ChangeSetName=change_set_name,
        Parameters=[
            {"ParameterKey": "SecretParameter", "ParameterValue": "NewSecretValue2"},
        ],
    )
    aws_client.cloudformation.get_waiter("change_set_create_complete").wait(
        StackName=stack_name,
        ChangeSetName=change_set_name,
    )
    change_sets = aws_client.cloudformation.describe_change_set(
        StackName=stack_id,
        ChangeSetName=change_set_name,
    )
    snapshot.match("describe_updated_change_set_no_echo_false", change_sets)
    describe_stacks = aws_client.cloudformation.describe_stacks(StackName=stack_id)
    snapshot.match("describe_updated_stacks_no_echo_false", describe_stacks)


@markers.aws.validated
def test_stack_resource_not_found(deploy_cfn_template, aws_client, snapshot):
    stack = deploy_cfn_template(
        template_path=os.path.join(
            os.path.dirname(__file__), "../../../templates/sns_topic_simple.yaml"
        ),
        parameters={"TopicName": f"topic{short_uid()}"},
    )

    with pytest.raises(botocore.exceptions.ClientError) as ex:
        aws_client.cloudformation.describe_stack_resource(
            StackName=stack.stack_name, LogicalResourceId="NonExistentResource"
        )

    snapshot.add_transformer(snapshot.transform.regex(stack.stack_name, "<stack-name>"))
    snapshot.match("Error", ex.value.response)


@markers.aws.validated
def test_non_existing_stack_message(aws_client, snapshot):
    with pytest.raises(botocore.exceptions.ClientError) as ex:
        aws_client.cloudformation.describe_stacks(StackName="non-existing")

    snapshot.add_transformer(snapshot.transform.regex("non-existing", "<stack-name>"))
    snapshot.match("Error", ex.value.response)


@skip_if_legacy_engine()
@markers.aws.validated
def test_no_parameters_given(aws_client, deploy_cfn_template, snapshot):
    template_path = os.path.join(
        os.path.dirname(__file__), "../../../templates/ssm_parameter_defaultname.yaml"
    )
    with pytest.raises(ClientError) as exc_info:
        deploy_cfn_template(template_path=template_path)
    snapshot.match("deploy-error", exc_info.value)


@markers.aws.validated
def test_blank_parameter_value(aws_client: ServiceLevelClientFactory, cleanups):
    """
    While testing the new engine, I found that we don't handle the parameter value being blank well
    """
    template = {
        "Parameters": {
            "MyFoo": {
                "Type": "String",
            },
        },
        "Resources": {
            "MyTopic": {
                "Type": "AWS::SNS::Topic",
                "Properties": {
                    "DisplayName": {"Ref": "MyFoo"},
                },
            },
        },
    }

    res = aws_client.cloudformation.create_stack(
        TemplateBody=json.dumps(template),
        StackName=f"stack-{short_uid()}",
        Parameters=[{"ParameterKey": "MyFoo", "ParameterValue": ""}],
    )
    stack_id = res["StackId"]
    cleanups.append(lambda: aws_client.cloudformation.delete_stack(StackName=stack_id))

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