import json

import boto3
import pytest
from botocore.exceptions import ClientError

from moto import mock_aws
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID


@mock_aws
def test_publish_batch_unknown_topic():
    client = boto3.client("sns", region_name="us-east-1")
    with pytest.raises(ClientError) as exc:
        client.publish_batch(
            TopicArn=f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:unknown",
            PublishBatchRequestEntries=[{"Id": "id_1", "Message": "1"}],
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "NotFound"
    assert err["Message"] == "Topic does not exist"


@mock_aws
def test_publish_batch_too_many_items():
    client = boto3.client("sns", region_name="eu-north-1")
    topic = client.create_topic(Name="some-topic")

    with pytest.raises(ClientError) as exc:
        client.publish_batch(
            TopicArn=topic["TopicArn"],
            PublishBatchRequestEntries=[
                {"Id": f"id_{idx}", "Message": f"{idx}"} for idx in range(11)
            ],
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "TooManyEntriesInBatchRequest"
    assert err["Message"] == "The batch request contains more entries than permissible."


@mock_aws
def test_publish_batch_non_unique_ids():
    client = boto3.client("sns", region_name="us-west-2")
    topic = client.create_topic(Name="some-topic")

    with pytest.raises(ClientError) as exc:
        client.publish_batch(
            TopicArn=topic["TopicArn"],
            PublishBatchRequestEntries=[
                {"Id": "id", "Message": f"{idx}"} for idx in range(5)
            ],
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "BatchEntryIdsNotDistinct"
    assert (
        err["Message"] == "Two or more batch entries in the request have the same Id."
    )


@mock_aws
def test_publish_batch_fifo_without_message_group_id():
    client = boto3.client("sns", region_name="us-east-1")
    topic = client.create_topic(
        Name="fifo_without_msg.fifo",
        Attributes={"FifoTopic": "true", "ContentBasedDeduplication": "true"},
    )

    with pytest.raises(ClientError) as exc:
        client.publish_batch(
            TopicArn=topic["TopicArn"],
            PublishBatchRequestEntries=[{"Id": "id_2", "Message": "2"}],
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameter"
    assert err["Message"] == (
        "Invalid parameter: The MessageGroupId parameter is required for FIFO topics"
    )


@mock_aws
def test_publish_batch_standard_with_message_group_id():
    client = boto3.client("sns", region_name="us-east-1")
    topic_arn = client.create_topic(Name="standard_topic")["TopicArn"]
    entries = [
        {"Id": "id_1", "Message": "1"},
        {"Id": "id_2", "Message": "2", "MessageGroupId": "mgid"},
        {"Id": "id_3", "Message": "3"},
    ]
    resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)

    assert len(resp["Successful"]) == 2
    for message_status in resp["Successful"]:
        assert "MessageId" in message_status
    assert [m["Id"] for m in resp["Successful"]] == ["id_1", "id_3"]

    assert len(resp["Failed"]) == 1
    assert resp["Failed"][0] == {
        "Id": "id_2",
        "Code": "InvalidParameter",
        "Message": (
            "Invalid parameter: MessageGroupId Reason: The request includes "
            "MessageGroupId parameter that is not valid for this topic type"
        ),
        "SenderFault": True,
    }


@mock_aws
def test_publish_batch_to_sqs():
    client = boto3.client("sns", region_name="us-east-1")
    topic_arn = client.create_topic(Name="standard_topic")["TopicArn"]
    entries = [
        {"Id": "id_1", "Message": "1"},
        {"Id": "id_2", "Message": "2", "Subject": "subj2"},
        {
            "Id": "id_3",
            "Message": "3",
            "MessageAttributes": {"a": {"DataType": "String", "StringValue": "v"}},
        },
    ]

    sqs_conn = boto3.resource("sqs", region_name="us-east-1")
    queue = sqs_conn.create_queue(QueueName="test-queue")

    queue_url = f"arn:aws:sqs:us-east-1:{ACCOUNT_ID}:test-queue"
    client.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_url)

    resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)

    assert len(resp["Successful"]) == 3

    messages = queue.receive_messages(MaxNumberOfMessages=3)
    assert len(messages) == 3

    messages = [json.loads(m.body) for m in messages]
    for message in messages:
        for key in list(message.keys()):
            if key not in ["Message", "Subject", "MessageAttributes"]:
                del message[key]

    assert {"Message": "1"} in messages
    assert {"Message": "2", "Subject": "subj2"} in messages
    assert (
        {"Message": "3", "MessageAttributes": {"a": {"Type": "String", "Value": "v"}}}
    ) in messages


@mock_aws
def test_publish_batch_to_sqs_raw():
    client = boto3.client("sns", region_name="us-east-1")
    topic_arn = client.create_topic(Name="standard_topic")["TopicArn"]
    sqs = boto3.resource("sqs", region_name="us-east-1")
    queue = sqs.create_queue(QueueName="test-queue")

    queue_url = f"arn:aws:sqs:us-east-1:{ACCOUNT_ID}:test-queue"
    client.subscribe(
        TopicArn=topic_arn,
        Protocol="sqs",
        Endpoint=queue_url,
        Attributes={"RawMessageDelivery": "true"},
    )

    entries = [
        {"Id": "1", "Message": "foo"},
        {
            "Id": "2",
            "Message": "bar",
            "MessageAttributes": {"a": {"DataType": "String", "StringValue": "v"}},
        },
    ]
    resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)

    assert len(resp["Successful"]) == 2

    received = queue.receive_messages(
        MaxNumberOfMessages=10, MessageAttributeNames=["All"]
    )

    messages = [(message.body, message.message_attributes) for message in received]

    assert ("foo", None) in messages
    assert ("bar", {"a": {"StringValue": "v", "DataType": "String"}}) in messages


@mock_aws
def test_publish_with_with_message_structure_json():
    sns = boto3.client("sns", region_name="us-east-1")

    topic = sns.create_topic(Name="some-topic")

    sqs = boto3.resource("sqs", region_name="us-east-1")
    queue = sqs.create_queue(QueueName="test-queue")

    sns.subscribe(
        TopicArn=topic["TopicArn"],
        Protocol="sqs",
        Endpoint=queue.attributes["QueueArn"],
    )

    entries = [
        {
            "Id": "no-message-structure",
            "Message": "no-message-structure",
        },
        {
            "Id": "json-message-structure",
            "Message": json.dumps(
                {
                    "default": "default-message",
                    "sqs": "queue-message",
                }
            ),
            "MessageStructure": "json",
        },
        {
            "Id": "json-message-structure-fallback-to-default",
            "Message": json.dumps(
                {
                    "default": "default-message",
                    "email": "email-message",
                }
            ),
            "MessageStructure": "json",
        },
    ]

    sns.publish_batch(
        TopicArn=topic["TopicArn"],
        PublishBatchRequestEntries=entries,
    )

    queue_messages = queue.receive_messages(MaxNumberOfMessages=10)

    assert len(queue_messages) == 3

    first_queue_message = json.loads(queue_messages[0].body)
    assert first_queue_message["Message"] == "no-message-structure"

    second_queue_message = json.loads(queue_messages[1].body)
    assert second_queue_message["Message"] == "queue-message"

    third_queue_message = json.loads(queue_messages[2].body)
    assert third_queue_message["Message"] == "default-message"


@mock_aws
def test_publish_with_upper_cased_message_attributes():
    sns = boto3.client("sns", "us-east-1")
    sqs = boto3.resource("sqs", "us-east-1")

    topic_arn = sns.create_topic(Name="topic")["TopicArn"]
    queue = sqs.create_queue(QueueName="queue")
    queue_arn = sqs.meta.client.get_queue_attributes(
        QueueUrl=queue.url, AttributeNames=["QueueArn"]
    )["Attributes"]["QueueArn"]

    sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn)

    sns.publish_batch(
        TopicArn=topic_arn,
        PublishBatchRequestEntries=[
            {
                "Id": "some id",
                "Message": json.dumps({"type": "test"}),
                "MessageAttributes": {
                    "SomeCamelCaseAttr": {"DataType": "String", "StringValue": "value"}
                },
            }
        ],
    )

    msg = queue.receive_messages()[0]
    body = json.loads(msg.body)

    assert body["MessageAttributes"] == {
        "SomeCamelCaseAttr": {"Type": "String", "Value": "value"}
    }