import boto3
from botocore.exceptions import ClientError
import sure  # noqa # pylint: disable=unused-import
from datetime import datetime
import pytest

from moto import mock_dynamodb, settings
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
from moto.dynamodb.models import dynamodb_backends


@mock_dynamodb
def test_create_table_standard():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="messages",
        KeySchema=[
            {"AttributeName": "id", "KeyType": "HASH"},
            {"AttributeName": "subject", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
            {"AttributeName": "subject", "AttributeType": "S"},
        ],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 5},
    )
    actual = client.describe_table(TableName="messages")["Table"]

    actual.should.have.key("AttributeDefinitions").equal(
        [
            {"AttributeName": "id", "AttributeType": "S"},
            {"AttributeName": "subject", "AttributeType": "S"},
        ]
    )
    actual.should.have.key("CreationDateTime").be.a(datetime)
    actual.should.have.key("GlobalSecondaryIndexes").equal([])
    actual.should.have.key("LocalSecondaryIndexes").equal([])
    actual.should.have.key("ProvisionedThroughput").equal(
        {"NumberOfDecreasesToday": 0, "ReadCapacityUnits": 1, "WriteCapacityUnits": 5}
    )
    actual.should.have.key("TableSizeBytes").equal(0)
    actual.should.have.key("TableName").equal("messages")
    actual.should.have.key("TableStatus").equal("ACTIVE")
    actual.should.have.key("TableArn").equal(
        f"arn:aws:dynamodb:us-east-1:{ACCOUNT_ID}:table/messages"
    )
    actual.should.have.key("KeySchema").equal(
        [
            {"AttributeName": "id", "KeyType": "HASH"},
            {"AttributeName": "subject", "KeyType": "RANGE"},
        ]
    )
    actual.should.have.key("ItemCount").equal(0)


@mock_dynamodb
def test_create_table_with_local_index():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="messages",
        KeySchema=[
            {"AttributeName": "id", "KeyType": "HASH"},
            {"AttributeName": "subject", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
            {"AttributeName": "subject", "AttributeType": "S"},
            {"AttributeName": "threads", "AttributeType": "S"},
        ],
        LocalSecondaryIndexes=[
            {
                "IndexName": "threads_index",
                "KeySchema": [
                    {"AttributeName": "id", "KeyType": "HASH"},
                    {"AttributeName": "threads", "KeyType": "RANGE"},
                ],
                "Projection": {"ProjectionType": "ALL"},
            }
        ],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 5},
    )
    actual = client.describe_table(TableName="messages")["Table"]

    actual.should.have.key("AttributeDefinitions").equal(
        [
            {"AttributeName": "id", "AttributeType": "S"},
            {"AttributeName": "subject", "AttributeType": "S"},
            {"AttributeName": "threads", "AttributeType": "S"},
        ]
    )
    actual.should.have.key("CreationDateTime").be.a(datetime)
    actual.should.have.key("GlobalSecondaryIndexes").equal([])
    actual.should.have.key("LocalSecondaryIndexes").equal(
        [
            {
                "IndexName": "threads_index",
                "KeySchema": [
                    {"AttributeName": "id", "KeyType": "HASH"},
                    {"AttributeName": "threads", "KeyType": "RANGE"},
                ],
                "Projection": {"ProjectionType": "ALL"},
            }
        ]
    )
    actual.should.have.key("ProvisionedThroughput").equal(
        {"NumberOfDecreasesToday": 0, "ReadCapacityUnits": 1, "WriteCapacityUnits": 5}
    )
    actual.should.have.key("TableSizeBytes").equal(0)
    actual.should.have.key("TableName").equal("messages")
    actual.should.have.key("TableStatus").equal("ACTIVE")
    actual.should.have.key("TableArn").equal(
        f"arn:aws:dynamodb:us-east-1:{ACCOUNT_ID}:table/messages"
    )
    actual.should.have.key("KeySchema").equal(
        [
            {"AttributeName": "id", "KeyType": "HASH"},
            {"AttributeName": "subject", "KeyType": "RANGE"},
        ]
    )
    actual.should.have.key("ItemCount").equal(0)


@mock_dynamodb
def test_create_table_with_gsi():
    dynamodb = boto3.client("dynamodb", region_name="us-east-1")

    table = dynamodb.create_table(
        TableName="users",
        KeySchema=[
            {"AttributeName": "forum_name", "KeyType": "HASH"},
            {"AttributeName": "subject", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "forum_name", "AttributeType": "S"},
            {"AttributeName": "subject", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
        GlobalSecondaryIndexes=[
            {
                "IndexName": "test_gsi",
                "KeySchema": [{"AttributeName": "subject", "KeyType": "HASH"}],
                "Projection": {"ProjectionType": "ALL"},
            }
        ],
    )
    table["TableDescription"]["GlobalSecondaryIndexes"].should.equal(
        [
            {
                "KeySchema": [{"KeyType": "HASH", "AttributeName": "subject"}],
                "IndexName": "test_gsi",
                "Projection": {"ProjectionType": "ALL"},
                "IndexStatus": "ACTIVE",
                "ProvisionedThroughput": {
                    "ReadCapacityUnits": 0,
                    "WriteCapacityUnits": 0,
                },
            }
        ]
    )

    table = dynamodb.create_table(
        TableName="users2",
        KeySchema=[
            {"AttributeName": "forum_name", "KeyType": "HASH"},
            {"AttributeName": "subject", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "forum_name", "AttributeType": "S"},
            {"AttributeName": "subject", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
        GlobalSecondaryIndexes=[
            {
                "IndexName": "test_gsi",
                "KeySchema": [{"AttributeName": "subject", "KeyType": "HASH"}],
                "Projection": {"ProjectionType": "ALL"},
                "ProvisionedThroughput": {
                    "ReadCapacityUnits": 3,
                    "WriteCapacityUnits": 5,
                },
            }
        ],
    )
    table["TableDescription"]["GlobalSecondaryIndexes"].should.equal(
        [
            {
                "KeySchema": [{"KeyType": "HASH", "AttributeName": "subject"}],
                "IndexName": "test_gsi",
                "Projection": {"ProjectionType": "ALL"},
                "IndexStatus": "ACTIVE",
                "ProvisionedThroughput": {
                    "ReadCapacityUnits": 3,
                    "WriteCapacityUnits": 5,
                },
            }
        ]
    )


@mock_dynamodb
def test_create_table_with_stream_specification():
    conn = boto3.client("dynamodb", region_name="us-east-1")

    resp = conn.create_table(
        TableName="test-streams",
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
        StreamSpecification={
            "StreamEnabled": True,
            "StreamViewType": "NEW_AND_OLD_IMAGES",
        },
    )

    resp["TableDescription"].should.have.key("StreamSpecification")
    resp["TableDescription"]["StreamSpecification"].should.equal(
        {"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"}
    )
    resp["TableDescription"].should.contain("LatestStreamLabel")
    resp["TableDescription"].should.contain("LatestStreamArn")

    resp = conn.delete_table(TableName="test-streams")

    resp["TableDescription"].should.contain("StreamSpecification")


@mock_dynamodb
def test_create_table_with_tags():
    client = boto3.client("dynamodb", region_name="us-east-1")

    resp = client.create_table(
        TableName="test-streams",
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
        Tags=[{"Key": "tk", "Value": "tv"}],
    )

    resp = client.list_tags_of_resource(
        ResourceArn=resp["TableDescription"]["TableArn"]
    )
    resp.should.have.key("Tags").equals([{"Key": "tk", "Value": "tv"}])


@mock_dynamodb
def test_create_table_pay_per_request():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="test1",
        AttributeDefinitions=[
            {"AttributeName": "client", "AttributeType": "S"},
            {"AttributeName": "app", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "client", "KeyType": "HASH"},
            {"AttributeName": "app", "KeyType": "RANGE"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )

    actual = client.describe_table(TableName="test1")["Table"]
    actual.should.have.key("BillingModeSummary").equals(
        {"BillingMode": "PAY_PER_REQUEST"}
    )
    actual.should.have.key("ProvisionedThroughput").equals(
        {"NumberOfDecreasesToday": 0, "ReadCapacityUnits": 0, "WriteCapacityUnits": 0}
    )


@mock_dynamodb
def test_create_table__provisioned_throughput():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="test1",
        AttributeDefinitions=[
            {"AttributeName": "client", "AttributeType": "S"},
            {"AttributeName": "app", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "client", "KeyType": "HASH"},
            {"AttributeName": "app", "KeyType": "RANGE"},
        ],
        ProvisionedThroughput={"ReadCapacityUnits": 2, "WriteCapacityUnits": 3},
    )

    actual = client.describe_table(TableName="test1")["Table"]
    actual.should.have.key("BillingModeSummary").equals({"BillingMode": "PROVISIONED"})
    actual.should.have.key("ProvisionedThroughput").equals(
        {"NumberOfDecreasesToday": 0, "ReadCapacityUnits": 2, "WriteCapacityUnits": 3}
    )


@mock_dynamodb
def test_create_table_without_specifying_throughput():
    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")

    with pytest.raises(ClientError) as exc:
        dynamodb_client.create_table(
            TableName="my-table",
            AttributeDefinitions=[
                {"AttributeName": "some_field", "AttributeType": "S"}
            ],
            KeySchema=[{"AttributeName": "some_field", "KeyType": "HASH"}],
            BillingMode="PROVISIONED",
            StreamSpecification={"StreamEnabled": False, "StreamViewType": "NEW_IMAGE"},
        )
    err = exc.value.response["Error"]
    err["Code"].should.equal("ValidationException")
    err["Message"].should.equal(
        "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
    )


@mock_dynamodb
def test_create_table_error_pay_per_request_with_provisioned_param():
    client = boto3.client("dynamodb", region_name="us-east-1")

    with pytest.raises(ClientError) as exc:
        client.create_table(
            TableName="test1",
            AttributeDefinitions=[
                {"AttributeName": "client", "AttributeType": "S"},
                {"AttributeName": "app", "AttributeType": "S"},
            ],
            KeySchema=[
                {"AttributeName": "client", "KeyType": "HASH"},
                {"AttributeName": "app", "KeyType": "RANGE"},
            ],
            ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
            BillingMode="PAY_PER_REQUEST",
        )
    err = exc.value.response["Error"]
    err["Code"].should.equal("ValidationException")
    err["Message"].should.equal(
        "ProvisionedThroughput cannot be specified when BillingMode is PAY_PER_REQUEST"
    )


@mock_dynamodb
def test_create_table_with_ssespecification__false():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="test1",
        AttributeDefinitions=[
            {"AttributeName": "client", "AttributeType": "S"},
            {"AttributeName": "app", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "client", "KeyType": "HASH"},
            {"AttributeName": "app", "KeyType": "RANGE"},
        ],
        BillingMode="PAY_PER_REQUEST",
        SSESpecification={"Enabled": False},
    )

    actual = client.describe_table(TableName="test1")["Table"]
    actual.shouldnt.have.key("SSEDescription")


@mock_dynamodb
def test_create_table_with_ssespecification__true():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="test1",
        AttributeDefinitions=[
            {"AttributeName": "client", "AttributeType": "S"},
            {"AttributeName": "app", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "client", "KeyType": "HASH"},
            {"AttributeName": "app", "KeyType": "RANGE"},
        ],
        BillingMode="PAY_PER_REQUEST",
        SSESpecification={"Enabled": True},
    )

    actual = client.describe_table(TableName="test1")["Table"]
    actual.should.have.key("SSEDescription")
    actual["SSEDescription"].should.have.key("Status").equals("ENABLED")
    actual["SSEDescription"].should.have.key("SSEType").equals("KMS")
    actual["SSEDescription"].should.have.key("KMSMasterKeyArn").match(
        "^arn:aws:kms"
    )  # Default KMS key for DynamoDB


@mock_dynamodb
def test_create_table_with_ssespecification__custom_kms_key():
    client = boto3.client("dynamodb", region_name="us-east-1")
    client.create_table(
        TableName="test1",
        AttributeDefinitions=[
            {"AttributeName": "client", "AttributeType": "S"},
            {"AttributeName": "app", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "client", "KeyType": "HASH"},
            {"AttributeName": "app", "KeyType": "RANGE"},
        ],
        BillingMode="PAY_PER_REQUEST",
        SSESpecification={"Enabled": True, "KMSMasterKeyId": "custom-kms-key"},
    )

    actual = client.describe_table(TableName="test1")["Table"]
    actual.should.have.key("SSEDescription")
    actual["SSEDescription"].should.have.key("Status").equals("ENABLED")
    actual["SSEDescription"].should.have.key("SSEType").equals("KMS")
    actual["SSEDescription"].should.have.key("KMSMasterKeyArn").equals("custom-kms-key")


@mock_dynamodb
def test_create_table__specify_non_key_column():
    client = boto3.client("dynamodb", "us-east-2")
    client.create_table(
        TableName="tab",
        KeySchema=[
            {"AttributeName": "PK", "KeyType": "HASH"},
            {"AttributeName": "SomeColumn", "KeyType": "N"},
        ],
        BillingMode="PAY_PER_REQUEST",
        AttributeDefinitions=[
            {"AttributeName": "PK", "AttributeType": "S"},
            {"AttributeName": "SomeColumn", "AttributeType": "N"},
        ],
    )

    actual = client.describe_table(TableName="tab")["Table"]
    actual["KeySchema"].should.equal(
        [
            {"AttributeName": "PK", "KeyType": "HASH"},
            {"AttributeName": "SomeColumn", "KeyType": "N"},
        ]
    )

    if not settings.TEST_SERVER_MODE:
        ddb = dynamodb_backends[ACCOUNT_ID]["us-east-2"]
        ddb.tables["tab"].attr.should.contain(
            {"AttributeName": "PK", "AttributeType": "S"}
        )
        ddb.tables["tab"].attr.should.contain(
            {"AttributeName": "SomeColumn", "AttributeType": "N"}
        )
        # It should recognize PK is the Hash Key
        ddb.tables["tab"].hash_key_attr.should.equal("PK")
        # It should recognize that SomeColumn is not a Range Key
        ddb.tables["tab"].has_range_key.should.equal(False)
        ddb.tables["tab"].range_key_names.should.equal([])