"""Test Dynamo Utils."""

from typing import Any, Dict
from unittest.mock import AsyncMock, MagicMock, call, patch

import pytest
from botocore.config import Config

from pdp.utils import dynamo


@pytest.fixture()
def botocore_config() -> Config:
    """Create a botocore.Config object."""
    return Config(tcp_keepalive=False)


@patch("pdp.utils.dynamo.get_config")
def test_get_opts_returns_nothing(
    mock_get_config: MagicMock, botocore_config: Config
) -> None:
    """Test opts returns empty dict when no endpoint url is configured."""
    mock_config = type(
        "config",
        (object,),
        {"DYNAMODB_ENDPOINT_URL": None, "DYNAMODB_TCP_KEEP_ALIVE": False},
    )()
    mock_get_config.return_value = botocore_config
    result = dynamo.get_opts(mock_config)
    assert result == {"config": botocore_config}


@patch("pdp.utils.dynamo.get_config")
def test_get_opts_returns_endpoint_url_configuration(
    mock_get_config: MagicMock, botocore_config: Config
) -> None:
    """Test opts returns endpoint_url url configuration."""
    mock_config = type(
        "config",
        (object,),
        {
            "DYNAMODB_ENDPOINT_URL": "http://dynamodb-here.com",
            "DYNAMODB_TCP_KEEP_ALIVE": False,
        },
    )()
    mock_get_config.return_value = botocore_config
    result = dynamo.get_opts(mock_config)
    assert result == {
        "endpoint_url": "http://dynamodb-here.com",
        "config": botocore_config,
    }


@patch("pdp.utils.dynamo.Config")
def test_get_config(mock_config_cls: MagicMock) -> None:
    """Test botocore configuration returned by get_config."""
    mock_config_instance = mock_config_cls.return_value
    mock_config = type(
        "config",
        (object,),
        {
            "DYNAMODB_ENDPOINT_URL": "http://dynamodb-here.com",
            "DYNAMODB_TCP_KEEP_ALIVE": False,
        },
    )()

    result_boto_config = dynamo.get_config(mock_config)
    assert result_boto_config == mock_config_instance
    mock_config_cls.assert_called_once_with(tcp_keepalive=False)


def test_encode_pagination_cursor(dynamo_last_evaluated_key: Dict[str, Any]) -> None:
    """Test encode pagination cursor."""
    result = dynamo.encode_pagination_cursor(dynamo_last_evaluated_key)
    assert (
        result
        == "eyJpZGVudGl0eV91dWlkIjogeyJTIjogImM1ODc5MzY1LTJjMDctNGMxOS1hM2YwLWNjMDMwYzAxYjZhNSJ9LCAidGVuYW50X3V1aWQiOiB7IlMiOiAidGVzdC11dWlkLTEyMzQ1NiJ9fQ=="  # noqa: E501
    )


def test_encode_pagination_shorthand(dynamo_last_evaluated_key: Dict[str, Any]) -> None:
    """Test encode pagination as shorthand."""
    result = dynamo.encode_pagination_shorthand(
        dynamo_last_evaluated_key, "identity_uuid"
    )
    assert result == "c5879365-2c07-4c19-a3f0-cc030c01b6a5"

    result = dynamo.encode_pagination_shorthand(
        dynamo_last_evaluated_key, "identity_uuid", range_key="tenant_uuid"
    )
    assert result == "c5879365-2c07-4c19-a3f0-cc030c01b6a5#test-uuid-123456"


def test_decode_pagination_cursor(dynamo_last_evaluated_key: Dict[str, Any]) -> None:
    """Test decode pagination cursor."""
    encoded_key = "eyJpZGVudGl0eV91dWlkIjogeyJTIjogImM1ODc5MzY1LTJjMDctNGMxOS1hM2YwLWNjMDMwYzAxYjZhNSJ9LCAidGVuYW50X3V1aWQiOiB7IlMiOiAidGVzdC11dWlkLTEyMzQ1NiJ9fQ=="  # noqa: E501

    result = dynamo.decode_pagination_cursor(encoded_key)
    assert result == dynamo_last_evaluated_key


def test_deserialize_dynamo_item_array() -> None:
    """Test deserialize_dynamo_item_array."""
    dynamo_item_arr = [
        {
            "tenant_type": {"S": "account"},
            "tenant_uuid": {"S": "b87b9586-03dc-47be-a3b0-53ca84aa4145"},
            "roles": {"L": [{"M": {"role": {"S": "settings_admin"}}}]},
            "expires_at": {"N": "1733761755"},
        }
    ]
    result = dynamo.deserialize_dynamo_item_array(dynamo_item_arr)
    assert result == [
        {
            "tenant_type": "account",
            "tenant_uuid": "b87b9586-03dc-47be-a3b0-53ca84aa4145",
            "roles": [{"role": "settings_admin"}],
            "expires_at": 1733761755,
        }
    ]


def test_serialize_to_dynamo_item() -> None:
    """Test serialize_to_dynamo_item."""
    python_item = {
        "tenant_type": "account",
        "tenant_uuid": "b87b9586-03dc-47be-a3b0-53ca84aa4145",
        "roles": [{"role": "settings_admin"}],
        "expires_at": 1733761755,
    }
    result = dynamo.serialize_to_dynamo_item(python_item)
    assert result == {
        "M": {
            "tenant_type": {"S": "account"},
            "tenant_uuid": {"S": "b87b9586-03dc-47be-a3b0-53ca84aa4145"},
            "roles": {"L": [{"M": {"role": {"S": "settings_admin"}}}]},
            "expires_at": {"N": "1733761755"},
        }
    }


@pytest.fixture
def identity_tenant_dict() -> Dict[str, Any]:
    """Resuable dictionary representing an identity's tenant permissions."""
    return {
        "roles": [{"role": "settings_admin"}],
        "version": "1",
        "tenant_type": "account",
        "created_at": "time",
        "created_by": "cby",
        "updated_at": "time",
        "updated_by": "uby",
    }


@pytest.fixture
def identity_tenant_dynamo_item() -> Dict[str, Any]:
    """Reusable representation of a dynamo item.

    It is the equivalent of `identity_tenant_dict`
    """
    return {
        "created_at": {"S": "time"},
        "created_by": {"S": "cby"},
        "roles": {"L": [{"M": {"role": {"S": "settings_admin"}}}]},
        "tenant_type": {"S": "account"},
        "updated_at": {"S": "time"},
        "updated_by": {"S": "uby"},
        "version": {"S": "1"},
    }


def test_serialize_dict_to_dynamo_item_for_identity_tenant_dict(
    identity_tenant_dict: Dict[str, Any],
    identity_tenant_dynamo_item: Dict[str, Any],
) -> None:
    """Test serialize_dict_to_dynamo_item.

    Note the difference in output compared to the result in
    test_serialize_to_dynamo_item_for_identity_tenant_dict
    """
    result = dynamo.serialize_dict_to_dynamo_item(identity_tenant_dict)
    assert result == identity_tenant_dynamo_item


def test_serialize_to_dynamo_item_for_identity_tenant_dict(
    identity_tenant_dict: Dict[str, Any],
    identity_tenant_dynamo_item: Dict[str, Any],
) -> None:
    """Test serialize_dict_to_dynamo_item.

    Note the difference in output compared to the result in
    test_serialize_dict_to_dynamo_item_for_identity_tenant_dict
    """
    result = dynamo.serialize_to_dynamo_item(identity_tenant_dict)
    assert result == {
        "M": identity_tenant_dynamo_item,
    }


def test_deserialize_dynamo_item_to_dict(
    identity_tenant_dynamo_item: Dict[str, Any],
    identity_tenant_dict: Dict[str, Any],
) -> None:
    """Test deserialize_dynamo_item_to_dict.

    Note this is the reverse operation tested in
    test_serialize_dict_to_dynamo_item_for_identity_tenant_dict.
    """
    result = dynamo.deserialize_dynamo_item_to_dict(identity_tenant_dynamo_item)
    assert result == identity_tenant_dict


@pytest.mark.parametrize(
    "request_items, expected",
    [
        pytest.param({}, 0, id="empty dict has no items"),
        pytest.param({"table1": [], "table2": []}, 0, id="empty keys has no items"),
        pytest.param({"table1": ["item1"]}, 1, id="there is one item"),
        pytest.param({"table1": ["item1", "item2"]}, 2, id="there are two items"),
        pytest.param(
            {"table1": ["item1"], "table2": ["item2"]},
            2,
            id="there are two items across two table requests",
        ),
    ],
)
def test_count_batch_write_item_request_items(
    request_items: Dict[str, Any],
    expected: int,
) -> None:
    """Test count_batch_write_item_request_items."""
    actual = dynamo.count_batch_write_item_request_items(request_items)
    assert actual == expected


@pytest.mark.parametrize(
    "request_items",
    [
        pytest.param({}, id="request_items can't be empty"),
        pytest.param(
            {"table": [str(i) for i in range(26)]},
            id="request_items can't have more than 25 items",
        ),
        pytest.param(
            {
                "table1": [str(i) for i in range(20)],
                "table2": [str(i) for i in range(6)],
            },
            id="request_items can't have more than 25 items across tables",
        ),
    ],
)
async def test_batch_write_item_assertions(request_items: Dict[str, Any]) -> None:
    """Test batch_write_item validates range_keys size."""
    mock_dynamo_client = MagicMock()
    with pytest.raises(AssertionError):
        await dynamo.batch_write_item(mock_dynamo_client, request_items)


def request_items(val: Any) -> Dict[str, Any]:
    """Build and return mocked request_items."""
    return {
        "table1": [
            {"DeleteRequest": {"some_key": str(val)}},
        ]
    }


async def test_batch_write_item_finishes_right_away() -> None:
    """Test batch_write_item returns immediately if all items are processed."""

    mock_dynamo_client = MagicMock()
    mock_dynamo_client.batch_write_item.side_effect = [
        {
            "UnprocessedItems": {},
        },
    ]

    remaining_items = await dynamo.batch_write_item(
        mock_dynamo_client, request_items(65)
    )

    assert remaining_items == {}
    mock_dynamo_client.batch_write_item.assert_called_once_with(
        RequestItems=request_items(65)
    )


@patch("pdp.utils.dynamo.sleep")
@patch("pdp.utils.dynamo.backoff")
async def test_batch_write_item_attempts_until_no_unprocessed_items(
    mock_backoff: MagicMock,
    mock_sleep: AsyncMock,
) -> None:
    """Test batch_write_item attempts until unprocessed items is empty."""

    mock_dynamo_client = MagicMock()
    mock_dynamo_client.batch_write_item.side_effect = [
        {
            "UnprocessedItems": request_items(32),
        },
        {
            "UnprocessedItems": {},
        },
    ]
    mock_backoff.get_backoff_with_full_jitter.return_value = 297

    remaining_items = await dynamo.batch_write_item(
        mock_dynamo_client, request_items(1)
    )

    assert remaining_items == {}
    assert mock_dynamo_client.batch_write_item.call_count == 2
    mock_dynamo_client.batch_write_item.assert_has_calls(
        [
            call(RequestItems=request_items(1)),
            call(RequestItems=request_items(32)),
        ]
    )
    assert mock_backoff.get_backoff_with_full_jitter.call_count == 1
    mock_backoff.get_backoff_with_full_jitter.assert_called_once_with(1, 3, 1)
    assert mock_sleep.call_count == 1
    mock_sleep.assert_called_once_with(297)


@patch("pdp.utils.dynamo.sleep")
@patch("pdp.utils.dynamo.backoff")
@patch("pdp.utils.dynamo.config")
async def test_batch_write_item_attempts_max(
    mock_config: MagicMock,
    mock_backoff: MagicMock,
    mock_sleep: AsyncMock,
) -> None:
    """Test batch_write_item max attempts is reached."""

    mock_config.DYNAMODB_BATCH_UNPROCESSED_ITEMS_MAX_ATTEMPTS = 5

    mock_dynamo_client = MagicMock()
    mock_dynamo_client.batch_write_item.side_effect = [
        {"UnprocessedItems": request_items(1)},
        {"UnprocessedItems": request_items(2)},
        {"UnprocessedItems": request_items(3)},
        {"UnprocessedItems": request_items(4)},
        {"UnprocessedItems": request_items(5)},
        {"UnprocessedItems": request_items(6)},
    ]
    mock_backoff.get_backoff_with_full_jitter.side_effect = [0.1, 0.2, 0.3, 0.4, 0.5]

    unprocessed_items = await dynamo.batch_write_item(
        mock_dynamo_client, request_items(99)
    )

    assert unprocessed_items == request_items(6)
    assert mock_dynamo_client.batch_write_item.call_count == 6
    mock_dynamo_client.batch_write_item.assert_has_calls(
        [
            # This is the first call
            call(RequestItems=request_items(99)),
            # These are the subsequent attempts
            call(RequestItems=request_items(1)),
            call(RequestItems=request_items(2)),
            call(RequestItems=request_items(3)),
            call(RequestItems=request_items(4)),
            call(RequestItems=request_items(5)),
        ]
    )
    assert mock_backoff.get_backoff_with_full_jitter.call_count == 5
    mock_backoff.get_backoff_with_full_jitter.assert_has_calls(
        [
            call(1, 3, 1),
            call(1, 3, 2),
            call(1, 3, 3),
            call(1, 3, 4),
            call(1, 3, 5),
        ]
    )
    assert mock_sleep.call_count == 5
    mock_sleep.assert_has_calls(
        [
            call(0.1),
            call(0.2),
            call(0.3),
            call(0.4),
            call(0.5),
        ]
    )


def test_get_batch_write_item_keys() -> None:
    """Test get_batch_write_item_keys."""
    delete_keys = dynamo.get_batch_write_item_keys(is_delete=True)
    assert delete_keys == dynamo.DynamoDbBatchWriteItemKey(
        request_type="DeleteRequest",
        subelement="Key",
    )

    put_keys = dynamo.get_batch_write_item_keys(is_delete=False)
    assert put_keys == dynamo.DynamoDbBatchWriteItemKey(
        request_type="PutRequest",
        subelement="Item",
    )