import base64
import contextlib
import importlib
import json
import logging
import queue
import random
import time
import uuid
from io import BytesIO
from operator import itemgetter

import pytest
import requests
import xmltodict
from botocore.auth import SigV4Auth
from botocore.config import Config
from botocore.exceptions import ClientError
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from localstack_snapshot.snapshots.transformer import RegexTransformer
from pytest_httpserver import HTTPServer
from werkzeug import Response

from localstack import config
from localstack.aws.api.lambda_ import Runtime
from localstack.config import external_service_url
from localstack.constants import (
    AWS_REGION_US_EAST_1,
)
from localstack.services.sns.constants import (
    PLATFORM_ENDPOINT_MSGS_ENDPOINT,
    SMS_MSGS_ENDPOINT,
    SMS_PHONE_NUMBER_OPT_OUT_ENDPOINT,
    SUBSCRIPTION_TOKENS_ENDPOINT,
)
from localstack.testing.aws.util import is_aws_cloud
from localstack.testing.config import TEST_AWS_ACCESS_KEY_ID, TEST_AWS_SECRET_ACCESS_KEY
from localstack.testing.pytest import markers
from localstack.testing.snapshots.transformer_utility import TransformerUtility
from localstack.utils import testutil
from localstack.utils.aws.arns import get_partition, parse_arn, sqs_queue_arn
from localstack.utils.net import wait_for_port_closed, wait_for_port_open
from localstack.utils.strings import short_uid, to_bytes, to_str
from localstack.utils.sync import poll_condition, retry
from localstack.utils.testutil import check_expected_lambda_log_events_length
from tests.aws.services.lambda_.functions import lambda_integration
from tests.aws.services.lambda_.test_lambda import TEST_LAMBDA_PYTHON, TEST_LAMBDA_PYTHON_ECHO
from tests.aws.services.sns.conftest import is_sns_v1_provider

LOG = logging.getLogger(__name__)

PUBLICATION_TIMEOUT = 0.500
PUBLICATION_RETRIES = 4


@pytest.fixture(autouse=True)
def sns_snapshot_transformer(snapshot):
    snapshot.add_transformer(snapshot.transform.sns_api())


@pytest.fixture
def sns_create_platform_application(aws_client):
    platform_applications = []

    def factory(**kwargs):
        if "Name" not in kwargs:
            kwargs["Name"] = f"platform-app-{short_uid()}"
        response = aws_client.sns.create_platform_application(**kwargs)
        platform_applications.append(response["PlatformApplicationArn"])
        return response

    yield factory

    for platform_application in platform_applications:
        try:
            endpoints = aws_client.sns.list_endpoints_by_platform_application(
                PlatformApplicationArn=platform_application
            )
        except Exception as e:
            LOG.debug(
                "Error cleaning up platform app '%s': %s",
                platform_application,
                e,
            )
            return
        for endpoint in endpoints["Endpoints"]:
            try:
                aws_client.sns.delete_endpoint(EndpointArn=endpoint["EndpointArn"])
            except Exception as e:
                LOG.debug(
                    "Error cleaning up platform endpoint '%s' for platform app '%s': %s",
                    endpoint["EndpointArn"],
                    platform_application,
                    e,
                )
        try:
            aws_client.sns.delete_platform_application(PlatformApplicationArn=platform_application)
        except Exception as e:
            LOG.debug("Error cleaning up platform application '%s': %s", platform_application, e)


@pytest.fixture
def sns_create_platform_endpoint(aws_client):
    platform_endpoints = []

    def factory(platform_application_arn: str, token: str, **kwargs):
        response = aws_client.sns.create_platform_endpoint(
            PlatformApplicationArn=platform_application_arn, Token=token, **kwargs
        )
        platform_endpoints.append(response["EndpointArn"])
        return response

    yield factory

    for endpoint in platform_endpoints:
        try:
            aws_client.sns.delete_endpoint(EndpointArn=endpoint["EndpointArn"])
        except Exception as e:
            LOG.debug(
                "Error cleaning up platform endpoint '%s': %s",
                endpoint,
                e,
            )


@pytest.fixture
def sns_provider():
    def factory(**kwargs):
        if is_sns_v1_provider():
            provider = "localstack.services.sns.provider"
        else:
            provider = "localstack.services.sns.v2.provider"
        # TODO: remove once legacy provider is retired completely and import normally
        module = importlib.import_module(provider)
        SnsProvider = module.SnsProvider
        return SnsProvider

    return factory


class TestSNSTopicCrud:
    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.get-topic-attrs.Attributes.DeliveryPolicy",  # TODO: remove this with the v2 provider switch
            "$.get-topic-attrs.Attributes.EffectiveDeliveryPolicy",
        ],
        condition=is_sns_v1_provider,
    )
    def test_create_topic_with_attributes(self, sns_create_topic, snapshot, aws_client):
        create_topic = sns_create_topic(
            Name="topictest.fifo",
            Attributes={
                "DisplayName": "TestTopic",
                "SignatureVersion": "2",
                "FifoTopic": "true",
            },
        )
        topic_arn = create_topic["TopicArn"]

        get_attrs_resp = aws_client.sns.get_topic_attributes(
            TopicArn=topic_arn,
        )
        snapshot.match("get-topic-attrs", get_attrs_resp)

        with pytest.raises(ClientError) as e:
            wrong_topic_arn = f"{topic_arn[:-8]}{short_uid()}"
            aws_client.sns.get_topic_attributes(TopicArn=wrong_topic_arn)

        snapshot.match("get-attrs-nonexistent-topic", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.get_topic_attributes(TopicArn="test-topic")

        snapshot.match("get-attrs-malformed-topic", e.value.response)

    @markers.aws.validated
    def test_tags(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]
        with pytest.raises(ClientError) as exc:
            aws_client.sns.tag_resource(
                ResourceArn=topic_arn,
                Tags=[
                    {"Key": "k1", "Value": "v1"},
                    {"Key": "k2", "Value": "v2"},
                    {"Key": "k2", "Value": "v2"},
                ],
            )
        snapshot.match("duplicate-key-error", exc.value.response)

        aws_client.sns.tag_resource(
            ResourceArn=topic_arn,
            Tags=[
                {"Key": "k1", "Value": "v1"},
                {"Key": "k2", "Value": "v2"},
            ],
        )

        tags = aws_client.sns.list_tags_for_resource(ResourceArn=topic_arn)
        # could not figure out the logic for tag order in AWS, so resorting to sorting it manually in place
        tags["Tags"].sort(key=itemgetter("Key"))
        snapshot.match("list-created-tags", tags)

        aws_client.sns.untag_resource(ResourceArn=topic_arn, TagKeys=["k1"])
        tags = aws_client.sns.list_tags_for_resource(ResourceArn=topic_arn)
        snapshot.match("list-after-delete-tags", tags)

        # test update tag
        aws_client.sns.tag_resource(ResourceArn=topic_arn, Tags=[{"Key": "k2", "Value": "v2b"}])
        tags = aws_client.sns.list_tags_for_resource(ResourceArn=topic_arn)
        snapshot.match("list-after-update-tags", tags)

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.get-topic-attrs.Attributes.DeliveryPolicy",  # TODO: remove this with the v2 provider switch
            "$.get-topic-attrs.Attributes.EffectiveDeliveryPolicy",
        ],
        condition=is_sns_v1_provider,
    )
    def test_create_topic_test_arn(self, sns_create_topic, snapshot, aws_client, account_id):
        topic_name = "topic-test-create"
        response = sns_create_topic(Name=topic_name)
        snapshot.match("create-topic", response)
        topic_arn = response["TopicArn"]
        topic_arn_params = topic_arn.split(":")
        testutil.response_arn_matches_partition(aws_client.sns, topic_arn)
        # we match the response but need to be sure the resource name is the same
        assert topic_arn_params[5] == topic_name

        if not is_aws_cloud():
            assert topic_arn_params[4] == account_id

        topic_attrs = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-topic-attrs", topic_attrs)

        response = aws_client.sns.delete_topic(TopicArn=topic_arn)
        snapshot.match("delete-topic", response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("topic-not-exists", e.value.response)

    @markers.aws.validated
    def test_delete_topic_idempotency(self, sns_create_topic, aws_client, snapshot):
        topic_arn = sns_create_topic()["TopicArn"]

        response = aws_client.sns.delete_topic(TopicArn=topic_arn)
        snapshot.match("delete-topic", response)

        with pytest.raises(ClientError):
            aws_client.sns.get_topic_attributes(TopicArn=topic_arn)

        delete_topic = aws_client.sns.delete_topic(TopicArn=topic_arn)
        snapshot.match("delete-topic-again", delete_topic)

    @markers.aws.validated
    def test_create_duplicate_topic_with_more_tags(self, sns_create_topic, snapshot, aws_client):
        topic_name = "test-duplicated-topic-more-tags"
        sns_create_topic(Name=topic_name)

        with pytest.raises(ClientError) as e:
            aws_client.sns.create_topic(Name=topic_name, Tags=[{"Key": "key1", "Value": "value1"}])

        snapshot.match("exception-duplicate", e.value.response)

    @markers.aws.validated
    def test_create_duplicate_topic_check_idempotency(self, sns_create_topic, snapshot):
        topic_name = f"test-{short_uid()}"
        tags = [{"Key": "a", "Value": "1"}, {"Key": "b", "Value": "2"}]
        kwargs = [
            {"Tags": tags},  # to create the same topic again with same tags
            {"Tags": [tags[0]]},  # to create the same topic again with one of the tags from above
            {"Tags": []},  # to create the same topic again with no tags
        ]

        # create topic with two tags
        response = sns_create_topic(Name=topic_name, Tags=tags)
        snapshot.match("response-created", response)

        for index, arg in enumerate(kwargs):
            response = sns_create_topic(Name=topic_name, **arg)
            # we check in the snapshot that they all have the same <resource:1> tag (original topic)
            snapshot.match(f"response-same-arn-{index}", response)

    @markers.aws.validated
    def test_create_topic_after_delete_with_new_tags(self, sns_create_topic, snapshot, aws_client):
        topic_name = f"test-{short_uid()}"
        topic = sns_create_topic(Name=topic_name, Tags=[{"Key": "Name", "Value": "pqr"}])
        snapshot.match("topic-0", topic)
        aws_client.sns.delete_topic(TopicArn=topic["TopicArn"])

        topic1 = sns_create_topic(Name=topic_name, Tags=[{"Key": "Name", "Value": "abc"}])
        snapshot.match("topic-1", topic1)

    @markers.aws.validated
    @pytest.mark.skip(reason="Not properly implemented in Moto, only mocked")
    def test_topic_delivery_policy_crud(self, sns_create_topic, snapshot, aws_client):
        # https://docs.aws.amazon.com/sns/latest/dg/sns-message-delivery-retries.html
        create_topic = sns_create_topic(
            Name="topictest.fifo",
            Attributes={
                "DisplayName": "TestTopic",
                "SignatureVersion": "2",
                "FifoTopic": "true",
                "DeliveryPolicy": json.dumps(
                    {
                        "http": {
                            "defaultRequestPolicy": {"headerContentType": "application/json"},
                        }
                    }
                ),
            },
        )
        topic_arn = create_topic["TopicArn"]

        get_attrs = aws_client.sns.get_topic_attributes(
            TopicArn=topic_arn,
        )
        snapshot.match("get-topic-attrs", get_attrs)

        set_attrs = aws_client.sns.set_topic_attributes(
            TopicArn=topic_arn,
            AttributeName="DeliveryPolicy",
            AttributeValue=json.dumps(
                {
                    "http": {
                        "defaultHealthyRetryPolicy": {
                            "minDelayTarget": 5,
                            "maxDelayTarget": 6,
                            "numRetries": 1,
                        }
                    }
                }
            ),
        )
        snapshot.match("set-topic-attrs", set_attrs)

        get_attrs_updated = aws_client.sns.get_topic_attributes(
            TopicArn=topic_arn,
        )
        snapshot.match("get-topic-attrs-after-update", get_attrs_updated)

        set_attrs_none = aws_client.sns.set_topic_attributes(
            TopicArn=topic_arn,
            AttributeName="DeliveryPolicy",
            AttributeValue=json.dumps(
                {
                    "http": {"defaultHealthyRetryPolicy": None},
                }
            ),
        )
        snapshot.match("set-topic-attrs-none", set_attrs_none)

        get_attrs_updated = aws_client.sns.get_topic_attributes(
            TopicArn=topic_arn,
        )
        snapshot.match("get-topic-attrs-after-none", get_attrs_updated)

        set_attrs_none_full = aws_client.sns.set_topic_attributes(
            TopicArn=topic_arn,
            AttributeName="DeliveryPolicy",
            AttributeValue="",
        )
        snapshot.match("set-topic-attrs-full-none", set_attrs_none_full)

        get_attrs_updated = aws_client.sns.get_topic_attributes(
            TopicArn=topic_arn,
        )
        snapshot.match("get-topic-attrs-after-delete", get_attrs_updated)


class TestSNSTopicCrudV2:
    @markers.aws.validated
    def test_delete_non_existent_topic(self, snapshot, aws_client, account_id, region_name):
        response = aws_client.sns.delete_topic(
            TopicArn=f"arn:aws:sns:{region_name}:{account_id}:non-existent-{short_uid()}"
        )

        snapshot.match("delete-non-existent-topic", response)

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        # skipped only for v1
        condition=is_sns_v1_provider,
        paths=[
            "$..Attributes.DeliveryPolicy",
            "$..Attributes.EffectiveDeliveryPolicy",
        ],
    )
    def test_create_topic_should_be_idempotent(self, snapshot, sns_create_topic, aws_client):
        topic_name = f"test-idempotent-{short_uid()}"

        resp1 = sns_create_topic(Name=topic_name)
        aws_client.sns.set_topic_attributes(
            TopicArn=resp1["TopicArn"], AttributeName="DisplayName", AttributeValue="AlreadySet"
        )
        attrs = aws_client.sns.get_topic_attributes(TopicArn=resp1["TopicArn"])
        snapshot.match("topic-attrs-idempotent-1", attrs)

        resp2 = sns_create_topic(Name=topic_name)
        attrs = aws_client.sns.get_topic_attributes(TopicArn=resp2["TopicArn"])
        snapshot.match("topic-attrs-idempotent-2", attrs)

    @markers.snapshot.skip_snapshot_verify(
        # skipped only for v1
        condition=is_sns_v1_provider,
        paths=["$..Error.Message"],
    )
    @markers.aws.validated
    def test_create_topic_name_constraints(self, snapshot, sns_create_topic):
        # Valid names within length constraints
        valid_name = "a" * 256
        resp = sns_create_topic(Name=valid_name)
        snapshot.match("valid-name-max-length", resp)

        # Invalid names: too short / too long
        with pytest.raises(ClientError) as e:
            sns_create_topic(Name="")
        snapshot.match("name-too-short", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_create_topic(Name="a" * 257)
        snapshot.match("name-too-long", e.value.response)

        # Invalid chars
        # TODO: the index is used because using special characters in the matched name doesn't work for all chars
        #  -> we should write a snapshot test to investigate further
        for index, c in enumerate([":", ";", "!", "@", "|", "^", "%", " "]):
            with pytest.raises(ClientError) as e:
                sns_create_topic(Name=f"bad{c}name")
            snapshot.match(f"name-invalid-char-{index}", e.value.response)

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        # skipped only for v1
        condition=is_sns_v1_provider,
        paths=[
            "$..Attributes.DeliveryPolicy",
            "$..Attributes.EffectiveDeliveryPolicy",
        ],
    )
    def test_create_topic_in_multiple_regions(
        self, aws_client, aws_client_factory, snapshot, region_name, secondary_region_name
    ):
        topic_name = f"multiregion-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(region_name, "<region-primary>"))
        snapshot.add_transformer(RegexTransformer(secondary_region_name, "<region-secondary>"))

        sns_primary_region = aws_client_factory(region_name=region_name).sns
        topic_arn_primary_region = sns_primary_region.create_topic(Name=topic_name)["TopicArn"]

        sns_secondary_region = aws_client_factory(region_name=secondary_region_name).sns
        topic_arn_secondary_region = sns_secondary_region.create_topic(Name=topic_name)["TopicArn"]

        list_topics_primary = sns_primary_region.list_topics()
        snapshot.match("list-primary", list_topics_primary)
        list_topics_secondary = sns_secondary_region.list_topics()
        snapshot.match("list-secondary", list_topics_secondary)

        snapshot.match(
            "topic-east", sns_primary_region.get_topic_attributes(TopicArn=topic_arn_primary_region)
        )
        snapshot.match(
            "topic-west",
            sns_secondary_region.get_topic_attributes(TopicArn=topic_arn_secondary_region),
        )

    @markers.aws.validated
    def test_list_topic_paging(self, aws_client, sns_create_topic):
        topic_arns = []
        page_size = 100
        for i in range(page_size + 20):  # > default page size
            resp = sns_create_topic(Name=f"paging-{i}-{short_uid()}")
            topic_arns.append(resp["TopicArn"])

        resp = aws_client.sns.list_topics()
        assert len(resp["Topics"]) == page_size

        assert "NextToken" in resp
        token = resp["NextToken"]

        resp2 = aws_client.sns.list_topics(NextToken=token)

        # Collect all returned ARNs to ensure no duplicates / missing
        all_returned_arns = [t["TopicArn"] for t in resp["Topics"]] + [
            t["TopicArn"] for t in resp2["Topics"]
        ]
        assert set(topic_arns).issubset(set(all_returned_arns))

    @pytest.mark.skipif(is_sns_v1_provider(), reason="not covered in v1")
    @markers.aws.validated
    def test_topic_get_attributes_with_fifo_false(self, sns_create_topic, aws_client, snapshot):
        resp = sns_create_topic(
            Name=f"standard-topic-{short_uid()}", Attributes={"FifoTopic": "false"}
        )
        topic_arn = resp["TopicArn"]

        attrs = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-attrs-standard-topic", attrs)

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_topic_attributes(
                TopicArn=topic_arn, AttributeName="FifoTopic", AttributeValue="false"
            )
        snapshot.match("set-fifo-false-after-creation", e.value.response)

    @markers.aws.validated
    def test_topic_add_permission(self, sns_create_topic, aws_client, snapshot, account_id):
        topic_arn = sns_create_topic()["TopicArn"]
        resp = aws_client.sns.add_permission(
            TopicArn=topic_arn, Label="test", AWSAccountId=[account_id], ActionName=["Publish"]
        )
        snapshot.match("add-permission-response", resp)

        attributes_resp = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        policy_str = attributes_resp["Attributes"]["Policy"]
        policy_json = json.loads(policy_str)
        snapshot.match("topic-policy-after-permission", policy_json)

    @markers.aws.validated
    def test_topic_add_multiple_permissions(
        self, sns_create_topic, aws_client, snapshot, account_id
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        resp = aws_client.sns.add_permission(
            TopicArn=topic_arn,
            Label="test",
            AWSAccountId=[account_id],
            ActionName=["Publish", "Subscribe"],
        )
        snapshot.match("add-permission-response", resp)

        attributes_resp = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        policy_str = attributes_resp["Attributes"]["Policy"]
        policy_json = json.loads(policy_str)
        snapshot.match("topic-policy-after-permission", policy_json)

    @markers.aws.validated
    def test_topic_remove_permission(self, sns_create_topic, aws_client, snapshot, account_id):
        topic_arn = sns_create_topic()["TopicArn"]
        label = "test"
        aws_client.sns.add_permission(
            TopicArn=topic_arn, Label=label, AWSAccountId=[account_id], ActionName=["Publish"]
        )

        aws_client.sns.remove_permission(TopicArn=topic_arn, Label=label)
        attributes_resp = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        policy_str = attributes_resp["Attributes"]["Policy"]
        policy_json = json.loads(policy_str)
        snapshot.match("topic-policy", policy_json)

    @markers.snapshot.skip_snapshot_verify(paths=["$..Error.Message"], condition=is_sns_v1_provider)
    @markers.aws.validated
    def test_add_permission_errors(self, snapshot, aws_client, account_id):
        topic_name = f"topic-{short_uid()}"
        topic_arn = aws_client.sns.create_topic(Name=topic_name)["TopicArn"]

        aws_client.sns.add_permission(
            TopicArn=topic_arn,
            Label="test",
            AWSAccountId=[account_id],
            ActionName=["Publish"],
        )

        with pytest.raises(ClientError) as e:
            aws_client.sns.add_permission(
                TopicArn=topic_arn,
                Label="test",
                AWSAccountId=[account_id],
                ActionName=["AddPermission"],
            )
        snapshot.match("duplicate-label", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.add_permission(
                TopicArn=f"{topic_arn}-not-existing",
                Label="test-2",
                AWSAccountId=[account_id],
                ActionName=["AddPermission"],
            )
        snapshot.match("topic-not-found", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.add_permission(
                TopicArn=topic_arn,
                Label="test-2",
                AWSAccountId=[account_id],
                ActionName=["InvalidAction"],
            )
        snapshot.match("invalid-action", e.value.response)

    @markers.aws.validated
    def test_remove_permission_errors(self, snapshot, aws_client, account_id):
        topic_name = f"topic-{short_uid()}"
        topic_arn = aws_client.sns.create_topic(Name=topic_name)["TopicArn"]
        aws_client.sns.add_permission(
            TopicArn=topic_arn,
            Label="test",
            AWSAccountId=[account_id],
            ActionName=["Publish"],
        )

        with pytest.raises(ClientError) as e:
            aws_client.sns.remove_permission(TopicArn=f"{topic_arn}-not-existing", Label="test")

        snapshot.match("topic-not-found", e.value.response)

    @markers.aws.validated
    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Not implemented in moto")
    def test_data_protection_policy_crud(self, snapshot, aws_client, region_name):
        topic_name = f"topic-{short_uid()}"
        topic_arn = aws_client.sns.create_topic(Name=topic_name)["TopicArn"]

        policy_doc = {
            "Name": "data_protection_policy",
            "Description": "Test Policy",
            "Version": "2021-06-01",
            "Statement": [
                {
                    "Sid": "test-statement",
                    "DataDirection": "Inbound",
                    "Principal": ["*"],
                    "DataIdentifier": [
                        f"arn:aws:dataprotection:{region_name}::data-identifier/EmailAddress"
                    ],
                    "Operation": {"Deny": {}},
                }
            ],
        }

        response = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-topic-attributes-before-policy", response)

        response = aws_client.sns.put_data_protection_policy(
            ResourceArn=topic_arn, DataProtectionPolicy=json.dumps(policy_doc)
        )
        snapshot.match("put-data-protection-policy", response)

        response = aws_client.sns.get_data_protection_policy(ResourceArn=topic_arn)
        snapshot.match("get-data-protection-policy", response)

        # check if policy shows up in the attributes
        response = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-topic-attributes-after-policy", response)


class TestSNSPublishCrud:
    """
    This class contains tests related to the global `Publish` validation, not tied to a particular kind of subscription
    """

    @markers.aws.validated
    def test_publish_by_path_parameters(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        aws_http_client_factory,
        snapshot,
        aws_client,
        region_name,
    ):
        message = "test message direct post request"
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        client = aws_http_client_factory(
            "sns",
            signer_factory=SigV4Auth,
            region=region_name,
            aws_access_key_id=TEST_AWS_ACCESS_KEY_ID,
            aws_secret_access_key=TEST_AWS_SECRET_ACCESS_KEY,
        )

        if is_aws_cloud():
            endpoint_url = f"https://sns.{region_name}.amazonaws.com"
        else:
            endpoint_url = config.internal_service_url()

        response = client.post(
            endpoint_url,
            params={
                "Action": "Publish",
                "Version": "2010-03-31",
                "TopicArn": topic_arn,
                "Message": message,
            },
        )

        json_response = xmltodict.parse(response.content)
        json_response["PublishResponse"].pop("@xmlns")
        json_response["PublishResponse"]["ResponseMetadata"]["HTTPStatusCode"] = (
            response.status_code
        )
        json_response["PublishResponse"]["ResponseMetadata"]["HTTPHeaders"] = dict(response.headers)
        snapshot.match("post-request", json_response)

        assert response.status_code == 200
        assert b"<PublishResponse" in response.content

        rs = aws_client.sqs.receive_message(
            QueueUrl=queue_url, VisibilityTimeout=0, WaitTimeSeconds=5
        )
        snapshot.match("messages", rs)
        msg_body = json.loads(rs["Messages"][0]["Body"])
        assert msg_body["TopicArn"] == topic_arn
        assert msg_body["Message"] == message

    @markers.aws.validated
    def test_publish_wrong_arn_format(self, snapshot, aws_client):
        message = "Good news everyone!"
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(Message=message, TopicArn="randomstring")

        snapshot.match("invalid-topic-arn", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(Message=message, TopicArn="randomstring:1")

        snapshot.match("invalid-topic-arn-1", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(Message=message, TopicArn="")

        snapshot.match("empty-topic", e.value.response)

    @markers.aws.validated
    def test_publish_message_by_target_arn(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        # using an SQS subscription to test TopicArn/TargetArn as it is easier to check against AWS
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        aws_client.sns.publish(TopicArn=topic_arn, Message="test-msg-1")

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            MessageAttributeNames=["All"],
            VisibilityTimeout=0,
            WaitTimeSeconds=4,
        )

        snapshot.match("receive-topic-arn", response)

        message = response["Messages"][0]
        aws_client.sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"])

        # publish with TargetArn instead of TopicArn
        aws_client.sns.publish(TargetArn=topic_arn, Message="test-msg-2")

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            MessageAttributeNames=["All"],
            VisibilityTimeout=0,
            WaitTimeSeconds=4,
        )
        snapshot.match("receive-target-arn", response)

    @markers.aws.validated
    def test_publish_message_before_subscribe_topic(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        rs = aws_client.sns.publish(
            TopicArn=topic_arn, Subject="test-subject-before-sub", Message="test_message_before"
        )
        snapshot.match("publish-before-subscribing", rs)

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        message_subject = "test-subject-after-sub"
        message_body = "test_message_after"

        rs = aws_client.sns.publish(
            TopicArn=topic_arn, Subject=message_subject, Message=message_body
        )
        snapshot.match("publish-after-subscribing", rs)

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url, VisibilityTimeout=0, WaitTimeSeconds=5
        )
        # nothing was subscribing to the topic, so the first message is lost
        snapshot.match("receive-messages", response)

    @markers.aws.validated
    def test_unknown_topic_publish(self, sns_create_topic, snapshot, aws_client):
        # create topic to get the basic arn structure
        # otherwise you get InvalidClientTokenId exception because of account id
        topic_arn = sns_create_topic()["TopicArn"]
        # append to get an unknown topic
        fake_arn = f"{topic_arn}-fake"
        message = "This is a test message"

        # test to send a message with no subscribers
        response = aws_client.sns.publish(TopicArn=topic_arn, Message=message)
        snapshot.match("success", response)

        # test to send to a nonexistent topic
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(TopicArn=fake_arn, Message=message)

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

    @markers.aws.validated
    def test_topic_publish_another_region(
        self, sns_create_topic, snapshot, aws_client, aws_client_factory, secondary_region_name
    ):
        # create the topic in the default region, so that it's easier to clean up with the fixture
        topic_arn = sns_create_topic()["TopicArn"]

        # create a client in another region
        sns_client_region_2 = aws_client_factory.get_client(
            service_name="sns",
            region_name=secondary_region_name,
        )

        message = "This is a test message"

        # test to send a message with the client from the same region
        response = aws_client.sns.publish(TopicArn=topic_arn, Message=message)
        snapshot.match("success", response)

        # test to send from the second region client
        with pytest.raises(ClientError) as e:
            sns_client_region_2.publish(TopicArn=topic_arn, Message=message)

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

        # test to send batch from the second region client
        with pytest.raises(ClientError) as e:
            sns_client_region_2.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": message,
                    }
                ],
            )

        snapshot.match("error-batch", e.value.response)

    @markers.aws.validated
    def test_publish_non_existent_target(self, sns_create_topic, snapshot, aws_client, region_name):
        topic_arn = sns_create_topic()["TopicArn"]
        account_id = parse_arn(topic_arn)["account"]
        with pytest.raises(ClientError) as ex:
            aws_client.sns.publish(
                TargetArn=f"arn:{get_partition(region_name)}:sns:us-east-1:{account_id}:endpoint/APNS/abcdef/0f7d5971-aa8b-4bd5-b585-0826e9f93a66",
                Message="This is a push notification",
            )
        snapshot.match("non-existent-endpoint", ex.value.response)

    @markers.aws.validated
    def test_publish_with_empty_subject(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]

        # Publish without subject
        rs = aws_client.sns.publish(
            TopicArn=topic_arn, Message=json.dumps({"message": "test_publish"})
        )
        snapshot.match("response-without-subject", rs)
        assert rs["ResponseMetadata"]["HTTPStatusCode"] == 200

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(
                TopicArn=topic_arn,
                Subject="",
                Message=json.dumps({"message": "test_publish"}),
            )

        snapshot.match("response-with-empty-subject", e.value.response)

    @markers.aws.validated
    def test_empty_sns_message(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(Message="", TopicArn=topic_arn)

        snapshot.match("empty-msg-error", e.value.response)

        queue_attrs = aws_client.sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["ApproximateNumberOfMessages"]
        )
        snapshot.match("queue-attrs", queue_attrs)

    @markers.aws.validated
    def test_publish_too_long_message(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]
        # simulate payload over 256kb
        message = "This is a test message" * 12000

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(TopicArn=topic_arn, Message=message)

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

        # maximum size: 262144
        # craft a message body that is under the limit, but goes over with the message attributes
        message_attrs = {"attr1": {"DataType": "Number", "StringValue": "1"}}
        counted_values = [*message_attrs.keys(), *message_attrs["attr1"].values()]
        message_attrs_len = sum(len(to_bytes(value)) for value in counted_values)
        message_with_attrs = (262144 - message_attrs_len + 1) * "a"

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(
                TopicArn=topic_arn, Message=message_with_attrs, MessageAttributes=message_attrs
            )

        snapshot.match("error-with-attrs", e.value.response)

        # assert that it goes through with the right size, remove the last char
        publish = aws_client.sns.publish(
            TopicArn=topic_arn, Message=message_with_attrs[:-1], MessageAttributes=message_attrs
        )
        assert publish["ResponseMetadata"]["HTTPStatusCode"] == 200

    @markers.aws.validated
    def test_publish_batch_too_long_message(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]
        # simulate payload over 256kb
        max_size = 262144

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": "a" * (max_size // 2),
                    },
                    {
                        "Id": "2",
                        "Message": "b" * (1 + max_size // 2),
                    },
                ],
            )

        snapshot.match("error-no-attrs", e.value.response)

        # craft a message body that is under the limit, but goes over with the message attributes
        message_attrs = {"attr1": {"DataType": "Number", "StringValue": "1"}}
        counted_values = [*message_attrs.keys(), *message_attrs["attr1"].values()]
        message_attrs_len = sum(len(to_bytes(value)) for value in counted_values)
        message_with_attrs = (262144 - message_attrs_len) * "a"

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": message_with_attrs,
                        "MessageAttributes": message_attrs,
                    },
                    {
                        "Id": "2",
                        "Message": "b",
                    },
                ],
            )

        snapshot.match("error-with-attrs", e.value.response)

        # assert that it goes through with the right size, remove the last char
        publish_batch = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=[
                {
                    "Id": "1",
                    "Message": message_with_attrs[:-1],
                    "MessageAttributes": message_attrs,
                },
                {
                    "Id": "2",
                    "Message": "b",
                },
            ],
        )
        assert publish_batch["ResponseMetadata"]["HTTPStatusCode"] == 200

    @markers.aws.validated
    def test_publish_batch_invalid_entry_id(self, sns_create_topic, snapshot, aws_client):
        """Test validation of batch entry ID according to AWS specifications.

        Entry IDs must:
        - Only contain alphanumeric characters, hyphens, and underscores
        - Be at most 80 characters long
        """
        topic_arn = sns_create_topic()["TopicArn"]

        # Test 1: ID with invalid character (dot)
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "message.id",
                        "Message": "test message",
                    }
                ],
            )
        snapshot.match("invalid-char-dot", e.value.response)

        # Test 2: ID with multiple invalid characters
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "msg@123#test",
                        "Message": "test message",
                    }
                ],
            )
        snapshot.match("invalid-char-special", e.value.response)

        # Test 3: ID that is too long (81 characters)
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "a" * 81,
                        "Message": "test message",
                    }
                ],
            )
        snapshot.match("id-too-long-81", e.value.response)

        # Test 4: ID that is way too long (91 characters)
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "myreallyreallyreallyreallyreallyreallyreallyreallyreallyreallyreallyreallyreallylongid",
                        "Message": "test message",
                    }
                ],
            )
        snapshot.match("id-too-long-91", e.value.response)

        # Test 5: Valid ID with exactly 80 characters (should succeed)
        publish_batch_response = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=[
                {
                    "Id": "a" * 80,
                    "Message": "test message",
                }
            ],
        )
        snapshot.match("valid-id-80-chars", publish_batch_response)

        # Test 6: Valid ID with allowed characters (should succeed)
        publish_batch_response = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=[
                {
                    "Id": "valid-message_123",
                    "Message": "test message",
                }
            ],
        )
        snapshot.match("valid-id-with-hyphen-underscore", publish_batch_response)

    @markers.aws.validated
    def test_message_structure_json_exc(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]
        # TODO: add batch

        # missing `default` key for the JSON
        with pytest.raises(ClientError) as e:
            message = json.dumps({"sqs": "Test message"})
            aws_client.sns.publish(
                TopicArn=topic_arn,
                Message=message,
                MessageStructure="json",
            )
        snapshot.match("missing-default-key", e.value.response)

        # invalid JSON
        with pytest.raises(ClientError) as e:
            message = '{"default": "This is a default message"} }'
            aws_client.sns.publish(
                TopicArn=topic_arn,
                Message=message,
                MessageStructure="json",
            )
        snapshot.match("invalid-json", e.value.response)

        # duplicate keys: from SNS docs, should fail but does work
        # https://docs.aws.amazon.com/sns/latest/api/API_Publish.html
        # `Duplicate keys are not allowed.`
        message = '{"default": "This is a default message", "default": "Duplicate"}'
        resp = aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageStructure="json",
        )
        snapshot.match("duplicate-json-keys", resp)

        with pytest.raises(ClientError) as e:
            message = json.dumps({"default": {"object": "test"}})
            aws_client.sns.publish(
                TopicArn=topic_arn,
                Message=message,
                MessageStructure="json",
            )
        snapshot.match("key-is-not-string", e.value.response)


class TestSNSSubscriptionCrud:
    @markers.aws.validated
    def test_subscribe_with_invalid_protocol(self, sns_create_topic, sns_subscription, snapshot):
        topic_arn = sns_create_topic()["TopicArn"]

        with pytest.raises(ClientError) as e:
            sns_subscription(
                TopicArn=topic_arn, Protocol="test-protocol", Endpoint="localstack@yopmail.com"
            )

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

    @markers.aws.validated
    def test_unsubscribe_from_non_existing_subscription(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        aws_client.sns.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])
        # unsubscribing a second time
        response = aws_client.sns.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])
        snapshot.match("empty-unsubscribe", response)

    @markers.aws.validated
    def test_create_subscriptions_with_attributes(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        snapshot,
        aws_client,
        sns_subscription,
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        queue_arn = sqs_get_queue_arn(queue_url)

        with pytest.raises(ClientError) as e:
            sns_subscription(
                TopicArn=topic_arn,
                Protocol="sqs",
                Endpoint=queue_arn,
                Attributes={
                    "RawMessageDelivery": "wrongvalue",  # set an weird case value, SNS will lower it
                    "FilterPolicyScope": "MessageBody",
                    "FilterPolicy": "",
                },
                ReturnSubscriptionArn=True,
            )
        snapshot.match("subscribe-wrong-attr", e.value.response)

        subscribe_resp = sns_subscription(
            TopicArn=topic_arn,
            Protocol="sqs",
            Endpoint=queue_arn,
            Attributes={
                "RawMessageDelivery": "TrUe",  # set an weird case value, SNS will lower it
                "FilterPolicyScope": "MessageBody",
                "FilterPolicy": "",
            },
            ReturnSubscriptionArn=True,
        )
        snapshot.match("subscribe", subscribe_resp)

        get_attrs_resp = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscribe_resp["SubscriptionArn"]
        )
        snapshot.match("get-attrs", get_attrs_resp)

        with pytest.raises(ClientError) as e:
            wrong_sub_arn = f"{subscribe_resp['SubscriptionArn'][:-8]}{short_uid()}"
            aws_client.sns.get_subscription_attributes(SubscriptionArn=wrong_sub_arn)

        snapshot.match("get-attrs-nonexistent-sub", e.value.response)

    @markers.aws.validated
    def test_not_found_error_on_set_subscription_attributes(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_subscription,
        snapshot,
        aws_client,
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        queue_arn = sqs_get_queue_arn(queue_url)
        subscription = sns_subscription(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn)
        snapshot.match("sub", subscription)
        subscription_arn = subscription["SubscriptionArn"]

        response = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        subscription_attributes = response["Attributes"]
        snapshot.match("sub-attrs", response)

        assert subscription_attributes["SubscriptionArn"] == subscription_arn

        subscriptions_by_topic = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("subscriptions-for-topic-before-unsub", subscriptions_by_topic)
        assert len(subscriptions_by_topic["Subscriptions"]) == 1

        aws_client.sns.unsubscribe(SubscriptionArn=subscription_arn)

        def check_subscription_deleted():
            try:
                # AWS doesn't give NotFound error on GetSubscriptionAttributes for a while, might be cached
                aws_client.sns.set_subscription_attributes(
                    SubscriptionArn=subscription_arn,
                    AttributeName="RawMessageDelivery",
                    AttributeValue="true",
                )
                raise Exception("Subscription is not deleted")
            except ClientError as e:
                assert e.response["Error"]["Code"] == "NotFound"
                assert e.response["ResponseMetadata"]["HTTPStatusCode"] == 404
                snapshot.match("sub-not-found", e.response)

        retry(check_subscription_deleted, retries=10, sleep_before=0.2, sleep=3)
        subscriptions_by_topic = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("subscriptions-for-topic-after-unsub", subscriptions_by_topic)
        assert len(subscriptions_by_topic["Subscriptions"]) == 0

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.invalid-json-redrive-policy.Error.Message",  # message contains java trace in AWS, assert instead
            "$.invalid-json-filter-policy.Error.Message",  # message contains java trace in AWS, assert instead
        ]
    )
    def test_validate_set_sub_attributes(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        aws_client,
    ):
        topic_name = f"topic-{short_uid()}"
        queue_name = f"queue-{short_uid()}"
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]
        queue_url = sqs_create_queue(QueueName=queue_name)
        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        sub_arn = subscription["SubscriptionArn"]

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=sub_arn,
                AttributeName="FakeAttribute",
                AttributeValue="test-value",
            )
        snapshot.match("fake-attribute", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=sub_arn,
                AttributeName="RawMessageDelivery",
                AttributeValue="test-ValUe",
            )
        snapshot.match("raw-message-wrong-value", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=sub_arn,
                AttributeName="RawMessageDelivery",
                AttributeValue="",
            )
        snapshot.match("raw-message-empty-value", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=sub_arn,
                AttributeName="RedrivePolicy",
                AttributeValue=json.dumps({"deadLetterTargetArn": "fake-arn"}),
            )
        snapshot.match("fake-arn-redrive-policy", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=sub_arn,
                AttributeName="RedrivePolicy",
                AttributeValue="{invalidjson}",
            )
        snapshot.match("invalid-json-redrive-policy", e.value.response)
        assert e.value.response["Error"]["Message"].startswith(
            "Invalid parameter: RedrivePolicy: failed to parse JSON."
        )

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=sub_arn,
                AttributeName="FilterPolicy",
                AttributeValue="{invalidjson}",
            )
        snapshot.match("invalid-json-filter-policy", e.value.response)
        assert e.value.response["Error"]["Message"].startswith(
            "Invalid parameter: FilterPolicy: failed to parse JSON."
        )

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=["$.invalid-token.Error.Message"]  # validate the token shape
    )
    def test_sns_confirm_subscription_wrong_token(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]

        with pytest.raises(ClientError) as e:
            wrong_topic = topic_arn[:-1] + "i"
            aws_client.sns.confirm_subscription(
                TopicArn=wrong_topic,
                Token="51b2ff3edb475b7d91550e0ab6edf0c1de2a34e6ebaf6c2262a001bcb7e051c43aa00022ceecce70bd2a67b2042da8d8eb47fef7a4e4e942d23e7fa56146b9ee35da040b4b8af564cc4184a7391c834cb75d75c22981f776ad1ce8805e9bab29da2329985337bb8095627907b46c8577c8440556b6f86582a954758026f41fc62041c4b3f67b0f5921232b5dae5aaca1",
            )

        snapshot.match("topic-not-exists", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token="randomtoken")

        snapshot.match("invalid-token", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.confirm_subscription(
                TopicArn=topic_arn,
                Token="51b2ff3edb475b7d91550e0ab6edf0c1de2a34e6ebaf6c2262a001bcb7e051c43aa00022ceecce70bd2a67b2042da8d8eb47fef7a4e4e942d23e7fa56146b9ee35da040b4b8af564cc4184a7391c834cb75d75c22981f776ad1ce8805e9bab29da2329985337bb8095627907b46c8577c8440556b6f86582a954758026f41fc62041c4b3f67b0f5921232b5dae5aaca1",
            )

        snapshot.match("token-not-exists", e.value.response)

    @markers.aws.validated
    def test_list_subscriptions(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_subscription,
        aws_client,
    ):
        topic = sns_create_topic()
        topic_arn = topic["TopicArn"]
        topic_2 = sns_create_topic()
        topic_arn_2 = topic_2["TopicArn"]
        created_subscriptions = []
        for i in range(3):
            queue_url = sqs_create_queue()
            queue_arn = sqs_get_queue_arn(queue_url)
            sns_subscription(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn)
            created_subscriptions.append((topic_arn, queue_arn))
        for i in range(3):
            queue_url = sqs_create_queue()
            queue_arn = sqs_get_queue_arn(queue_url)
            sns_subscription(TopicArn=topic_arn_2, Protocol="sqs", Endpoint=queue_arn)
            created_subscriptions.append((topic_arn_2, queue_arn))

        list_subs = aws_client.sns.list_subscriptions()
        all_subs = list_subs["Subscriptions"]
        if list_subs.get("NextToken"):
            while next_token := list_subs.get("NextToken"):
                list_subs = aws_client.sns.list_subscriptions(NextToken=next_token)
                all_subs.extend(list_subs["Subscriptions"])

        all_sub_tuples = [(sub["TopicArn"], sub["Endpoint"]) for sub in all_subs]
        assert all(sub in all_sub_tuples for sub in created_subscriptions)

    @markers.aws.validated
    def test_list_subscriptions_by_topic_pagination(
        self, sns_create_topic, sns_subscription, snapshot, aws_client
    ):
        # ordering of the listing seems to be not consistent, so we can transform its value
        snapshot.add_transformers_list(
            [
                snapshot.transform.key_value("Endpoint"),
                snapshot.transform.key_value("NextToken"),
            ]
        )

        base_phone_number = "+12312312"
        topic_arn = sns_create_topic()["TopicArn"]
        for phone_suffix in range(101):
            phone_number = f"{base_phone_number}{phone_suffix}"
            sns_subscription(TopicArn=topic_arn, Protocol="sms", Endpoint=phone_number)

        response = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        # not snapshotting the results, it contains 100 entries
        assert "NextToken" in response
        # seems to be b64 encoded
        assert base64.b64decode(response["NextToken"])
        assert len(response["Subscriptions"]) == 100
        # keep the page 1 subscriptions ARNs
        page_1_subs = {sub["SubscriptionArn"] for sub in response["Subscriptions"]}

        response = aws_client.sns.list_subscriptions_by_topic(
            TopicArn=topic_arn, NextToken=response["NextToken"]
        )
        snapshot.match("list-sub-per-topic-page-2", response)
        assert "NextToken" not in response
        assert len(response["Subscriptions"]) == 1
        # assert that the last Subscription is not present in page 1
        assert response["Subscriptions"][0]["SubscriptionArn"] not in page_1_subs

        response = aws_client.sns.list_subscriptions()
        # not snapshotting because there might be subscriptions outside the topic, this is all the requester subs
        assert "NextToken" in response
        assert len(response["Subscriptions"]) <= 100

    @markers.aws.validated
    def test_subscribe_idempotency(
        self, aws_client, sns_create_topic, sqs_create_queue, sqs_get_queue_arn, snapshot
    ):
        """
        Test the idempotency of SNS subscribe calls for a given endpoint and its attributes
        """
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        queue_arn = sqs_get_queue_arn(queue_url)

        def subscribe_queue_to_topic(attributes: dict = None) -> dict:
            kwargs = {}
            if attributes is not None:
                kwargs["Attributes"] = attributes
            response = aws_client.sns.subscribe(
                TopicArn=topic_arn,
                Protocol="sqs",
                Endpoint=queue_arn,
                ReturnSubscriptionArn=True,
                **kwargs,
            )
            return response

        subscribe_resp = subscribe_queue_to_topic(
            {
                "RawMessageDelivery": "True",
            }
        )
        snapshot.match("subscribe", subscribe_resp)

        get_attrs_resp = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscribe_resp["SubscriptionArn"]
        )
        snapshot.match("get-sub-attrs", get_attrs_resp)

        subscribe_resp = subscribe_queue_to_topic(
            {
                "RawMessageDelivery": "true",
            }
        )
        snapshot.match("subscribe-exact-same-raw", subscribe_resp)

        subscribe_resp = subscribe_queue_to_topic(
            {
                "RawMessageDelivery": "true",
                "FilterPolicyScope": "MessageAttributes",  # test if it also matches default values
            }
        )

        snapshot.match("subscribe-idempotent", subscribe_resp)

        # no attributes and empty attributes are working as well
        subscribe_resp = subscribe_queue_to_topic()
        snapshot.match("subscribe-idempotent-no-attributes", subscribe_resp)

        subscribe_resp = subscribe_queue_to_topic({})
        snapshot.match("subscribe-idempotent-empty-attributes", subscribe_resp)

        subscribe_resp = subscribe_queue_to_topic({"FilterPolicyScope": "MessageAttributes"})
        snapshot.match("subscribe-missing-attributes", subscribe_resp)

        with pytest.raises(ClientError) as e:
            subscribe_queue_to_topic(
                {
                    "RawMessageDelivery": "false",
                    "FilterPolicyScope": "MessageBody",
                }
            )
        snapshot.match("subscribe-diff-attributes", e.value.response)

    @markers.aws.validated
    def test_unsubscribe_idempotency(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_name = f"topic-{short_uid()}"
        queue_name = f"queue-{short_uid()}"
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]
        queue_url = sqs_create_queue(QueueName=queue_name)
        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        sub_arn = subscription["SubscriptionArn"]

        unsubscribe_1 = aws_client.sns.unsubscribe(SubscriptionArn=sub_arn)
        snapshot.match("unsubscribe-1", unsubscribe_1)
        unsubscribe_2 = aws_client.sns.unsubscribe(SubscriptionArn=sub_arn)
        snapshot.match("unsubscribe-2", unsubscribe_2)

    @markers.aws.validated
    def test_unsubscribe_wrong_arn_format(self, snapshot, aws_client_factory, region_name):
        sns_client = aws_client_factory(
            region_name=region_name, config=Config(parameter_validation=False)
        ).sns

        with pytest.raises(ClientError) as e:
            sns_client.unsubscribe(SubscriptionArn="randomstring")

        snapshot.match("invalid-unsubscribe-arn-1", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_client.unsubscribe(SubscriptionArn="arn:aws:sns:us-east-1:random")

        snapshot.match("invalid-unsubscribe-arn-2", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_client.unsubscribe(SubscriptionArn="arn:aws:sns:us-east-1:111111111111:random")

        snapshot.match("invalid-unsubscribe-arn-3", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_client.unsubscribe()

        snapshot.match("invalid-unsubscribe-arn-4", e.value.response)

    @markers.aws.validated
    def test_subscribe_with_invalid_topic(self, sns_create_topic, sns_subscription, snapshot):
        with pytest.raises(ClientError) as e:
            sns_subscription(
                TopicArn="randomstring", Protocol="email", Endpoint="localstack@yopmail.com"
            )

        snapshot.match("invalid-subscribe-arn-1", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_subscription(
                TopicArn="arn:aws:sns:us-east-1:random",
                Protocol="email",
                Endpoint="localstack@yopmail.com",
            )

        snapshot.match("invalid-subscribe-arn-2", e.value.response)

        topic_arn = sns_create_topic()["TopicArn"]
        bad_topic_arn = topic_arn + "aaa"
        with pytest.raises(ClientError) as e:
            sns_subscription(
                TopicArn=bad_topic_arn, Protocol="email", Endpoint="localstack@yopmail.com"
            )

        snapshot.match("non-existent-topic", e.value.response)


class TestSNSSubscriptionCrudV2:
    @markers.aws.validated
    def test_subscribe_sms(self, sns_create_topic, aws_client, snapshot):
        topic_name = f"test-topic-{short_uid()}"
        snapshot.add_transformer(snapshot.transform.regex(topic_name, "<topic>"))
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]

        resp = aws_client.sns.subscribe(
            TopicArn=topic_arn,
            Protocol="sms",
            Endpoint="+1234567890",
        )
        snapshot.match("subscribe-sms-1", resp)

    # FIXME
    @pytest.mark.skipif(not is_aws_cloud(), reason="not accepting valid sms endpoint")
    @markers.aws.validated
    def test_subscribe_sms_obscure_phone_number(self, sns_create_topic, aws_client, snapshot):
        topic_name = f"test-topic-{short_uid()}"
        snapshot.add_transformer(snapshot.transform.regex(topic_name, "<topic>"))
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]
        resp = aws_client.sns.subscribe(
            TopicArn=topic_arn,
            Protocol="sms",
            Endpoint="+12/34-567.890",
        )
        snapshot.match("subscribe-sms-2", resp)

    @markers.aws.validated
    def test_subscribe_unknown_topic(self, aws_client, snapshot, account_id, region_name):
        with pytest.raises(ClientError) as e:
            aws_client.sns.subscribe(
                TopicArn=f"arn:aws:sns:{region_name}:{account_id}:unknown-{short_uid()}",
                Protocol="sms",
                Endpoint="+1234567890",
            )
        snapshot.match("subscribe-unknown-topic", e.value.response)

    @markers.aws.validated
    def test_subscribe_unknown_sqs_queue(
        self, sns_create_topic, aws_client, snapshot, account_id, region_name
    ):
        topic_name = f"test-topic-{short_uid()}"
        snapshot.add_transformer(snapshot.transform.regex(topic_name, "<topic>"))
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]
        resp = aws_client.sns.subscribe(
            TopicArn=topic_arn,
            Protocol="sqs",
            Endpoint=f"arn:aws:sqs:{region_name}:{account_id}:unknown-{short_uid()}",
        )
        snapshot.match("subscribe-unknown-sqs", resp)

        with pytest.raises(ClientError) as e:
            aws_client.sns.subscribe(
                TopicArn=topic_arn,
                Protocol="sqs",
                Endpoint="unknown",
            )
        snapshot.match("subscribe-unknown-sqs-invalid_arn", e.value.response)

    @markers.aws.validated
    def test_subscribe_sqs_queue_url(
        self, sns_create_topic, sqs_create_queue, aws_client, snapshot
    ):
        topic_arn = sns_create_topic(Name=f"sqs-topic-{short_uid()}")["TopicArn"]
        queue_url = sqs_create_queue(QueueName=f"queue-{short_uid()}")

        with pytest.raises(ClientError) as e:
            aws_client.sns.subscribe(
                TopicArn=topic_arn,
                Protocol="sqs",
                Endpoint=queue_url,
            )

        snapshot.match("subscribe-sqs-via-url-error", e.value.response)

    @markers.aws.validated
    def test_subscribe_invalid_sms(self, sns_create_topic, aws_client, snapshot):
        topic_arn = sns_create_topic(Name=f"bad-sms-{short_uid()}")["TopicArn"]

        invalid_numbers = ["+15--551234567", "NAA+15551234567", "+15551234567.", "/+15551234567"]

        for number in invalid_numbers:
            with pytest.raises(ClientError) as e:
                aws_client.sns.subscribe(
                    TopicArn=topic_arn,
                    Protocol="sms",
                    Endpoint=number,
                )
            snapshot.match(f"subscribe-bad-sms-{number}", e.value.response)

    # TODO: Parametrize for email protocol as well
    @markers.aws.validated
    def test_creating_subscription(self, sns_create_topic, aws_client, snapshot):
        topic_arn = sns_create_topic(Name=f"create-sub-{short_uid()}")["TopicArn"]

        resp = aws_client.sns.subscribe(
            TopicArn=topic_arn, Protocol="http", Endpoint="http://example.com/"
        )
        # TODO: investigate why it leaves the subscription in `SNS.ListSubscriptions`, we maybe need to clean up
        #  after deleting topics, not fully in parity as AWS cleans up instantly
        snapshot.match("create-subscription", resp)

    # TODO: parametrize for email protocol
    @markers.aws.validated
    def test_unsubscribe_from_deleted_topic(self, sns_create_topic, aws_client, snapshot):
        topic_arn = sns_create_topic(Name=f"del-topic-sub-{short_uid()}")["TopicArn"]

        sub_arn = aws_client.sns.subscribe(
            TopicArn=topic_arn,
            Protocol="http",
            Endpoint="http://example.com/",
        )["SubscriptionArn"]

        aws_client.sns.delete_topic(TopicArn=topic_arn)

        with pytest.raises(ClientError) as e:
            aws_client.sns.unsubscribe(SubscriptionArn=sub_arn)
        snapshot.match("unsubscribe-deleted-topic", e.value.response)

    # TODO: parametrize for http and email protocol
    @markers.aws.validated
    def test_getting_subscriptions_by_topic(
        self, sns_create_topic, sqs_create_queue, sqs_get_queue_arn, aws_client, snapshot
    ):
        topic_arn = sns_create_topic(Name=f"get-subs-{short_uid()}")["TopicArn"]
        queue_url = sqs_create_queue(QueueName=f"queue-{short_uid()}")

        aws_client.sns.subscribe(
            TopicArn=topic_arn,
            Protocol="sqs",
            Endpoint=sqs_get_queue_arn(queue_url),
        )

        resp = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("list-subscriptions-by-topic", resp)

    @markers.aws.validated
    def test_subscription_paging(self, sns_create_topic, aws_client, snapshot, sns_subscription):
        topic_arn = sns_create_topic(Name=f"paging-sub-{short_uid()}")["TopicArn"]

        sub_arns = []
        for i in range(120):
            sub_arns.append(
                sns_subscription(
                    TopicArn=topic_arn,
                    Protocol="email",
                    Endpoint=f"user{i}@example.com",
                )["SubscriptionArn"]
            )

        resp = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        assert "NextToken" in resp

    # TODO: parametrize for http and email protocol
    @markers.aws.validated
    def test_subscribe_attributes(
        self,
        sns_create_topic,
        aws_client,
        snapshot,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_subscription,
    ):
        topic_arn = sns_create_topic(Name=f"get-subs-{short_uid()}")["TopicArn"]
        queue_url = sqs_create_queue(QueueName=f"queue-{short_uid()}")

        subscribe_response = sns_subscription(
            TopicArn=topic_arn,
            Protocol="sqs",
            Endpoint=sqs_get_queue_arn(queue_url),
        )
        sub_arn = subscribe_response["SubscriptionArn"]

        resp = aws_client.sns.get_subscription_attributes(SubscriptionArn=sub_arn)
        snapshot.match("get-subscription-attributes", resp)

    # TODO: parametrize for http and email protocol
    @markers.aws.validated
    def test_creating_subscription_with_attributes(
        self,
        sns_create_topic,
        aws_client,
        snapshot,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_subscription,
    ):
        topic_arn = sns_create_topic(Name=f"get-subs-{short_uid()}")["TopicArn"]
        queue_url = sqs_create_queue(QueueName=f"queue-{short_uid()}")

        sub_arn = sns_subscription(
            TopicArn=topic_arn,
            Protocol="sqs",
            Endpoint=sqs_get_queue_arn(queue_url),
            Attributes={"RawMessageDelivery": "true"},
        )["SubscriptionArn"]

        attrs = aws_client.sns.get_subscription_attributes(SubscriptionArn=sub_arn)
        snapshot.match("subscription-with-attributes", attrs)

    # TODO: parametrize for http and email protocol
    @markers.aws.validated
    def test_set_subscription_attributes(
        self,
        sns_create_topic,
        aws_client,
        snapshot,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_subscription,
    ):
        topic_arn = sns_create_topic(Name=f"get-subs-{short_uid()}")["TopicArn"]
        queue_url = sqs_create_queue(QueueName=f"queue-{short_uid()}")

        sub_arn = sns_subscription(
            TopicArn=topic_arn,
            Protocol="sqs",
            Endpoint=sqs_get_queue_arn(queue_url),
        )["SubscriptionArn"]

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=sub_arn,
            AttributeName="RawMessageDelivery",
            AttributeValue="true",
        )

        attrs = aws_client.sns.get_subscription_attributes(SubscriptionArn=sub_arn)
        snapshot.match("set-subscription-attributes", attrs)

    @markers.snapshot.skip_snapshot_verify(
        paths=["$..Error.Message"],
        # AWS adds a strange stacktrace that contains REDACTED at the end. The "regular" error message matches
    )
    @markers.aws.validated
    def test_subscribe_invalid_filter_policy(self, sns_create_topic, aws_client, snapshot):
        topic_arn = sns_create_topic(Name=f"filter-policy-{short_uid()}")["TopicArn"]

        with pytest.raises(ClientError) as e:
            aws_client.sns.subscribe(
                TopicArn=topic_arn,
                Protocol="email",
                Endpoint="test@example.com",
                Attributes={"FilterPolicy": "invalid-json"},
            )
        snapshot.match("subscribe-invalid-filter-policy", e.value.response)

    # FIXME
    @pytest.mark.skip(
        reason="Https for aws lambda URL required but not implemented in localstack lambda"
    )
    @markers.aws.validated
    def test_confirm_subscription(
        self, sns_create_topic, aws_client, snapshot, create_lambda_function
    ):
        snapshot.add_transformer(TransformerUtility.key_value("SubscriptionArn"))
        topic_arn = sns_create_topic(Name=f"confirm-sub-{short_uid()}")["TopicArn"]
        function_name = f"lambda-{short_uid()}"
        create_lambda_function(
            handler_file=TEST_LAMBDA_PYTHON_ECHO,
            func_name=function_name,
            runtime=Runtime.python3_12,
        )

        url_config = aws_client.lambda_.create_function_url_config(
            FunctionName=function_name, AuthType="NONE"
        )
        aws_client.lambda_.add_permission(
            FunctionName=function_name,
            Action="lambda:InvokeFunctionUrl",
            StatementId="AllowSNSInvoke",
            Principal="*",
            FunctionUrlAuthType="NONE",
        )

        aws_client.sns.subscribe(
            TopicArn=topic_arn,
            Protocol="https",
            Endpoint=url_config["FunctionUrl"],
        )

        def _get_token():
            token_message = aws_client.logs.filter_log_events(
                logGroupName=f"/aws/lambda/{function_name}", filterPattern="Token="
            )["events"][0]["message"]
            confirmation_url = json.loads(json.loads(token_message)["body"])["SubscribeURL"]
            token = confirmation_url.split("Token=")[1]
            return token

        token = retry(_get_token, retries=5, sleep=2)
        resp = aws_client.sns.confirm_subscription(
            TopicArn=topic_arn, Token=token, AuthenticateOnUnsubscribe="true"
        )
        snapshot.match("confirm-subscription", resp)

    @markers.aws.validated
    def test_get_subscription_attributes_error_not_exists(
        self, aws_client, snapshot, account_id, region_name
    ):
        with pytest.raises(ClientError) as e:
            aws_client.sns.get_subscription_attributes(
                SubscriptionArn=f"arn:aws:sns:{region_name}:{account_id}:nonexistent:{uuid.uuid4()}"
            )
        snapshot.match("get-sub-attrs-nonexistent", e.value.response)


class TestSNSSubscriptionLambda:
    @markers.aws.validated
    def test_python_lambda_subscribe_sns_topic(
        self,
        sns_create_topic,
        sns_subscription,
        lambda_su_role,
        create_lambda_function,
        snapshot,
        aws_client,
    ):
        function_name = f"lambda-function-{short_uid()}"
        permission_id = f"test-statement-{short_uid()}"
        subject = "[Subject] Test subject"
        message = "Hello world."
        topic_arn = sns_create_topic()["TopicArn"]

        lambda_creation_response = create_lambda_function(
            func_name=function_name,
            handler_file=TEST_LAMBDA_PYTHON_ECHO,
            runtime=Runtime.python3_12,
            role=lambda_su_role,
        )
        lambda_arn = lambda_creation_response["CreateFunctionResponse"]["FunctionArn"]
        aws_client.lambda_.add_permission(
            FunctionName=function_name,
            StatementId=permission_id,
            Action="lambda:InvokeFunction",
            Principal="sns.amazonaws.com",
            SourceArn=topic_arn,
        )

        subscription = sns_subscription(
            TopicArn=topic_arn,
            Protocol="lambda",
            Endpoint=lambda_arn,
        )

        def check_subscription():
            subscription_arn = subscription["SubscriptionArn"]
            subscription_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=subscription_arn
            )
            assert subscription_attrs["Attributes"]["PendingConfirmation"] == "false"

        retry(check_subscription, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

        aws_client.sns.publish(TopicArn=topic_arn, Subject=subject, Message=message)

        # access events sent by lambda
        events = retry(
            check_expected_lambda_log_events_length,
            retries=10,
            sleep=1,
            function_name=function_name,
            expected_length=1,
            regex_filter="Records.*Sns",
            logs_client=aws_client.logs,
        )
        notification = events[0]["Records"][0]["Sns"]
        snapshot.match("notification", notification)

    @markers.aws.validated
    def test_sns_topic_as_lambda_dead_letter_queue(
        self,
        lambda_su_role,
        create_lambda_function,
        sns_create_topic,
        sqs_create_queue,
        sns_subscription,
        sns_create_sqs_subscription,
        snapshot,
        aws_client,
    ):
        """Tests an async event chain: SNS => Lambda => SNS DLQ => SQS
        1) SNS => Lambda: An SNS subscription triggers the Lambda function asynchronously.
        2) Lambda => SNS DLQ: A failing Lambda function triggers the SNS DLQ after all retries are exhausted.
        3) SNS DLQ => SQS: An SNS subscription forwards the DLQ message to SQS.
        """
        snapshot.add_transformer(
            snapshot.transform.jsonpath(
                "$..Messages..MessageAttributes.RequestID.Value", "request-id"
            )
        )

        # create an SNS topic that will be used as a DLQ by the lambda
        dlq_topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        # sqs_subscription
        sns_create_sqs_subscription(topic_arn=dlq_topic_arn, queue_url=queue_url)

        # create an SNS topic that will be used to invoke the lambda
        lambda_topic_arn = sns_create_topic()["TopicArn"]

        function_name = f"lambda-function-{short_uid()}"
        lambda_creation_response = create_lambda_function(
            func_name=function_name,
            handler_file=TEST_LAMBDA_PYTHON,
            runtime=Runtime.python3_12,
            role=lambda_su_role,
            DeadLetterConfig={"TargetArn": dlq_topic_arn},
        )
        snapshot.match(
            "lambda-response-dlq-config",
            lambda_creation_response["CreateFunctionResponse"]["DeadLetterConfig"],
        )
        lambda_arn = lambda_creation_response["CreateFunctionResponse"]["FunctionArn"]

        # allow the SNS topic to invoke the lambda
        permission_id = f"test-statement-{short_uid()}"
        aws_client.lambda_.add_permission(
            FunctionName=function_name,
            StatementId=permission_id,
            Action="lambda:InvokeFunction",
            Principal="sns.amazonaws.com",
            SourceArn=lambda_topic_arn,
        )

        # subscribe the lambda to the SNS topic: lambda_subscription
        sns_subscription(
            TopicArn=lambda_topic_arn,
            Protocol="lambda",
            Endpoint=lambda_arn,
        )

        # Set retries to zero to speed up the test
        aws_client.lambda_.put_function_event_invoke_config(
            FunctionName=function_name,
            MaximumRetryAttempts=0,
        )

        payload = {
            lambda_integration.MSG_BODY_RAISE_ERROR_FLAG: 1,
        }
        aws_client.sns.publish(TopicArn=lambda_topic_arn, Message=json.dumps(payload))

        def receive_dlq():
            result = aws_client.sqs.receive_message(
                QueueUrl=queue_url, MessageAttributeNames=["All"], VisibilityTimeout=0
            )
            assert len(result["Messages"]) > 0
            return result

        sleep = 3 if is_aws_cloud() else 1
        messages = retry(receive_dlq, retries=30, sleep=sleep)

        messages["Messages"][0]["Body"] = json.loads(messages["Messages"][0]["Body"])
        messages["Messages"][0]["Body"]["Message"] = json.loads(
            messages["Messages"][0]["Body"]["Message"]
        )

        snapshot.match("messages", messages)

    @markers.aws.validated
    def test_redrive_policy_lambda_subscription(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        create_lambda_function,
        lambda_su_role,
        sns_subscription,
        sns_allow_topic_sqs_queue,
        snapshot,
        aws_client,
    ):
        dlq_url = sqs_create_queue()
        dlq_arn = sqs_get_queue_arn(dlq_url)
        topic_arn = sns_create_topic()["TopicArn"]
        sns_allow_topic_sqs_queue(
            sqs_queue_url=dlq_url, sqs_queue_arn=dlq_arn, sns_topic_arn=topic_arn
        )

        lambda_name = f"test-{short_uid()}"
        lambda_arn = create_lambda_function(
            func_name=lambda_name,
            handler_file=TEST_LAMBDA_PYTHON,
            runtime=Runtime.python3_12,
            role=lambda_su_role,
        )["CreateFunctionResponse"]["FunctionArn"]

        subscription = sns_subscription(TopicArn=topic_arn, Protocol="lambda", Endpoint=lambda_arn)

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
            AttributeName="RedrivePolicy",
            AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
        )
        response_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"]
        )

        snapshot.match("subscription-attributes", response_attributes)

        aws_client.lambda_.delete_function(FunctionName=lambda_name)

        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test_redrive_policy",
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": "1"}},
        )

        response = aws_client.sqs.receive_message(
            QueueUrl=dlq_url, WaitTimeSeconds=10, MessageAttributeNames=["All"]
        )
        snapshot.match("messages", response)

    @markers.aws.validated
    @pytest.mark.parametrize("signature_version", ["1", "2"])
    def test_publish_lambda_verify_signature(
        self,
        aws_client,
        sns_create_topic,
        create_lambda_function,
        sns_subscription,
        lambda_su_role,
        snapshot,
        signature_version,
    ):
        # Lambda always returns SignatureVersion=1 in messages, however, it can be v2 and the signature needs to be
        # verified against v2 (SHA256). Weird bug on AWS side, we will do the same for now.

        function_name = f"lambda-function-{short_uid()}"
        permission_id = f"test-statement-{short_uid()}"
        subject = f"[Subject] Test subject Signature v{signature_version}"
        message = "Hello world."
        topic_arn = sns_create_topic(
            Attributes={
                "DisplayName": "TestTopicSignatureLambda",
                "SignatureVersion": signature_version,
            },
        )["TopicArn"]

        lambda_creation_response = create_lambda_function(
            func_name=function_name,
            handler_file=TEST_LAMBDA_PYTHON_ECHO,
            runtime=Runtime.python3_12,
            role=lambda_su_role,
        )
        lambda_arn = lambda_creation_response["CreateFunctionResponse"]["FunctionArn"]
        aws_client.lambda_.add_permission(
            FunctionName=function_name,
            StatementId=permission_id,
            Action="lambda:InvokeFunction",
            Principal="sns.amazonaws.com",
            SourceArn=topic_arn,
        )

        subscription = sns_subscription(
            TopicArn=topic_arn,
            Protocol="lambda",
            Endpoint=lambda_arn,
        )

        def check_subscription():
            subscription_arn = subscription["SubscriptionArn"]
            subscription_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=subscription_arn
            )
            assert subscription_attrs["Attributes"]["PendingConfirmation"] == "false"

        retry(check_subscription, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

        aws_client.sns.publish(TopicArn=topic_arn, Subject=subject, Message=message)

        # access events sent by lambda
        events = retry(
            check_expected_lambda_log_events_length,
            retries=10,
            sleep=1,
            function_name=function_name,
            expected_length=1,
            regex_filter="Records.*Sns",
            logs_client=aws_client.logs,
        )

        message = events[0]["Records"][0]["Sns"]
        snapshot.match("notification", message)

        cert_url = message["SigningCertUrl"]
        get_cert_req = requests.get(cert_url)
        assert get_cert_req.ok

        cert = x509.load_pem_x509_certificate(get_cert_req.content)
        message_signature = message["Signature"]
        # create the canonical string
        fields = ["Message", "MessageId", "Subject", "Timestamp", "TopicArn", "Type"]
        # Build the string to be signed.
        string_to_sign = "".join(
            [f"{field}\n{message[field]}\n" for field in fields if field in message]
        )

        # decode the signature from base64.
        decoded_signature = base64.b64decode(message_signature)

        message_sig_version = message["SignatureVersion"]
        # this is a bug on AWS side, assert our behaviour is the same for now, this might get fixed
        assert message_sig_version == "1"
        signature_hash = hashes.SHA1() if signature_version == "1" else hashes.SHA256()

        # calculate signature value with cert
        is_valid = cert.public_key().verify(
            decoded_signature,
            to_bytes(string_to_sign),
            padding=padding.PKCS1v15(),
            algorithm=signature_hash,
        )
        # if the verification failed, it would raise `InvalidSignature`
        assert is_valid is None


class TestSNSSubscriptionSQS:
    @markers.aws.validated
    def test_subscribe_sqs_queue(
        self, sqs_create_queue, sns_create_topic, sns_create_sqs_subscription, snapshot, aws_client
    ):
        # TODO: check with non default external port

        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        # create subscription with filter policy
        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        filter_policy = {"attr1": [{"numeric": [">", 0, "<=", 100]}]}
        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
            AttributeName="FilterPolicy",
            AttributeValue=json.dumps(filter_policy),
        )

        response_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
        )
        snapshot.match("subscription-attributes", response_attributes)

        # publish message that satisfies the filter policy
        message = "This is a test message"
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": "99.12"}},
        )

        # assert that message is received
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            VisibilityTimeout=0,
            MessageAttributeNames=["All"],
            WaitTimeSeconds=4,
        )
        snapshot.match("messages", response)

    @markers.aws.validated
    def test_publish_unicode_chars(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        # publish message to SNS, receive it from SQS, assert that messages are equal
        message = 'ö§a1"_!?,. £$-'
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url, VisibilityTimeout=0, WaitTimeSeconds=4
        )

        snapshot.match("received-message", response)

    @markers.aws.validated
    def test_attribute_raw_subscribe(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        # the hash isn't the same because of the Binary attributes (maybe decoding order?)
        snapshot.add_transformer(
            snapshot.transform.key_value(
                "MD5OfMessageAttributes",
                value_replacement="<md5-hash>",
                reference_replacement=False,
            )
        )
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        subscription = sns_create_sqs_subscription(
            topic_arn=topic_arn, queue_url=queue_url, Attributes={"RawMessageDelivery": "true"}
        )
        subscription_arn = subscription["SubscriptionArn"]

        response_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscription_arn
        )
        snapshot.match("subscription-attributes", response_attributes)

        # publish message to SNS, receive it from SQS, assert that messages are equal and that they are Raw
        message = "This is a test message"
        binary_attribute = b"\x02\x03\x04"
        # extending this test case to test support for binary message attribute data
        # https://github.com/localstack/localstack/issues/2432
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageAttributes={"store": {"DataType": "Binary", "BinaryValue": binary_attribute}},
        )

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            MessageAttributeNames=["All"],
            VisibilityTimeout=0,
            WaitTimeSeconds=4,
        )
        snapshot.match("messages-response", response)

    @markers.aws.validated
    def test_sqs_topic_subscription_confirmation(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        subscription_attrs = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        def check_subscription():
            nonlocal subscription_attrs
            if not subscription_attrs["PendingConfirmation"] == "false":
                subscription_arn = subscription_attrs["SubscriptionArn"]
                subscription_attrs = aws_client.sns.get_subscription_attributes(
                    SubscriptionArn=subscription_arn
                )["Attributes"]
            else:
                snapshot.match("subscription-attrs", subscription_attrs)

            return subscription_attrs["PendingConfirmation"] == "false"

        # SQS subscriptions are auto confirmed if the endpoint and the topic are in the same AWS account
        assert poll_condition(check_subscription, timeout=5)

    @markers.aws.validated
    def test_publish_sqs_from_sns(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        subscription_arn = subscription["SubscriptionArn"]

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="RawMessageDelivery",
            AttributeValue="true",
        )
        response = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("sub-attrs-raw-true", response)

        string_value = "99.12"
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="Test msg",
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": string_value}},
        )

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            MessageAttributeNames=["All"],
            VisibilityTimeout=0,
            WaitTimeSeconds=4,
        )
        snapshot.match("message-raw-true", response)
        # format is of SQS MessageAttributes when RawDelivery is set to "true"
        assert response["Messages"][0]["MessageAttributes"] == {
            "attr1": {"DataType": "Number", "StringValue": string_value}
        }

        aws_client.sqs.delete_message(
            QueueUrl=queue_url, ReceiptHandle=response["Messages"][0]["ReceiptHandle"]
        )

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="RawMessageDelivery",
            AttributeValue="false",
        )
        response = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("sub-attrs-raw-false", response)

        string_value = "100.12"
        aws_client.sns.publish(
            TargetArn=topic_arn,
            Message="Test msg",
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": string_value}},
        )
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            MessageAttributeNames=["All"],
            VisibilityTimeout=0,
            WaitTimeSeconds=4,
        )
        snapshot.match("message-raw-false", response)
        message_body = json.loads(response["Messages"][0]["Body"])
        # format is SNS MessageAttributes when RawDelivery is "false"
        assert message_body["MessageAttributes"] == {
            "attr1": {"Type": "Number", "Value": string_value}
        }

    @markers.aws.validated
    def test_publish_batch_messages_from_sns_to_sqs(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()
        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        subscription_arn = subscription["SubscriptionArn"]

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="RawMessageDelivery",
            AttributeValue="true",
        )
        response = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("sub-attrs-raw-true", response)

        publish_batch_response = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=[
                {
                    "Id": "1",
                    "Message": "Test Message with two attributes",
                    "Subject": "Subject",
                    "MessageAttributes": {
                        "attr1": {"DataType": "Number", "StringValue": "99.12"},
                        "attr2": {"DataType": "Number", "StringValue": "109.12"},
                    },
                },
                {
                    "Id": "2",
                    "Message": "Test Message with one attribute",
                    "Subject": "Subject",
                    "MessageAttributes": {"attr1": {"DataType": "Number", "StringValue": "19.12"}},
                },
                {
                    "Id": "3",
                    "Message": "Test Message without attribute",
                    "Subject": "Subject",
                },
                {
                    "Id": "4",
                    "Message": "Test Message without subject",
                },
                {
                    "Id": "5",
                    "Message": json.dumps({"default": "test default", "sqs": "test sqs"}),
                    "MessageStructure": "json",
                },
            ],
        )
        snapshot.match("publish-batch", publish_batch_response)

        messages = []

        def get_messages():
            # due to the random nature of receiving SQS messages, we need to consolidate a single object to match
            sqs_response = aws_client.sqs.receive_message(
                QueueUrl=queue_url,
                WaitTimeSeconds=1,
                VisibilityTimeout=0,
                MessageAttributeNames=["All"],
                AttributeNames=["All"],
            )
            for message in sqs_response["Messages"]:
                messages.append(message)
                aws_client.sqs.delete_message(
                    QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"]
                )

            assert len(messages) == 5

        retry(get_messages, retries=10, sleep=0.1)
        # we need to sort the list (the order does not matter as we're not using FIFO)
        messages.sort(key=itemgetter("Body"))
        snapshot.match("messages", {"Messages": messages})

    @markers.aws.validated
    def test_publish_batch_messages_without_topic(self, sns_create_topic, snapshot, aws_client):
        topic_arn = sns_create_topic()["TopicArn"]
        fake_topic_arn = topic_arn + "fake-topic"

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=fake_topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": "Test Message with two attributes",
                        "Subject": "Subject",
                    }
                ],
            )
        snapshot.match("publish-batch-no-topic", e.value.response)

    @markers.aws.validated
    def test_publish_batch_exceptions(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        fifo_topic_name = f"topic-{short_uid()}.fifo"
        topic_arn = sns_create_topic(Name=fifo_topic_name, Attributes={"FifoTopic": "true"})[
            "TopicArn"
        ]

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": "Test message without Group ID",
                    }
                ],
            )
        snapshot.match("no-group-id", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {"Id": f"Id_{i}", "Message": "Too many messages"} for i in range(11)
                ],
            )
        snapshot.match("too-many-msg", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {"Id": "1", "Message": "Messages with the same ID"} for _ in range(2)
                ],
            )
        snapshot.match("same-msg-id", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": "Test message without MessageDeduplicationId",
                        "MessageGroupId": "msg1",
                    }
                ],
            )
        snapshot.match("no-dedup-id", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "Message": json.dumps({"sqs": "test sqs"}),
                        "MessageStructure": "json",
                    }
                ],
            )
        snapshot.match("no-default-key-json", e.value.response)

    @markers.aws.validated
    def test_subscribe_to_sqs_with_queue_url(
        self, sns_create_topic, sqs_create_queue, sns_subscription, snapshot
    ):
        topic = sns_create_topic()
        topic_arn = topic["TopicArn"]
        queue_url = sqs_create_queue()
        with pytest.raises(ClientError) as e:
            sns_subscription(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_url)
        snapshot.match("sub-queue-url", e.value.response)

    @markers.aws.validated
    def test_publish_sqs_from_sns_with_xray_propagation(
        self, sns_create_topic, sqs_create_queue, sns_create_sqs_subscription, snapshot, aws_client
    ):
        def add_xray_header(request, **_kwargs):
            request.headers["X-Amzn-Trace-Id"] = (
                "Root=1-3152b799-8954dae64eda91bc9a23a7e8;Parent=7fa8c0f79203be72;Sampled=1"
            )

        try:
            aws_client.sns.meta.events.register("before-send.sns.Publish", add_xray_header)

            topic = sns_create_topic()
            topic_arn = topic["TopicArn"]
            queue_url = sqs_create_queue()
            sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
            aws_client.sns.publish(TargetArn=topic_arn, Message="X-Ray propagation test msg")

            response = aws_client.sqs.receive_message(
                QueueUrl=queue_url,
                AttributeNames=["SentTimestamp", "AWSTraceHeader"],
                MaxNumberOfMessages=1,
                MessageAttributeNames=["All"],
                VisibilityTimeout=2,
                WaitTimeSeconds=2,
            )

            assert len(response["Messages"]) == 1
            message = response["Messages"][0]
            snapshot.match("xray-msg", message)
            assert (
                message["Attributes"]["AWSTraceHeader"]
                == "Root=1-3152b799-8954dae64eda91bc9a23a7e8;Parent=7fa8c0f79203be72;Sampled=1"
            )
        finally:
            aws_client.sns.meta.events.unregister("before-send.sns.Publish", add_xray_header)

    @pytest.mark.parametrize("raw_message_delivery", [True, False])
    @markers.aws.validated
    def test_redrive_policy_sqs_queue_subscription(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        sqs_queue_exists,
        sns_create_sqs_subscription,
        sns_allow_topic_sqs_queue,
        raw_message_delivery,
        snapshot,
        aws_client,
    ):
        # the hash isn't the same because of the Binary attributes (maybe decoding order?)
        snapshot.add_transformer(
            snapshot.transform.key_value(
                "MD5OfMessageAttributes",
                value_replacement="<md5-hash>",
                reference_replacement=False,
            )
        )
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        dlq_url = sqs_create_queue()
        dlq_arn = sqs_get_queue_arn(dlq_url)

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
            AttributeName="RedrivePolicy",
            AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
        )

        if raw_message_delivery:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=subscription["SubscriptionArn"],
                AttributeName="RawMessageDelivery",
                AttributeValue="true",
            )

        sns_allow_topic_sqs_queue(
            sqs_queue_url=dlq_url,
            sqs_queue_arn=dlq_arn,
            sns_topic_arn=topic_arn,
        )

        aws_client.sqs.delete_queue(QueueUrl=queue_url)

        # AWS takes some time to delete the queue, which make the test fails as it delivers the message correctly
        assert poll_condition(lambda: not sqs_queue_exists(queue_url), timeout=5)

        message = "test_dlq_after_sqs_endpoint_deleted"
        message_attr = {
            "attr1": {
                "DataType": "Number",
                "StringValue": "111",
            },
            "attr2": {
                "DataType": "Binary",
                "BinaryValue": b"\x02\x03\x04",
            },
        }
        aws_client.sns.publish(TopicArn=topic_arn, Message=message, MessageAttributes=message_attr)

        response = aws_client.sqs.receive_message(
            QueueUrl=dlq_url,
            WaitTimeSeconds=10,
            AttributeNames=["All"],
            MessageAttributeNames=["All"],
        )
        snapshot.match("messages", response)

    @markers.aws.validated
    def test_message_attributes_not_missing(
        self, sns_create_sqs_subscription, sns_create_topic, sqs_create_queue, snapshot, aws_client
    ):
        # the hash isn't the same because of the Binary attributes (maybe decoding order?)
        snapshot.add_transformer(
            snapshot.transform.key_value(
                "MD5OfMessageAttributes",
                value_replacement="<md5-hash>",
                reference_replacement=False,
            )
        )
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
            AttributeName="RawMessageDelivery",
            AttributeValue="true",
        )
        attributes = {
            "an-attribute-key": {"DataType": "String", "StringValue": "an-attribute-value"},
            "binary-attribute": {"DataType": "Binary", "BinaryValue": b"\x02\x03\x04"},
        }

        publish_response = aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="text",
            MessageAttributes=attributes,
        )
        snapshot.match("publish-msg-raw", publish_response)

        msg = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            AttributeNames=["All"],
            MessageAttributeNames=["All"],
            WaitTimeSeconds=3,
        )
        # as SNS piggybacks on SQS MessageAttributes when RawDelivery is true
        # BinaryValue depends on SQS implementation, and is decoded automatically
        snapshot.match("raw-delivery-msg-attrs", msg)

        aws_client.sqs.delete_message(
            QueueUrl=queue_url, ReceiptHandle=msg["Messages"][0]["ReceiptHandle"]
        )

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
            AttributeName="RawMessageDelivery",
            AttributeValue="false",
        )

        publish_response = aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="text",
            MessageAttributes=attributes,
        )
        snapshot.match("publish-msg-json", publish_response)

        msg = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            AttributeNames=["All"],
            MessageAttributeNames=["All"],
            WaitTimeSeconds=3,
        )
        snapshot.match("json-delivery-msg-attrs", msg)
        # binary payload in base64 encoded by AWS, UTF-8 for JSON
        # https://docs.aws.amazon.com/sns/latest/api/API_MessageAttributeValue.html

    @markers.aws.validated
    def test_subscription_after_failure_to_deliver(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        sqs_queue_exists,
        sns_create_sqs_subscription,
        sns_allow_topic_sqs_queue,
        sqs_receive_num_messages,
        snapshot,
        aws_client,
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_name = f"test-queue-{short_uid()}"
        queue_url = sqs_create_queue(QueueName=queue_name)

        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        subscription_arn = subscription["SubscriptionArn"]

        dlq_url = sqs_create_queue()
        dlq_arn = sqs_get_queue_arn(dlq_url)

        sns_allow_topic_sqs_queue(
            sqs_queue_url=dlq_url,
            sqs_queue_arn=dlq_arn,
            sns_topic_arn=topic_arn,
        )

        sub_attrs = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("subscriptions-attrs", sub_attrs)

        message = "test_dlq_before_sqs_endpoint_deleted"
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url, WaitTimeSeconds=10, MaxNumberOfMessages=4
        )
        snapshot.match("messages-before-delete", response)
        aws_client.sqs.delete_message(
            QueueUrl=queue_url, ReceiptHandle=response["Messages"][0]["ReceiptHandle"]
        )

        aws_client.sqs.delete_queue(QueueUrl=queue_url)

        # setting up a second queue to be able to poll and know approximately when the message on the deleted queue
        # have been published
        queue_test_url = sqs_create_queue()
        test_subscription = sns_create_sqs_subscription(
            topic_arn=topic_arn, queue_url=queue_test_url
        )
        test_subscription_arn = test_subscription["SubscriptionArn"]
        # try to send a message before setting a DLQ
        message = "test_dlq_after_sqs_endpoint_deleted"
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)

        # to avoid race condition, publish is async and the redrive policy can be in effect before the actual publish
        # we wait until the 2nd subscription received the message
        poll_condition(
            lambda: sqs_receive_num_messages(
                queue_url=queue_test_url, expected_messages=1, max_iterations=2
            ),
            timeout=10,
        )
        aws_client.sns.unsubscribe(SubscriptionArn=test_subscription_arn)
        # we still wait a bit to be sure the message is well published
        time.sleep(1)

        # check the subscription is still there after we deleted the queue
        subscriptions = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("subscriptions", subscriptions)

        # set the RedrivePolicy with a DLQ. Subsequent failing messages to the subscription should go there
        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="RedrivePolicy",
            AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
        )

        sub_attrs = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("subscriptions-attrs-with-redrive", sub_attrs)

        # AWS takes some time to delete the queue, which make the test fails as it delivers the message correctly
        assert poll_condition(lambda: not sqs_queue_exists(queue_url), timeout=5)

        # test sending and receiving multiple messages
        for i in range(2):
            message = f"test_dlq_after_sqs_endpoint_deleted_{i}"

            aws_client.sns.publish(TopicArn=topic_arn, Message=message)
            response = aws_client.sqs.receive_message(
                QueueUrl=dlq_url, WaitTimeSeconds=10, MaxNumberOfMessages=4
            )
            aws_client.sqs.delete_message(
                QueueUrl=dlq_url, ReceiptHandle=response["Messages"][0]["ReceiptHandle"]
            )

            snapshot.match(f"message-{i}-after-delete", response)

    @markers.aws.validated
    def test_empty_or_wrong_message_attributes(
        self,
        sns_create_sqs_subscription,
        sns_create_topic,
        sqs_create_queue,
        snapshot,
        aws_client_factory,
        region_name,
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        client_no_validation = aws_client_factory(
            region_name=region_name, config=Config(parameter_validation=False)
        ).sns

        wrong_message_attributes = {
            "missing_string_attr": {"attr1": {"DataType": "String", "StringValue": ""}},
            "fully_missing_string_attr": {"attr1": {"DataType": "String"}},
            "fully_missing_data_type": {"attr1": {"StringValue": "value"}},
            "missing_binary_attr": {"attr1": {"DataType": "Binary", "BinaryValue": b""}},
            "str_attr_binary_value": {"attr1": {"DataType": "String", "BinaryValue": b"123"}},
            "int_attr_binary_value": {"attr1": {"DataType": "Number", "BinaryValue": b"123"}},
            "binary_attr_string_value": {"attr1": {"DataType": "Binary", "StringValue": "123"}},
            "invalid_attr_string_value": {
                "attr1": {"DataType": "InvalidType", "StringValue": "123"}
            },
            "too_long_name": {"a" * 257: {"DataType": "String", "StringValue": "123"}},
            "invalid_name": {"a^*?": {"DataType": "String", "StringValue": "123"}},
            "invalid_name_2": {".abc": {"DataType": "String", "StringValue": "123"}},
            "invalid_name_3": {"abc.": {"DataType": "String", "StringValue": "123"}},
            "invalid_name_4": {"a..bc": {"DataType": "String", "StringValue": "123"}},
        }

        for error_type, msg_attrs in wrong_message_attributes.items():
            with pytest.raises(ClientError) as e:
                client_no_validation.publish(
                    TopicArn=topic_arn,
                    Message="test message",
                    MessageAttributes=msg_attrs,
                )

            snapshot.match(error_type, e.value.response)

            with pytest.raises(ClientError) as e:
                client_no_validation.publish_batch(
                    TopicArn=topic_arn,
                    PublishBatchRequestEntries=[
                        {
                            "Id": "1",
                            "Message": "test-batch",
                            "MessageAttributes": msg_attrs,
                        },
                        {
                            "Id": "3",
                            "Message": "valid-batch",
                        },
                    ],
                )
            snapshot.match(f"batch-{error_type}", e.value.response)

    @markers.aws.validated
    def test_message_attributes_prefixes(
        self, sns_create_sqs_subscription, sns_create_topic, sqs_create_queue, snapshot, aws_client
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_url = sqs_create_queue()

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(
                TopicArn=topic_arn,
                Message="test message",
                MessageAttributes={"attr1": {"DataType": "String.", "StringValue": "prefixed-1"}},
            )
        snapshot.match("publish-error", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(
                TopicArn=topic_arn,
                Message="test message",
                MessageAttributes={
                    "attr1": {"DataType": "Stringprefixed", "StringValue": "prefixed-1"}
                },
            )
        snapshot.match("publish-error-2", e.value.response)

        response = aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test message",
            MessageAttributes={
                "attr1": {"DataType": "String.prefixed", "StringValue": "prefixed-1"}
            },
        )
        snapshot.match("publish-ok-1", response)

        response = aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test message",
            MessageAttributes={
                "attr1": {"DataType": "String.  prefixed.", "StringValue": "prefixed-1"}
            },
        )
        snapshot.match("publish-ok-2", response)

    @markers.aws.validated
    def test_message_structure_json_to_sqs(
        self, aws_client, sns_create_topic, sqs_create_queue, snapshot, sns_create_sqs_subscription
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        queue_name = f"test-queue-{short_uid()}"
        queue_url = sqs_create_queue(QueueName=queue_name)

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        message = json.dumps({"default": "default field", "sqs": json.dumps({"field": "value"})})
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageStructure="json",
        )
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url, WaitTimeSeconds=10, MaxNumberOfMessages=1
        )
        snapshot.match("get-msg-json-sqs", response)
        receipt_handle = response["Messages"][0]["ReceiptHandle"]
        aws_client.sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)

        # don't json dumps the SQS field, it will be ignored, and the message received will be the `default`
        message = json.dumps({"default": "default field", "sqs": {"field": "value"}})
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageStructure="json",
        )
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url, WaitTimeSeconds=10, MaxNumberOfMessages=1
        )
        snapshot.match("get-msg-json-default", response)

    @markers.aws.validated
    @pytest.mark.parametrize("signature_version", ["1", "2"])
    def test_publish_sqs_verify_signature(
        self,
        aws_client,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        signature_version,
    ):
        topic_arn = sns_create_topic(
            Attributes={
                "DisplayName": "TestTopicSignature",
                "SignatureVersion": signature_version,
            },
        )["TopicArn"]

        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test signature value with attributes",
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": "1"}},
        )
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=10,
            AttributeNames=["All"],
            MessageAttributeNames=["All"],
        )
        snapshot.match("messages", response)
        message = json.loads(response["Messages"][0]["Body"])

        cert_url = message["SigningCertURL"]
        get_cert_req = requests.get(cert_url)
        assert get_cert_req.ok

        cert = x509.load_pem_x509_certificate(get_cert_req.content)
        message_signature = message["Signature"]
        # create the canonical string
        fields = ["Message", "MessageId", "Subject", "Timestamp", "TopicArn", "Type"]
        # Build the string to be signed.
        string_to_sign = "".join(
            [f"{field}\n{message[field]}\n" for field in fields if field in message]
        )

        # decode the signature from base64.
        decoded_signature = base64.b64decode(message_signature)

        message_sig_version = message["SignatureVersion"]
        assert message_sig_version == signature_version
        signature_hash = hashes.SHA1() if message_sig_version == "1" else hashes.SHA256()

        # calculate signature value with cert
        is_valid = cert.public_key().verify(
            decoded_signature,
            to_bytes(string_to_sign),
            padding=padding.PKCS1v15(),
            algorithm=signature_hash,
        )
        # if the verification failed, it would raise `InvalidSignature`
        assert is_valid is None

    @markers.aws.validated
    def test_publish_message_group_id(
        self,
        aws_client,
        sns_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
    ):
        topic_arn = sns_topic["Attributes"]["TopicArn"]
        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test signature value with attributes",
            MessageGroupId="my-group-id-1",
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": "1"}},
        )
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=10,
            AttributeNames=["All"],
            MessageAttributeNames=["All"],
        )
        snapshot.match("messages", response)


class TestSNSSubscriptionSQSFifo:
    @markers.aws.validated
    @pytest.mark.parametrize("content_based_deduplication", [True, False])
    def test_message_to_fifo_sqs(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        content_based_deduplication,
        aws_client,
    ):
        topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}.fifo"
        topic_attributes = {"FifoTopic": "true"}
        queue_attributes = {"FifoQueue": "true"}
        if content_based_deduplication:
            topic_attributes["ContentBasedDeduplication"] = "true"
            queue_attributes["ContentBasedDeduplication"] = "true"

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]
        queue_url = sqs_create_queue(
            QueueName=queue_name,
            Attributes=queue_attributes,
        )

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        message = "Test"
        kwargs = {"MessageGroupId": "message-group-id-1"}
        if not content_based_deduplication:
            kwargs["MessageDeduplicationId"] = "message-deduplication-id-1"

        aws_client.sns.publish(TopicArn=topic_arn, Message=message, **kwargs)

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=10,
            AttributeNames=["All"],
        )
        snapshot.match("messages", response)

        aws_client.sqs.delete_message(
            QueueUrl=queue_url, ReceiptHandle=response["Messages"][0]["ReceiptHandle"]
        )
        # republish the message, to check deduplication
        aws_client.sns.publish(TopicArn=topic_arn, Message=message, **kwargs)
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=1,
            AttributeNames=["All"],
        )
        snapshot.match("dedup-messages", response)

    @markers.aws.validated
    @pytest.mark.parametrize("content_based_deduplication", [True, False])
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.dedup-messages.Messages"
        ],  # FIXME: introduce deduplication at Topic level, not only SQS
    )
    def test_fifo_topic_to_regular_sqs(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        content_based_deduplication,
        aws_client,
    ):
        # it seems change is coming on AWS, as FIFO topic do not require FIFO queues anymore. This might mean that
        # the FIFO logic is being migrated to SNS and do not rely on SQS FIFO anymore? or that the FIFO is only
        # guaranteed with FIFO queues, but you can also subscribe with regular subscribers for deduplication for ex. ?
        # The change in error message suggest the latter:
        # "RedrivePolicy: must use a FIFO queue as DLQ for a FIFO topic" became:
        # -> "RedrivePolicy: must use a FIFO queue as DLQ for a FIFO Subscription to a FIFO Topic."

        topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}"
        topic_attributes = {"FifoTopic": "true"}
        if content_based_deduplication:
            topic_attributes["ContentBasedDeduplication"] = "true"

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]
        queue_url = sqs_create_queue(QueueName=queue_name)

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        message = "Test"
        kwargs = {"MessageGroupId": "message-group-id-1"}
        if not content_based_deduplication:
            kwargs["MessageDeduplicationId"] = "message-deduplication-id-1"

        aws_client.sns.publish(TopicArn=topic_arn, Message=message, **kwargs)

        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=10,
            AttributeNames=["All"],
        )
        snapshot.match("messages", response)

        aws_client.sqs.delete_message(
            QueueUrl=queue_url, ReceiptHandle=response["Messages"][0]["ReceiptHandle"]
        )
        # republish the message, to check deduplication
        # TODO: not implemented in LocalStack yet, only deduplication with FIFO SQS queues
        aws_client.sns.publish(TopicArn=topic_arn, Message=message, **kwargs)
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=3,
            AttributeNames=["All"],
        )
        snapshot.match("dedup-messages", response)

    @markers.aws.validated
    def test_validations_for_fifo(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_create_sqs_subscription,
        snapshot,
        aws_client,
    ):
        topic_name = f"topic-{short_uid()}"
        fifo_topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}"
        fifo_queue_name = f"queue-{short_uid()}.fifo"
        not_fifo_dlq_name = f"queue-dlq-{short_uid()}"

        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]

        fifo_topic_arn = sns_create_topic(Name=fifo_topic_name, Attributes={"FifoTopic": "true"})[
            "TopicArn"
        ]

        fifo_queue_url = sqs_create_queue(
            QueueName=fifo_queue_name, Attributes={"FifoQueue": "true"}
        )

        queue_url = sqs_create_queue(QueueName=queue_name)
        not_fifo_dlq_url = sqs_create_queue(QueueName=not_fifo_dlq_name)

        with pytest.raises(ClientError) as e:
            sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=fifo_queue_url)

        assert e.match("standard SNS topic")
        snapshot.match("not-fifo-topic", e.value.response)

        # SNS does not reject a regular SQS queue subscribed to a FIFO topic anymore
        subscription_not_fifo = sns_create_sqs_subscription(
            topic_arn=fifo_topic_arn, queue_url=queue_url
        )
        snapshot.match("not-fifo-queue", subscription_not_fifo)

        not_fifo_queue_arn = sqs_get_queue_arn(not_fifo_dlq_url)
        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_not_fifo["SubscriptionArn"],
            AttributeName="RedrivePolicy",
            AttributeValue=json.dumps({"deadLetterTargetArn": not_fifo_queue_arn}),
        )

        subscription = sns_create_sqs_subscription(
            topic_arn=fifo_topic_arn, queue_url=fifo_queue_url
        )
        queue_arn = sqs_get_queue_arn(queue_url)

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=subscription["SubscriptionArn"],
                AttributeName="RedrivePolicy",
                AttributeValue=json.dumps({"deadLetterTargetArn": queue_arn}),
            )
        snapshot.match("regular-queue-for-dlq-of-fifo-topic", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(TopicArn=fifo_topic_arn, Message="test")

        assert e.match("MessageGroupId")
        snapshot.match("no-msg-group-id", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(
                TopicArn=fifo_topic_arn, Message="test", MessageGroupId=short_uid()
            )
        # if ContentBasedDeduplication is not set at the topic level, it needs MessageDeduplicationId for each msg
        assert e.match("MessageDeduplicationId")
        assert e.match("ContentBasedDeduplication")
        snapshot.match("no-dedup-policy", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(
                TopicArn=topic_arn, Message="test", MessageDeduplicationId=short_uid()
            )
        assert e.match("MessageDeduplicationId")
        snapshot.match("no-msg-dedup-regular-topic", e.value.response)

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.topic-attrs.Attributes.DeliveryPolicy",
            "$.topic-attrs.Attributes.EffectiveDeliveryPolicy",
            "$.topic-attrs.Attributes.Policy.Statement..Action",  # SNS:Receive is added by moto but not returned in AWS
        ],
        condition=is_sns_v1_provider,
    )
    @pytest.mark.parametrize("raw_message_delivery", [True, False])
    def test_publish_fifo_messages_to_dlq(
        self,
        sns_create_topic,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_create_sqs_subscription,
        sns_allow_topic_sqs_queue,
        snapshot,
        raw_message_delivery,
        aws_client,
    ):
        # the hash isn't the same because of the Binary attributes (maybe decoding order?)
        snapshot.add_transformer(
            snapshot.transform.key_value(
                "MD5OfMessageAttributes",
                value_replacement="<md5-hash>",
                reference_replacement=False,
            )
        )

        topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}.fifo"
        dlq_name = f"dlq-{short_uid()}.fifo"

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes={"FifoTopic": "true"},
        )["TopicArn"]

        response = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("topic-attrs", response)

        queue_url = sqs_create_queue(
            QueueName=queue_name,
            Attributes={"FifoQueue": "true"},
        )

        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        subscription_arn = subscription["SubscriptionArn"]

        if raw_message_delivery:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=subscription_arn,
                AttributeName="RawMessageDelivery",
                AttributeValue="true",
            )

        dlq_url = sqs_create_queue(
            QueueName=dlq_name,
            Attributes={"FifoQueue": "true"},
        )
        dlq_arn = sqs_get_queue_arn(dlq_url)

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"],
            AttributeName="RedrivePolicy",
            AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
        )

        sns_allow_topic_sqs_queue(
            sqs_queue_url=dlq_url,
            sqs_queue_arn=dlq_arn,
            sns_topic_arn=topic_arn,
        )

        aws_client.sqs.delete_queue(QueueUrl=queue_url)

        message_group_id = "complexMessageGroupId"
        publish_batch_request_entries = [
            {
                "Id": "1",
                "MessageGroupId": message_group_id,
                "Message": "Test Message with two attributes",
                "Subject": "Subject",
                "MessageAttributes": {
                    "attr1": {"DataType": "Number", "StringValue": "99.12"},
                    "attr2": {"DataType": "Number", "StringValue": "109.12"},
                },
                "MessageDeduplicationId": "MessageDeduplicationId-1",
            },
            {
                "Id": "2",
                "MessageGroupId": message_group_id,
                "Message": "Test Message with one attribute",
                "Subject": "Subject",
                "MessageAttributes": {"attr1": {"DataType": "Number", "StringValue": "19.12"}},
                "MessageDeduplicationId": "MessageDeduplicationId-2",
            },
            {
                "Id": "3",
                "MessageGroupId": message_group_id,
                "Message": "Test Message without attribute",
                "Subject": "Subject",
                "MessageDeduplicationId": "MessageDeduplicationId-3",
            },
        ]

        publish_batch_response = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=publish_batch_request_entries,
        )

        snapshot.match("publish-batch-response-fifo", publish_batch_response)

        assert "Successful" in publish_batch_response
        assert "Failed" in publish_batch_response

        for successful_resp in publish_batch_response["Successful"]:
            assert "Id" in successful_resp
            assert "MessageId" in successful_resp

        message_ids_received = set()
        messages = []

        def get_messages_from_dlq(amount_msg: int):
            # due to the random nature of receiving SQS messages, we need to consolidate a single object to match
            # MaxNumberOfMessages could return less than 3 messages
            sqs_response = aws_client.sqs.receive_message(
                QueueUrl=dlq_url,
                MessageAttributeNames=["All"],
                AttributeNames=["All"],
                MaxNumberOfMessages=10,
                WaitTimeSeconds=1,
                VisibilityTimeout=1,
            )

            for message in sqs_response["Messages"]:
                LOG.debug("Message received %s", message)
                if message["MessageId"] in message_ids_received:
                    continue

                message_ids_received.add(message["MessageId"])
                messages.append(message)
                aws_client.sqs.delete_message(
                    QueueUrl=dlq_url, ReceiptHandle=message["ReceiptHandle"]
                )

            assert len(messages) == amount_msg

        retry(get_messages_from_dlq, retries=5, sleep=1, amount_msg=3)
        snapshot.match("batch-messages-in-dlq", {"Messages": messages})
        messages.clear()

        publish_response = aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test-message",
            MessageGroupId="message-group-id-1",
            MessageDeduplicationId="message-deduplication-id-1",
        )
        snapshot.match("publish-response-fifo", publish_response)
        retry(get_messages_from_dlq, retries=5, sleep=1, amount_msg=1)
        snapshot.match("messages-in-dlq", {"Messages": messages})

    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.republish-batch-response-fifo.Successful..MessageId",  # TODO: SNS doesnt keep track of duplicate
            "$.republish-batch-response-fifo.Successful..SequenceNumber",  # TODO: SNS doesnt keep track of duplicate
        ],
    )
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.topic-attrs.Attributes.DeliveryPolicy",
            "$.topic-attrs.Attributes.EffectiveDeliveryPolicy",
            "$.topic-attrs.Attributes.Policy.Statement..Action",  # SNS:Receive is added by moto but not returned in AWS
        ],
        condition=is_sns_v1_provider,
    )
    @pytest.mark.parametrize("content_based_deduplication", [True, False])
    def test_publish_batch_messages_from_fifo_topic_to_fifo_queue(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        content_based_deduplication,
        aws_client,
    ):
        topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}.fifo"
        topic_attributes = {"FifoTopic": "true"}
        queue_attributes = {"FifoQueue": "true"}
        if content_based_deduplication:
            topic_attributes["ContentBasedDeduplication"] = "true"
            queue_attributes["ContentBasedDeduplication"] = "true"

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]

        response = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("topic-attrs", response)

        queue_url = sqs_create_queue(
            QueueName=queue_name,
            Attributes=queue_attributes,
        )

        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
        subscription_arn = subscription["SubscriptionArn"]

        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="RawMessageDelivery",
            AttributeValue="true",
        )

        response = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("sub-attrs-raw-true", response)
        message_group_id = "complexMessageGroupId"
        publish_batch_request_entries = [
            {
                "Id": "1",
                "MessageGroupId": message_group_id,
                "Message": "Test Message with two attributes",
                "Subject": "Subject",
                "MessageAttributes": {
                    "attr1": {"DataType": "Number", "StringValue": "99.12"},
                    "attr2": {"DataType": "Number", "StringValue": "109.12"},
                },
            },
            {
                "Id": "2",
                "MessageGroupId": message_group_id,
                "Message": "Test Message with one attribute",
                "Subject": "Subject",
                "MessageAttributes": {"attr1": {"DataType": "Number", "StringValue": "19.12"}},
            },
            {
                "Id": "3",
                "MessageGroupId": message_group_id,
                "Message": "Test Message without attribute",
                "Subject": "Subject",
            },
        ]

        if not content_based_deduplication:
            for index, message in enumerate(publish_batch_request_entries):
                message["MessageDeduplicationId"] = f"MessageDeduplicationId-{index}"

        publish_batch_response = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=publish_batch_request_entries,
        )

        snapshot.match("publish-batch-response-fifo", publish_batch_response)

        assert "Successful" in publish_batch_response
        assert "Failed" in publish_batch_response

        for successful_resp in publish_batch_response["Successful"]:
            assert "Id" in successful_resp
            assert "MessageId" in successful_resp

        message_ids_received = set()
        messages = []

        def get_messages():
            # due to the random nature of receiving SQS messages, we need to consolidate a single object to match
            # MaxNumberOfMessages could return less than 3 messages
            sqs_response = aws_client.sqs.receive_message(
                QueueUrl=queue_url,
                MessageAttributeNames=["All"],
                AttributeNames=["All"],
                MaxNumberOfMessages=10,
                WaitTimeSeconds=1,
                VisibilityTimeout=10,
            )

            for _message in sqs_response["Messages"]:
                if _message["MessageId"] in message_ids_received:
                    continue

                message_ids_received.add(_message["MessageId"])
                messages.append(_message)
                aws_client.sqs.delete_message(
                    QueueUrl=queue_url, ReceiptHandle=_message["ReceiptHandle"]
                )

            assert len(messages) == 3

        retry(get_messages, retries=5, sleep=1)
        snapshot.match("messages", {"Messages": messages})

        publish_batch_response = aws_client.sns.publish_batch(
            TopicArn=topic_arn,
            PublishBatchRequestEntries=publish_batch_request_entries,
        )

        snapshot.match("republish-batch-response-fifo", publish_batch_response)
        get_deduplicated_messages = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            MessageAttributeNames=["All"],
            AttributeNames=["All"],
            MaxNumberOfMessages=10,
            WaitTimeSeconds=3,
            VisibilityTimeout=0,
        )
        # there should not be any messages here, as they are duplicate
        # see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html
        snapshot.match("duplicate-messages", get_deduplicated_messages)

    @markers.aws.validated
    @pytest.mark.parametrize("raw_message_delivery", [True, False])
    def test_publish_to_fifo_topic_to_sqs_queue_no_content_dedup(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        raw_message_delivery,
        aws_client,
    ):
        topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}.fifo"
        topic_attributes = {"FifoTopic": "true", "ContentBasedDeduplication": "true"}
        queue_attributes = {"FifoQueue": "true"}

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]
        queue_url = sqs_create_queue(
            QueueName=queue_name,
            Attributes=queue_attributes,
        )

        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        if raw_message_delivery:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=subscription["SubscriptionArn"],
                AttributeName="RawMessageDelivery",
                AttributeValue="true",
            )

        # Topic has ContentBasedDeduplication set to true, the queue should receive only one message
        # SNS will create a MessageDeduplicationId for the SQS queue, as it does not have ContentBasedDeduplication
        for _ in range(2):
            aws_client.sns.publish(
                TopicArn=topic_arn, Message="Test single", MessageGroupId="message-group-id-1"
            )
            aws_client.sns.publish_batch(
                TopicArn=topic_arn,
                PublishBatchRequestEntries=[
                    {
                        "Id": "1",
                        "MessageGroupId": "message-group-id-1",
                        "Message": "Test batched",
                    }
                ],
            )

        messages = []
        message_ids_received = set()

        def get_messages():
            # due to the random nature of receiving SQS messages, we need to consolidate a single object to match
            # MaxNumberOfMessages could return less than 2 messages
            sqs_response = aws_client.sqs.receive_message(
                QueueUrl=queue_url,
                MessageAttributeNames=["All"],
                AttributeNames=["All"],
                MaxNumberOfMessages=10,
                WaitTimeSeconds=1,
                VisibilityTimeout=10,
            )

            for message in sqs_response["Messages"]:
                if message["MessageId"] in message_ids_received:
                    continue

                message_ids_received.add(message["MessageId"])
                messages.append(message)
                aws_client.sqs.delete_message(
                    QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"]
                )

            assert len(messages) == 2

        retry(get_messages, retries=5, sleep=1)
        messages.sort(key=lambda x: x["Attributes"]["MessageDeduplicationId"])
        snapshot.match("messages", {"Messages": messages})

    @markers.aws.validated
    def test_publish_to_fifo_topic_deduplication_on_topic_level(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        aws_client,
    ):
        topic_name = f"topic-{short_uid()}.fifo"
        queue_name = f"queue-{short_uid()}.fifo"
        topic_attributes = {"FifoTopic": "true", "ContentBasedDeduplication": "true"}
        queue_attributes = {"FifoQueue": "true"}

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]
        queue_url = sqs_create_queue(
            QueueName=queue_name,
            Attributes=queue_attributes,
        )

        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        # TODO: for message deduplication, we are using the underlying features of the SQS queue
        # however, SQS queue only deduplicate at the Queue level, where the SNS topic deduplicate on the topic level
        # we will need to implement this
        # TODO: add a test with 2 subscriptions and a filter, to validate deduplication at topic level
        message = "Test"
        aws_client.sns.publish(
            TopicArn=topic_arn, Message=message, MessageGroupId="message-group-id-1"
        )
        time.sleep(
            0.5
        )  # this is to ensure order of arrival, because we do not deduplicate at SNS level yet
        aws_client.sns.publish(
            TopicArn=topic_arn, Message=message, MessageGroupId="message-group-id-2"
        )

        # get the deduplicated message and delete it
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            VisibilityTimeout=10,
            WaitTimeSeconds=10,
            AttributeNames=["All"],
        )
        snapshot.match("messages", response)
        aws_client.sqs.delete_message(
            QueueUrl=queue_url, ReceiptHandle=response["Messages"][0]["ReceiptHandle"]
        )
        # assert there are no more messages in the queue
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            VisibilityTimeout=10,
            WaitTimeSeconds=1,
            AttributeNames=["All"],
        )
        snapshot.match("dedup-messages", response)

    @markers.aws.validated
    def test_publish_to_fifo_with_target_arn(self, sns_create_topic, aws_client):
        topic_name = f"topic-{short_uid()}.fifo"
        topic_attributes = {
            "FifoTopic": "true",
            "ContentBasedDeduplication": "true",
        }

        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]

        message = {"foo": "bar"}
        response = aws_client.sns.publish(
            TargetArn=topic_arn,
            Message=json.dumps({"default": json.dumps(message)}),
            MessageStructure="json",
            MessageGroupId="123",
        )
        assert "MessageId" in response

    @markers.aws.validated
    def test_message_to_fifo_sqs_ordering(
        self,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        snapshot,
        aws_client,
        sqs_collect_messages,
    ):
        topic_name = f"topic-{short_uid()}.fifo"
        topic_attributes = {"FifoTopic": "true", "ContentBasedDeduplication": "true"}
        topic_arn = sns_create_topic(
            Name=topic_name,
            Attributes=topic_attributes,
        )["TopicArn"]

        queue_attributes = {"FifoQueue": "true", "ContentBasedDeduplication": "true"}
        queues = []
        queue_amount = 5
        message_amount = 10

        for _ in range(queue_amount):
            queue_name = f"queue-{short_uid()}.fifo"
            queue_url = sqs_create_queue(
                QueueName=queue_name,
                Attributes=queue_attributes,
            )
            sns_create_sqs_subscription(
                topic_arn=topic_arn, queue_url=queue_url, Attributes={"RawMessageDelivery": "true"}
            )
            queues.append(queue_url)

        for i in range(message_amount):
            aws_client.sns.publish(
                TopicArn=topic_arn, Message=str(i), MessageGroupId="message-group-id-1"
            )

        all_messages = []
        for queue_url in queues:
            messages = sqs_collect_messages(
                queue_url,
                expected=message_amount,
                timeout=10,
                max_number_of_messages=message_amount,
            )
            contents = [message["Body"] for message in messages]
            all_messages.append(contents)

        # we're expecting the order to be the same across all queues
        reference_order = all_messages[0]
        for received_content in all_messages[1:]:
            assert received_content == reference_order


class TestSNSSubscriptionSES:
    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_topic_email_subscription_confirmation(
        self, sns_create_topic, sns_subscription, aws_client, sns_provider
    ):
        # FIXME: we do not send the token to the email endpoint, so they cannot validate it
        # create AWS validated test for format
        # for now, access internals
        topic_arn = sns_create_topic()["TopicArn"]
        subscription = sns_subscription(
            TopicArn=topic_arn,
            Protocol="email",
            Endpoint="localstack@yopmail.com",
        )
        subscription_arn = subscription["SubscriptionArn"]
        parsed_arn = parse_arn(subscription_arn)
        SnsProvider = sns_provider()
        store = SnsProvider.get_store(parsed_arn["account"], parsed_arn["region"])

        sub_attr = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        assert sub_attr["Attributes"]["PendingConfirmation"] == "true"

        def check_subscription():
            for token, sub_arn in store.subscription_tokens.items():
                if sub_arn == subscription_arn:
                    aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token=token)

            sub_attributes = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=subscription_arn
            )
            assert sub_attributes["Attributes"]["PendingConfirmation"] == "false"

        retry(check_subscription, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_email_sender(
        self,
        sns_create_topic,
        sns_subscription,
        aws_client,
        monkeypatch,
    ):
        # make sure to reset all received emails in SES
        requests.delete("http://localhost:4566/_aws/ses")

        topic_arn = sns_create_topic()["TopicArn"]
        sns_subscription(
            TopicArn=topic_arn,
            Protocol="email",
            Endpoint="localstack@yopmail.com",
        )

        aws_client.sns.publish(
            Message="Test message",
            TopicArn=topic_arn,
        )

        def _get_messages(amount: int) -> list[dict]:
            response = requests.get("http://localhost:4566/_aws/ses").json()
            assert len(response["messages"]) == amount
            return response["messages"]

        messages = retry(lambda: _get_messages(1), retries=PUBLICATION_RETRIES, sleep=1)
        # legacy default value, should be replaced at some point
        assert messages[0]["Source"] == "admin@localstack.com"
        requests.delete("http://localhost:4566/_aws/ses")

        sender_address = "no-reply@sns.localstack.cloud"
        monkeypatch.setattr(config, "SNS_SES_SENDER_ADDRESS", sender_address)

        aws_client.sns.publish(
            Message="Test message",
            TopicArn=topic_arn,
        )
        messages = retry(lambda: _get_messages(1), retries=PUBLICATION_RETRIES, sleep=1)
        assert messages[0]["Source"] == sender_address


@pytest.fixture(scope="class")
def platform_credentials() -> tuple[str, str]:
    # these values need to be extracted from a real amazon developer account if tested against AWS
    # https://developer.amazon.com/settings/console/securityprofile/overview.html
    client_id = "dummy"
    client_secret = "dummy"
    return client_id, client_secret


@pytest.fixture(scope="class")
def e_mail_address() -> str:
    # this address must be real and accessible if you want to test email subscriptions
    e_mail = "test@example.com"
    return e_mail


@pytest.fixture(scope="class")
def phone_number() -> str:
    # if you want to test phone number operations against AWS and a real phone number, replace this value
    # and use this fixture.
    # note: you might need to verify that number first in your AWS account due to the sms sandbox
    phone_number = "+430000000000"
    return phone_number


class TestSNSPlatformApplicationCrud:
    @markers.aws.manual_setup_required
    def test_create_platform_application(
        self, aws_client, snapshot, sns_create_platform_application, platform_credentials
    ):
        platform = "ADM"
        # if tested against AWS, the fixture needs to contain real credentials
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        response = sns_create_platform_application(Platform=platform, Attributes=attributes)
        snapshot.match("create-platform-application", response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.manual_setup_required
    def test_list_platform_applications(
        self, aws_client, snapshot, sns_create_platform_application, platform_credentials
    ):
        name = f"platform-application-{short_uid()}"
        platform = "ADM"
        # if tested against AWS, the fixture needs to contain real credentials
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        sns_create_platform_application(Name=name, Platform=platform, Attributes=attributes)

        response = aws_client.sns.list_platform_applications()
        snapshot.match("list-platform-applications", response)

    @pytest.mark.parametrize(
        "attributes",
        [
            {},
            {"PlatformPrincipal": "dummy"},
            {"PlatformCredential": "dummy"},
        ],
        ids=["no-args", "missing-credential", "missing-principal"],
    )
    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.validated
    def test_create_platform_application_invalid_attributes(self, aws_client, snapshot, attributes):
        # We cannot actually verify the validity of the passed credentials, which is why it is not part of this test
        name = f"platform-application-{short_uid()}"
        platform = "ADM"
        with pytest.raises(ClientError) as e:
            aws_client.sns.create_platform_application(
                Name=name, Platform=platform, Attributes=attributes
            )
        snapshot.match("platform-application-no-attributes", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @pytest.mark.parametrize(
        "name", [f"{'a' * 257}", "", "@name"], ids=["too-long", "empty", "invalid-char"]
    )
    @markers.aws.validated
    def test_create_platform_application_invalid_name(self, aws_client, snapshot, name):
        attributes = {"PlatformPrincipal": "dummy", "PlatformCredential": "dummy"}
        platform = "ADM"
        with pytest.raises(ClientError) as e:
            aws_client.sns.create_platform_application(
                Name=name, Platform=platform, Attributes=attributes
            )
        snapshot.match("invalid-application-name", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.validated
    def test_create_platform_application_invalid_platform(self, aws_client, snapshot):
        attributes = {"PlatformPrincipal": "dummy", "PlatformCredential": "dummy"}

        name = f"platform-application-{short_uid()}"
        invalid_platform = "AAA"
        with pytest.raises(ClientError) as e:
            aws_client.sns.create_platform_application(
                Name=name, Platform=invalid_platform, Attributes=attributes
            )
        snapshot.match("invalid-platform", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.manual_setup_required
    def test_get_platform_application_attributes(
        self, aws_client, snapshot, sns_create_platform_application, platform_credentials
    ):
        name = f"platform-application-{short_uid()}"
        platform = "ADM"
        # if tested against AWS, the fixture needs to contain real credentials
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        platform_application_arn = sns_create_platform_application(
            Name=name, Platform=platform, Attributes=attributes
        )["PlatformApplicationArn"]
        response = aws_client.sns.get_platform_application_attributes(
            PlatformApplicationArn=platform_application_arn
        )
        snapshot.match("get-application-attributes", response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.validated
    def test_get_platform_application_attributes_invalid_arn(self, aws_client, snapshot):
        with pytest.raises(ClientError) as e:
            aws_client.sns.get_platform_application_attributes(PlatformApplicationArn="invalid-arn")
        snapshot.match("invalid-application-arn", e.value.response)

    @markers.aws.validated
    def test_get_platform_application_attributes_non_existing_app(
        self, aws_client, snapshot, account_id, region_name
    ):
        with pytest.raises(ClientError) as e:
            aws_client.sns.get_platform_application_attributes(
                PlatformApplicationArn=f"arn:aws:sns:{region_name}:{account_id}:app/ADM/NonExistingApp"
            )
        snapshot.match("non_existing-application-arn", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.manual_setup_required
    def test_set_platform_application_attributes(
        self, aws_client, snapshot, sns_create_platform_application, platform_credentials
    ):
        name = f"platform-application-{short_uid()}"
        platform = "ADM"
        # if tested against AWS, the fixture needs to contain real credentials
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        platform_application_arn = sns_create_platform_application(
            Name=name, Platform=platform, Attributes=attributes
        )["PlatformApplicationArn"]

        attributes = {"SuccessFeedbackSampleRate": "50"}
        aws_client.sns.set_platform_application_attributes(
            Attributes=attributes, PlatformApplicationArn=platform_application_arn
        )
        # give the attributes time to propagate
        if is_aws_cloud():
            time.sleep(30)
        response = aws_client.sns.get_platform_application_attributes(
            PlatformApplicationArn=platform_application_arn
        )
        snapshot.match("set-get-application-attributes", response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.validated
    def test_set_platform_application_attributes_invalid_arn(self, aws_client, snapshot):
        with pytest.raises(ClientError) as e:
            aws_client.sns.set_platform_application_attributes(
                PlatformApplicationArn="invalid-arn", Attributes={}
            )
        snapshot.match("invalid-application-arn", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @pytest.mark.parametrize(
        "attributes",
        [
            {},
            {"PlatformPrincipal": "dummy"},
        ],
        ids=["no-args", "dummy-args"],
    )
    @markers.aws.validated
    def test_set_platform_application_attributes_non_existing_app(
        self, aws_client, snapshot, account_id, region_name, attributes
    ):
        with pytest.raises(ClientError) as e:
            aws_client.sns.set_platform_application_attributes(
                PlatformApplicationArn=f"arn:aws:sns:{region_name}:{account_id}:app/ADM/NonExistingApp",
                Attributes=attributes,
            )
        snapshot.match("non_existing-application-arn", e.value.response)


class TestSNSPlatformEndpointCrud:
    @markers.aws.manual_setup_required
    def test_create_platform_endpoint(
        self,
        sns_create_platform_application,
        sns_create_platform_endpoint,
        aws_client,
        account_id,
        region_name,
        platform_credentials,
        snapshot,
    ):
        # if tested against AWS, the fixture needs to contain real credentials
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        response = sns_create_platform_endpoint(platform_application_arn=app_arn, token="token_1")
        snapshot.match("create-platform-endpoint", response)

    @markers.aws.manual_setup_required
    def test_create_platform_endpoint_idempotency(
        self,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        platform_credentials,
        sns_create_platform_endpoint,
        snapshot,
    ):
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        response = sns_create_platform_endpoint(platform_application_arn=app_arn, token="token_1")
        snapshot.match("create-platform-endpoint", response)

        # create again with the same attributes and token
        response = sns_create_platform_endpoint(platform_application_arn=app_arn, token="token_1")
        snapshot.match("create-platform-endpoint-idempotent", response)

        # create again with different attributes
        with pytest.raises(ClientError) as e:
            sns_create_platform_endpoint(
                platform_application_arn=app_arn, token="token_1", Attributes={"Enabled": "false"}
            )
        snapshot.match("create-platform-endpoint-different-token", e.value.response)

    @markers.aws.validated
    def test_create_platform_endpoint_non_existent_app(
        self,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        snapshot,
    ):
        platform_application_arn = "arn:aws:sns:%s:%s:app/ADM/non_existent_app"
        with pytest.raises(ClientError) as e:
            aws_client.sns.create_platform_endpoint(
                PlatformApplicationArn=platform_application_arn % (region_name, account_id),
                Token="token_1",
            )
        snapshot.match("create-platform-endpoint", e.value.response)

    @markers.aws.manual_setup_required
    def test_list_platform_endpoints(
        self,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        platform_credentials,
        sns_create_platform_endpoint,
        snapshot,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        sns_create_platform_endpoint(platform_application_arn=app_arn, token="token_1")
        response = aws_client.sns.list_endpoints_by_platform_application(
            PlatformApplicationArn=app_arn
        )

        snapshot.match("list-endpoints-by-app", response)

    @markers.aws.manual_setup_required
    def test_delete_platform_endpoint(
        self,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        platform_credentials,
        sns_create_platform_endpoint,
        snapshot,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1"
        )["EndpointArn"]
        response = aws_client.sns.list_endpoints_by_platform_application(
            PlatformApplicationArn=app_arn
        )
        snapshot.match("list-endpoints-by-app-pre-delete", response)

        response = aws_client.sns.delete_endpoint(EndpointArn=endpoint_arn)
        snapshot.match("delete-endpoint", response)
        if is_aws_cloud():
            time.sleep(10)
        response = aws_client.sns.list_endpoints_by_platform_application(
            PlatformApplicationArn=app_arn
        )
        snapshot.match("list-endpoints-by-app-post-delete", response)

    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$..NextToken"
        ],  # FIXME: investigate why AWS is not deterministic here, NextToken is sometimes present, sometimes not
    )
    @markers.aws.manual_setup_required
    def test_delete_platform_endpoint_with_subscription(
        self,
        sns_create_platform_application,
        sns_create_topic,
        sns_subscription,
        aws_client,
        account_id,
        region_name,
        platform_credentials,
        sns_create_platform_endpoint,
        snapshot,
    ):
        # from the docs https://docs.aws.amazon.com/cli/latest/reference/sns/delete-endpoint.html
        # "When you delete an endpoint that is also subscribed to a topic, then
        # you must also unsubscribe the endpoint from the topic."
        # This test validates this particular case

        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        snapshot.add_transformer(snapshot.transform.key_value("NextToken"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1"
        )["EndpointArn"]
        topic_arn = sns_create_topic()["TopicArn"]
        sns_subscription(TopicArn=topic_arn, Protocol="application", Endpoint=endpoint_arn)

        # time to let the changes propagate
        if is_aws_cloud():
            time.sleep(20)

        # we list subscriptions by topic, as some tests are not cleaning up properly
        response = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("list-subscriptions-pre-delete", response)

        response = aws_client.sns.delete_endpoint(EndpointArn=endpoint_arn)
        snapshot.match("delete-endpoint", response)

        # time to let the changes propagate
        if is_aws_cloud():
            time.sleep(20)

        response = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("list-subscriptions-post-delete", response)

    @markers.aws.manual_setup_required
    def test_delete_endpoints_of_deleted_app(
        self,
        sns_create_platform_application,
        sns_create_topic,
        sns_subscription,
        aws_client,
        account_id,
        region_name,
        platform_credentials,
        sns_create_platform_endpoint,
        snapshot,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1"
        )["EndpointArn"]

        response = aws_client.sns.delete_platform_application(PlatformApplicationArn=app_arn)
        snapshot.match("delete-application", response)
        if is_aws_cloud():
            time.sleep(10)
        with pytest.raises(ClientError) as e:
            aws_client.sns.list_endpoints_by_platform_application(PlatformApplicationArn=app_arn)
        snapshot.match("list-endpoints-after-app-delete", e.value.response)

        aws_client.sns.delete_endpoint(EndpointArn=endpoint_arn)

    @markers.aws.manual_setup_required
    def test_get_platform_endpoint_attributes(
        self,
        aws_client,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1"
        )["EndpointArn"]

        response = aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
        snapshot.match("get-platform-endpoint-attributes", response)

    @markers.aws.manual_setup_required
    def test_set_platform_endpoint_attributes(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=attributes
        )["PlatformApplicationArn"]
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1"
        )["EndpointArn"]

        new_attributes = {"Enabled": "false"}
        response = aws_client.sns.set_endpoint_attributes(
            EndpointArn=endpoint_arn, Attributes=new_attributes
        )
        snapshot.match("set-platform-endpoint-attributes", response)

        response = aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
        snapshot.match("get-platform-endpoint-attributes", response)

    @markers.aws.validated
    def test_get_platform_endpoint_attributes_non_existent_endpoint(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
    ):
        endpoint_arn = f"arn:aws:sns:{region_name}:{account_id}:endpoint/ADM/fakeapp/{uuid.uuid4()}"
        with pytest.raises(ClientError) as e:
            aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
        snapshot.match("set-platform-endpoint-attributes-non-existent-app", e.value.response)

    @markers.aws.validated
    def test_set_platform_endpoint_attributes_non_existent_endpoint(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
    ):
        endpoint_arn = f"arn:aws:sns:{region_name}:{account_id}:endpoint/ADM/fakeapp/{uuid.uuid4()}"
        attributes = {"Enabled": "false"}
        with pytest.raises(ClientError) as e:
            aws_client.sns.set_endpoint_attributes(EndpointArn=endpoint_arn, Attributes=attributes)
        snapshot.match("set-platform-endpoint-attributes-non-existent-app", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @pytest.mark.parametrize(
        "attributes",
        [
            {},
            {"PlatformPrincipal": "Principal"},
            {"PlatformCredential": "Credential"},
            {"InvalidKey": "Value"},
            {"CustomUserData": "A" * 2050},
        ],
        ids=[
            "Empty",
            "Invalid_Name_Principal",
            "Invalid_Name_Credential",
            "Invalid_Name_Generic",
            "Data_Too_Long",
        ],
    )
    @markers.aws.manual_setup_required
    def test_set_platform_endpoint_attributes_invalid_attributes(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
        attributes,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        initial_attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=initial_attributes
        )["PlatformApplicationArn"]
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1"
        )["EndpointArn"]

        with pytest.raises(ClientError) as e:
            aws_client.sns.set_endpoint_attributes(EndpointArn=endpoint_arn, Attributes=attributes)
        snapshot.match("set-platform-endpoint-invalid-attributes", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @pytest.mark.parametrize(
        "attributes",
        [
            {"PlatformPrincipal": "Principal"},
            {"PlatformCredential": "Credential"},
            {"InvalidKey": "Value"},
            {"CustomUserData": "A" * 2050},
        ],
        ids=[
            "Invalid_Name_Principal",
            "Invalid_Name_Credential",
            "Invalid_Name_Generic",
            "Data_Too_Long",
        ],
    )
    @markers.aws.manual_setup_required
    def test_create_platform_endpoint_with_invalid_attributes(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
        attributes,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        platform_attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=platform_attributes
        )["PlatformApplicationArn"]
        with pytest.raises(ClientError) as e:
            sns_create_platform_endpoint(
                platform_application_arn=app_arn, token="token_1", Attributes=attributes
            )
        snapshot.match("create-platform-endpoint-invalid-attr", e.value.response)

    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Parity gap with old provider")
    @markers.aws.manual_setup_required
    def test_create_platform_endpoint_custom_data(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
    ):
        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        platform_attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=platform_attributes
        )["PlatformApplicationArn"]

        custom_user_data = "bar"
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn, token="token_1", CustomUserData=custom_user_data
        )["EndpointArn"]

        response = aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
        snapshot.match("create-endpoint-double-custom-data", response)

    @markers.aws.manual_setup_required
    def test_create_platform_endpoint_double_custom_data(
        self,
        snapshot,
        platform_credentials,
        sns_create_platform_endpoint,
        sns_create_platform_application,
        aws_client,
    ):
        # For some reason, CustomUserData can be specified both as parameter directly and inside attributes.

        app_name = f"platform-application-{short_uid()}"
        snapshot.add_transformer(RegexTransformer(app_name, "<platform_application>"))
        # if tested against AWS, the fixture needs to contain real credentials
        principal, credential = platform_credentials
        platform_attributes = {"PlatformPrincipal": principal, "PlatformCredential": credential}
        app_arn = sns_create_platform_application(
            Name=app_name, Platform="ADM", Attributes=platform_attributes
        )["PlatformApplicationArn"]

        attributes = {"CustomUserData": "foo"}
        custom_user_data = "bar"
        endpoint_arn = sns_create_platform_endpoint(
            platform_application_arn=app_arn,
            token="token_1",
            Attributes=attributes,
            CustomUserData=custom_user_data,
        )["EndpointArn"]

        response = aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
        snapshot.match("create-endpoint-double-custom-data", response)


class TestSNSPlatformEndpoint:
    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_subscribe_platform_endpoint(
        self,
        sns_create_topic,
        sns_subscription,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        sns_provider,
        platform_credentials,
    ):
        SnsProvider = sns_provider()
        sns_backend = SnsProvider.get_store(account_id, region_name)
        topic_arn = sns_create_topic()["TopicArn"]

        platform = "ADM"
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        app_arn = sns_create_platform_application(
            Name="app1", Platform=platform, Attributes=attributes
        )["PlatformApplicationArn"]
        platform_arn = aws_client.sns.create_platform_endpoint(
            PlatformApplicationArn=app_arn, Token="token_1"
        )["EndpointArn"]

        # create subscription with filter policy
        filter_policy = {"attr1": [{"numeric": [">", 0, "<=", 100]}]}
        sns_subscription(
            TopicArn=topic_arn,
            Protocol="application",
            Endpoint=platform_arn,
            Attributes={"FilterPolicy": json.dumps(filter_policy)},
        )
        # publish message that satisfies the filter policy
        message = "This is a test message"
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": "99.12"}},
        )

        # assert that message has been received
        def check_message():
            assert len(sns_backend.platform_endpoint_messages[platform_arn]) > 0

        retry(check_message, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

    @markers.aws.needs_fixing
    @pytest.mark.skip(reason="Test asserts wrong behaviour")
    # AWS validating this is hard because we need real credentials for a GCM/Apple mobile app
    # TODO: AWS validate this test
    # See https://github.com/getmoto/moto/pull/6953 where Moto updated errors.
    def test_create_platform_endpoint_check_idempotency(
        self, sns_create_platform_application, aws_client
    ):
        response = sns_create_platform_application(
            Name=f"test-{short_uid()}",
            Platform="GCM",
            Attributes={"PlatformCredential": "123"},
        )
        token = "test1"
        # TODO: As per AWS docs:
        # > The CreatePlatformEndpoint action is idempotent, so if the requester already owns an endpoint
        # > with the same device token and attributes, that endpoint's ARN is returned without creating a new endpoint.
        # The 'Token' and 'Attributes' are critical to idempotent behaviour.
        kwargs_list = [
            {"Token": token, "CustomUserData": "test-data"},
            {"Token": token, "CustomUserData": "test-data"},
            {"Token": token},
            {"Token": token},
        ]
        platform_arn = response["PlatformApplicationArn"]
        responses = []
        for kwargs in kwargs_list:
            responses.append(
                aws_client.sns.create_platform_endpoint(
                    PlatformApplicationArn=platform_arn, **kwargs
                )
            )
        # Assert EndpointArn is returned in every call create platform call
        assert all("EndpointArn" in response for response in responses)
        endpoint_arn = responses[0]["EndpointArn"]

        with pytest.raises(ClientError) as e:
            aws_client.sns.create_platform_endpoint(
                PlatformApplicationArn=platform_arn,
                Token=token,
                CustomUserData="different-user-data",
            )
        assert e.value.response["Error"]["Code"] == "InvalidParameter"
        assert (
            e.value.response["Error"]["Message"]
            == f"Endpoint {endpoint_arn} already exists with the same Token, but different attributes."
        )

    @markers.aws.needs_fixing
    # AWS validating this is hard because we need real credentials for a GCM/Apple mobile app
    def test_publish_disabled_endpoint(
        self, sns_create_platform_application, aws_client, platform_credentials
    ):
        platform = "ADM"
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        response = sns_create_platform_application(
            Name=f"test-{short_uid()}",
            Platform=platform,
            Attributes=attributes,
        )
        platform_arn = response["PlatformApplicationArn"]
        response = aws_client.sns.create_platform_endpoint(
            PlatformApplicationArn=platform_arn,
            Token="test1",
        )
        endpoint_arn = response["EndpointArn"]

        get_attrs = aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
        assert get_attrs["Attributes"]["Enabled"] == "true"

        aws_client.sns.set_endpoint_attributes(
            EndpointArn=endpoint_arn, Attributes={"Enabled": "false"}
        )

        retries = 3
        sleep = 1
        if is_aws_cloud():
            retries = 30
            sleep = 2

        def _assert_endpoint_disabled():
            get_attrs = aws_client.sns.get_endpoint_attributes(EndpointArn=endpoint_arn)
            assert get_attrs["Attributes"]["Enabled"] == "false"

        retry(_assert_endpoint_disabled, retries=retries, sleep=sleep)
        with pytest.raises(ClientError) as e:
            message = {
                platform: '{ "notification": {"title": "Title of notification", "body": "It works" } }'
            }
            aws_client.sns.publish(
                TargetArn=endpoint_arn, MessageStructure="json", Message=json.dumps(message)
            )

        assert e.value.response["Error"]["Code"] == "EndpointDisabled"
        assert e.value.response["Error"]["Message"] == "Endpoint is disabled"

    @markers.aws.only_localstack  # needs real credentials for GCM/FCM
    @pytest.mark.skip(reason="Need to implement credentials validation when creating platform")
    def test_publish_to_gcm(self, sns_create_platform_application, aws_client):
        key = "mock_server_key"
        token = "mock_token"

        response = sns_create_platform_application(
            Name="firebase", Platform="GCM", Attributes={"PlatformCredential": key}
        )

        platform_app_arn = response["PlatformApplicationArn"]

        response = aws_client.sns.create_platform_endpoint(
            PlatformApplicationArn=platform_app_arn,
            Token=token,
        )
        endpoint_arn = response["EndpointArn"]

        message = {
            "GCM": '{ "notification": {"title": "Title of notification", "body": "It works" } }'
        }

        with pytest.raises(ClientError) as ex:
            aws_client.sns.publish(
                TargetArn=endpoint_arn, MessageStructure="json", Message=json.dumps(message)
            )
        assert ex.value.response["Error"]["Code"] == "InvalidParameter"

    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_publish_to_platform_endpoint_is_dispatched(
        self,
        sns_create_topic,
        sns_subscription,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        sns_provider,
        platform_credentials,
    ):
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        topic_arn = sns_create_topic()["TopicArn"]
        endpoints_arn = {}
        for platform_type in ["APNS", "GCM"]:
            application_platform_name = f"app-platform-{platform_type}-{short_uid()}"

            # Create an Apple platform application
            app_arn = sns_create_platform_application(
                Name=application_platform_name, Platform=platform_type, Attributes=attributes
            )["PlatformApplicationArn"]

            endpoint_arn = aws_client.sns.create_platform_endpoint(
                PlatformApplicationArn=app_arn, Token=short_uid()
            )["EndpointArn"]

            # store the endpoint for checking results
            endpoints_arn[platform_type] = endpoint_arn

            # subscribe this endpoint to a topic
            sns_subscription(
                TopicArn=topic_arn,
                Protocol="application",
                Endpoint=endpoint_arn,
            )

        # now we have two platform endpoints subscribed to the same topic
        message = {
            "default": "This is the default message which must be present when publishing a message to a topic.",
            "APNS": '{"aps":{"alert": "Check out these awesome deals!","url":"www.amazon.com"} }',
            "GCM": '{"data":{"message":"Check out these awesome deals!","url":"www.amazon.com"}}',
        }

        # publish to the topic
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=json.dumps(message),
            MessageStructure="json",
        )
        SnsProvider = sns_provider()
        sns_backend = SnsProvider.get_store(account_id, region_name)
        platform_endpoint_msgs = sns_backend.platform_endpoint_messages

        # assert that message has been received
        def check_message():
            assert len(platform_endpoint_msgs[endpoint_arn]) > 0

        retry(check_message, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

        # each endpoint should only receive the message that was directed to them
        assert platform_endpoint_msgs[endpoints_arn["GCM"]][0]["Message"] == message["GCM"]
        assert platform_endpoint_msgs[endpoints_arn["APNS"]][0]["Message"] == message["APNS"]


class TestSNSSMS:
    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_publish_sms(self, aws_client, account_id, region_name, sns_provider):
        phone_number = "+33000000000"
        response = aws_client.sns.publish(PhoneNumber=phone_number, Message="This is a SMS")
        assert "MessageId" in response
        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200

        SnsProvider = sns_provider()
        sns_backend = SnsProvider.get_store(
            account_id,
            region_name,
        )

        def check_messages():
            sms_was_found = False
            for message in sns_backend.sms_messages:
                if message["PhoneNumber"] == phone_number:
                    sms_was_found = True
                    break

            assert sms_was_found

        retry(check_messages, sleep=0.5)

    @markers.aws.validated
    def test_subscribe_sms_endpoint(self, sns_create_topic, sns_subscription, snapshot, aws_client):
        phone_number = "+123123123"
        topic_arn = sns_create_topic()["TopicArn"]
        response = sns_subscription(TopicArn=topic_arn, Protocol="sms", Endpoint=phone_number)
        snapshot.match("subscribe-sms-endpoint", response)

        sub_attrs = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=response["SubscriptionArn"]
        )
        snapshot.match("subscribe-sms-attrs", sub_attrs)

    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_publish_sms_endpoint(
        self, sns_create_topic, sns_subscription, aws_client, account_id, region_name, sns_provider
    ):
        list_of_contacts = [
            f"+{random.randint(100000000, 9999999999)}",
            f"+{random.randint(100000000, 9999999999)}",
            f"+{random.randint(100000000, 9999999999)}",
        ]
        message = "Good news everyone!"
        topic_arn = sns_create_topic()["TopicArn"]
        for number in list_of_contacts:
            sns_subscription(TopicArn=topic_arn, Protocol="sms", Endpoint=number)

        aws_client.sns.publish(Message=message, TopicArn=topic_arn)

        SnsProvider = sns_provider()
        sns_backend = SnsProvider.get_store(account_id, region_name)

        def check_messages():
            sms_messages = sns_backend.sms_messages
            for contact in list_of_contacts:
                sms_was_found = False
                for _message in sms_messages:
                    if _message["PhoneNumber"] == contact:
                        sms_was_found = True
                        break

                assert sms_was_found

        retry(check_messages, sleep=0.5)

    @markers.aws.validated
    def test_publish_wrong_phone_format(
        self, sns_create_topic, sns_subscription, snapshot, aws_client
    ):
        message = "Good news everyone!"
        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(Message=message, PhoneNumber="+1a234")

        snapshot.match("invalid-number", e.value.response)

        with pytest.raises(ClientError) as e:
            aws_client.sns.publish(Message=message, PhoneNumber="NAA+15551234567")

        snapshot.match("wrong-format", e.value.response)

        topic_arn = sns_create_topic()["TopicArn"]
        with pytest.raises(ClientError) as e:
            sns_subscription(TopicArn=topic_arn, Protocol="sms", Endpoint="NAA+15551234567")
        snapshot.match("wrong-endpoint", e.value.response)

    @markers.aws.manual_setup_required
    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Not implemented in v1")
    def test_set_get_sms_attributes(self, aws_client, account_id, snapshot):
        if is_aws_cloud():
            LOG.warning(
                "Warning: this test will permanently set values for sms attributes in this region. \n Removing is not possible, only overwriting.\n Remove the exception if you wish to proceed"
            )
            raise Exception("Check logs to enable this test against AWS")

        aws_client.sns.set_sms_attributes(
            attributes={
                "DeliveryStatusSuccessSamplingRate": "100",
                "DefaultSenderID": "LSTest",
                "DefaultSMSType": "Promotional",
            }
        )

        response = aws_client.sns.get_sms_attributes()
        snapshot.match("get-sms-all-attributes", response)

        response = aws_client.sns.get_sms_attributes(
            attributes=["DefaultSenderID", "DefaultSMSType"]
        )
        snapshot.match("get-sms-some-attributes", response)

    @markers.aws.manual_setup_required
    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Not implemented in v1")
    def test_get_sms_attributes_from_unmodified_region(
        self, aws_client_factory, secondary_region_name, snapshot
    ):
        # if the secondary region is already modified, you can use SECONDARY_TEST_AWS_PROFILE and similar
        # variables to change it

        default_spend_limit = "1"

        sns_unmodified_region = aws_client_factory(region_name=secondary_region_name).sns
        response = sns_unmodified_region.get_sms_attributes()
        assert response["attributes"]["MonthlySpendLimit"] == default_spend_limit
        snapshot.match("get-sms-attributes_unmodified_region", response)

    @pytest.mark.parametrize(
        "attribute_key_value",
        [
            ("InvalidAttribute", "invalid"),
            ("DefaultSendID", "a" * 12),
            ("DefaultSendID", "123456789"),
            ("DefaultSMSType", "invalid"),
        ],
        ids=["InvalidAttributeName", "TooLongID", "NoLetterID", "InvalidSMSType"],
    )
    @markers.aws.validated
    @pytest.mark.skipif(condition=is_sns_v1_provider(), reason="Not implemented in v1")
    def test_set_invalid_sms_attributes(self, aws_client, snapshot, attribute_key_value):
        with pytest.raises(ClientError) as e:
            aws_client.sns.set_sms_attributes(
                attributes={
                    attribute_key_value[0]: attribute_key_value[1],
                }
            )
        snapshot.match("invalid-attribute", e.value.response)

    @markers.aws.manual_setup_required
    @pytest.mark.skipif(is_sns_v1_provider(), reason="Not correctly implemented in v1")
    def test_is_phone_number_opted_out(
        self, phone_number, aws_client, snapshot, sns_provider, account_id, region_name, cleanups
    ):
        # this test expects the fixture-provided phone number to be opted out
        # if you want to test against AWS, you need to manually opt out a number
        # https://us-east-1.console.aws.amazon.com/sms-voice/home?region=us-east-1#/opt-out-lists?name=Default&tab=opt-out-list-opted-out-numbers
        sns_store = sns_provider().get_store(account_id, region_name)

        def cleanup_store():
            sns_store.PHONE_NUMBERS_OPTED_OUT.remove(phone_number)

        if not is_aws_cloud():
            sns_store.PHONE_NUMBERS_OPTED_OUT.add(phone_number)
            cleanups.append(cleanup_store)

        response = aws_client.sns.check_if_phone_number_is_opted_out(phoneNumber=phone_number)
        snapshot.match("phone-number-opted-out", response)

    @markers.aws.manual_setup_required
    @pytest.mark.skipif(is_sns_v1_provider(), reason="Not correctly implemented in v1")
    def test_list_phone_numbers_opted_out(
        self, phone_number, aws_client, snapshot, sns_provider, account_id, region_name, cleanups
    ):
        # this test expects exactly one phone number opted out
        # if you want to test against AWS, you need to manually opt out a number
        # https://us-east-1.console.aws.amazon.com/sms-voice/home?region=us-east-1#/opt-out-lists?name=Default&tab=opt-out-list-opted-out-numbers
        sns_store = sns_provider().get_store(account_id, region_name)

        def cleanup_store():
            sns_store.PHONE_NUMBERS_OPTED_OUT.remove(phone_number)

        if not is_aws_cloud():
            sns_store.PHONE_NUMBERS_OPTED_OUT.add(phone_number)
            cleanups.append(cleanup_store)

        snapshot.add_transformer(
            TransformerUtility.jsonpath(
                jsonpath="$..phoneNumbers[*]",
                value_replacement="phone-number",
            )
        )
        response = aws_client.sns.list_phone_numbers_opted_out()
        snapshot.match("list-phone-numbers-opted-out", response)

    @markers.aws.manual_setup_required
    @pytest.mark.skipif(is_sns_v1_provider(), reason="Not correctly implemented in v1")
    def test_opt_in_phone_number(
        self, phone_number, aws_client, snapshot, sns_provider, account_id, region_name, cleanups
    ):
        # this test expects exactly one phone number opted out
        # if you want to test against AWS, you need to manually opt out a number
        # https://us-east-1.console.aws.amazon.com/sms-voice/home?region=us-east-1#/opt-out-lists?name=Default&tab=opt-out-list-opted-out-numbers
        # IMPORTANT: a phone number can only be opted in once every 30 days on AWS.
        # Make sure everything else is set up and taken care of properly before trying to validate this.
        sns_store = sns_provider().get_store(account_id, region_name)

        def cleanup_store():
            sns_store.PHONE_NUMBERS_OPTED_OUT.remove(phone_number)

        if not is_aws_cloud():
            sns_store.PHONE_NUMBERS_OPTED_OUT.add(phone_number)
            cleanups.append(cleanup_store)
        response = aws_client.sns.check_if_phone_number_is_opted_out(phoneNumber=phone_number)
        assert response["isOptedOut"]

        response = aws_client.sns.opt_in_phone_number(phoneNumber=phone_number)
        snapshot.match("opt-in-phone-number", response)

    @markers.aws.only_localstack
    @markers.requires_in_process
    @pytest.mark.skipif(is_sns_v1_provider(), reason="Not implemented in v1")
    def test_opt_out_phone_number_via_endpoint(
        self, phone_number, aws_client, snapshot, sns_provider, account_id, cleanups
    ):
        response = aws_client.sns.check_if_phone_number_is_opted_out(phoneNumber=phone_number)
        assert not response["isOptedOut"]
        data = {"phoneNumber": phone_number, "accountId": account_id}
        phone_url = config.external_service_url() + SMS_PHONE_NUMBER_OPT_OUT_ENDPOINT
        requests.post(phone_url, data=json.dumps(data))

        response = aws_client.sns.check_if_phone_number_is_opted_out(phoneNumber=phone_number)
        assert response["isOptedOut"]

    @markers.aws.validated
    def test_opt_in_non_existing_phone_number(
        self, phone_number, aws_client, snapshot, sns_provider, account_id, region_name
    ):
        non_existing_number = "+4411111111"
        response = aws_client.sns.opt_in_phone_number(phoneNumber=non_existing_number)

        snapshot.match("opt-in-non-existing-number", response)

    @pytest.mark.skipif(is_sns_v1_provider(), reason="Not correctly implemented in v1")
    @markers.aws.validated
    def test_opt_in_invalid_number(self, phone_number, aws_client, snapshot):
        invalid_number = "invalid"
        with pytest.raises(ClientError) as e:
            aws_client.sns.opt_in_phone_number(phoneNumber=invalid_number)

        snapshot.match("opt-in-non-existing-number", e.value.response)


class TestSNSSubscriptionHttp:
    @markers.aws.validated
    def test_http_subscription_response(
        self,
        sns_create_topic,
        sns_subscription,
        aws_client,
        snapshot,
    ):
        topic_arn = sns_create_topic()["TopicArn"]
        snapshot.match("topic-arn", {"TopicArn": topic_arn})

        # we need to hit whatever URL, even external, the publishing is async, but we need an endpoint who won't
        # confirm the subscription
        subscription = sns_subscription(
            TopicArn=topic_arn,
            Protocol="http",
            Endpoint="http://example.com",
            ReturnSubscriptionArn=False,
        )
        snapshot.match("subscription", subscription)

        subscription_with_arn = sns_subscription(
            TopicArn=topic_arn,
            Protocol="http",
            Endpoint="http://example.com",
            ReturnSubscriptionArn=True,
        )
        snapshot.match("subscription-with-arn", subscription_with_arn)

    @markers.requires_in_process  # uses pytest httpserver
    @markers.aws.manual_setup_required
    def test_redrive_policy_http_subscription(
        self, sns_create_topic, sqs_create_queue, sqs_get_queue_arn, sns_subscription, aws_client
    ):
        dlq_name = f"dlq-{short_uid()}"
        dlq_url = sqs_create_queue(QueueName=dlq_name)
        dlq_arn = sqs_get_queue_arn(dlq_url)
        topic_arn = sns_create_topic()["TopicArn"]

        # create HTTP endpoint and connect it to SNS topic
        with HTTPServer() as server:
            server.expect_request("/subscription").respond_with_data(b"", 200)
            http_endpoint = server.url_for("/subscription")
            wait_for_port_open(server.port)

            subscription = sns_subscription(
                TopicArn=topic_arn, Protocol="http", Endpoint=http_endpoint
            )
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=subscription["SubscriptionArn"],
                AttributeName="RedrivePolicy",
                AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
            )

            # wait for subscription notification to arrive at http endpoint
            poll_condition(lambda: len(server.log) >= 1, timeout=10)
            request, _ = server.log[0]
            event = request.get_json(force=True)
            assert request.path.endswith("/subscription")
            assert event["Type"] == "SubscriptionConfirmation"
            assert event["TopicArn"] == topic_arn
            aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token=event["Token"])

        wait_for_port_closed(server.port)

        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=json.dumps({"message": "test_redrive_policy"}),
        )

        response = aws_client.sqs.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=10)
        assert len(response["Messages"]) == 1, (
            f"invalid number of messages in DLQ response {response}"
        )
        message = json.loads(response["Messages"][0]["Body"])
        assert message["Type"] == "Notification"
        assert json.loads(message["Message"])["message"] == "test_redrive_policy"

    @markers.requires_in_process  # uses pytest httpserver
    @markers.aws.manual_setup_required
    def test_multiple_subscriptions_http_endpoint(
        self, sns_create_topic, sns_subscription, aws_client
    ):
        # create a topic
        topic_arn = sns_create_topic()["TopicArn"]

        # build fake http server endpoints
        _requests = queue.Queue()

        # create HTTP endpoint and connect it to SNS topic
        def handler(_request):
            _requests.put(_request)
            return Response(status=429)

        number_of_endpoints = 4

        servers = []
        try:
            for _ in range(number_of_endpoints):
                server = HTTPServer()
                server.start()
                servers.append(server)
                server.expect_request("/").respond_with_handler(handler)
                http_endpoint = server.url_for("/")
                wait_for_port_open(http_endpoint)

                sns_subscription(TopicArn=topic_arn, Protocol="http", Endpoint=http_endpoint)

            # fetch subscription information
            subscription_list = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
            assert subscription_list["ResponseMetadata"]["HTTPStatusCode"] == 200
            assert len(subscription_list["Subscriptions"]) == number_of_endpoints, (
                f"unexpected number of subscriptions {subscription_list}"
            )

            tokens = []
            for _ in range(number_of_endpoints):
                request = _requests.get(timeout=2)
                request_data = request.get_json(True)
                tokens.append(request_data["Token"])
                assert request_data["TopicArn"] == topic_arn

            with pytest.raises(queue.Empty):
                # make sure only four requests are received
                _requests.get(timeout=1)

            # assert the first subscription is pending confirmation
            sub_1 = subscription_list["Subscriptions"][0]
            sub_1_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=sub_1["SubscriptionArn"]
            )
            assert sub_1_attrs["Attributes"]["PendingConfirmation"] == "true"

            # assert the second subscription is pending confirmation
            sub_2 = subscription_list["Subscriptions"][1]
            sub_2_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=sub_2["SubscriptionArn"]
            )
            assert sub_2_attrs["Attributes"]["PendingConfirmation"] == "true"

            # confirm the first subscription
            response = aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token=tokens[0])
            # assert the confirmed subscription is the first one
            assert response["SubscriptionArn"] == sub_1["SubscriptionArn"]

            # assert the first subscription is confirmed
            sub_1_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=sub_1["SubscriptionArn"]
            )
            assert sub_1_attrs["Attributes"]["PendingConfirmation"] == "false"

            # assert the second subscription is NOT confirmed
            sub_2_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=sub_2["SubscriptionArn"]
            )
            assert sub_2_attrs["Attributes"]["PendingConfirmation"] == "true"

        finally:
            subscription_list = aws_client.sns.list_subscriptions_by_topic(TopicArn=topic_arn)
            for subscription in subscription_list["Subscriptions"]:
                aws_client.sns.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])
            for server in servers:
                server.stop()

    @markers.requires_in_process  # uses pytest httpserver
    @markers.aws.manual_setup_required
    @pytest.mark.parametrize("raw_message_delivery", [True, False])
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.http-message-headers.Accept",  # requests adds the header but not SNS, not very important
            "$.http-message-headers-raw.Accept",
            "$.http-confirm-sub-headers.Accept",
        ]
    )
    def test_subscribe_external_http_endpoint(
        self, sns_create_http_endpoint, raw_message_delivery, aws_client, snapshot
    ):
        def _get_snapshot_requests_response(response: requests.Response) -> dict:
            parsed_xml_body = xmltodict.parse(response.content)
            for root_tag, fields in parsed_xml_body.items():
                fields.pop("@xmlns", None)
                if "ResponseMetadata" in fields:
                    fields["ResponseMetadata"]["HTTPHeaders"] = dict(response.headers)
                    fields["ResponseMetadata"]["HTTPStatusCode"] = response.status_code
            return parsed_xml_body

        def _clean_headers(response_headers: dict):
            return {key: val for key, val in response_headers.items() if "Forwarded" not in key}

        snapshot.add_transformer(
            [
                snapshot.transform.key_value("RequestId"),
                snapshot.transform.key_value("Token"),
                snapshot.transform.key_value("Host"),
                snapshot.transform.key_value(
                    "Content-Length", reference_replacement=False
                ),  # might change depending on compression
                snapshot.transform.key_value(
                    "Connection", reference_replacement=False
                ),  # casing might change
                snapshot.transform.regex(
                    r"(?i)(?<=SubscribeURL[\"|']:\s[\"|'])(https?.*?)(?=/\?Action=ConfirmSubscription)",
                    replacement="<subscribe-domain>",
                ),
            ]
        )

        # Necessitate manual set up to allow external access to endpoint, only in local testing
        topic_arn, subscription_arn, endpoint_url, server = sns_create_http_endpoint(
            raw_message_delivery
        )
        assert poll_condition(
            lambda: len(server.log) >= 1,
            timeout=5,
        )
        sub_request, _ = server.log[0]
        payload = sub_request.get_json(force=True)
        snapshot.match("subscription-confirmation", payload)
        assert payload["Type"] == "SubscriptionConfirmation"
        assert sub_request.headers["x-amz-sns-message-type"] == "SubscriptionConfirmation"
        assert "Signature" in payload
        assert "SigningCertURL" in payload

        snapshot.match("http-confirm-sub-headers", _clean_headers(sub_request.headers))

        token = payload["Token"]
        subscribe_url = payload["SubscribeURL"]
        service_url, subscribe_url_path = payload["SubscribeURL"].rsplit("/", maxsplit=1)
        assert subscribe_url == (
            f"{service_url}/?Action=ConfirmSubscription&TopicArn={topic_arn}&Token={token}"
        )

        test_broken_confirm_url = (
            f"{service_url}/?Action=ConfirmSubscription&TopicArn=not-an-arn&Token={token}"
        )
        broken_confirm_subscribe_request = requests.get(test_broken_confirm_url)
        snapshot.match(
            "broken-topic-arn-confirm",
            _get_snapshot_requests_response(broken_confirm_subscribe_request),
        )

        test_broken_token_confirm_url = (
            f"{service_url}/?Action=ConfirmSubscription&TopicArn={topic_arn}&Token=abc123"
        )
        broken_token_confirm_subscribe_request = requests.get(test_broken_token_confirm_url)
        snapshot.match(
            "broken-token-confirm",
            _get_snapshot_requests_response(broken_token_confirm_subscribe_request),
        )

        # using the right topic name with a different region will fail when confirming the subscription
        parsed_arn = parse_arn(topic_arn)
        different_region = "eu-central-1" if parsed_arn["region"] != "eu-central-1" else "us-east-1"
        different_region_topic = topic_arn.replace(parsed_arn["region"], different_region)
        different_region_topic_confirm_url = f"{service_url}/?Action=ConfirmSubscription&TopicArn={different_region_topic}&Token={token}"
        region_topic_confirm_subscribe_request = requests.get(different_region_topic_confirm_url)
        snapshot.match(
            "different-region-arn-confirm",
            _get_snapshot_requests_response(region_topic_confirm_subscribe_request),
        )

        # but a nonexistent topic in the right region will succeed
        last_fake_topic_char = "a" if topic_arn[-1] != "a" else "b"
        nonexistent = topic_arn[:-1] + last_fake_topic_char
        assert nonexistent != topic_arn
        test_wrong_topic_confirm_url = (
            f"{service_url}/?Action=ConfirmSubscription&TopicArn={nonexistent}&Token={token}"
        )
        wrong_topic_confirm_subscribe_request = requests.get(test_wrong_topic_confirm_url)
        snapshot.match(
            "nonexistent-token-confirm",
            _get_snapshot_requests_response(wrong_topic_confirm_subscribe_request),
        )

        # weirdly, even with a wrong topic, SNS will confirm the topic
        subscription_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscription_arn
        )
        assert subscription_attributes["Attributes"]["PendingConfirmation"] == "false"

        confirm_subscribe_request = requests.get(subscribe_url)
        confirm_subscribe = xmltodict.parse(confirm_subscribe_request.content)
        assert (
            confirm_subscribe["ConfirmSubscriptionResponse"]["ConfirmSubscriptionResult"][
                "SubscriptionArn"
            ]
            == subscription_arn
        )
        # also confirm that ConfirmSubscription is idempotent
        snapshot.match(
            "confirm-subscribe", _get_snapshot_requests_response(confirm_subscribe_request)
        )

        subscription_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscription_arn
        )
        assert subscription_attributes["Attributes"]["PendingConfirmation"] == "false"

        message = "test_external_http_endpoint"
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)

        assert poll_condition(
            lambda: len(server.log) >= 2,
            timeout=5,
        )
        notification_request, _ = server.log[1]
        assert notification_request.headers["x-amz-sns-message-type"] == "Notification"

        expected_unsubscribe_url = (
            f"{service_url}/?Action=Unsubscribe&SubscriptionArn={subscription_arn}"
        )
        if raw_message_delivery:
            payload = notification_request.data.decode()
            assert payload == message
            snapshot.match("http-message-headers-raw", _clean_headers(notification_request.headers))
        else:
            payload = notification_request.get_json(force=True)
            assert payload["Type"] == "Notification"
            assert "Signature" in payload
            assert "SigningCertURL" in payload
            assert payload["Message"] == message
            assert payload["UnsubscribeURL"] == expected_unsubscribe_url
            snapshot.match("http-message", payload)
            snapshot.match("http-message-headers", _clean_headers(notification_request.headers))

        unsub_request = requests.get(expected_unsubscribe_url)
        unsubscribe_confirmation = xmltodict.parse(unsub_request.content)
        assert "UnsubscribeResponse" in unsubscribe_confirmation
        snapshot.match("unsubscribe-response", _get_snapshot_requests_response(unsub_request))

        assert poll_condition(
            lambda: len(server.log) >= 3,
            timeout=5,
        )
        unsub_request, _ = server.log[2]

        payload = unsub_request.get_json(force=True)
        assert payload["Type"] == "UnsubscribeConfirmation"
        assert unsub_request.headers["x-amz-sns-message-type"] == "UnsubscribeConfirmation"
        assert "Signature" in payload
        assert "SigningCertURL" in payload
        token = payload["Token"]
        assert payload["SubscribeURL"] == (
            f"{service_url}/?Action=ConfirmSubscription&TopicArn={topic_arn}&Token={token}"
        )
        snapshot.match("unsubscribe-request", payload)

    @markers.requires_in_process  # uses pytest httpserver
    @markers.aws.manual_setup_required
    @pytest.mark.parametrize("raw_message_delivery", [True, False])
    def test_dlq_external_http_endpoint(
        self,
        sqs_create_queue,
        sqs_get_queue_arn,
        sns_create_http_endpoint,
        sns_allow_topic_sqs_queue,
        raw_message_delivery,
        aws_client,
    ):
        # Necessitate manual set up to allow external access to endpoint, only in local testing
        topic_arn, http_subscription_arn, endpoint_url, server = sns_create_http_endpoint(
            raw_message_delivery
        )

        dlq_url = sqs_create_queue()
        dlq_arn = sqs_get_queue_arn(dlq_url)

        sns_allow_topic_sqs_queue(
            sqs_queue_url=dlq_url, sqs_queue_arn=dlq_arn, sns_topic_arn=topic_arn
        )
        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=http_subscription_arn,
            AttributeName="RedrivePolicy",
            AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
        )
        assert poll_condition(
            lambda: len(server.log) >= 1,
            timeout=5,
        )
        sub_request, _ = server.log[0]
        payload = sub_request.get_json(force=True)
        assert payload["Type"] == "SubscriptionConfirmation"
        assert sub_request.headers["x-amz-sns-message-type"] == "SubscriptionConfirmation"

        subscribe_url = payload["SubscribeURL"]
        service_url, subscribe_url_path = payload["SubscribeURL"].rsplit("/", maxsplit=1)

        confirm_subscribe_request = requests.get(subscribe_url)
        confirm_subscribe = xmltodict.parse(confirm_subscribe_request.content)
        assert (
            confirm_subscribe["ConfirmSubscriptionResponse"]["ConfirmSubscriptionResult"][
                "SubscriptionArn"
            ]
            == http_subscription_arn
        )

        subscription_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=http_subscription_arn
        )
        assert subscription_attributes["Attributes"]["PendingConfirmation"] == "false"

        server.stop()
        wait_for_port_closed(server.port)

        message = "test_dlq_external_http_endpoint"
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)

        response = aws_client.sqs.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=3)
        assert len(response["Messages"]) == 1, (
            f"invalid number of messages in DLQ response {response}"
        )

        if raw_message_delivery:
            assert response["Messages"][0]["Body"] == message
        else:
            received_message = json.loads(response["Messages"][0]["Body"])
            assert received_message["Type"] == "Notification"
            assert received_message["Message"] == message

        receipt_handle = response["Messages"][0]["ReceiptHandle"]
        aws_client.sqs.delete_message(QueueUrl=dlq_url, ReceiptHandle=receipt_handle)

        expected_unsubscribe_url = (
            f"{service_url}/?Action=Unsubscribe&SubscriptionArn={http_subscription_arn}"
        )

        unsub_request = requests.get(expected_unsubscribe_url)
        unsubscribe_confirmation = xmltodict.parse(unsub_request.content)
        assert "UnsubscribeResponse" in unsubscribe_confirmation

        response = aws_client.sqs.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=2)
        # AWS doesn't send to the DLQ if the UnsubscribeConfirmation fails to be delivered
        assert "Messages" not in response or response["Messages"] == []

    @markers.requires_in_process  # uses pytest httpserver
    @markers.aws.manual_setup_required
    @pytest.mark.parametrize("raw_message_delivery", [True, False])
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$.http-message-headers.Accept",  # requests adds the header but not SNS, not very important
            "$.http-message-headers-raw.Accept",
            "$.http-confirm-sub-headers.Accept",
            # TODO: we need to fix this parity in Moto, in order to make the retrieval logic of those values easier
            "$.sub-attrs.Attributes.ConfirmationWasAuthenticated",
            "$.sub-attrs.Attributes.DeliveryPolicy",
            "$.sub-attrs.Attributes.EffectiveDeliveryPolicy",
            "$.topic-attrs.Attributes.DeliveryPolicy",
            "$.topic-attrs.Attributes.EffectiveDeliveryPolicy",
            "$.topic-attrs.Attributes.Policy.Statement..Action",
        ]
    )
    def test_subscribe_external_http_endpoint_content_type(
        self,
        sns_create_http_endpoint,
        raw_message_delivery,
        aws_client,
        snapshot,
    ):
        def _clean_headers(response_headers: dict):
            return {key: val for key, val in response_headers.items() if "Forwarded" not in key}

        snapshot.add_transformer(
            [
                snapshot.transform.key_value("RequestId"),
                snapshot.transform.key_value("Token"),
                snapshot.transform.key_value("Host"),
                snapshot.transform.key_value(
                    "Content-Length", reference_replacement=False
                ),  # might change depending on compression
                snapshot.transform.key_value(
                    "Connection", reference_replacement=False
                ),  # casing might change
                snapshot.transform.regex(
                    r"(?i)(?<=SubscribeURL[\"|']:\s[\"|'])(https?.*?)(?=/\?Action=ConfirmSubscription)",
                    replacement="<subscribe-domain>",
                ),
            ]
        )

        # Necessitate manual set up to allow external access to endpoint, only in local testing
        topic_arn, subscription_arn, endpoint_url, server = sns_create_http_endpoint(
            raw_message_delivery
        )

        # try both setting the Topic attribute or Subscription attribute
        # https://docs.aws.amazon.com/sns/latest/dg/sns-message-delivery-retries.html#creating-delivery-policy
        if raw_message_delivery:
            aws_client.sns.set_subscription_attributes(
                SubscriptionArn=subscription_arn,
                AttributeName="DeliveryPolicy",
                AttributeValue=json.dumps(
                    {
                        "requestPolicy": {"headerContentType": "text/csv"},
                    }
                ),
            )
        else:
            aws_client.sns.set_topic_attributes(
                TopicArn=topic_arn,
                AttributeName="DeliveryPolicy",
                AttributeValue=json.dumps(
                    {
                        "http": {
                            "defaultRequestPolicy": {"headerContentType": "application/json"},
                        }
                    }
                ),
            )

        topic_attrs = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("topic-attrs", topic_attrs)

        sub_attrs = aws_client.sns.get_subscription_attributes(SubscriptionArn=subscription_arn)
        snapshot.match("sub-attrs", sub_attrs)

        assert poll_condition(
            lambda: len(server.log) >= 1,
            timeout=5,
        )
        sub_request, _ = server.log[0]
        payload = sub_request.get_json(force=True)
        snapshot.match("subscription-confirmation", payload)
        snapshot.match("http-confirm-sub-headers", _clean_headers(sub_request.headers))

        token = payload["Token"]
        aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token=token)

        message = "test_external_http_endpoint"
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)

        assert poll_condition(
            lambda: len(server.log) >= 2,
            timeout=5,
        )
        notification_request, _ = server.log[1]
        assert notification_request.headers["x-amz-sns-message-type"] == "Notification"
        if raw_message_delivery:
            payload = notification_request.data.decode()
            assert payload == message
            snapshot.match("http-message-headers-raw", _clean_headers(notification_request.headers))
        else:
            payload = notification_request.get_json(force=True)
            snapshot.match("http-message", payload)
            snapshot.match("http-message-headers", _clean_headers(notification_request.headers))

    @markers.aws.validated
    def test_subscribe_external_http_endpoint_lambda_url_sig_validation(
        self,
        create_sns_http_endpoint_and_queue,
        sns_create_topic,
        sns_subscription,
        aws_client,
        snapshot,
        sqs_collect_messages,
    ):
        def _get_snapshot_from_lambda_url_msg(events: list[dict]) -> dict:
            formatted_events = []

            def _filter_headers(headers: dict) -> dict:
                filtered_headers = {}
                for key, value in headers.items():
                    l_key = key.lower()
                    if l_key.startswith("x-amz-sns") or key in (
                        "content-type",
                        "accept-encoding",
                        "user-agent",
                    ):
                        filtered_headers[key] = value

                return filtered_headers

            for event in events:
                msg = json.loads(event["Body"])["event"]
                formatted_events.append(
                    {"headers": _filter_headers(msg["headers"]), "body": json.loads(msg["body"])}
                )

            return {"events": formatted_events}

        def validate_message_signature(msg_event: dict, msg_type: str):
            cert_url = msg_event["SigningCertURL"]
            get_cert_req = requests.get(cert_url)
            assert get_cert_req.ok

            cert = x509.load_pem_x509_certificate(get_cert_req.content)
            message_signature = msg_event["Signature"]
            # create the canonical string
            if msg_type == "Notification":
                fields = ["Message", "MessageId", "Subject", "Timestamp", "TopicArn", "Type"]
            else:
                fields = [
                    "Message",
                    "MessageId",
                    "SubscribeURL",
                    "Timestamp",
                    "Token",
                    "TopicArn",
                    "Type",
                ]

            # Build the string to be signed.
            string_to_sign = "".join(
                [f"{field}\n{msg_event[field]}\n" for field in fields if field in msg_event]
            )

            # decode the signature from base64.
            decoded_signature = base64.b64decode(message_signature)

            message_sig_version = msg_event["SignatureVersion"]
            # this is a bug on AWS side, assert our behaviour is the same for now, this might get fixed
            assert message_sig_version == "1"
            signature_hash = hashes.SHA1() if message_sig_version == "1" else hashes.SHA256()

            # calculate signature value with cert
            # if the signature is invalid, this will raise an exception
            cert.public_key().verify(
                decoded_signature,
                to_bytes(string_to_sign),
                padding=padding.PKCS1v15(),
                algorithm=signature_hash,
            )

        snapshot.add_transformer(
            [
                snapshot.transform.key_value("RequestId"),
                snapshot.transform.key_value("Token"),
                snapshot.transform.key_value("Host"),
                snapshot.transform.regex(
                    r"(?i)(?<=SubscribeURL[\"|']:\s[\"|'])(https?.*?)(?=/\?Action=ConfirmSubscription)",
                    replacement="<subscribe-domain>",
                ),
            ]
        )
        http_endpoint_url, queue_url = create_sns_http_endpoint_and_queue()
        topic_arn = sns_create_topic()["TopicArn"]
        sns_protocol = http_endpoint_url.split("://")[0]
        subscription = sns_subscription(
            TopicArn=topic_arn, Protocol=sns_protocol, Endpoint=http_endpoint_url
        )
        subscription_arn = subscription["SubscriptionArn"]
        delivery_policy = {
            "healthyRetryPolicy": {
                "minDelayTarget": 1,
                "maxDelayTarget": 1,
                "numRetries": 0,
                "numNoDelayRetries": 0,
                "numMinDelayRetries": 0,
                "numMaxDelayRetries": 0,
                "backoffFunction": "linear",
            },
            "sicklyRetryPolicy": None,
            "throttlePolicy": {"maxReceivesPerSecond": 1000},
            "guaranteed": False,
        }
        aws_client.sns.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="DeliveryPolicy",
            AttributeValue=json.dumps(delivery_policy),
        )

        messages = sqs_collect_messages(queue_url, expected=1, timeout=10)
        subscribe_event = _get_snapshot_from_lambda_url_msg(messages)
        snapshot.match("subscription-confirmation", subscribe_event)

        subscribe_payload = subscribe_event["events"][0]["body"]

        validate_message_signature(
            subscribe_payload,
            msg_type=subscribe_event["events"][0]["headers"]["x-amz-sns-message-type"],
        )

        token = subscribe_payload["Token"]
        subscribe_url = subscribe_payload["SubscribeURL"]
        service_url, subscribe_url_path = subscribe_url.rsplit("/", maxsplit=1)
        # we manually assert here to be sure the format is right, as it hard to verify with snapshots
        assert subscribe_url == (
            f"{service_url}/?Action=ConfirmSubscription&TopicArn={topic_arn}&Token={token}"
        )

        confirm_subscription = aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token=token)
        snapshot.match("confirm-subscription", confirm_subscription)

        subscription_attributes = aws_client.sns.get_subscription_attributes(
            SubscriptionArn=subscription_arn
        )
        assert subscription_attributes["Attributes"]["PendingConfirmation"] == "false"

        message = "test_external_http_endpoint"
        aws_client.sns.publish(TopicArn=topic_arn, Message=message)

        messages = sqs_collect_messages(queue_url, expected=1, timeout=10)
        publish_event = _get_snapshot_from_lambda_url_msg(messages)
        snapshot.match("publish-event", publish_event)
        publish_payload = publish_event["events"][0]["body"]
        validate_message_signature(
            publish_payload,
            msg_type=publish_event["events"][0]["headers"]["x-amz-sns-message-type"],
        )

        unsub_request = requests.get(publish_payload["UnsubscribeURL"])
        assert b"UnsubscribeResponse" in unsub_request.content

        messages = sqs_collect_messages(queue_url, expected=1, timeout=10)
        unsubscribe_event = _get_snapshot_from_lambda_url_msg(messages)
        snapshot.match("unsubscribe-event", unsubscribe_event)

        unsubscribe_payload = unsubscribe_event["events"][0]["body"]
        validate_message_signature(
            unsubscribe_payload,
            msg_type=unsubscribe_event["events"][0]["headers"]["x-amz-sns-message-type"],
        )


class TestSNSSubscriptionFirehose:
    @markers.aws.validated
    def test_publish_to_firehose_with_s3(
        self,
        create_role,
        s3_create_bucket,
        firehose_create_delivery_stream,
        sns_create_topic,
        sns_subscription,
        aws_client,
        region_name,
    ):
        role_name = f"test-role-{short_uid()}"
        stream_name = f"test-stream-{short_uid()}"
        bucket_name = f"test-bucket-{short_uid()}"
        topic_name = f"test_topic_{short_uid()}"

        trust_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "s3.amazonaws.com"},
                    "Action": "sts:AssumeRole",
                },
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "firehose.amazonaws.com"},
                    "Action": "sts:AssumeRole",
                },
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "sns.amazonaws.com"},
                    "Action": "sts:AssumeRole",
                },
            ],
        }

        role = create_role(RoleName=role_name, AssumeRolePolicyDocument=json.dumps(trust_policy))

        aws_client.iam.attach_role_policy(
            RoleName=role_name,
            PolicyArn=f"arn:{get_partition(region_name)}:iam::aws:policy/AmazonKinesisFirehoseFullAccess",
        )

        aws_client.iam.attach_role_policy(
            RoleName=role_name,
            PolicyArn=f"arn:{get_partition(region_name)}:iam::aws:policy/AmazonS3FullAccess",
        )
        subscription_role_arn = role["Role"]["Arn"]

        if is_aws_cloud():
            time.sleep(10)

        s3_create_bucket(Bucket=bucket_name)

        stream = firehose_create_delivery_stream(
            DeliveryStreamName=stream_name,
            DeliveryStreamType="DirectPut",
            S3DestinationConfiguration={
                "RoleARN": subscription_role_arn,
                "BucketARN": f"arn:{get_partition(region_name)}:s3:::{bucket_name}",
                "BufferingHints": {"SizeInMBs": 1, "IntervalInSeconds": 60},
            },
        )

        topic = sns_create_topic(Name=topic_name)
        sns_subscription(
            TopicArn=topic["TopicArn"],
            Protocol="firehose",
            Endpoint=stream["DeliveryStreamARN"],
            Attributes={"SubscriptionRoleArn": subscription_role_arn},
            ReturnSubscriptionArn=True,
        )

        message = json.dumps({"message": "hello world"})
        message_attributes = {
            "testAttribute": {"DataType": "String", "StringValue": "valueOfAttribute"}
        }
        aws_client.sns.publish(
            TopicArn=topic["TopicArn"], Message=message, MessageAttributes=message_attributes
        )

        def validate_content():
            files = aws_client.s3.list_objects(Bucket=bucket_name)["Contents"]
            f = BytesIO()
            aws_client.s3.download_fileobj(bucket_name, files[0]["Key"], f)
            content = to_str(f.getvalue())

            sns_message = json.loads(content.split("\n")[0])

            assert "Type" in sns_message
            assert "MessageId" in sns_message
            assert "Message" in sns_message
            assert "Timestamp" in sns_message

            assert message == sns_message["Message"]

        retries = 5
        sleep = 1
        sleep_before = 0
        if is_aws_cloud():
            retries = 30
            sleep = 10
            sleep_before = 10

        retry(validate_content, retries=retries, sleep_before=sleep_before, sleep=sleep)


class TestSNSMultiAccounts:
    @pytest.fixture
    def sns_primary_client(self, aws_client):
        return aws_client.sns

    @pytest.fixture
    def sns_secondary_client(self, secondary_aws_client):
        return secondary_aws_client.sns

    @pytest.fixture
    def sqs_primary_client(self, aws_client):
        return aws_client.sqs

    @pytest.fixture
    def sqs_secondary_client(self, secondary_aws_client):
        return secondary_aws_client.sqs

    @markers.aws.only_localstack
    def test_cross_account_access(self, sns_primary_client, sns_secondary_client, sns_create_topic):
        # Cross-account access is supported for below operations.
        # This list is taken from ActionName param of the AddPermissions operation
        #
        # - GetTopicAttributes
        # - SetTopicAttributes
        # - AddPermission
        # - RemovePermission
        # - Publish
        # - Subscribe
        # - ListSubscriptionsByTopic
        # - DeleteTopic

        topic_name = f"topic-{short_uid()}"
        # sns_create_topic uses the primary client by default
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]

        assert sns_secondary_client.set_topic_attributes(
            TopicArn=topic_arn, AttributeName="DisplayName", AttributeValue="xenon"
        )

        response = sns_secondary_client.get_topic_attributes(TopicArn=topic_arn)
        assert response["Attributes"]["DisplayName"] == "xenon"

        assert sns_secondary_client.add_permission(
            TopicArn=topic_arn,
            Label="foo",
            AWSAccountId=["666666666666"],
            ActionName=["AddPermission"],
        )
        assert sns_secondary_client.remove_permission(TopicArn=topic_arn, Label="foo")

        assert sns_secondary_client.publish(TopicArn=topic_arn, Message="hello world")

        subscription_arn = sns_secondary_client.subscribe(
            TopicArn=topic_arn, Protocol="email", Endpoint="devil@hell.com"
        )["SubscriptionArn"]

        response = sns_secondary_client.list_subscriptions_by_topic(TopicArn=topic_arn)
        subscriptions = [s["SubscriptionArn"] for s in response["Subscriptions"]]
        assert subscription_arn in subscriptions

        response = sns_primary_client.set_subscription_attributes(
            SubscriptionArn=subscription_arn,
            AttributeName="RawMessageDelivery",
            AttributeValue="true",
        )
        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
        response = sns_primary_client.get_subscription_attributes(SubscriptionArn=subscription_arn)
        assert response["Attributes"]["RawMessageDelivery"] == "true"

        assert sns_secondary_client.delete_topic(TopicArn=topic_arn)

    @markers.aws.only_localstack
    def test_cross_account_publish_to_sqs(
        self,
        sns_create_topic,
        secondary_account_id,
        region_name,
        sns_primary_client,
        sns_secondary_client,
        sqs_primary_client,
        sqs_secondary_client,
        sqs_get_queue_arn,
        cleanups,
    ):
        """
        This test validates that we can publish to SQS queues that are not in the default account, and that another
        account can publish to the topic as well

        Note: we are not setting Queue policies here as it's only in localstack and IAM is not enforced, for the sake
        of simplicity
        """

        topic_name = "sample_topic"
        topic_1 = sns_create_topic(Name=topic_name)
        topic_1_arn = topic_1["TopicArn"]

        # create a queue with the primary AccountId
        queue_name = "sample_queue"
        queue_1 = sqs_primary_client.create_queue(QueueName=queue_name)
        queue_1_url = queue_1["QueueUrl"]
        cleanups.append(lambda: sqs_primary_client.delete_queue(QueueUrl=queue_1_url))
        queue_1_arn = sqs_get_queue_arn(queue_1_url)

        # create a queue with the secondary AccountId
        queue_2 = sqs_secondary_client.create_queue(QueueName=queue_name)
        queue_2_url = queue_2["QueueUrl"]
        cleanups.append(lambda: sqs_secondary_client.delete_queue(QueueUrl=queue_2_url))
        # test that we get the right queue URL at the same time, even if we use the primary client
        queue_2_arn = sqs_queue_arn(
            queue_2_url,
            secondary_account_id,
            region_name,
        )

        # create a second queue with the secondary AccountId
        queue_name_2 = "sample_queue_two"
        queue_3 = sqs_secondary_client.create_queue(QueueName=queue_name_2)
        queue_3_url = queue_3["QueueUrl"]
        cleanups.append(lambda: sqs_secondary_client.delete_queue(QueueUrl=queue_3_url))
        # test that we get the right queue URL at the same time, even if we use the primary client
        queue_3_arn = sqs_queue_arn(
            queue_3_url,
            secondary_account_id,
            region_name,
        )

        # test that we can subscribe with the primary client to a queue from the same account
        sns_primary_client.subscribe(
            TopicArn=topic_1_arn,
            Protocol="sqs",
            Endpoint=queue_1_arn,
        )

        # test that we can subscribe with the primary client to a queue from the secondary account
        sns_primary_client.subscribe(
            TopicArn=topic_1_arn,
            Protocol="sqs",
            Endpoint=queue_2_arn,
        )

        # test that we can subscribe with the secondary client (not owning the topic) to a queue of the secondary client
        sns_secondary_client.subscribe(
            TopicArn=topic_1_arn,
            Protocol="sqs",
            Endpoint=queue_3_arn,
        )

        # now, we have 3 subscriptions in topic_1, one to the queue_1 located in the same account, and 2 to queue_2 and
        # queue_3 located in the secondary account
        subscriptions = sns_primary_client.list_subscriptions_by_topic(TopicArn=topic_1_arn)
        assert len(subscriptions["Subscriptions"]) == 3

        sns_primary_client.publish(TopicArn=topic_1_arn, Message="TestMessageOwner")

        def get_messages_from_queues(message_content: str):
            for client, queue_url in (
                (sqs_primary_client, queue_1_url),
                (sqs_secondary_client, queue_2_url),
                (sqs_secondary_client, queue_3_url),
            ):
                response = client.receive_message(
                    QueueUrl=queue_url,
                    VisibilityTimeout=0,
                    WaitTimeSeconds=5,
                )
                messages = response["Messages"]
                assert len(messages) == 1
                assert topic_1_arn in messages[0]["Body"]
                assert message_content in messages[0]["Body"]
                client.delete_message(
                    QueueUrl=queue_url, ReceiptHandle=messages[0]["ReceiptHandle"]
                )

        get_messages_from_queues("TestMessageOwner")

        # assert that we can also publish to the topic 1 from the secondary account
        sns_secondary_client.publish(TopicArn=topic_1_arn, Message="TestMessageSecondary")

        get_messages_from_queues("TestMessageSecondary")


class TestSNSMultiRegions:
    @pytest.fixture
    def sns_region1_client(self, aws_client):
        return aws_client.sns

    @pytest.fixture
    def sns_region2_client(self, aws_client_factory, secondary_region_name):
        return aws_client_factory(region_name=secondary_region_name).sns

    @pytest.fixture
    def sqs_region2_client(self, aws_client_factory, secondary_region_name):
        return aws_client_factory(region_name=secondary_region_name).sqs

    @markers.aws.validated
    def test_cross_region_access(self, sns_region1_client, sns_region2_client, snapshot, cleanups):
        # We do not have a list of supported Cross-region access for operations.
        # This test is validating that Cross-account does not mean Cross-region most of the time

        topic_name = f"topic-{short_uid()}"
        topic_arn = sns_region1_client.create_topic(Name=topic_name)["TopicArn"]
        cleanups.append(lambda: sns_region1_client.delete_topic(TopicArn=topic_arn))

        with pytest.raises(ClientError) as e:
            sns_region2_client.set_topic_attributes(
                TopicArn=topic_arn, AttributeName="DisplayName", AttributeValue="xenon"
            )
        snapshot.match("set-topic-attrs", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_region2_client.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-topic-attrs", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_region2_client.publish(TopicArn=topic_arn, Message="hello world")
        snapshot.match("cross-region-publish-forbidden", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_region2_client.subscribe(
                TopicArn=topic_arn, Protocol="email", Endpoint="devil@hell.com"
            )
        snapshot.match("cross-region-subscribe", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_region2_client.list_subscriptions_by_topic(TopicArn=topic_arn)
        snapshot.match("list-subs", e.value.response)

        with pytest.raises(ClientError) as e:
            sns_region2_client.delete_topic(TopicArn=topic_arn)
        snapshot.match("delete-topic", e.value.response)

    @markers.aws.validated
    def test_cross_region_delivery_sqs(
        self,
        sns_region1_client,
        sns_region2_client,
        sqs_region2_client,
        sns_create_topic,
        sqs_create_queue,
        sns_allow_topic_sqs_queue,
        cleanups,
        snapshot,
    ):
        topic_arn = sns_create_topic()["TopicArn"]

        queue_url = sqs_create_queue()
        response = sqs_region2_client.create_queue(QueueName=f"queue-{short_uid()}")
        queue_url = response["QueueUrl"]
        cleanups.append(lambda: sqs_region2_client.delete_queue(QueueUrl=queue_url))

        queue_arn = sqs_region2_client.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]

        # allow topic to write to sqs queue
        sqs_region2_client.set_queue_attributes(
            QueueUrl=queue_url,
            Attributes={
                "Policy": json.dumps(
                    {
                        "Statement": [
                            {
                                "Effect": "Allow",
                                "Principal": {"Service": "sns.amazonaws.com"},
                                "Action": "sqs:SendMessage",
                                "Resource": queue_arn,
                                "Condition": {"ArnEquals": {"aws:SourceArn": topic_arn}},
                            }
                        ]
                    }
                )
            },
        )

        # connect sns topic to sqs
        with pytest.raises(ClientError) as e:
            sns_region2_client.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn)
        snapshot.match("subscribe-cross-region", e.value.response)

        subscription = sns_region1_client.subscribe(
            TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn
        )
        snapshot.match("subscribe-same-region", subscription)

        message = "This is a test message"
        # we already test that publishing from another region is forbidden with `test_topic_publish_another_region`
        sns_region1_client.publish(
            TopicArn=topic_arn,
            Message=message,
            MessageAttributes={"attr1": {"DataType": "Number", "StringValue": "99.12"}},
        )

        # assert that message is received
        response = sqs_region2_client.receive_message(
            QueueUrl=queue_url,
            VisibilityTimeout=0,
            MessageAttributeNames=["All"],
            WaitTimeSeconds=4,
        )
        snapshot.match("messages", response)


class TestSNSPublishDelivery:
    @markers.aws.validated
    @markers.snapshot.skip_snapshot_verify(
        paths=[
            "$..Attributes.DeliveryPolicy",
            "$..Attributes.EffectiveDeliveryPolicy",
            "$..Attributes.Policy.Statement..Action",  # SNS:Receive is added by moto but not returned in AWS
        ],
        condition=is_sns_v1_provider,
    )
    def test_delivery_lambda(
        self,
        sns_create_topic,
        sns_subscription,
        lambda_su_role,
        create_lambda_function,
        create_role,
        create_policy,
        snapshot,
        aws_client,
    ):
        snapshot.add_transformers_list(
            [
                snapshot.transform.key_value(
                    "dwellTimeMs", reference_replacement=False, value_replacement="<time-ms>"
                ),
                snapshot.transform.key_value("nextBackwardToken"),
                snapshot.transform.key_value("nextForwardToken"),
            ]
        )
        function_name = f"lambda-function-{short_uid()}"
        permission_id = f"test-statement-{short_uid()}"
        subject = "[Subject] Test subject"
        message_fail = "Should not be received"
        message_success = "Should be received"
        topic_name = f"test-topic-{short_uid()}"
        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]
        parsed_arn = parse_arn(topic_arn)
        account_id = parsed_arn["account"]
        region = parsed_arn["region"]
        role_name = f"SNSSuccessFeedback-{short_uid()}"
        policy_name = f"SNSSuccessFeedback-policy-{short_uid()}"

        # enable Success Feedback from SNS to be sent to CloudWatch
        trust_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "sns.amazonaws.com"},
                    "Action": "sts:AssumeRole",
                }
            ],
        }
        cloudwatch_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "logs:CreateLogGroup",
                        "logs:CreateLogStream",
                        "logs:PutLogEvents",
                        "logs:PutMetricFilter",
                        "logs:PutRetentionPolicy",
                    ],
                    "Resource": ["*"],
                }
            ],
        }

        role_response = create_role(
            RoleName=role_name, AssumeRolePolicyDocument=json.dumps(trust_policy)
        )
        role_arn = role_response["Role"]["Arn"]
        policy_arn = create_policy(
            PolicyName=policy_name, PolicyDocument=json.dumps(cloudwatch_policy)
        )["Policy"]["Arn"]
        aws_client.iam.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn)
        if is_aws_cloud():
            # wait for the policy to be properly attached
            time.sleep(20)

        topic_attributes = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-topic-attrs", topic_attributes)

        lambda_creation_response = create_lambda_function(
            func_name=function_name,
            handler_file=TEST_LAMBDA_PYTHON_ECHO,
            runtime=Runtime.python3_12,
            role=lambda_su_role,
        )
        lambda_arn = lambda_creation_response["CreateFunctionResponse"]["FunctionArn"]
        aws_client.lambda_.add_permission(
            FunctionName=function_name,
            StatementId=permission_id,
            Action="lambda:InvokeFunction",
            Principal="sns.amazonaws.com",
            SourceArn=topic_arn,
        )

        subscription = sns_subscription(
            TopicArn=topic_arn,
            Protocol="lambda",
            Endpoint=lambda_arn,
        )

        def check_subscription():
            subscription_arn = subscription["SubscriptionArn"]
            subscription_attrs = aws_client.sns.get_subscription_attributes(
                SubscriptionArn=subscription_arn
            )
            assert subscription_attrs["Attributes"]["PendingConfirmation"] == "false"

        retry(check_subscription, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

        publish_no_logs = aws_client.sns.publish(
            TopicArn=topic_arn, Subject=subject, Message=message_fail
        )
        snapshot.match("publish-no-logs", publish_no_logs)

        # Then enable the SNS Delivery Logs for Lambda on the topic
        aws_client.sns.set_topic_attributes(
            TopicArn=topic_arn,
            AttributeName="LambdaSuccessFeedbackRoleArn",
            AttributeValue=role_arn,
        )

        aws_client.sns.set_topic_attributes(
            TopicArn=topic_arn,
            AttributeName="LambdaSuccessFeedbackSampleRate",
            AttributeValue="100",
        )

        topic_attributes = aws_client.sns.get_topic_attributes(TopicArn=topic_arn)
        snapshot.match("get-topic-attrs-with-success-feedback", topic_attributes)

        publish_logs = aws_client.sns.publish(
            TopicArn=topic_arn, Subject=subject, Message=message_success
        )
        # we snapshot the publish call to match the messageId to the events
        snapshot.match("publish-logs", publish_logs)

        # TODO: Wait until Lambda function actually executes and not only for SNS logs
        log_group_name = f"sns/{region}/{account_id}/{topic_name}"

        def get_log_events():
            log_streams = aws_client.logs.describe_log_streams(logGroupName=log_group_name)[
                "logStreams"
            ]
            assert len(log_streams) == 1
            log_events = aws_client.logs.get_log_events(
                logGroupName=log_group_name,
                logStreamName=log_streams[0]["logStreamName"],
            )
            assert len(log_events["events"]) == 1
            # the default retention is 30 days, so delete the logGroup to clean up AWS
            with contextlib.suppress(ClientError):
                aws_client.logs.delete_log_group(logGroupName=log_group_name)
            return log_events

        sleep_time = 5 if is_aws_cloud() else 0.3
        events = retry(get_log_events, retries=10, sleep=sleep_time)

        # we need to decode the providerResponse to be able to properly match on the response
        # test would raise an error anyway if it's not a JSON string
        msg = json.loads(events["events"][0]["message"])
        events["events"][0]["message"] = msg
        events["events"][0]["message"]["delivery"]["providerResponse"] = json.loads(
            msg["delivery"]["providerResponse"]
        )

        snapshot.match("delivery-events", events)


class TestSNSCertEndpoint:
    @markers.requires_in_process
    @markers.aws.only_localstack
    @pytest.mark.parametrize("cert_host", ["", "sns.us-east-1.amazonaws.com"])
    def test_cert_endpoint_host(
        self,
        aws_client,
        sns_create_topic,
        sqs_create_queue,
        sns_create_sqs_subscription,
        monkeypatch,
        cert_host,
    ):
        """
        Some SDK will validate the Cert URL matches a certain regex pattern. We validate the user can set the value
        to arbitrary host, but those will obviously not resolve / return a valid certificate.
        """
        monkeypatch.setattr(config, "SNS_CERT_URL_HOST", cert_host)
        topic_arn = sns_create_topic(
            Attributes={
                "DisplayName": "TestTopicSignature",
                "SignatureVersion": "1",
            },
        )["TopicArn"]

        queue_url = sqs_create_queue()
        sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)

        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message="test cert host",
        )
        response = aws_client.sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=10,
        )
        message = json.loads(response["Messages"][0]["Body"])

        cert_url = message["SigningCertURL"]
        if not cert_host:
            assert external_service_url() in cert_url
        else:
            assert cert_host in cert_url
            assert external_service_url() not in cert_url


@pytest.mark.usefixtures("openapi_validate")
class TestSNSRetrospectionEndpoints:
    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_publish_to_platform_endpoint_can_retrospect(
        self,
        sns_create_topic,
        sns_subscription,
        sns_create_platform_application,
        aws_client,
        account_id,
        region_name,
        secondary_region_name,
        sns_provider,
        platform_credentials,
    ):
        platform = "APNS"
        client_id, client_secret = platform_credentials
        attributes = {"PlatformPrincipal": client_id, "PlatformCredential": client_secret}
        SnsProvider = sns_provider()
        sns_backend = SnsProvider.get_store(account_id, region_name)
        # clean up the saved messages
        sns_backend_endpoint_arns = list(sns_backend.platform_endpoint_messages.keys())
        for saved_endpoint_arn in sns_backend_endpoint_arns:
            sns_backend.platform_endpoint_messages.pop(saved_endpoint_arn, None)

        topic_arn = sns_create_topic()["TopicArn"]
        application_platform_name = f"app-platform-{short_uid()}"

        app_arn = sns_create_platform_application(
            Name=application_platform_name, Platform=platform, Attributes=attributes
        )["PlatformApplicationArn"]

        endpoint_arn = aws_client.sns.create_platform_endpoint(
            PlatformApplicationArn=app_arn, Token=short_uid()
        )["EndpointArn"]

        endpoint_arn_2 = aws_client.sns.create_platform_endpoint(
            PlatformApplicationArn=app_arn, Token=short_uid()
        )["EndpointArn"]

        sns_subscription(
            TopicArn=topic_arn,
            Protocol="application",
            Endpoint=endpoint_arn,
        )

        # example message from
        # https://docs.aws.amazon.com/sns/latest/dg/sns-send-custom-platform-specific-payloads-mobile-devices.html
        message = json.dumps({"APNS": json.dumps({"aps": {"content-available": 1}})})
        message_for_topic = {
            "default": "This is the default message which must be present when publishing a message to a topic.",
            "APNS": json.dumps({"aps": {"content-available": 1}}),
        }
        message_for_topic_string = json.dumps(message_for_topic)
        message_attributes = {
            "AWS.SNS.MOBILE.APNS.TOPIC": {
                "DataType": "String",
                "StringValue": "com.amazon.mobile.messaging.myapp",
            },
            "AWS.SNS.MOBILE.APNS.PUSH_TYPE": {
                "DataType": "String",
                "StringValue": "background",
            },
            "AWS.SNS.MOBILE.APNS.PRIORITY": {
                "DataType": "String",
                "StringValue": "5",
            },
        }
        # publish to a topic which has a platform subscribed to it
        aws_client.sns.publish(
            TopicArn=topic_arn,
            Message=message_for_topic_string,
            MessageAttributes=message_attributes,
            MessageStructure="json",
        )
        # publish directly to the platform endpoint
        aws_client.sns.publish(
            TargetArn=endpoint_arn_2,
            Message=message,
            MessageAttributes=message_attributes,
            MessageStructure="json",
        )

        # assert that message has been received
        def check_message():
            assert len(sns_backend.platform_endpoint_messages[endpoint_arn]) > 0

        retry(check_message, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

        msgs_url = config.internal_service_url() + PLATFORM_ENDPOINT_MSGS_ENDPOINT
        api_contents = requests.get(
            msgs_url, params={"region": region_name, "accountId": account_id}
        ).json()
        api_platform_endpoints_msgs = api_contents["platform_endpoint_messages"]

        assert len(api_platform_endpoints_msgs) == 2
        assert len(api_platform_endpoints_msgs[endpoint_arn]) == 1
        assert len(api_platform_endpoints_msgs[endpoint_arn_2]) == 1
        assert api_contents["region"] == region_name

        assert api_platform_endpoints_msgs[endpoint_arn][0]["Message"] == json.dumps(
            message_for_topic["APNS"]
        )
        assert (
            api_platform_endpoints_msgs[endpoint_arn][0]["MessageAttributes"] == message_attributes
        )

        # Ensure you can select the region
        msg_with_region = requests.get(
            msgs_url,
            params={"region": secondary_region_name, "accountId": account_id},
        ).json()
        assert len(msg_with_region["platform_endpoint_messages"]) == 0
        assert msg_with_region["region"] == secondary_region_name

        # Ensure default region is us-east-1
        msg_with_region = requests.get(msgs_url).json()
        assert msg_with_region["region"] == AWS_REGION_US_EAST_1

        # Ensure messages can be filtered by EndpointArn
        api_contents_with_endpoint = requests.get(
            msgs_url,
            params={
                "endpointArn": endpoint_arn,
                "region": region_name,
                "accountId": account_id,
            },
        ).json()
        msgs_with_endpoint = api_contents_with_endpoint["platform_endpoint_messages"]
        assert len(msgs_with_endpoint) == 1
        assert len(msgs_with_endpoint[endpoint_arn]) == 1
        assert api_contents_with_endpoint["region"] == region_name

        # Ensure you can reset the saved messages by EndpointArn
        delete_res = requests.delete(
            msgs_url,
            params={
                "endpointArn": endpoint_arn,
                "region": region_name,
                "accountId": account_id,
            },
        )
        assert delete_res.status_code == 204
        api_contents_with_endpoint = requests.get(
            msgs_url,
            params={
                "endpointArn": endpoint_arn,
                "region": region_name,
                "accountId": account_id,
            },
        ).json()
        msgs_with_endpoint = api_contents_with_endpoint["platform_endpoint_messages"]
        assert len(msgs_with_endpoint[endpoint_arn]) == 0

        # Ensure you can reset the saved messages by region
        delete_res = requests.delete(
            msgs_url, params={"region": region_name, "accountId": account_id}
        )
        assert delete_res.status_code == 204
        msg_with_region = requests.get(
            msgs_url, params={"region": region_name, "accountId": account_id}
        ).json()
        assert not msg_with_region["platform_endpoint_messages"]

    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_publish_sms_can_retrospect(
        self,
        sns_create_topic,
        sns_subscription,
        aws_client,
        account_id,
        region_name,
        secondary_region_name,
        sns_provider,
    ):
        SnsProvider = sns_provider()
        sns_store = SnsProvider.get_store(account_id, region_name)

        list_of_contacts = [
            f"+{random.randint(100000000, 9999999999)}",
            f"+{random.randint(100000000, 9999999999)}",
            f"+{random.randint(100000000, 9999999999)}",
        ]
        phone_number_1 = list_of_contacts[0]
        message = "Good news everyone!"
        topic_arn = sns_create_topic()["TopicArn"]
        for number in list_of_contacts:
            sns_subscription(TopicArn=topic_arn, Protocol="sms", Endpoint=number)

        # clean up the saved messages
        sns_store.sms_messages.clear()

        # publish to a topic which has a PhoneNumbers subscribed to it
        aws_client.sns.publish(Message=message, TopicArn=topic_arn)

        # publish directly to the PhoneNumber
        aws_client.sns.publish(
            PhoneNumber=phone_number_1,
            Message=message,
        )

        # assert that message has been received
        def check_message():
            assert len(sns_store.sms_messages) == 4

        retry(check_message, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)

        msgs_url = config.internal_service_url() + SMS_MSGS_ENDPOINT
        api_contents = requests.get(
            msgs_url, params={"region": region_name, "accountId": account_id}
        ).json()
        api_sms_msgs = api_contents["sms_messages"]

        assert len(api_sms_msgs) == 3
        assert len(api_sms_msgs[phone_number_1]) == 2
        assert len(api_sms_msgs[list_of_contacts[1]]) == 1
        assert len(api_sms_msgs[list_of_contacts[2]]) == 1

        assert api_contents["region"] == region_name

        assert api_sms_msgs[phone_number_1][0]["Message"] == "Good news everyone!"

        # Ensure you can select the region
        msg_with_region = requests.get(msgs_url, params={"region": secondary_region_name}).json()
        assert len(msg_with_region["sms_messages"]) == 0
        assert msg_with_region["region"] == secondary_region_name

        # Ensure default region is us-east-1
        msg_with_region = requests.get(msgs_url).json()
        assert msg_with_region["region"] == AWS_REGION_US_EAST_1

        # Ensure messages can be filtered by EndpointArn
        api_contents_with_number = requests.get(
            msgs_url,
            params={
                "phoneNumber": phone_number_1,
                "accountId": account_id,
                "region": region_name,
            },
        ).json()
        msgs_with_number = api_contents_with_number["sms_messages"]
        assert len(msgs_with_number) == 1
        assert len(msgs_with_number[phone_number_1]) == 2
        assert api_contents_with_number["region"] == region_name

        # Ensure you can reset the saved messages by EndpointArn
        delete_res = requests.delete(
            msgs_url,
            params={
                "phoneNumber": phone_number_1,
                "accountId": account_id,
                "region": region_name,
            },
        )
        assert delete_res.status_code == 204
        api_contents_with_number = requests.get(
            msgs_url, params={"phoneNumber": phone_number_1}
        ).json()
        msgs_with_number = api_contents_with_number["sms_messages"]
        assert len(msgs_with_number[phone_number_1]) == 0

        # Ensure you can reset the saved messages by region
        delete_res = requests.delete(
            msgs_url, params={"region": region_name, "accountId": account_id}
        )
        assert delete_res.status_code == 204
        msg_with_region = requests.get(msgs_url, params={"region": region_name}).json()
        assert not msg_with_region["sms_messages"]

    @markers.requires_in_process
    @markers.aws.only_localstack
    def test_subscription_tokens_can_retrospect(
        self,
        sns_create_topic,
        sns_subscription,
        sns_create_http_endpoint,
        aws_client,
        account_id,
        region_name,
        sns_provider,
    ):
        SnsProvider = sns_provider()
        sns_store = SnsProvider.get_store(account_id, region_name)
        # clean up the saved tokens
        sns_store.subscription_tokens.clear()

        message = "Good news everyone!"
        # Necessitate manual set up to allow external access to endpoint, only in local testing
        topic_arn, subscription_arn, endpoint_url, server = sns_create_http_endpoint()
        assert poll_condition(
            lambda: len(server.log) >= 1,
            timeout=5,
        )
        sub_request, _ = server.log[0]
        payload = sub_request.get_json(force=True)
        assert payload["Type"] == "SubscriptionConfirmation"
        token = payload["Token"]
        server.clear()

        # we won't confirm the subscription, to simulate an external provider that wouldn't be able to access LocalStack
        # try to access the internal to confirm the Token is there
        tokens_base_url = config.internal_service_url() + SUBSCRIPTION_TOKENS_ENDPOINT
        api_contents = requests.get(f"{tokens_base_url}/{subscription_arn}").json()
        assert api_contents["subscription_token"] == token
        assert api_contents["subscription_arn"] == subscription_arn

        # try to send a message to an unconfirmed subscription, assert that the message isn't received
        aws_client.sns.publish(Message=message, TopicArn=topic_arn)

        assert poll_condition(
            lambda: len(server.log) == 0,
            timeout=1,
        )

        aws_client.sns.confirm_subscription(TopicArn=topic_arn, Token=token)
        aws_client.sns.publish(Message=message, TopicArn=topic_arn)
        assert poll_condition(
            lambda: len(server.log) == 1,
            timeout=2,
        )

        wrong_sub_arn = subscription_arn.replace(
            region_name,
            "il-central-1" if region_name != "il-central-1" else "me-south-1",
        )
        wrong_region_req = requests.get(f"{tokens_base_url}/{wrong_sub_arn}")
        assert wrong_region_req.status_code == 404
        assert wrong_region_req.json() == {
            "error": "The provided SubscriptionARN is not found",
            "subscription_arn": wrong_sub_arn,
        }

        # Ensure proper error is raised with wrong ARN
        incorrect_arn_req = requests.get(f"{tokens_base_url}/randomarnhere")
        assert incorrect_arn_req.status_code == 400
        assert incorrect_arn_req.json() == {
            "error": "The provided SubscriptionARN is invalid",
            "subscription_arn": "randomarnhere",
        }
