# coding: utf-8
# -------------------------------------------------------------------------
# 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 datetime import datetime, timedelta
from dateutil.tz import tzutc, tzoffset
from math import isnan

from devtools_testutils import AzureTestCase

from azure.data.tables import (
    TableEntity,
    EntityProperty,
    EdmType,
    UpdateMode,
    generate_table_sas,
    TableSasPermissions,
    TableAccessPolicy,
)
from azure.data.tables.aio import TableServiceClient
from azure.core import MatchConditions
from azure.core.credentials import AzureSasCredential
from azure.core.exceptions import (
    HttpResponseError,
    ResourceNotFoundError,
    ResourceExistsError,
    ResourceModifiedError
)

from _shared.asynctestcase import AsyncTableTestCase
from async_preparers import cosmos_decorator_async
# ------------------------------------------------------------------------------
# TODO: change to `with table_client as client:` to close sessions
# ------------------------------------------------------------------------------

class StorageTableEntityTest(AzureTestCase, AsyncTableTestCase):
    @cosmos_decorator_async
    async def test_url_encoding_at_symbol(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):

        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = {
                u"PartitionKey": u"PK",
                u"RowKey": u"table@storage.com",
                u"Value": 100
            }

            await self.table.create_entity(entity)

            f = u"RowKey eq '{}'".format(entity["RowKey"])
            entities = self.table.query_entities(f)

            count = 0
            async for e in entities:
                assert e['PartitionKey'] == entity[u"PartitionKey"]
                assert e['RowKey'] == entity[u"RowKey"]
                assert e['Value'] == entity[u"Value"]
                await self.table.delete_entity(e['PartitionKey'], e['RowKey'])
                count += 1

            assert count == 1

            entities = self.table.query_entities(f)
            count = 0
            async for e in entities:
                count += 1
            assert count == 0

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_dictionary(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_entity_dict()

            # Act
            resp = await self.table.create_entity(entity=entity)

            # Assert
            assert resp is not None
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_with_hook(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_entity_dict()

            # Act
            resp = await self.table.create_entity(entity=entity)
            received_entity = await self.table.get_entity(
                partition_key=entity["PartitionKey"],
                row_key=entity["RowKey"]
            )
            # Assert
            self._assert_valid_metadata(resp)
            self._assert_default_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_with_no_metadata(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_entity_dict()
            headers = {'Accept': 'application/json;odata=nometadata'}

            # Act
            resp = await self.table.create_entity(
                entity=entity,
                headers={'Accept': 'application/json;odata=nometadata'},
            )
            received_entity = await self.table.get_entity(
                partition_key=entity["PartitionKey"],
                row_key=entity["RowKey"],
                headers=headers
            )

            # Assert
            self._assert_valid_metadata(resp)
            self._assert_default_entity_json_no_metadata(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_with_full_metadata(self, tables_cosmos_account_name,
                                                    tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_entity_dict()
            headers = {'Accept': 'application/json;odata=fullmetadata'}

            # Act
            resp = await self.table.create_entity(
                entity=entity,
                headers=headers
            )
            received_entity = await self.table.get_entity(
                partition_key=entity["PartitionKey"],
                row_key=entity["RowKey"],
                headers=headers
            )

            # Assert
            self._assert_valid_metadata(resp)
            self._assert_default_entity_json_full_metadata(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_conflict(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            with pytest.raises(ResourceExistsError):
                await self.table.create_entity(entity=entity)

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_with_large_int32_value_throws(self, tables_cosmos_account_name,
                                                               tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Act
            dict32 = self._create_random_base_entity_dict()
            dict32['large'] = EntityProperty(2 ** 31, EdmType.INT32) # TODO: this is outside the range of int32

            # Assert
            with pytest.raises(TypeError):
                await self.table.create_entity(entity=dict32)

            dict32['large'] = EntityProperty(-(2 ** 31 + 1), EdmType.INT32)  # TODO: this is outside the range of int32
            with pytest.raises(TypeError):
                await self.table.create_entity(entity=dict32)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_with_large_int64_value_throws(self, tables_cosmos_account_name,
                                                               tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Act
            dict64 = self._create_random_base_entity_dict()
            dict64['large'] = EntityProperty(2 ** 63, EdmType.INT64)

            # Assert
            with pytest.raises(TypeError):
                await self.table.create_entity(entity=dict64)

            dict64['large'] = EntityProperty(-(2 ** 63 + 1), EdmType.INT64)
            with pytest.raises(TypeError):
                await self.table.create_entity(entity=dict64)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_with_large_int_success(self, tables_cosmos_account_name,
                                                         tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Act
            dict64 = self._create_random_base_entity_dict()
            dict64['large'] = EntityProperty(2 ** 50, EdmType.INT64)

            # Assert
            await self.table.create_entity(entity=dict64)

            received_entity = await self.table.get_entity(dict64['PartitionKey'], dict64['RowKey'])
            assert received_entity['large'].value == dict64['large'].value

            dict64['RowKey'] = 'negative'
            dict64['large'] = EntityProperty(-(2 ** 50 + 1), EdmType.INT64)
            await self.table.create_entity(entity=dict64)

            received_entity = await self.table.get_entity(dict64['PartitionKey'], dict64['RowKey'])
            assert received_entity['large'].value == dict64['large'].value

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_missing_pk(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = {'RowKey': 'rk'}

            # Act
            with pytest.raises(ValueError) as error:
                resp = await self.table.create_entity(entity=entity)
                assert str(error).contains("PartitionKey must be present in an entity")
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_empty_string_pk(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = {'RowKey': 'rk', 'PartitionKey': ''}

            # Act
            with pytest.raises(HttpResponseError):
                await self.table.create_entity(entity=entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_missing_rk(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = {'PartitionKey': 'pk'}

            # Act
            with pytest.raises(ValueError) as error:
                resp = await self.table.create_entity(entity=entity)
                assert str(error).contains("PartitionKey must be present in an entity")
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_entity_empty_string_rk(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = {'PartitionKey': 'pk', 'RowKey': ''}

            # Act
            with pytest.raises(HttpResponseError):
                await self.table.create_entity(entity=entity)

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            resp = await self.table.get_entity(partition_key=entity['PartitionKey'],
                                               row_key=entity['RowKey'])

            # Assert
            assert resp['PartitionKey'] ==  entity['PartitionKey']
            assert resp['RowKey'] ==  entity['RowKey']
            self._assert_default_entity(resp)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_with_select(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            resp = await self.table.get_entity(partition_key=entity['PartitionKey'],
                                               row_key=entity['RowKey'],
                                               select=['age', 'ratio'])
            resp.pop('_metadata', None)
            assert resp == {'age': 39, 'ratio': 3.1}
            resp = await self.table.get_entity(partition_key=entity['PartitionKey'],
                                               row_key=entity['RowKey'],
                                               select='age,ratio')
            resp.pop('_metadata', None)
            assert resp == {'age': 39, 'ratio': 3.1}

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_with_hook(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            resp = await self.table.get_entity(
                partition_key=entity['PartitionKey'],
                row_key=entity['RowKey'],
            )

            # Assert
            assert resp['PartitionKey'] ==  entity['PartitionKey']
            assert resp['RowKey'] ==  entity['RowKey']
            self._assert_default_entity(resp)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_if_match(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, etag = await self._insert_random_entity()

            entity = await self.table.get_entity(
                partition_key=entity['PartitionKey'],
                row_key=entity['RowKey']
            )

            await self.table.delete_entity(
                entity,
                etag=etag,
                match_condition=MatchConditions.IfNotModified
            )

            with pytest.raises(ResourceNotFoundError):
                await self.table.get_entity(
                    partition_key=entity['PartitionKey'],
                    row_key=entity['RowKey']
                )
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_if_match_entity_bad_etag(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, old_etag = await self._insert_random_entity()

            entity["value"] = 10
            await self.table.update_entity(entity)

            # Get Entity and set old etag
            e = await self.table.get_entity(entity["PartitionKey"], entity["RowKey"])
            new_etag = e.metadata["etag"]
            e.metadata["etag"] = old_etag

            with pytest.raises(ResourceModifiedError):
                await self.table.delete_entity(e, match_condition=MatchConditions.IfNotModified)

            # Try delete with correct etag
            await self.table.delete_entity(e, etag=new_etag, match_condition=MatchConditions.IfNotModified)

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity_if_match_table_entity(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, etag = await self._insert_random_entity()
            table_entity = TableEntity(**entity)

            entity = await self.table.get_entity(
                partition_key=entity['PartitionKey'],
                row_key=entity['RowKey']
            )

            with pytest.raises(ValueError):
                await self.table.delete_entity(table_entity, match_condition=MatchConditions.IfNotModified)

            await self.table.delete_entity(table_entity, etag=etag, match_condition=MatchConditions.IfNotModified)

            with pytest.raises(ResourceNotFoundError):
                await self.table.get_entity(entity["PartitionKey"], entity["RowKey"])

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_full_metadata(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            resp = await self.table.get_entity(
                entity['PartitionKey'],
                entity['RowKey'],
                headers={'accept': 'application/json;odata=fullmetadata'})

            # Assert
            assert resp['PartitionKey'] ==  entity['PartitionKey']
            assert resp['RowKey'] ==  entity['RowKey']
            self._assert_default_entity_json_full_metadata(resp)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_no_metadata(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            resp = await self.table.get_entity(
                partition_key=entity['PartitionKey'],
                row_key=entity['RowKey'],
                headers={'accept': 'application/json;odata=nometadata'})

            # Assert
            assert resp['PartitionKey'] ==  entity['PartitionKey']
            assert resp['RowKey'] ==  entity['RowKey']
            self._assert_default_entity_json_no_metadata(resp)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_not_existing(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_entity_dict()

            # Act
            with pytest.raises(ResourceNotFoundError):
                await self.table.get_entity(partition_key=entity['PartitionKey'],
                                            row_key=entity['RowKey'])

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_get_entity_with_special_doubles(self, tables_cosmos_account_name,
                                                   tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()
            entity.update({
                'inf': float('inf'),
                'negativeinf': float('-inf'),
                'nan': float('nan')
            })
            await self.table.create_entity(entity=entity)

            # Act
            resp = await self.table.get_entity(partition_key=entity['PartitionKey'],
                                               row_key=entity['RowKey'])

            # Assert
            assert resp['inf'] ==  float('inf')
            assert resp['negativeinf'] ==  float('-inf')
            assert isnan(resp['nan'])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_update_entity(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])

            resp = await self.table.update_entity(mode=UpdateMode.REPLACE, entity=sent_entity)

            # Assert
            #  assert resp
            received_entity = await self.table.get_entity(
                partition_key=entity['PartitionKey'],
                row_key=entity['RowKey'])

            self._assert_updated_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_update_entity_not_existing(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            with pytest.raises(ResourceNotFoundError):
                await self.table.update_entity(mode=UpdateMode.REPLACE, entity=sent_entity)

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_update_entity_with_if_matches(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, etag = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            await self.table.update_entity(
                mode=UpdateMode.REPLACE,
                entity=sent_entity, etag=etag,
                match_condition=MatchConditions.IfNotModified)

            # Assert
            # assert resp
            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])
            self._assert_updated_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_update_entity_with_if_doesnt_match(self, tables_cosmos_account_name,
                                                      tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            with pytest.raises(ResourceModifiedError):
                await self.table.update_entity(
                    mode=UpdateMode.REPLACE,
                    entity=sent_entity,
                    etag=u'W/"datetime\'2012-06-15T22%3A51%3A44.9662825Z\'"',
                    match_condition=MatchConditions.IfNotModified)

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_or_merge_entity_with_existing_entity(self, tables_cosmos_account_name,
                                                               tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            await self.table.upsert_entity(mode=UpdateMode.MERGE, entity=sent_entity)

            # Assert
            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])
            self._assert_merged_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_or_merge_entity_with_non_existing_entity(self, tables_cosmos_account_name,
                                                                   tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            await self.table.upsert_entity(mode=UpdateMode.MERGE, entity=sent_entity)

            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])
            self._assert_updated_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_or_replace_entity_with_existing_entity(self, tables_cosmos_account_name,
                                                                 tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            resp = await self.table.upsert_entity(mode=UpdateMode.REPLACE, entity=sent_entity)

            # Assert
            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])
            self._assert_updated_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_insert_or_replace_entity_with_non_existing_entity(self, tables_cosmos_account_name,
                                                                     tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            await self.table.upsert_entity(mode=UpdateMode.REPLACE, entity=sent_entity)

            received_entity = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])
            self._assert_updated_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_merge_entity(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            await self.table.update_entity(mode=UpdateMode.MERGE, entity=sent_entity)

            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])
            self._assert_merged_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_merge_entity_not_existing(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            with pytest.raises(ResourceNotFoundError):
                await self.table.update_entity(mode=UpdateMode.MERGE, entity=sent_entity)

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_merge_entity_with_if_matches(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, etag = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            resp = await self.table.update_entity(mode=UpdateMode.MERGE,
                                                  entity=sent_entity, etag=etag,
                                                  match_condition=MatchConditions.IfNotModified)

            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])
            self._assert_merged_entity(received_entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_merge_entity_with_if_doesnt_match(self, tables_cosmos_account_name,
                                                     tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            with pytest.raises(ResourceModifiedError):
                await self.table.update_entity(mode=UpdateMode.MERGE,
                                               entity=sent_entity,
                                               etag='W/"datetime\'2012-06-15T22%3A51%3A44.9662825Z\'"',
                                               match_condition=MatchConditions.IfNotModified)

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            await self.table.delete_entity(partition_key=entity['PartitionKey'], row_key=entity['RowKey'])

            with pytest.raises(ResourceNotFoundError):
                await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity_not_existing(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()
            await self.table.delete_entity(entity['PartitionKey'], entity['RowKey'])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity_with_if_matches(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, etag = await self._insert_random_entity()

            await self.table.delete_entity(
                entity['PartitionKey'],
                entity['RowKey'],
                etag=etag,
                match_condition=MatchConditions.IfNotModified
            )

            with pytest.raises(ResourceNotFoundError):
                await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity_with_if_doesnt_match(self, tables_cosmos_account_name,
                                                      tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Act
            with pytest.raises(ResourceModifiedError):
                await self.table.delete_entity(
                    entity['PartitionKey'],
                    entity['RowKey'],
                    etag=u'W/"datetime\'2012-06-15T22%3A51%3A44.9662825Z\'"',
                    match_condition=MatchConditions.IfNotModified
                )
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity_overloads(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Test passing in an entity
            await self.table.delete_entity(entity)

            pk, rk = self._create_pk_rk("pk", "rk")
            pk, rk = pk + u"2", rk + u"2"
            entity2 = {
                u"PartitionKey": pk,
                u"RowKey": rk,
                u"Value": 100
            }
            await self.table.create_entity(entity2)

            # Test passing in a partition key and row key
            await self.table.delete_entity(pk, rk)

            count = 0
            async for entity in self.table.list_entities():
                count += 1
            assert count == 0
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_delete_entity_overloads_kwargs(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()

            # Test passing in an entity
            await self.table.delete_entity(entity=entity)

            pk, rk = self._create_pk_rk("pk", "rk")
            pk, rk = pk + u"2", rk + u"2"
            entity2 = {
                u"PartitionKey": pk,
                u"RowKey": rk,
                u"Value": 100
            }
            await self.table.create_entity(entity2)

            # Test passing in a partition key and row key
            await self.table.delete_entity(partition_key=pk, row_key=rk)

            count = 0
            async for entity in self.table.list_entities():
                count += 1
            assert count == 0
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_unicode_property_value(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        ''' regression test for github issue #57'''
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()
            entity1 = entity.copy()
            entity1.update({'Description': u'ꀕ'})
            entity2 = entity.copy()
            entity2.update({'RowKey': 'test2', 'Description': 'ꀕ'})

            # Act
            await self.table.create_entity(entity=entity1)
            await self.table.create_entity(entity=entity2)
            entities = []
            async for e in self.table.query_entities(
                    "PartitionKey eq '{}'".format(entity['PartitionKey'])):
                entities.append(e)

            # Assert
            assert len(entities) ==  2
            assert entities[0]['Description'] ==  u'ꀕ'
            assert entities[1]['Description'] ==  u'ꀕ'
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_unicode_property_name(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()
            entity1 = entity.copy()
            entity1.update({u'啊齄丂狛狜': u'ꀕ'})
            entity2 = entity.copy()
            entity2.update({'RowKey': 'test2', u'啊齄丂狛狜': 'hello'})

            # Act
            await self.table.create_entity(entity=entity1)
            await self.table.create_entity(entity=entity2)
            entities = []
            async for e in self.table.query_entities(
                    "PartitionKey eq '{}'".format(entity['PartitionKey'])):
                entities.append(e)

            # Assert
            assert len(entities) ==  2
            assert entities[0][u'啊齄丂狛狜'] ==  u'ꀕ'
            assert entities[1][u'啊齄丂狛狜'] ==  u'hello'
        finally:
            await self._tear_down()

    @pytest.mark.skip("Bad Request: Cosmos cannot handle single quotes in a PK/RK (confirm)")
    @cosmos_decorator_async
    async def test_operations_on_entity_with_partition_key_having_single_quote(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        partition_key_with_single_quote = "a''''b"
        row_key_with_single_quote = "a''''b"
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity(pk=partition_key_with_single_quote, rk=row_key_with_single_quote)

            sent_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            await self.table.upsert_entity(mode=UpdateMode.REPLACE, entity=sent_entity)

            # row key here only has 2 quotes
            received_entity = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])
            self._assert_updated_entity(received_entity)

            sent_entity['newField'] = 'newFieldValue'
            await self.table.update_entity(mode=UpdateMode.REPLACE, entity=sent_entity)

            received_entity = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])
            self._assert_updated_entity(received_entity)
            assert received_entity['newField'] ==  'newFieldValue'

            await self.table.delete_entity(entity['PartitionKey'], entity['RowKey'])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_empty_and_spaces_property_value(self, tables_cosmos_account_name,
                                                   tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()
            entity.update({
                'EmptyByte': b'',
                'EmptyUnicode': u'',
                'SpacesOnlyByte': b'   ',
                'SpacesOnlyUnicode': u'   ',
                'SpacesBeforeByte': b'   Text',
                'SpacesBeforeUnicode': u'   Text',
                'SpacesAfterByte': b'Text   ',
                'SpacesAfterUnicode': u'Text   ',
                'SpacesBeforeAndAfterByte': b'   Text   ',
                'SpacesBeforeAndAfterUnicode': u'   Text   ',
            })

            # Act
            await self.table.create_entity(entity=entity)
            resp = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])

            # Assert
            assert resp is not None
            assert resp['EmptyByte'] ==  b''
            assert resp['EmptyUnicode'] ==  u''
            assert resp['SpacesOnlyByte'] ==  b'   '
            assert resp['SpacesOnlyUnicode'] ==  u'   '
            assert resp['SpacesBeforeByte'] ==  b'   Text'
            assert resp['SpacesBeforeUnicode'] ==  u'   Text'
            assert resp['SpacesAfterByte'] ==  b'Text   '
            assert resp['SpacesAfterUnicode'] ==  u'Text   '
            assert resp['SpacesBeforeAndAfterByte'] ==  b'   Text   '
            assert resp['SpacesBeforeAndAfterUnicode'] ==  u'   Text   '
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_none_property_value(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_base_entity_dict()
            entity.update({'NoneValue': None})

            # Act
            await self.table.create_entity(entity=entity)
            resp = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])

            # Assert
            assert resp is not None
            assert not hasattr(resp, 'NoneValue')
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_binary_property_value(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            binary_data = b'\x01\x02\x03\x04\x05\x06\x07\x08\t\n'
            entity = self._create_random_base_entity_dict()
            entity.update({'binary': b'\x01\x02\x03\x04\x05\x06\x07\x08\t\n'})

            # Act
            await self.table.create_entity(entity=entity)
            resp = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])

            # Assert
            assert resp is not None
            assert resp['binary'] ==  binary_data
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_timezone(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            local_tz = tzoffset('BRST', -10800)
            local_date = datetime(2003, 9, 27, 9, 52, 43, tzinfo=local_tz)
            entity = self._create_random_base_entity_dict()
            entity.update({'date': local_date})

            # Act
            await self.table.create_entity(entity=entity)
            resp = await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])

            # Assert
            assert resp is not None
            # times are not equal because request is made after
            assert resp['date'].astimezone(tzutc()) ==  local_date.astimezone(tzutc())
            assert resp['date'].astimezone(local_tz) ==  local_date
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            async with await self._create_query_table(2) as table:

                # Act
                entities = []
                async for t in table.list_entities():
                    entities.append(t)

            # Assert
            assert len(entities) ==  2
            for entity in entities:
                self._assert_default_entity(entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_each_page(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            base_entity = {
                "PartitionKey": u"pk",
                "RowKey": u"1",
            }

            for i in range(10):
                if i > 5:
                    base_entity['PartitionKey'] += str(i)
                base_entity['RowKey'] += str(i)
                base_entity['value'] = i
                try:
                    await self.table.create_entity(base_entity)
                except ResourceExistsError:
                    pass

            query_filter = u"PartitionKey eq 'pk'"

            entity_count = 0
            page_count = 0
            async for entity_page in self.table.query_entities(query_filter, results_per_page=2).by_page():

                temp_count = 0
                async for ent in entity_page:
                    temp_count += 1

                assert temp_count <= 2
                page_count += 1
                entity_count += temp_count

            assert entity_count == 6
            assert page_count == 3

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = await self._insert_two_opposite_entities()

            # Act
            entities = self.table.query_entities("married eq @my_param", parameters={'my_param': True})

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_multiple_params(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()

            # Act
            parameters = {
                'my_param': True,
                'rk': entity['RowKey']
            }
            entities = self.table.query_entities("married eq @my_param and RowKey eq @rk", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_integers(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()

            # Act
            parameters = {
                'my_param': 40,
            }
            entities = self.table.query_entities("age lt @my_param", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_floats(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()

            # Act
            parameters = {
                'my_param': entity['ratio'] + 1.0,
            }
            entities = self.table.query_entities("ratio lt @my_param", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_datetimes(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()

            # Act
            parameters = {
                'my_param': entity['birthday'],
            }
            entities = self.table.query_entities("birthday eq @my_param", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_guids(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()

            # Act
            parameters = {
                'my_param': entity['clsid']
            }
            entities = self.table.query_entities("clsid eq @my_param", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_binary(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()

            # Act
            parameters = {
                'my_param': entity['binary']
            }
            entities = self.table.query_entities("binary eq @my_param", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                self._assert_default_entity(entity)
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_user_filter_int64(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_two_opposite_entities()
            large_entity = {
                u"PartitionKey": u"pk001",
                u"RowKey": u"rk001",
                u"large_int": EntityProperty(2 ** 40, EdmType.INT64),
            }
            await self.table.create_entity(large_entity)

            # Act
            parameters = {
                'my_param': large_entity['large_int'].value
            }
            entities = self.table.query_entities("large_int eq @my_param", parameters=parameters)

            length = 0
            assert entities is not None
            async for entity in entities:
                # self._assert_default_entity(entity)
                assert large_entity['large_int'] == entity['large_int']
                length += 1

            assert length == 1
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_zero_entities(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            async with await self._create_query_table(0) as table:

                # Act
                entities = []
                async for t in table.list_entities():
                    entities.append(t)

            # Assert
            assert len(entities) ==  0
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_full_metadata(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            async with await self._create_query_table(2) as table:

                # Act
                entities = []
                async for t in table.list_entities(headers={'accept': 'application/json;odata=fullmetadata'}):
                    entities.append(t)

            # Assert
            assert len(entities) ==  2
            for entity in entities:
                self._assert_default_entity_json_full_metadata(entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_no_metadata(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            async with await self._create_query_table(2) as table:

                # Act
                entities = []
                async for t in table.list_entities(headers={'accept': 'application/json;odata=nometadata'}):
                    entities.append(t)

            # Assert
            assert len(entities) ==  2
            for entity in entities:
                self._assert_default_entity_json_no_metadata(entity)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_with_filter(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity, _ = await self._insert_random_entity()
            entity2, _ = await self._insert_random_entity(pk="foo" + entity['PartitionKey'])
            entity3, _ = await self._insert_random_entity(pk="bar" + entity['PartitionKey'])

            # Act
            entities = []
            async for t in self.table.query_entities(
                    "PartitionKey eq '{}'".format(entity['PartitionKey'])):
                entities.append(t)

            # Assert
            assert len(entities) ==  1
            assert entity['PartitionKey'] ==  entities[0]['PartitionKey']
            self._assert_default_entity(entities[0])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_injection_async(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            table_name = self.get_resource_name('querytable')
            table = await self.ts.create_table_if_not_exists(table_name)
            entity_a = {'PartitionKey': 'foo', 'RowKey': 'bar1', 'IsAdmin': 'admin'}
            entity_b = {'PartitionKey': 'foo', 'RowKey': 'bar2', 'IsAdmin': ''}
            await table.create_entity(entity_a)
            await table.create_entity(entity_b)

            is_user_admin = "PartitionKey eq @first and IsAdmin eq 'admin'"
            entity_query = table.query_entities(is_user_admin, parameters={'first': 'foo'})
            entities = []
            async for e in entity_query:
                entities.append(e)
            assert len(entities) ==  1

            injection = "foo' or RowKey eq 'bar2"
            injected_query = "PartitionKey eq '{}' and IsAdmin eq 'admin'".format(injection)
            entity_query = table.query_entities(injected_query)
            entities = []
            async for e in entity_query:
                entities.append(e)
            assert len(entities) ==  2

            entity_query = table.query_entities(is_user_admin, parameters={'first': injection})
            entities = []
            async for e in entity_query:
                entities.append(e)
            assert len(entities) ==  0
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_special_chars(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            table_name = self.get_resource_name('querytable')
            table = await self.ts.create_table_if_not_exists(table_name)
            entity_a = {'PartitionKey': u'foo', 'RowKey': u'bar1', 'Chars': u":@?'/!_^#+,$"}
            entity_b = {'PartitionKey': u'foo', 'RowKey': u'bar2', 'Chars': u'=& ?"\\{}<>%'}
            await table.create_entity(entity_a)
            await table.create_entity(entity_b)

            entities = []
            all_entities = table.query_entities("PartitionKey eq 'foo'")
            async for e in all_entities:
                entities.append(e)
            assert len(entities) == 2

            entities = []
            parameters = {'key': 'foo'}
            all_entities = table.query_entities("PartitionKey eq @key", parameters=parameters)
            async for e in all_entities:
                entities.append(e)
            assert len(entities) == 2

            entities = []
            query = "PartitionKey eq 'foo' and RowKey eq 'bar1' and Chars eq ':@?''/!_^#+,$'"
            query_entities = table.query_entities(query)
            async for e in query_entities:
                entities.append(e)
            assert len(entities) == 1

            entities = []
            query = "PartitionKey eq @key and RowKey eq @row and Chars eq @quote"
            parameters = {'key': 'foo', 'row': 'bar1', 'quote': ":@?'/!_^#+,$"}
            query_entities = table.query_entities(query, parameters=parameters)
            async for e in query_entities:
                entities.append(e)
            assert len(entities) == 1

            entities = []
            query = "PartitionKey eq 'foo' and RowKey eq 'bar2' and Chars eq '=& ?\"\\{}<>%'"
            query_entities = table.query_entities(query)
            async for e in query_entities:
                entities.append(e)
            assert len(entities) == 1

            entities = []
            query = "PartitionKey eq @key and RowKey eq @row and Chars eq @quote"
            parameters = {'key': 'foo', 'row': 'bar2', 'quote': r'=& ?"\{}<>%'}
            query_entities = table.query_entities(query, parameters=parameters)
            async for e in query_entities:
                entities.append(e)
            assert len(entities) == 1

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_invalid_filter(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            base_entity = {
                u"PartitionKey": u"pk",
                u"RowKey": u"rk",
                u"value": 1
            }

            for i in range(5):
                base_entity[u"RowKey"] += str(i)
                base_entity[u"value"] += i
                await self.table.create_entity(base_entity)
            # Act
            with pytest.raises(HttpResponseError):
                async for t in self.table.query_entities("aaa bbb ccc"):
                    _ = t
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_with_select(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            table = await self._create_query_table(2)

            # Act
            entities = []
            async for entity in table.list_entities(select=['age', 'sex']):
                entities.append(entity)
                assert entities[0]['age'] ==  39
                assert entities[0]['sex'] ==  'male'
                assert not "birthday" in entities[0]
                assert not "married" in entities[0]
                assert not "deceased" in entities[0]

            # Assert
            assert len(entities) == 2
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_with_top(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            async with await self._create_query_table(3) as table:

                entities = []
                async for t in table.list_entities(results_per_page=2).by_page():
                    entities.append(t)
            # Assert
            assert len(entities) ==  2
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_query_entities_with_top_and_next(self, tables_cosmos_account_name,
                                                    tables_primary_cosmos_account_key):
        # Arrange
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            async with await self._create_query_table(5) as table:

                # Act
                resp1 = table.list_entities(results_per_page=2).by_page()
                entities1 = []
                async for el in await resp1.__anext__():
                    entities1.append(el)
                resp2 = table.list_entities(results_per_page=2).by_page(
                    continuation_token=resp1.continuation_token)
                entities2 = []
                async for el in await resp2.__anext__():
                    entities2.append(el)
                resp3 = table.list_entities(results_per_page=2).by_page(
                    continuation_token=resp2.continuation_token)
                entities3 = []
                async for el in await resp3.__anext__():
                    entities3.append(el)

            # Assert
            assert len(entities1) ==  2
            assert len(entities2) ==  2
            assert len(entities3) ==  1
            self._assert_default_entity(entities1[0])
            self._assert_default_entity(entities1[1])
            self._assert_default_entity(entities2[0])
            self._assert_default_entity(entities2[1])
            self._assert_default_entity(entities3[0])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_datetime_milliseconds(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            entity = self._create_random_entity_dict()

            entity['milliseconds'] = datetime(2011, 11, 4, 0, 5, 23, 283000, tzinfo=tzutc())

            await self.table.create_entity(entity)

            received_entity = await self.table.get_entity(
                partition_key=entity['PartitionKey'],
                row_key=entity['RowKey']
            )

            assert entity['milliseconds'] == received_entity['milliseconds']

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_datetime_str_passthrough(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        partition, row = self._create_pk_rk(None, None)

        dotnet_timestamp = "2013-08-22T01:12:06.2608595Z"
        entity = {
            'PartitionKey': partition,
            'RowKey': row,
            'datetime1': EntityProperty(dotnet_timestamp, EdmType.DATETIME)
        }
        try:
            await self.table.create_entity(entity)
            received = await self.table.get_entity(partition, row)
            assert isinstance(received['datetime1'], datetime)
            assert received['datetime1'].tables_service_value == dotnet_timestamp

            received['datetime2'] = received['datetime1'].replace(year=2020)
            assert received['datetime2'].tables_service_value == ""

            await self.table.update_entity(received, mode=UpdateMode.REPLACE)
            updated = await self.table.get_entity(partition, row)
            assert isinstance(updated['datetime1'], datetime)
            assert isinstance(updated['datetime2'], datetime)
            assert updated['datetime1'].tables_service_value == dotnet_timestamp
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_sas_query(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        url = self.account_url(tables_cosmos_account_name, "cosmos")

        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Arrange
            entity, _ = await self._insert_random_entity()
            token = self.generate_sas(
                generate_table_sas,
                tables_primary_cosmos_account_key,
                self.table_name,
                permission=TableSasPermissions(read=True),
                expiry=datetime.utcnow() + timedelta(hours=1),
                start=datetime.utcnow() - timedelta(minutes=1),
            )

            # Act
            service = TableServiceClient(
                self.account_url(tables_cosmos_account_name, "cosmos"),
                credential=AzureSasCredential(token),
            )
            table = service.get_table_client(self.table_name)
            entities = []
            async for t in table.query_entities(
                    "PartitionKey eq '{}'".format(entity['PartitionKey'])):
                entities.append(t)

            # Assert
            assert len(entities) ==  1
            self._assert_default_entity(entities[0])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_sas_add(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        url = self.account_url(tables_cosmos_account_name, "cosmos")
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Arrange
            token = self.generate_sas(
                generate_table_sas,
                tables_primary_cosmos_account_key,
                self.table_name,
                permission=TableSasPermissions(add=True),
                expiry=datetime.utcnow() + timedelta(hours=1),
                start=datetime.utcnow() - timedelta(minutes=1),
            )

            # Act
            service = TableServiceClient(
                self.account_url(tables_cosmos_account_name, "cosmos"),
                credential=AzureSasCredential(token),
            )
            table = service.get_table_client(self.table_name)

            entity = self._create_random_entity_dict()
            await table.create_entity(entity=entity)

            # Assert
            resp = await self.table.get_entity(partition_key=entity['PartitionKey'],
                                               row_key=entity['RowKey'])
            self._assert_default_entity(resp)
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_sas_add_outside_range(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        url = self.account_url(tables_cosmos_account_name, "cosmos")
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Arrange
            token = self.generate_sas(
                generate_table_sas,
                tables_primary_cosmos_account_key,
                self.table_name,
                permission=TableSasPermissions(add=True),
                expiry=datetime.utcnow() + timedelta(hours=1),
                start_pk='test', start_rk='test1',
                end_pk='test', end_rk='test1',
            )

            # Act
            service = TableServiceClient(
                self.account_url(tables_cosmos_account_name, "cosmos"),
                credential=AzureSasCredential(token),
            )
            table = service.get_table_client(self.table_name)
            with pytest.raises(HttpResponseError):
                entity = self._create_random_entity_dict()
                await table.create_entity(entity=entity)

            # Assert
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_sas_update(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        url = self.account_url(tables_cosmos_account_name, "cosmos")
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Arrange
            entity, _ = await self._insert_random_entity()
            token = self.generate_sas(
                generate_table_sas,
                tables_primary_cosmos_account_key,
                self.table_name,
                permission=TableSasPermissions(update=True),
                expiry=datetime.utcnow() + timedelta(hours=1),
            )

            # Act
            service = TableServiceClient(
                self.account_url(tables_cosmos_account_name, "cosmos"),
                credential=AzureSasCredential(token),
            )
            table = service.get_table_client(self.table_name)
            updated_entity = self._create_updated_entity_dict(entity['PartitionKey'], entity['RowKey'])
            resp = await table.update_entity(mode=UpdateMode.REPLACE, entity=updated_entity)
            received_entity = await self.table.get_entity(entity['PartitionKey'],
                                                          entity['RowKey'])

            # Assert
            self._assert_updated_entity(received_entity)
            assert resp is not None

        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_sas_delete(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        url = self.account_url(tables_cosmos_account_name, "cosmos")
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        try:
            # Arrange
            entity, _ = await self._insert_random_entity()
            token = self.generate_sas(
                generate_table_sas,
                tables_primary_cosmos_account_key,
                self.table_name,
                permission=TableSasPermissions(delete=True),
                expiry=datetime.utcnow() + timedelta(hours=1),
            )

            # Act
            service = TableServiceClient(
                self.account_url(tables_cosmos_account_name, "cosmos"),
                credential=AzureSasCredential(token),
            )
            table = service.get_table_client(self.table_name)
            await table.delete_entity(entity['PartitionKey'], entity['RowKey'])

            # Assert
            with pytest.raises(ResourceNotFoundError):
                await self.table.get_entity(entity['PartitionKey'], entity['RowKey'])
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_datetime_duplicate_field(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        partition, row = self._create_pk_rk(None, None)

        entity = {
            'PartitionKey': partition,
            'RowKey': row,
            'Timestamp': datetime(year=1999, month=9, day=9, hour=9, minute=9)
        }
        try:
            await self.table.create_entity(entity)
            received = await self.table.get_entity(partition, row)

            assert 'Timestamp' not in received
            assert 'timestamp' in received.metadata
            assert isinstance(received.metadata['timestamp'], datetime)
            assert received.metadata['timestamp'].year > 2020
        
            received['timestamp'] = datetime(year=1999, month=9, day=9, hour=9, minute=9)
            await self.table.update_entity(received, mode=UpdateMode.REPLACE)
            received = await self.table.get_entity(partition, row)

            assert 'timestamp' in received
            assert isinstance(received['timestamp'], datetime)
            assert received['timestamp'].year == 1999
            assert isinstance(received.metadata['timestamp'], datetime)
            assert received.metadata['timestamp'].year > 2020
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_etag_duplicate_field(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        partition, row = self._create_pk_rk(None, None)

        entity = {
            'PartitionKey': partition,
            'RowKey': row,
            #'ETag': u'foo',
            'etag': u'bar',
            'Etag': u'baz',
        }
        try:
            await self.table.create_entity(entity)
            created = await self.table.get_entity(partition, row)

            #assert created['ETag'] == u'foo'
            assert created['etag'] == u'bar'
            assert 'Etag' not in created
            assert created.metadata['etag'].startswith(u'W/"datetime\'')

            #entity['ETag'] = u'one'
            entity['etag'] = u'two'
            entity['Etag'] = u'three'
            with pytest.raises(ValueError):
                await self.table.update_entity(entity, match_condition=MatchConditions.IfNotModified)
        
            #created['ETag'] = u'one'
            created['etag'] = u'two'
            created['Etag'] = u'three'
            await self.table.update_entity(created, match_condition=MatchConditions.IfNotModified)

            updated = await self.table.get_entity(partition, row)
            #assert updated['ETag'] == u'one'
            assert updated['etag'] == u'two'
            assert 'Etag' not in updated
            assert updated.metadata['etag'].startswith(u'W/"datetime\'')
            assert updated.metadata['etag'] != created.metadata['etag']
        finally:
            await self._tear_down()

    @cosmos_decorator_async
    async def test_entity_create_response_echo(self, tables_cosmos_account_name, tables_primary_cosmos_account_key):
        await self._set_up(tables_cosmos_account_name, tables_primary_cosmos_account_key, url="cosmos")
        partition, row = self._create_pk_rk(None, None)

        entity = {
            'PartitionKey': partition,
            'RowKey': row,
            'Value': 'foobar',
            'Answer': 42
        }
        try:
            result = await self.table.create_entity(entity)
            assert 'preference_applied' not in result
            assert 'content' not in result
            await self.table.delete_entity(entity)

            result = await self.table.create_entity(entity, headers={'Prefer': 'return-no-content'})
            assert 'preference_applied' in result
            assert result['preference_applied'] == 'return-no-content'
            assert 'content' in result
            assert result['content'] is None
            await self.table.delete_entity(entity)

            result = await self.table.create_entity(entity, headers={'Prefer': 'return-content'})
            assert 'preference_applied' in result
            assert result['preference_applied'] == 'return-content'
            assert 'content' in result
            assert result['content']['PartitionKey'] == partition
            assert result['content']['Value'] == 'foobar'
            assert result['content']['Answer'] == 42
        finally:
            await self._tear_down()