# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
import pytest

from devtools_testutils import AzureRecordedTestCase
from devtools_testutils.aio import recorded_by_proxy_async

from azure.core.exceptions import ResourceExistsError, HttpResponseError
from azure.data.tables.aio import TableServiceClient, TableClient

from _shared.asynctestcase import AsyncTableTestCase
from async_preparers import cosmos_decorator_async


class TestTableCosmosAsync(AzureRecordedTestCase, AsyncTableTestCase):
    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_create_table(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table_name = self._get_table_reference()

        # Act
        table = ts.get_table_client(table_name)
        created = await table.create_table()

        # Assert
        assert created.name == table_name

        await ts.delete_table(table_name=table_name)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_create_table_fail_on_exist(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table_name = self._get_table_reference()

        # Act
        created = await ts.create_table(table_name=table_name)
        with pytest.raises(ResourceExistsError):
            await ts.create_table(table_name=table_name)

        # Assert
        assert created
        await ts.delete_table(table_name=table_name)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_query_tables_per_page(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )

        table_name = "myasynctable"

        for i in range(5):
            await ts.create_table(table_name + str(i))

        query_filter = "TableName eq 'myasynctable0' or TableName eq 'myasynctable1' or TableName eq 'myasynctable2'"
        table_count = 0
        page_count = 0
        async for table_page in ts.query_tables(query_filter, results_per_page=2).by_page():

            temp_count = 0
            async for table in table_page:
                temp_count += 1
            assert temp_count <= 2
            page_count += 1
            table_count += temp_count

        assert page_count == 2
        assert table_count == 3

        await self._delete_all_tables(tables_cosmos_account_name, tables_primary_cosmos_account_key)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_list_tables(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table = await self._create_table(ts)

        # Act
        tables = []
        async for t in ts.list_tables():
            tables.append(t)

        # Assert
        assert tables is not None
        assert len(tables) >= 1
        assert tables[0] is not None

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_query_tables_with_filter(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table = await self._create_table(ts)

        # Act
        name_filter = "TableName eq '{}'".format(table.table_name)
        tables = []
        async for t in ts.query_tables(name_filter):
            tables.append(t)

        # Assert
        assert tables is not None
        assert len(tables) == 1
        await ts.delete_table(table.table_name)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_list_tables_with_num_results(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._delete_all_tables(tables_cosmos_account_name, tables_primary_cosmos_account_key)
        prefix = "listtable"
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table_list = []
        for i in range(0, 4):
            await self._create_table(ts, prefix + str(i), table_list)

        # Act
        all_tables = 0
        async for t in ts.list_tables():
            all_tables += 1

        small_page = 0
        async for page in ts.list_tables(results_per_page=3).by_page():
            page_size = 0
            async for table in page:
                page_size += 1
            assert page_size <= 3
            small_page += 1

        assert small_page == 2
        assert all_tables == 4

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_list_tables_with_marker(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        prefix = "listtable"
        table_names = []
        for i in range(0, 4):
            await self._create_table(ts, prefix + str(i), table_names)

        # Act
        generator1 = ts.list_tables(results_per_page=2).by_page()
        await generator1.__anext__()

        generator2 = ts.list_tables(results_per_page=2).by_page(continuation_token=generator1.continuation_token)
        await generator2.__anext__()

        tables1 = generator1._current_page
        tables2 = generator2._current_page

        tables1_len = 0
        async for _ in tables1:
            tables1_len += 1
        tables2_len = 0
        async for _ in tables2:
            tables2_len += 1

        # Assert
        assert tables1_len == 2
        assert tables2_len == 2
        assert tables1 != tables2

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_delete_table_with_existing_table(
        self, tables_cosmos_account_name, tables_primary_cosmos_account_key
    ):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table = await self._create_table(ts)

        # Act
        deleted = await ts.delete_table(table_name=table.table_name)

        # Assert
        assert deleted is None

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_delete_table_with_non_existing_table_fail_not_exist(
        self, tables_cosmos_account_name, tables_primary_cosmos_account_key
    ):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table_name = self._get_table_reference()
        await ts.delete_table(table_name)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_create_table_underscore_name(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table_name = "my_table"

        client = await ts.create_table(table_name)
        assert client.table_name == table_name

        await ts.delete_table(table_name)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_create_table_unicode_name(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        ts = TableServiceClient(
            self.account_url(tables_cosmos_account_name, "cosmos"), credential=tables_primary_cosmos_account_key
        )
        table_name = "啊齄丂狛狜"

        client = await ts.create_table(table_name)
        assert client.table_name == table_name

        await ts.delete_table(table_name)

    @cosmos_decorator_async
    @recorded_by_proxy_async
    async def test_client_with_url_ends_with_table_name(
        self, tables_cosmos_account_name, tables_primary_cosmos_account_key
    ):
        url = self.account_url(tables_cosmos_account_name, "cosmos")
        table_name = self.get_resource_name("mytable")
        invalid_url = url + "/" + table_name
        # test table client has the same table name as in url
        tc = TableClient(invalid_url, table_name, credential=tables_primary_cosmos_account_key)
        with pytest.raises(HttpResponseError) as exc:
            await tc.create_table()
        assert ("Request url is invalid") in str(exc.value)
        assert ("Please check your account URL.") in str(exc.value)
        # test table client has a different table name as in url
        table_name2 = self.get_resource_name("mytable2")
        tc2 = TableClient(invalid_url, table_name2, credential=tables_primary_cosmos_account_key)
        with pytest.raises(HttpResponseError) as exc:
            await tc2.create_table()
        assert ("Request url is invalid") in str(exc.value)
        assert ("Please check your account URL.") in str(exc.value)

        valid_tc = TableClient(url, table_name, credential=tables_primary_cosmos_account_key)
        await valid_tc.create_table()
        # test creating a table when it already exists
        with pytest.raises(HttpResponseError) as exc:
            await tc.create_table()
        assert ("Request url is invalid") in str(exc.value)
        assert ("Please check your account URL.") in str(exc.value)
        # test deleting a table when it already exists
        with pytest.raises(HttpResponseError) as exc:
            await tc.delete_table()
        assert ("Request url is invalid") in str(exc.value)
        assert ("Please check your account URL.") in str(exc.value)
        await valid_tc.delete_table()
        await valid_tc.close()
        await tc.close()
        await tc2.close()