text: ﻿#The MIT License (MIT)
  #Copyright (c) 2014 Microsoft Corporation
  
  #Permission is hereby granted, free of charge, to any person obtaining a copy
  #of this software and associated documentation files (the "Software"), to deal
  #in the Software without restriction, including without limitation the rights
  #to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  #copies of the Software, and to permit persons to whom the Software is
  #furnished to do so, subject to the following conditions:
  
  #The above copyright notice and this permission notice shall be included in all
  #copies or substantial portions of the Software.
  
  #THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  #IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  #FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  #AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  #LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  #OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  #SOFTWARE.
  
  """End to end test.
  """
  
  import json
  import logging
  import os.path
  import sys
  import unittest
  from six.moves import xrange
  from struct import unpack, pack
  # from six.moves.builtins import *
  import time
  import six
  if six.PY2:
      import urllib as urllib
  else:
      import urllib.parse as urllib
  import uuid
  import pytest
  import azure.cosmos.base as base
  import azure.cosmos.consistent_hash_ring as consistent_hash_ring
  import azure.cosmos.documents as documents
  import azure.cosmos.cosmos_client as cosmos_client
  import azure.cosmos.errors as errors
  import azure.cosmos.hash_partition_resolver as hash_partition_resolver
  from azure.cosmos.http_constants import HttpHeaders, StatusCodes, SubStatusCodes
  import azure.cosmos.murmur_hash as murmur_hash
  import azure.cosmos.range_partition_resolver as range_partition_resolver
  import azure.cosmos.range as partition_range
  import test_config
  import test_partition_resolver
  import azure.cosmos.base as base
  
  pytestmark = pytest.mark.cosmosEmulator
  
  #IMPORTANT NOTES: 
    
  #  	Most test cases in this file create collections in your Azure Cosmos account.
  #  	Collections are billing entities.  By running these test cases, you may incur monetary costs on your account.
    
  #  	To Run the test, replace the two member fields (masterKey and host) with values 
  #   associated with your Azure Cosmos account.
  
  @pytest.mark.usefixtures("teardown")
  class CRUDTests(unittest.TestCase):
      """Python CRUD Tests.
      """
  
      configs = test_config._test_config
      host = configs.host
      masterKey = configs.masterKey
      connectionPolicy = configs.connectionPolicy
  
      def __AssertHTTPFailureWithStatus(self, status_code, func, *args, **kwargs):
          """Assert HTTP failure with status.
  
          :Parameters:
              - `status_code`: int
              - `func`: function
          """
          try:
              func(*args, **kwargs)
              self.assertFalse(True, 'function should fail.')
          except errors.HTTPFailure as inst:
              self.assertEqual(inst.status_code, status_code)
  
      @classmethod
      def setUpClass(cls):
          if (cls.masterKey == '[YOUR_KEY_HERE]' or
                  cls.host == '[YOUR_ENDPOINT_HERE]'):
              raise Exception(
                  "You must specify your Azure Cosmos account values for "
                  "'masterKey' and 'host' at the top of this class to run the "
                  "tests.")
          cls.client = cosmos_client.CosmosClient(cls.host, {'masterKey': cls.masterKey}, cls.connectionPolicy)
          cls.databseForTest = cls.configs.create_database_if_not_exist(cls.client)
  
  
      def setUp(self):
          self.client = cosmos_client.CosmosClient(self.host, {'masterKey': self.masterKey}, self.connectionPolicy)
  
      def test_database_crud_self_link(self):
          self._test_database_crud(False)
  
      def test_database_crud_name_based(self):
          self._test_database_crud(True)
  
      def _test_database_crud(self, is_name_based):
          # read databases.
          databases = list(self.client.ReadDatabases())
          # create a database.
          before_create_databases_count = len(databases)
          database_definition = { 'id': str(uuid.uuid4()) }
          created_db = self.client.CreateDatabase(database_definition)
          self.assertEqual(created_db['id'], database_definition['id'])
          # Read databases after creation.
          databases = list(self.client.ReadDatabases())
          self.assertEqual(len(databases),
                           before_create_databases_count + 1,
                           'create should increase the number of databases')
          # query databases.
          databases = list(self.client.QueryDatabases({
              'query': 'SELECT * FROM root r WHERE r.id=@id',
              'parameters': [
                  { 'name':'@id', 'value': database_definition['id'] }
              ]
          }))
          self.assert_(databases,
                       'number of results for the query should be > 0')
  
          # read database.
          self.client.ReadDatabase(self.GetDatabaseLink(created_db, is_name_based))
  
          # delete database.
          self.client.DeleteDatabase(self.GetDatabaseLink(created_db, is_name_based))
          # read database after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadDatabase,
                                             self.GetDatabaseLink(created_db, is_name_based))
  
      def test_sql_query_crud(self):
          # create two databases.
          db1 = self.client.CreateDatabase({ 'id': 'database 1' })
          db2 = self.client.CreateDatabase({ 'id': 'database 2' })
          # query with parameters.
          databases = list(self.client.QueryDatabases({
              'query': 'SELECT * FROM root r WHERE r.id=@id',
              'parameters': [
                  { 'name':'@id', 'value': 'database 1' }
              ]
          }))
          self.assertEqual(1, len(databases), 'Unexpected number of query results.')
  
          # query without parameters.
          databases = list(self.client.QueryDatabases({
              'query': 'SELECT * FROM root r WHERE r.id="database non-existing"'
          }))
          self.assertEqual(0, len(databases), 'Unexpected number of query results.')
  
          # query with a string.
          databases = list(self.client.QueryDatabases('SELECT * FROM root r WHERE r.id="database 2"'))
          self.assertEqual(1, len(databases), 'Unexpected number of query results.')
  
          self.client.DeleteDatabase(db1['_self'])
          self.client.DeleteDatabase(db2['_self'])
  
      def test_collection_crud_self_link(self):
          self._test_collection_crud(False)
  
      def test_collection_crud_name_based(self):
          self._test_collection_crud(True)
  
      def _test_collection_crud(self, is_name_based):
          created_db = self.databseForTest
          collections = list(self.client.ReadContainers(self.GetDatabaseLink(created_db, is_name_based)))
          # create a collection
          before_create_collections_count = len(collections)
          collection_definition = { 'id': 'test_collection_crud ' + str(uuid.uuid4()), 'indexingPolicy': {'indexingMode': 'consistent'} }
          created_collection = self.client.CreateContainer(self.GetDatabaseLink(created_db, is_name_based),
                                                       collection_definition)
          self.assertEqual(collection_definition['id'], created_collection['id'])
          self.assertEqual('consistent', created_collection['indexingPolicy']['indexingMode'])
  
          # read collections after creation
          collections = list(self.client.ReadContainers(self.GetDatabaseLink(created_db, is_name_based)))
          self.assertEqual(len(collections),
                           before_create_collections_count + 1,
                           'create should increase the number of collections')
          # query collections
          collections = list(self.client.QueryContainers(
              self.GetDatabaseLink(created_db, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name':'@id', 'value': collection_definition['id'] }
                  ]
              }))
          # Replacing indexing policy is allowed.
          lazy_policy = {'indexingMode': 'lazy'}
          created_collection['indexingPolicy'] = lazy_policy
          replaced_collection = self.client.ReplaceContainer(self.GetDocumentCollectionLink(created_db, created_collection, is_name_based), created_collection)
          self.assertEqual('lazy', replaced_collection['indexingPolicy']['indexingMode'])
          # Replacing collection Id should fail.
          change_collection = created_collection.copy()
          change_collection['id'] = 'try_change_id'
          self.__AssertHTTPFailureWithStatus(StatusCodes.BAD_REQUEST,
                                             self.client.ReplaceContainer,
                                             self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
                                             change_collection)
  
          self.assertTrue(collections)
          # delete collection
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection, is_name_based))
          # read collection after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadContainer,
                                             self.GetDocumentCollectionLink(created_db, created_collection, is_name_based))
  
      
      def test_partitioned_collection(self):
          created_db = self.databseForTest
  
          collection_definition = {   'id': 'test_partitioned_collection ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/id'],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
  
          options = { 'offerThroughput': 10100 }
  
          created_collection = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition, 
                                  options)
          
          self.assertEqual(collection_definition.get('id'), created_collection.get('id'))
          self.assertEqual(collection_definition.get('partitionKey').get('paths')[0], created_collection.get('partitionKey').get('paths')[0])
          self.assertEqual(collection_definition.get('partitionKey').get('kind'), created_collection.get('partitionKey').get('kind'))
  
          offers = self.GetCollectionOffers(self.client, created_collection['_rid'])
          
          self.assertEqual(1, len(offers))
          expected_offer = offers[0]
          self.assertEqual(expected_offer.get('content').get('offerThroughput'), options.get('offerThroughput'))
  
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection))
  
      def test_partitioned_collection_quota(self):
          created_db = self.databseForTest
  
          options = { 'offerThroughput': 20000 }
  
          created_collection = self.configs.create_multi_partition_collection_if_not_exist(self.client)
  
          read_options = { 'populatePartitionKeyRangeStatistics': True, 'populateQuotaInfo': True}
  
          retrieved_collection = self.client.ReadContainer(created_collection.get('_self'), read_options)
  
          self.assertTrue(retrieved_collection.get("statistics") != None)
          self.assertTrue(self.client.last_response_headers.get("x-ms-resource-usage") != None)
  
      def test_partitioned_collection_partition_key_extraction(self):
          created_db = self.databseForTest
          
          collection_definition = {   'id': 'test_partitioned_collection_partition_key_extraction ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/address/state'],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
  
          created_collection = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition)
  
          document_definition = {'id': 'document1',
                                 'address' : { 'street' : '1 Microsoft Way',
                                               'city' : 'Redmond',
                                               'state' : 'WA',
                                               'zip code' : 98052
                                             }
                                 }
  
          # create document without partition key being specified
          created_document = self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          self.assertEqual(created_document.get('id'), document_definition.get('id'))
          self.assertEqual(created_document.get('address').get('state'), document_definition.get('address').get('state'))
  
          # create document by specifying a different partition key in options than what's in the document will result in BadRequest(status code 400)
          document_definition['id'] = 'document2'
          options = { 'partitionKey': 'NY' }
  
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.CreateItem,
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition,
              options)
  
          collection_definition1 = {   'id': 'test_partitioned_collection_partition_key_extraction1 ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/address'],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
  
          created_collection1 = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition1)
  
          # Create document with partitionkey not present as a leaf level property but a dict
          options = {}
          created_document = self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection1),
              document_definition, options)
  
          self.assertEqual(options['partitionKey'], documents.Undefined)
  
          collection_definition2 = {   'id': 'test_partitioned_collection_partition_key_extraction2 ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/address/state/city'],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
  
          created_collection2 = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition2)
  
          # Create document with partitionkey not present in the document
          options = {}
          created_document = self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection2),
              document_definition, options)
  
          self.assertEqual(options['partitionKey'], documents.Undefined)
  
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection))
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection1))
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection2))
          
      def test_partitioned_collection_partition_key_extraction_special_chars(self):
          created_db = self.databseForTest
  
          collection_definition1 = {   'id': 'test_partitioned_collection_partition_key_extraction_special_chars1 ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/\"level\' 1*()\"/\"le/vel2\"'],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
          
          created_collection1 = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition1)
  
          document_definition = {'id': 'document1',
                                 "level' 1*()" : { "le/vel2" : 'val1' }
                                }
  
          options = {}
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection1),
              document_definition, options)
  
          self.assertEqual(options['partitionKey'], 'val1')
  
          collection_definition2 = {   'id': 'test_partitioned_collection_partition_key_extraction_special_chars2 ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/\'level\" 1*()\'/\'le/vel2\''],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
          
          created_collection2 = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition2)
  
          document_definition = {'id': 'document2',
                                 'level\" 1*()' : { 'le/vel2' : 'val2' }
                                }
  
          options = {}
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection2),
              document_definition, options)
  
          self.assertEqual(options['partitionKey'], 'val2')
  
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection1))
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, created_collection2))
  
      def test_partitioned_collection_path_parser(self):
          test_dir = os.path.dirname(os.path.abspath(__file__))
          with open(os.path.join(test_dir, "BaselineTest.PathParser.json")) as json_file:
              entries = json.loads(json_file.read())
          for entry in entries:
              parts = base.ParsePaths([entry['path']])
              self.assertEqual(parts, entry['parts'])
  
          paths = ["/\"Ke \\ \\\" \\\' \\? \\a \\\b \\\f \\\n \\\r \\\t \\v y1\"/*"]
          parts = [ "Ke \\ \\\" \\\' \\? \\a \\\b \\\f \\\n \\\r \\\t \\v y1", "*" ]
          self.assertEqual(parts, base.ParsePaths(paths))
  
          paths = ["/'Ke \\ \\\" \\\' \\? \\a \\\b \\\f \\\n \\\r \\\t \\v y1'/*"]
          parts = [ "Ke \\ \\\" \\\' \\? \\a \\\b \\\f \\\n \\\r \\\t \\v y1", "*" ]
          self.assertEqual(parts, base.ParsePaths(paths))
  
      def test_partitioned_collection_document_crud_and_query(self):
          created_db = self.databseForTest
          
          created_collection = self.configs.create_multi_partition_collection_if_not_exist(self.client)
  
          document_definition = {'id': 'document',
                                 'key': 'value'}
  
          created_document = self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          self.assertEqual(created_document.get('id'), document_definition.get('id'))
          self.assertEqual(created_document.get('key'), document_definition.get('key'))
  
          # For ReadDocument, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          #self.__AssertHTTPFailureWithStatus(
          #    StatusCodes.BAD_REQUEST,
          #    client.ReadItem,
          #    self.GetDocumentLink(created_db, created_collection, created_document))
  
          # read document
          options = { 'partitionKey': document_definition.get('id') }
          read_document = self.client.ReadItem(
              self.GetDocumentLink(created_db, created_collection, created_document), 
              options)
  
          self.assertEqual(read_document.get('id'), created_document.get('id'))
          self.assertEqual(read_document.get('key'), created_document.get('key'))
  
          # Read document feed doesn't require partitionKey as it's always a cross partition query
          documentlist = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection)))
          self.assertEqual(1, len(documentlist))
  
          # replace document
          document_definition['key'] = 'new value'
  
          replaced_document = self.client.ReplaceItem(
              self.GetDocumentLink(created_db, created_collection, created_document),
              document_definition)
  
          self.assertEqual(replaced_document.get('key'), document_definition.get('key'))
  
          # upsert document(create scenario)
          document_definition['id'] = 'document2'
          document_definition['key'] = 'value2'
  
          upserted_document = self.client.UpsertItem(self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          self.assertEqual(upserted_document.get('id'), document_definition.get('id'))
          self.assertEqual(upserted_document.get('key'), document_definition.get('key'))
  
          documentlist = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection)))
          self.assertEqual(2, len(documentlist))
  
          # For DeleteDocument, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.DeleteItem,
              self.GetDocumentLink(created_db, created_collection, upserted_document))
  
          # delete document
          options = { 'partitionKey': upserted_document.get('id') }
          self.client.DeleteItem(
              self.GetDocumentLink(created_db, created_collection, upserted_document), 
              options)
  
          # query document on the partition key specified in the predicate will pass even without setting enableCrossPartitionQuery or passing in the partitionKey value
          documentlist = list(self.client.QueryItems(
              self.GetDocumentCollectionLink(created_db, created_collection),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=\'' + replaced_document.get('id') + '\''
              }))
          self.assertEqual(1, len(documentlist))
  
          # query document on any property other than partitionKey will fail without setting enableCrossPartitionQuery or passing in the partitionKey value
          try:
              list(self.client.QueryItems(
                  self.GetDocumentCollectionLink(created_db, created_collection),
                  {
                      'query': 'SELECT * FROM root r WHERE r.key=\'' + replaced_document.get('key') + '\''
                  }))
          except Exception:
              pass
  
          # cross partition query
          options = { 'enableCrossPartitionQuery': True }
          documentlist = list(self.client.QueryItems(
              self.GetDocumentCollectionLink(created_db, created_collection),
              {
                  'query': 'SELECT * FROM root r WHERE r.key=\'' + replaced_document.get('key') + '\''
              }, options))
  
          self.assertEqual(1, len(documentlist))
  
          # query document by providing the partitionKey value
          options = { 'partitionKey': replaced_document.get('id') }
          documentlist = list(self.client.QueryItems(
              self.GetDocumentCollectionLink(created_db, created_collection),
              {
                  'query': 'SELECT * FROM root r WHERE r.key=\'' + replaced_document.get('key') + '\''
              }, options))
  
          self.assertEqual(1, len(documentlist))
  
      def test_partitioned_collection_permissions(self):
          created_db = self.databseForTest
  
          collection_definition = {   'id': 'sample collection ' + str(uuid.uuid4()),
                                      'partitionKey': 
                                      {   
                                          'paths': ['/key'],
                                          'kind': documents.PartitionKind.Hash
                                      }
                                  }
  
          collection_definition['id'] = 'test_partitioned_collection_permissions all collection'
          
          all_collection = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition)
  
          collection_definition['id'] = 'test_partitioned_collection_permissions read collection'
  
          read_collection = self.client.CreateContainer(self.GetDatabaseLink(created_db),
                                  collection_definition)
  
          user = self.client.CreateUser(self.GetDatabaseLink(created_db), { 'id': 'user' })
  
          permission_definition = {
              'id': 'all permission',
              'permissionMode': documents.PermissionMode.All,
              'resource': self.GetDocumentCollectionLink(created_db, all_collection),
              'resourcePartitionKey' : [1]
          }
  
          all_permission = self.client.CreatePermission(self.GetUserLink(created_db, user), permission_definition)
  
          permission_definition = {
              'id': 'read permission',
              'permissionMode': documents.PermissionMode.Read,
              'resource': self.GetDocumentCollectionLink(created_db, read_collection),
              'resourcePartitionKey' : [1]
          }
  
          read_permission = self.client.CreatePermission(self.GetUserLink(created_db, user), permission_definition)
  
          resource_tokens = {}
          # storing the resource tokens based on Resource IDs
          resource_tokens[all_collection['_rid']] = (all_permission['_token'])
          resource_tokens[read_collection['_rid']] = (read_permission['_token'])
          
          restricted_client = cosmos_client.CosmosClient(
              CRUDTests.host, {'resourceTokens': resource_tokens}, CRUDTests.connectionPolicy)
  
          document_definition = {'id': 'document1',
                                 'key': 1
                                 }
          
          # Create document in all_collection should succeed since the partitionKey is 1 which is what specified as resourcePartitionKey in permission object and it has all permissions
          created_document = restricted_client.CreateItem(
              self.GetDocumentCollectionLink(created_db, all_collection, False),
              document_definition)
  
          # Create document in read_collection should fail since it has only read permissions for this collection
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.FORBIDDEN,
              restricted_client.CreateItem,
              self.GetDocumentCollectionLink(created_db, read_collection, False),
              document_definition)
  
          # Read document feed should succeed for this collection. Note that I need to pass in partitionKey here since permission has resourcePartitionKey defined
          options = { 'partitionKey': document_definition.get('key') }
          documentlist = list(restricted_client.ReadItems(
              self.GetDocumentCollectionLink(created_db, read_collection, False),
              options))
  
          self.assertEqual(0, len(documentlist))
  
          document_definition['key'] = 2
          options = { 'partitionKey': document_definition.get('key') }
          # Create document should fail since the partitionKey is 2 which is different that what is specified as resourcePartitionKey in permission object
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.FORBIDDEN,
              restricted_client.CreateItem,
              self.GetDocumentCollectionLink(created_db, all_collection, False),
              document_definition,
              options)
          
          document_definition['key'] = 1
          options = { 'partitionKey': document_definition.get('key') }
          # Delete document should succeed since the partitionKey is 1 which is what specified as resourcePartitionKey in permission object
          created_document = restricted_client.DeleteItem(
              self.GetDocumentLink(created_db, all_collection, created_document, False),
              options)
  
          # Delete document in read_collection should fail since it has only read permissions for this collection
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.FORBIDDEN,
              restricted_client.DeleteItem,
              self.GetDocumentCollectionLink(created_db, read_collection, False),
              options)
  
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, all_collection))
          self.client.DeleteContainer(self.GetDocumentCollectionLink(created_db, read_collection))
  
      def test_partitioned_collection_execute_stored_procedure(self):
          created_db = self.databseForTest
  
          created_collection = self.configs.create_multi_partition_collection_with_custom_pk_if_not_exist(self.client)
  
          sproc = {
              'id': 'storedProcedure' + str(uuid.uuid4()),
              'body': (
                  'function () {' +
                  '   var client = getContext().getCollection();' +
                  '   client.createDocument(client.getSelfLink(), { id: \'testDoc\', pk : 2}, {}, function(err, docCreated, options) { ' +
                  '   if(err) throw new Error(\'Error while creating document: \' + err.message);' +
                  '   else {' +
                           '   getContext().getResponse().setBody(1);' +
                  '        }' +
                  '   });}')
          }
  
          created_sproc = self.client.CreateStoredProcedure(self.GetDocumentCollectionLink(created_db, created_collection), sproc)
  
          # Partiton Key value same as what is specified in the stored procedure body
          self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(created_db, created_collection, created_sproc),
                                                 None, { 'partitionKey' : 2})
  
          # Partiton Key value different than what is specified in the stored procedure body will cause a bad request(400) error
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.ExecuteStoredProcedure,
              self.GetStoredProcedureLink(created_db, created_collection, created_sproc),
              None,
              { 'partitionKey' : 3})
  
      def test_partitioned_collection_attachment_crud_and_query(self):
          class ReadableStream(object):
              """Customized file-like stream.
              """
  
              def __init__(self, chunks = ['first chunk ', 'second chunk']):
                  """Initialization.
  
                  :Parameters:
                      - `chunks`: list
  
                  """
                  if six.PY2:
                      self._chunks = list(chunks)
                  else:
                      # python3: convert to bytes
                      self._chunks = [chunk.encode() for chunk in chunks]
  
              def read(self, n=-1):
                  """Simulates the read method in a file stream.
  
                  :Parameters:
                      - `n`: int
  
                  :Returns:
                      bytes or str
  
                  """
                  if self._chunks:
                      return self._chunks.pop(0)
                  else:
                      return ''
  
              def __len__(self):
                  """To make len(ReadableStream) work.
                  """
                  return sum([len(chunk) for chunk in self._chunks])
  
          db = self.databseForTest
          collection_definition = {'id': 'test_partitioned_collection_attachment_crud_and_query ' + str(uuid.uuid4()),
                                   'partitionKey': {'paths': ['/id'],'kind': 'Hash'}}
  
          collection = self.client.CreateContainer(db['_self'], collection_definition)
          
          document_definition = {'id': 'sample document' + str(uuid.uuid4()),
                                 'key': 'value'}
  
          document = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection),
                                           document_definition)
  
          content_stream = ReadableStream()
          options = { 'slug': 'sample attachment',
                      'contentType': 'application/text' }
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          #self.__AssertHTTPFailureWithStatus(
          #    StatusCodes.BAD_REQUEST,
          #    client.CreateAttachmentAndUploadMedia,
          #    self.GetDocumentLink(db, collection, document),
          #    content_stream,
          #    options)
  
          content_stream = ReadableStream()
          # Setting the partitionKey as part of options is required for attachment CRUD
          options = { 'slug': 'sample attachment' + str(uuid.uuid4()),
                      'contentType': 'application/text',
                      'partitionKey' :  document_definition.get('id') }
          
          # create attachment and upload media
          attachment = self.client.CreateAttachmentAndUploadMedia(
              self.GetDocumentLink(db, collection, document), content_stream, options)
  
          self.assertEqual(attachment['id'], options['slug'])
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          try:
              list(self.client.ReadAttachments(
              self.GetDocumentLink(db, collection, document)))
          except Exception:
              pass
  
          # Read attachment feed requires partitionKey to be passed
          options = { 'partitionKey': document_definition.get('id') }
          attachmentlist = list(self.client.ReadAttachments(
              self.GetDocumentLink(db, collection, document), options))
          self.assertEqual(1, len(attachmentlist))
  
          content_stream = ReadableStream()
          options = { 'slug': 'new attachment' + str(uuid.uuid4()),
                      'contentType': 'application/text' }
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.UpsertAttachmentAndUploadMedia,
              self.GetDocumentLink(db, collection, document),
              content_stream,
              options)
  
          content_stream = ReadableStream()
          # Setting the partitionKey as part of options is required for attachment CRUD
          options = { 'slug': 'new attachment' + str(uuid.uuid4()),
                      'contentType': 'application/text',
                      'partitionKey' :  document_definition.get('id') }
          
          # upsert attachment and upload media
          attachment = self.client.UpsertAttachmentAndUploadMedia(
              self.GetDocumentLink(db, collection, document), content_stream, options)
  
          self.assertEqual(attachment['id'], options['slug'])
  
          options = { 'partitionKey': document_definition.get('id') }
          attachmentlist = list(self.client.ReadAttachments(
              self.GetDocumentLink(db, collection, document), options))
          self.assertEqual(2, len(attachmentlist))
  
          # create attachment with media link
          dynamic_attachment = {
              'id': 'dynamic attachment' + str(uuid.uuid4()),
              'media': 'http://xstore.',
              'MediaType': 'Book',
              'Author':'My Book Author',
              'Title':'My Book Title',
              'contentType':'application/text'
          }
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.CreateAttachment,
              self.GetDocumentLink(db, collection, document),
              dynamic_attachment)
  
          # create dynamic attachment
          options = { 'partitionKey': document_definition.get('id') }
          attachment = self.client.CreateAttachment(self.GetDocumentLink(db, collection, document),
                                               dynamic_attachment, options)
  
          self.assertEqual(attachment['MediaType'], dynamic_attachment['MediaType'])
          self.assertEqual(attachment['Author'], dynamic_attachment['Author'])
  
          # Read Attachment feed
          options = { 'partitionKey': document_definition.get('id') }
          attachmentlist = list(self.client.ReadAttachments(
              self.GetDocumentLink(db, collection, document), options))
          self.assertEqual(3, len(attachmentlist))
          
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          #self.__AssertHTTPFailureWithStatus(
          #    StatusCodes.BAD_REQUEST,
          #    client.ReadAttachment,
          #    self.GetAttachmentLink(db, collection, document, attachment))
  
          # Read attachment
          options = { 'partitionKey': document_definition.get('id') }
          read_attachment = self.client.ReadAttachment(self.GetAttachmentLink(db, collection, document, attachment),
                                                  options)
  
          self.assertEqual(attachment['id'], read_attachment['id'])
  
          attachment['Author'] = 'new author'
          
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.ReplaceAttachment,
              self.GetAttachmentLink(db, collection, document, attachment),
              attachment)
  
          # replace the attachment
          options = { 'partitionKey': document_definition.get('id') }
          replaced_attachment = self.client.ReplaceAttachment(self.GetAttachmentLink(db, collection, document, attachment), attachment, options)
          
          self.assertEqual(attachment['id'], replaced_attachment['id'])
          self.assertEqual(attachment['Author'], replaced_attachment['Author'])
  
          attachment['id'] = 'new dynamic attachment' + str(uuid.uuid4())
          attachment['Title'] = 'new title'
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.UpsertAttachment,
              self.GetDocumentLink(db, collection, document),
              attachment)
  
          # upsert attachment(create scenario)
          options = { 'partitionKey': document_definition.get('id') }
          upserted_attachment = self.client.UpsertAttachment(self.GetDocumentLink(db, collection, document), attachment, options)
          
          self.assertEqual(attachment['id'], upserted_attachment['id'])
          self.assertEqual(attachment['Title'], upserted_attachment['Title'])
  
          # query attachments will fail without passing in the partitionKey value
          try:
              list(self.client.QueryAttachments(
                  self.GetDocumentLink(db, collection, document),
                  {
                      'query': 'SELECT * FROM root r WHERE r.MediaType=\'' + dynamic_attachment.get('MediaType') + '\''
                  }))
          except Exception:
              pass
  
          # query attachments by providing the partitionKey value
          options = { 'partitionKey': document_definition.get('id') }
          attachmentlist = list(self.client.QueryAttachments(
              self.GetDocumentLink(db, collection, document),
              {
                  'query': 'SELECT * FROM root r WHERE r.MediaType=\'' + dynamic_attachment.get('MediaType') + '\''
              }, options))
  
          self.assertEqual(2, len(attachmentlist))
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.DeleteAttachment,
              self.GetAttachmentLink(db, collection, document, attachment))
  
          # deleting attachment
          options = { 'partitionKey': document_definition.get('id') }
          self.client.DeleteAttachment(self.GetAttachmentLink(db, collection, document, attachment), options)
          self.client.DeleteContainer(collection['_self'])
  
      def test_partitioned_collection_partition_key_value_types(self):
          created_db = self.databseForTest
  
          created_collection = self.configs.create_multi_partition_collection_with_custom_pk_if_not_exist(self.client)
  
          document_definition = {'id': 'document1' + str(uuid.uuid4()),
                                 'pk' : None,
                                 'spam': 'eggs'}
  
          # create document with partitionKey set as None here
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          document_definition = {'id': 'document1' + str(uuid.uuid4()),
                                 'spam': 'eggs'}
  
          # create document with partitionKey set as Undefined here
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          document_definition = {'id': 'document1' + str(uuid.uuid4()),
                                 'pk' : True,
                                 'spam': 'eggs'}
  
          # create document with bool partitionKey
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          document_definition = {'id': 'document1' + str(uuid.uuid4()),
                                 'pk' : 'value',
                                 'spam': 'eggs'}
  
          # create document with string partitionKey
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          document_definition = {'id': 'document1' + str(uuid.uuid4()),
                                 'pk' : 100,
                                 'spam': 'eggs'}
  
          # create document with int partitionKey
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
          document_definition = {'id': 'document1' + str(uuid.uuid4()),
                                 'pk' : 10.50,
                                 'spam': 'eggs'}
  
          # create document with float partitionKey
          self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection),
              document_definition)
  
      def test_partitioned_collection_conflict_crud_and_query(self):
          created_db = self.databseForTest
  
          created_collection = self.configs.create_multi_partition_collection_if_not_exist(self.client)
  
          conflict_definition = {'id': 'new conflict',
                                 'resourceId' : 'doc1',
                                 'operationType' : 'create',
                                 'resourceType' : 'document'
                                }
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          #self.__AssertHTTPFailureWithStatus(
          #    StatusCodes.BAD_REQUEST,
          #    client.ReadConflict,
          #    self.GetConflictLink(created_db, created_collection, conflict_definition))
  
          # read conflict here will return resource not found(404) since there is no conflict here
          options = { 'partitionKey': conflict_definition.get('id') }
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.NOT_FOUND,
              self.client.ReadConflict,
              self.GetConflictLink(created_db, created_collection, conflict_definition),
              options)
  
          # Read conflict feed doesn't requires partitionKey to be specified as it's a cross partition thing
          conflictlist = list(self.client.ReadConflicts(self.GetDocumentCollectionLink(created_db, created_collection)))
          self.assertEqual(0, len(conflictlist))
  
          # Currently, we require to have the partitionKey to be specified as part of options otherwise we get BadRequest(status code 400)
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.DeleteConflict,
              self.GetConflictLink(created_db, created_collection, conflict_definition))
  
          # delete conflict here will return resource not found(404) since there is no conflict here
          options = { 'partitionKey': conflict_definition.get('id') }
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.NOT_FOUND,
              self.client.DeleteConflict,
              self.GetConflictLink(created_db, created_collection, conflict_definition),
              options)
  
          # query conflicts on any property other than partitionKey will fail without setting enableCrossPartitionQuery or passing in the partitionKey value
          try:
              list(self.client.QueryConflicts(
                  self.GetDocumentCollectionLink(created_db, created_collection),
                  {
                      'query': 'SELECT * FROM root r WHERE r.resourceType=\'' + conflict_definition.get('resourceType') + '\''
                  }))
          except Exception:
              pass
  
          # cross partition query
          options = { 'enableCrossPartitionQuery': True }
          conflictlist = list(self.client.QueryConflicts(
              self.GetDocumentCollectionLink(created_db, created_collection),
              {
                  'query': 'SELECT * FROM root r WHERE r.resourceType=\'' + conflict_definition.get('resourceType') + '\''
              }, options))
          
          self.assertEqual(0, len(conflictlist))
  
          # query conflicts by providing the partitionKey value
          options = { 'partitionKey': conflict_definition.get('id') }
          conflictlist = list(self.client.QueryConflicts(
              self.GetDocumentCollectionLink(created_db, created_collection),
              {
                  'query': 'SELECT * FROM root r WHERE r.resourceType=\'' + conflict_definition.get('resourceType') + '\''
              }, options))
  
          self.assertEqual(0, len(conflictlist))
  
      def test_document_crud_self_link(self):
          self._test_document_crud(False)
  
      def test_document_crud_name_based(self):
          self._test_document_crud(True)
          
      def _test_document_crud(self, is_name_based):
          # create database
          created_db = self.databseForTest
          # create collection
          created_collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          # read documents
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
  
          # create a document
          before_create_documents_count = len(documents)
          document_definition = {'name': 'sample document',
                                 'spam': 'eggs',
                                 'key': 'value'}
          # Should throw an error because automatic id generation is disabled.
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.CreateItem,
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              document_definition,
              {'disableAutomaticIdGeneration': True})
  
          created_document = self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              document_definition)
          self.assertEqual(created_document['name'],
                           document_definition['name'])
          self.assertTrue(created_document['id'] != None)
          # duplicated documents are allowed when 'id' is not provided.
          duplicated_document = self.client.CreateItem(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              document_definition)
          self.assertEqual(duplicated_document['name'],
                           document_definition['name'])
          self.assert_(duplicated_document['id'])
          self.assertNotEqual(duplicated_document['id'],
                              created_document['id'])
          # duplicated documents are not allowed when 'id' is provided.
          duplicated_definition_with_id = document_definition.copy()
          duplicated_definition_with_id['id'] = created_document['id']
          self.__AssertHTTPFailureWithStatus(StatusCodes.CONFLICT,
                                             self.client.CreateItem,
                                             self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
                                             duplicated_definition_with_id)
          # read documents after creation
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
          self.assertEqual(
              len(documents),
              before_create_documents_count + 2,
              'create should increase the number of documents')
          # query documents
          documents = list(self.client.QueryItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.name=@name',
                  'parameters': [
                      { 'name':'@name', 'value':document_definition['name'] }
                  ]
              }))
          self.assert_(documents)
          documents = list(self.client.QueryItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.name=@name',
                  'parameters': [
                      { 'name':'@name', 'value':document_definition['name'] }
                  ]
              },
              { 'enableScanInQuery': True}))
          self.assert_(documents)
          # replace document.
          created_document['name'] = 'replaced document'
          created_document['spam'] = 'not eggs'
          old_etag = created_document['_etag']
          replaced_document = self.client.ReplaceItem(
              self.GetDocumentLink(created_db, created_collection, created_document, is_name_based),
              created_document)
          self.assertEqual(replaced_document['name'],
                           'replaced document',
                           'document id property should change')
          self.assertEqual(replaced_document['spam'],
                           'not eggs',
                           'property should have changed')
          self.assertEqual(created_document['id'],
                           replaced_document['id'],
                           'document id should stay the same')
          
          #replace document based on condition
          replaced_document['name'] = 'replaced document based on condition'
          replaced_document['spam'] = 'new spam field'
  
          #should fail for stale etag
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.PRECONDITION_FAILED, self.client.ReplaceItem,
                  self.GetDocumentLink(created_db, created_collection, replaced_document, is_name_based),
                  replaced_document, { 'accessCondition' : { 'type': 'IfMatch', 'condition': old_etag } })
  
          #should pass for most recent etag
          replaced_document_conditional = self.client.ReplaceItem(
              self.GetDocumentLink(created_db, created_collection, replaced_document, is_name_based),
              replaced_document, { 'accessCondition' : { 'type': 'IfMatch', 'condition': replaced_document['_etag'] } })
          self.assertEqual(replaced_document_conditional['name'],
                           'replaced document based on condition',
                           'document id property should change')
          self.assertEqual(replaced_document_conditional['spam'],
                           'new spam field',
                           'property should have changed')
          self.assertEqual(replaced_document_conditional['id'],
                           replaced_document['id'],
                           'document id should stay the same')
          # read document
          one_document_from_read = self.client.ReadItem(
              self.GetDocumentLink(created_db, created_collection, replaced_document, is_name_based))
          self.assertEqual(replaced_document['id'],
                           one_document_from_read['id'])
          # delete document
          self.client.DeleteItem(self.GetDocumentLink(created_db, created_collection, replaced_document, is_name_based))
          # read documents after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadItem,
                                             self.GetDocumentLink(created_db, created_collection, replaced_document, is_name_based))
  
      def test_partitioning(self):
          # create test database
          created_db = self.databseForTest
          
          # Create bunch of collections participating in partitioning
          collection0 = self.client.CreateContainer(
              self.GetDatabaseLink(created_db, True),
              { 'id': 'test_partitioning coll_0' + str(uuid.uuid4()) })
          collection1 = self.client.CreateContainer(
              self.GetDatabaseLink(created_db, True),
              { 'id': 'test_partitioning coll_1' + str(uuid.uuid4())})
          collection2 = self.client.CreateContainer(
              self.GetDatabaseLink(created_db, True),
              { 'id': 'test_partitioning coll_2' + str(uuid.uuid4())})
  
          # Register the collection links for partitioning through partition resolver
          collection_links = [self.GetDocumentCollectionLink(created_db, collection0, True), self.GetDocumentCollectionLink(created_db, collection1, True), self.GetDocumentCollectionLink(created_db, collection2, True)]
          partition_resolver = test_partition_resolver.TestPartitionResolver(collection_links)
          self.client.RegisterPartitionResolver(self.GetDatabaseLink(created_db, True), partition_resolver)
  
          # create a document using the document definition
          document_definition = { 'id': '0',
                                  'name': 'sample document',
                                  'key': 'value' }
          
          self.client.CreateItem(
              self.GetDatabaseLink(created_db, True),
              document_definition)
  
          # Read the documents in collection1 and verify that the count is 1 now
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, collection0, True)))
          self.assertEqual(1, len(documents))
          
          # Verify that it contains the document with Id 0
          self.assertEqual('0', documents[0]['id'])
  
          document_definition['id'] = '1'
  
          self.client.CreateItem(
              self.GetDatabaseLink(created_db, True),
              document_definition)
  
          # Read the documents in collection1 and verify that the count is 1 now
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, collection1, True)))
          self.assertEqual(1, len(documents))
  
          # Verify that it contains the document with Id 1
          self.assertEqual('1', documents[0]['id'])
  
          document_definition['id'] = '2'
  
          self.client.CreateItem(
              self.GetDatabaseLink(created_db, True),
              document_definition)
  
          # Read the documents in collection2 and verify that the count is 1 now
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, collection2, True)))
          self.assertEqual(1, len(documents))
  
          # Verify that it contains the document with Id 2
          self.assertEqual('2', documents[0]['id'])
  
          # Updating the value of "key" property to test UpsertDocument(replace scenario)
          document_definition['id'] = '0'
          document_definition['key'] = 'new value'
  
          self.client.UpsertItem(
              self.GetDatabaseLink(created_db, True),
              document_definition)
  
          # Read the documents in collection0 and verify that the count is still 1
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, collection0, True)))
          self.assertEqual(1, len(documents))
  
          # Verify that it contains the document with new key value
          self.assertEqual(document_definition['key'], documents[0]['key'])
  
          # Query documents in all collections(since no partition key specified) using query string
          documents = list(self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=\'2\''
              }))
          self.assertEqual(1, len(documents))
  
          # Updating the value of id property to test UpsertDocument(create scenario)
          document_definition['id'] = '4'
  
          self.client.UpsertItem(
              self.GetDatabaseLink(created_db, True),
              document_definition)
  
          # Read the documents in collection1 and verify that the count is 2 now
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, collection1, True)))
          self.assertEqual(2, len(documents))
  
          # Query documents in all collections(since no partition key specified) using query spec
          documents = list(self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name':'@id', 'value':document_definition['id'] }
                  ]
              }))
          self.assertEqual(1, len(documents))
  
          # Query documents in collection(with partition key of '4' specified) which resolves to collection1
          documents = list(self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r'
              }, {}, document_definition['id']))
          self.assertEqual(2, len(documents))
  
          # Query documents in collection(with partition key '5' specified) which resolves to collection2 but non existent document in that collection
          documents = list(self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name':'@id', 'value':document_definition['id'] }
                  ]
              }, {}, '5'))
          self.assertEqual(0, len(documents))
  
          self.client.DeleteContainer(collection0['_self'])
          self.client.DeleteContainer(collection1['_self'])
          self.client.DeleteContainer(collection2['_self'])
      
      # Partitioning test(with paging)
      def test_partition_paging(self):
          # create test database
          created_db = self.databseForTest
          
          # Create bunch of collections participating in partitioning
          collection0 = self.client.CreateContainer(
              self.GetDatabaseLink(created_db, True),
              { 'id': 'test_partition_paging coll_0 ' + str(uuid.uuid4()) })
          collection1 = self.client.CreateContainer(
              self.GetDatabaseLink(created_db, True),
              { 'id': 'test_partition_paging coll_1 ' + str(uuid.uuid4()) })
          
          # Register the collection links for partitioning through partition resolver
          collection_links = [self.GetDocumentCollectionLink(created_db, collection0, True), self.GetDocumentCollectionLink(created_db, collection1, True)]
          partition_resolver = test_partition_resolver.TestPartitionResolver(collection_links)
          self.client.RegisterPartitionResolver(self.GetDatabaseLink(created_db, True), partition_resolver)
          
          # Create document definition used to create documents
          document_definition = { 'id': '0',
                                  'name': 'sample document',
                                  'key': 'value' }
          
          # Create 10 documents each with a different id starting from 0 to 9
          for i in xrange(0, 10):
              document_definition['id'] = str(i)
              self.client.CreateItem(
                  self.GetDatabaseLink(created_db, True),
                  document_definition)
  
          # Query the documents to ensure that you get the correct count(no paging)
          documents = list(self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id < \'7\''
              }))
          self.assertEqual(7, len(documents))
  
          # Query the documents with maxItemCount to restrict the max number of documents returned
          queryIterable = self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id < \'7\''
              }, {'maxItemCount':3})
  
          # Query again and count the number of documents(with paging)
          docCount = 0
          for _ in queryIterable:
              docCount += 1
  
          self.assertEqual(7, docCount)
  
          # Query again to test fetch_next_block to ensure that it returns the correct number of documents everytime it's called
          queryIterable = self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id < \'7\''
              }, {'maxItemCount':3})
  
          # Documents with id 0, 2, 4(in collection0)
          self.assertEqual(3, len(queryIterable.fetch_next_block()))
  
          # Documents with id 6(in collection0)
          self.assertEqual(1, len(queryIterable.fetch_next_block()))
  
          # Documents with id 1, 3, 5(in collection1)
          self.assertEqual(3, len(queryIterable.fetch_next_block()))
  
          # No more documents
          self.assertEqual(0, len(queryIterable.fetch_next_block()))
  
          # Set maxItemCount to -1 to lift the limit on max documents returned by the query
          queryIterable = self.client.QueryItems(
              self.GetDatabaseLink(created_db, True),
              {
                  'query': 'SELECT * FROM root r WHERE r.id < \'7\''
              }, {'maxItemCount':-1})
  
          # Documents with id 0, 2, 4, 6(all docs in collection0 adhering to query condition)
          self.assertEqual(4, len(queryIterable.fetch_next_block()))
  
          # Documents with id 1, 3, 5(all docs in collection1 adhering to query condition)
          self.assertEqual(3, len(queryIterable.fetch_next_block()))
  
          # No more documents
          self.assertEqual(0, len(queryIterable.fetch_next_block()))
  
          self.client.DeleteContainer(collection0['_self'])
          self.client.DeleteContainer(collection1['_self'])
          
      def test_hash_partition_resolver(self):
          created_db = self.databseForTest
          
          # Create bunch of collections participating in partitioning
          collection0 = { 'id': 'coll_0 ' + str(uuid.uuid4()) }
          collection1 = { 'id': 'coll_1 ' + str(uuid.uuid4()) }
  
          collection_links = [self.GetDocumentCollectionLink(created_db, collection0, True), self.GetDocumentCollectionLink(created_db, collection1, True)]
  
          id_partition_key_extractor = lambda document: document['id']
          
          hashpartition_resolver = hash_partition_resolver.HashPartitionResolver(id_partition_key_extractor, collection_links)
  
          # create a document using the document definition
          document_definition = { 'id': '0',
                                  'name': 'sample document',
                                  'key': 'value' }
          
          document_definition['id'] = '2'
  
          collection_link = hashpartition_resolver.ResolveForCreate(document_definition)
  
          read_collection_links = hashpartition_resolver.ResolveForRead(document_definition['id'])
  
          self.assertEqual(1, len(read_collection_links))
          self.assertEqual(collection_link, read_collection_links[0])
  
      def test_consistent_hash_ring(self):
          created_db = { 'id': 'db' }
  
          collection_links = list()
          expected_partition_list = list()
          
          total_collections_count = 2
  
          collection = { 'id': 'coll' }
  
          for i in xrange(0, total_collections_count):
              collection['id'] = 'coll' + str(i)
              collection_link = self.GetDocumentCollectionLink(created_db, collection, True)
              collection_links.append(collection_link)
  
          expected_partition_list.append(('dbs/db/colls/coll0', 1076200484))
          expected_partition_list.append(('dbs/db/colls/coll0', 1302652881))
          expected_partition_list.append(('dbs/db/colls/coll0', 2210251988))
          expected_partition_list.append(('dbs/db/colls/coll1', 2341558382))
          expected_partition_list.append(('dbs/db/colls/coll0', 2348251587))
          expected_partition_list.append(('dbs/db/colls/coll0', 2887945459))
          expected_partition_list.append(('dbs/db/colls/coll1', 2894403633))
          expected_partition_list.append(('dbs/db/colls/coll1', 3031617259))
          expected_partition_list.append(('dbs/db/colls/coll1', 3090861424))
          expected_partition_list.append(('dbs/db/colls/coll1', 4222475028))
  
          id_partition_key_extractor = lambda document: document['id']
          
          hashpartition_resolver = hash_partition_resolver.HashPartitionResolver(id_partition_key_extractor, collection_links, 5)
  
          actual_partition_list = hashpartition_resolver.consistent_hash_ring._GetSerializedPartitionList()
  
          self.assertEqual(len(expected_partition_list), len(actual_partition_list))
  
          for i in xrange(0, len(expected_partition_list)):
              self.assertEqual(actual_partition_list[i][0], expected_partition_list[i][0])
              self.assertEqual(actual_partition_list[i][1], expected_partition_list[i][1])
  
          # Querying for a document and verifying that it's in the expected collection
          read_collection_links = hashpartition_resolver.ResolveForRead("beadledom")
  
          self.assertEqual(1, len(read_collection_links))
  
          collection['id'] = 'coll1'
          collection_link = self.GetDocumentCollectionLink(created_db, collection, True)
  
          self.assertTrue(collection_link in read_collection_links)
  
          # Querying for a document and verifying that it's in the expected collection
          read_collection_links = hashpartition_resolver.ResolveForRead("999")
  
          self.assertEqual(1, len(read_collection_links))
  
          collection['id'] = 'coll0'
          collection_link = self.GetDocumentCollectionLink(created_db, collection, True)
  
          self.assertTrue(collection_link in read_collection_links)
  
      def test_murmur_hash(self):
          str = 'afdgdd'
          bytes = bytearray(str, encoding='utf-8')
  
          hash_value = murmur_hash._MurmurHash._ComputeHash(bytes)
          self.assertEqual(1099701186, hash_value)
  
          num = 374.0
          bytes = bytearray(pack('d', num))
  
          hash_value = murmur_hash._MurmurHash._ComputeHash(bytes)
          self.assertEqual(3717946798, hash_value)
  
          self._validate_bytes("", 0x1B873593, bytearray(b'\xEE\xA8\xA2\x67'), 1738713326)
          self._validate_bytes("1", 0xE82562E4, bytearray(b'\xD0\x92\x24\xED'), 3978597072)
          self._validate_bytes("00", 0xB4C39035, bytearray(b'\xFA\x09\x64\x1B'), 459540986)
          self._validate_bytes("eyetooth", 0x8161BD86, bytearray(b'\x98\x62\x1C\x6F'), 1864131224)
          self._validate_bytes("acid", 0x4DFFEAD7, bytearray(b'\x36\x92\xC0\xB9'), 3116405302)
          self._validate_bytes("elevation", 0x1A9E1828, bytearray(b'\xA9\xB6\x40\xDF'), 3745560233)
          self._validate_bytes("dent", 0xE73C4579, bytearray(b'\xD4\x59\xE1\xD3'), 3554761172)
          self._validate_bytes("homeland", 0xB3DA72CA, bytearray(b'\x06\x4D\x72\xBB'), 3144830214)
          self._validate_bytes("glamor", 0x8078A01B, bytearray(b'\x89\x89\xA2\xA7'), 2812447113)
          self._validate_bytes("flags", 0x4D16CD6C, bytearray(b'\x52\x87\x66\x02'), 40273746)
          self._validate_bytes("democracy", 0x19B4FABD, bytearray(b'\xE4\x55\xD6\xB0'), 2966836708)
          self._validate_bytes("bumble", 0xE653280E, bytearray(b'\xFE\xD7\xC3\x0C'), 214161406)
          self._validate_bytes("catch", 0xB2F1555F, bytearray(b'\x98\x4B\xB6\xCD'), 3451276184)
          self._validate_bytes("omnomnomnivore", 0x7F8F82B0, bytearray(b'\x38\xC4\xCD\xFF'), 4291675192)
          self._validate_bytes("The quick brown fox jumps over the lazy dog", 0x4C2DB001, bytearray(b'\x6D\xAB\x8D\xC9'), 3381504877)
  
      def _validate_bytes(self, str, seed, expected_hash_bytes, expected_value):
          hash_value = murmur_hash._MurmurHash._ComputeHash(bytearray(str, encoding='utf-8'), seed)
          bytes = bytearray(pack('I', hash_value))
          self.assertEqual(expected_value, hash_value)
          self.assertEqual(expected_hash_bytes, bytes)
  
      def test_get_bytes(self):
          actual_bytes = consistent_hash_ring._ConsistentHashRing._GetBytes("documentdb")
          expected_bytes = bytearray(b'\x64\x6F\x63\x75\x6D\x65\x6E\x74\x64\x62')
          self.assertEqual(expected_bytes, actual_bytes)
  
          actual_bytes = consistent_hash_ring._ConsistentHashRing._GetBytes("azure")
          expected_bytes = bytearray(b'\x61\x7A\x75\x72\x65')
          self.assertEqual(expected_bytes, actual_bytes)
  
          actual_bytes = consistent_hash_ring._ConsistentHashRing._GetBytes("json")
          expected_bytes = bytearray(b'\x6A\x73\x6F\x6E')
          self.assertEqual(expected_bytes, actual_bytes)
  
          actual_bytes = consistent_hash_ring._ConsistentHashRing._GetBytes("nosql")
          expected_bytes = bytearray(b'\x6E\x6F\x73\x71\x6C')
          self.assertEqual(expected_bytes, actual_bytes)
  
      def test_range_partition_resolver(self):
          # create test database
          created_db = self.databseForTest
          
          # Create bunch of collections participating in partitioning
          collection0 = { 'id': 'coll_0' }
          collection1 = { 'id': 'coll_1' }
          collection2 = { 'id': 'coll_2' }
  
          collection_links = [self.GetDocumentCollectionLink(created_db, collection0, True), self.GetDocumentCollectionLink(created_db, collection1, True), self.GetDocumentCollectionLink(created_db, collection2, True)]
          
          val_partition_key_extractor = lambda document: document['val']
  
          ranges =[partition_range.Range(0,400), partition_range.Range(401,800), partition_range.Range(501,1200)]
  
          partition_map = dict([(ranges[0],collection_links[0]), (ranges[1],collection_links[1]), (ranges[2],collection_links[2])])
  
          rangepartition_resolver = range_partition_resolver.RangePartitionResolver(val_partition_key_extractor, partition_map)
          
          # create a document using the document definition
          document_definition = { 'id': '0',
                                  'name': 'sample document',
                                  'val': 0 }
          
          document_definition['val'] = 400
  
          collection_link = rangepartition_resolver.ResolveForCreate(document_definition)
          self.assertEquals(collection_links[0], collection_link)
  
          read_collection_links = rangepartition_resolver.ResolveForRead(600)
  
          self.assertEqual(2, len(read_collection_links))
          self.assertTrue(collection_links[1] in read_collection_links)
          self.assertTrue(collection_links[2] in read_collection_links)
  
          read_collection_links = rangepartition_resolver.ResolveForRead(partition_range.Range(250, 500))
  
          self.assertEqual(2, len(read_collection_links))
          self.assertTrue(collection_links[0] in read_collection_links)
          self.assertTrue(collection_links[1] in read_collection_links)
  
          read_collection_links = rangepartition_resolver.ResolveForRead(list([partition_range.Range(250, 500), partition_range.Range(600, 1000)]))
  
          self.assertEqual(3, len(read_collection_links))
          self.assertTrue(collection_links[0] in read_collection_links)
          self.assertTrue(collection_links[1] in read_collection_links)
          self.assertTrue(collection_links[2] in read_collection_links)
  
          read_collection_links = rangepartition_resolver.ResolveForRead(list([50, 100, 600, 1000]))
  
          self.assertEqual(3, len(read_collection_links))
          self.assertTrue(collection_links[0] in read_collection_links)
          self.assertTrue(collection_links[1] in read_collection_links)
          self.assertTrue(collection_links[2] in read_collection_links)
  
          read_collection_links = rangepartition_resolver.ResolveForRead(list([100, None]))
  
          self.assertEqual(3, len(read_collection_links))
          self.assertTrue(collection_links[0] in read_collection_links)
          self.assertTrue(collection_links[1] in read_collection_links)
          self.assertTrue(collection_links[2] in read_collection_links)
  
  
      # Upsert test for Document resource - selflink version
      def test_document_upsert_self_link(self):
          self._test_document_upsert(False)
  
      # Upsert test for Document resource - name based routing version
      def test_document_upsert_name_based(self):
          self._test_document_upsert(True)
          
      def _test_document_upsert(self, is_name_based):
          # create database
          created_db = self.databseForTest
  
          # create collection
          created_collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
  
          # read documents and check count
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
          before_create_documents_count = len(documents)
  
          # create document definition
          document_definition = {'id': 'doc',
                                 'name': 'sample document',
                                 'spam': 'eggs',
                                 'key': 'value'}
  
          # create document using Upsert API
          created_document = self.client.UpsertItem(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              document_definition)
  
          # verify id property
          self.assertEqual(created_document['id'],
                           document_definition['id'])
  
          # read documents after creation and verify updated count
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
          self.assertEqual(
              len(documents),
              before_create_documents_count + 1,
              'create should increase the number of documents')
  
          # update document
          created_document['name'] = 'replaced document'
          created_document['spam'] = 'not eggs'
          
          # should replace document since it already exists
          upserted_document = self.client.UpsertItem(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              created_document)
          
          # verify the changed properties
          self.assertEqual(upserted_document['name'],
                           created_document['name'],
                           'document id property should change')
          self.assertEqual(upserted_document['spam'],
                           created_document['spam'],
                           'property should have changed')
  
          # verify id property
          self.assertEqual(upserted_document['id'],
                           created_document['id'],
                           'document id should stay the same')
          
          # read documents after upsert and verify count doesn't increases again
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
          self.assertEqual(
              len(documents),
              before_create_documents_count + 1,
              'number of documents should remain same')
  
          created_document['id'] = 'new id'
  
          # Upsert should create new document since the id is different
          new_document = self.client.UpsertItem(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based),
              created_document)
          
          # verify id property
          self.assertEqual(created_document['id'],
                           new_document['id'],
                           'document id should be same')
          
          # read documents after upsert and verify count increases
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
          self.assertEqual(
              len(documents),
              before_create_documents_count + 2,
              'upsert should increase the number of documents')
  
          # delete documents
          self.client.DeleteItem(self.GetDocumentLink(created_db, created_collection, upserted_document, is_name_based))
          self.client.DeleteItem(self.GetDocumentLink(created_db, created_collection, new_document, is_name_based))
  
          # read documents after delete and verify count is same as original
          documents = list(self.client.ReadItems(
              self.GetDocumentCollectionLink(created_db, created_collection, is_name_based)))
          self.assertEqual(
              len(documents),
              before_create_documents_count,
              'number of documents should remain same')
          
  
      def test_spatial_index_self_link(self):
          self._test_spatial_index(False)
  
      def test_spatial_index_name_based(self):
          self._test_spatial_index(True)
          
      def _test_spatial_index(self, is_name_based):
          db = self.databseForTest
          # partial policy specified
          collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              {
                  'id': 'collection with spatial index ' + str(uuid.uuid4()),
                  'indexingPolicy': {
                      'includedPaths': [
                          {
                              'path': '/"Location"/?',
                              'indexes': [
                                  {
                                      'kind': 'Spatial',
                                      'dataType': 'Point'
                                  }
                              ]
                          },
                          {
                              'path': '/'
                          }
                      ]
                  }
              })
          self.client.CreateItem(self.GetDocumentCollectionLink(db, collection, is_name_based), {
              'id': 'loc1',
              'Location': {
                  'type': 'Point',
                  'coordinates': [ 20.0, 20.0 ]
              }
          })
          self.client.CreateItem(self.GetDocumentCollectionLink(db, collection, is_name_based), {
              'id': 'loc2',
              'Location': {
                  'type': 'Point',
                  'coordinates': [ 100.0, 100.0 ]
              }
          })
          results = list(self.client.QueryItems(
              self.GetDocumentCollectionLink(db, collection, is_name_based),
              "SELECT * FROM root WHERE (ST_DISTANCE(root.Location, {type: 'Point', coordinates: [20.1, 20]}) < 20000) "))
          self.assertEqual(1, len(results))
          self.assertEqual('loc1', results[0]['id'])
  
      
      def test_attachment_crud_self_link(self):
          self._test_attachment_crud(False)
  
      def test_attachment_crud_name_based(self):
          self._test_attachment_crud(True)
          
      def _test_attachment_crud(self, is_name_based):
          class ReadableStream(object):
              """Customized file-like stream.
              """
  
              def __init__(self, chunks = ['first chunk ', 'second chunk']):
                  """Initialization.
  
                  :Parameters:
                      - `chunks`: list
  
                  """
                  if six.PY2:
                      self._chunks = list(chunks)
                  else:
                      # python3: convert to bytes
                      self._chunks = [chunk.encode() for chunk in chunks]
  
              def read(self, n=-1):
                  """Simulates the read method in a file stream.
  
                  :Parameters:
                      - `n`: int
  
                  :Returns:
                      str or bytes
  
                  """
                  if self._chunks:
                      return self._chunks.pop(0)
                  else:
                      return ''
  
              def __len__(self):
                  """To make len(ReadableStream) work.
                  """
                  return sum([len(chunk) for chunk in self._chunks])
  
  
          # Should do attachment CRUD operations successfully
          self.client.connection_policy.MediaReadMode = documents.MediaReadMode.Buffered
  
          # create database
          db = self.databseForTest
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          # create document
          document = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                           { 'id': 'sample document',
                                             'spam': 'eggs',
                                             'key': 'value' })
          # list all attachments
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          initial_count = len(attachments)
          valid_media_options = { 'slug': 'attachment name',
                                  'contentType': 'application/text' }
          invalid_media_options = { 'slug': 'attachment name',
                                    'contentType': 'junt/test' }
          # create attachment with invalid content-type
          content_stream = ReadableStream()
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST,
              self.client.CreateAttachmentAndUploadMedia,
              self.GetDocumentLink(db, collection, document, is_name_based),
              content_stream,
              invalid_media_options)
          content_stream = ReadableStream()
          # create streamed attachment with valid content-type
          valid_attachment = self.client.CreateAttachmentAndUploadMedia(
              self.GetDocumentLink(db, collection, document, is_name_based), content_stream, valid_media_options)
          self.assertEqual(valid_attachment['id'],
                           'attachment name',
                           'id of created attachment should be the'
                           ' same as the one in the request')
          content_stream = ReadableStream()
          # create colliding attachment
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.CONFLICT,
              self.client.CreateAttachmentAndUploadMedia,
              self.GetDocumentLink(db, collection, document, is_name_based),
              content_stream,
              valid_media_options)
  
          content_stream = ReadableStream()
          # create attachment with media link
          dynamic_attachment = {
              'id': 'dynamic attachment',
              'media': 'http://xstore.',
              'MediaType': 'Book',
              'Author':'My Book Author',
              'Title':'My Book Title',
              'contentType':'application/text'
          }
          attachment = self.client.CreateAttachment(self.GetDocumentLink(db, collection, document, is_name_based),
                                               dynamic_attachment)
          self.assertEqual(attachment['MediaType'],
                           'Book',
                           'invalid media type')
          self.assertEqual(attachment['Author'],
                           'My Book Author',
                           'invalid property value')
          # list all attachments
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments),
                           initial_count + 2,
                           'number of attachments should\'ve increased by 2')
          attachment['Author'] = 'new author'
          # replace the attachment
          self.client.ReplaceAttachment(self.GetAttachmentLink(db, collection, document, attachment, is_name_based), attachment)
          self.assertEqual(attachment['MediaType'],
                           'Book',
                           'invalid media type')
          self.assertEqual(attachment['Author'],
                           'new author',
                           'invalid property value')
          # read attachment media
          media_response = self.client.ReadMedia(valid_attachment['media'])
          self.assertEqual(media_response,
                           'first chunk second chunk')
          content_stream = ReadableStream(['modified first chunk ',
                                           'modified second chunk'])
          # update attachment media
          self.client.UpdateMedia(valid_attachment['media'],
                             content_stream,
                             valid_media_options)
          # read attachment media after update
          # read media buffered
          media_response = self.client.ReadMedia(valid_attachment['media'])
          self.assertEqual(media_response,
                           'modified first chunk modified second chunk')
          # read media streamed
          self.client.connection_policy.MediaReadMode = (
              documents.MediaReadMode.Streamed)
          media_response = self.client.ReadMedia(valid_attachment['media'])
          self.assertEqual(media_response.read(),
                           b'modified first chunk modified second chunk')
          # share attachment with a second document
          document = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                           {'id': 'document 2'})
          second_attachment = {
              'id': valid_attachment['id'],
              'contentType': valid_attachment['contentType'],
              'media': valid_attachment['media'] }
          attachment = self.client.CreateAttachment(self.GetDocumentLink(db, collection, document, is_name_based),
                                               second_attachment)
          self.assertEqual(valid_attachment['id'],
                           attachment['id'],
                           'id mismatch')
          self.assertEqual(valid_attachment['media'],
                           attachment['media'],
                           'media mismatch')
          self.assertEqual(valid_attachment['contentType'],
                           attachment['contentType'],
                           'contentType mismatch')
          # deleting attachment
          self.client.DeleteAttachment(self.GetAttachmentLink(db, collection, document, attachment, is_name_based))
          # read attachments after deletion
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments), 0)
  
      # Upsert test for Attachment resource - selflink version
      def test_attachment_upsert_self_link(self):
          self._test_attachment_upsert(False)
  
      # Upsert test for Attachment resource - name based routing version
      def test_attachment_upsert_name_based(self):
          self._test_attachment_upsert(True)
          
      def _test_attachment_upsert(self, is_name_based):
          class ReadableStream(object):
              """Customized file-like stream.
              """
  
              def __init__(self, chunks = ['first chunk ', 'second chunk']):
                  """Initialization.
  
                  :Parameters:
                      - `chunks`: list
  
                  """
                  if six.PY2:
                      self._chunks = list(chunks)
                  else:
                      # python3: convert to bytes
                      self._chunks = [chunk.encode() for chunk in chunks]
  
              def read(self, n=-1):
                  """Simulates the read method in a file stream.
  
                  :Parameters:
                      - `n`: int
  
                  :Returns:
                      str or bytes
  
                  """
                  if self._chunks:
                      return self._chunks.pop(0)
                  else:
                      return ''
  
              def __len__(self):
                  """To make len(ReadableStream) work.
                  """
                  return sum([len(chunk) for chunk in self._chunks])
  
          # create database
          db = self.databseForTest
          
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          
          # create document
          document = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                           { 'id': 'sample document' + str(uuid.uuid4()),
                                             'spam': 'eggs',
                                             'key': 'value' })
          
          # list all attachments and check count
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          initial_count = len(attachments)
          
          valid_media_options = { 'slug': 'attachment name',
                                  'contentType': 'application/text' }
          content_stream = ReadableStream()
          
          # create streamed attachment with valid content-type using Upsert API
          valid_attachment = self.client.UpsertAttachmentAndUploadMedia(
              self.GetDocumentLink(db, collection, document, is_name_based), content_stream, valid_media_options)
          
          # verify id property
          self.assertEqual(valid_attachment['id'],
                           'attachment name',
                           'id of created attachment should be the same')
  
          valid_media_options = { 'slug': 'new attachment name',
                                  'contentType': 'application/text' }
          content_stream = ReadableStream()
          
          # Upsert should create new attachment since since id is different
          new_valid_attachment = self.client.UpsertAttachmentAndUploadMedia(
              self.GetDocumentLink(db, collection, document, is_name_based), content_stream, valid_media_options)
          
          # verify id property
          self.assertEqual(new_valid_attachment['id'],
                           'new attachment name',
                           'id of new attachment should be the same')
  
          # read all attachments and verify updated count
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments),
                           initial_count + 2,
                           'number of attachments should have increased by 2')
          
          # create attachment with media link
          attachment_definition = {
              'id': 'dynamic attachment',
              'media': 'http://xstore.',
              'MediaType': 'Book',
              'Author':'My Book Author',
              'Title':'My Book Title',
              'contentType':'application/text'
          }
  
          # create dynamic attachment using Upsert API
          dynamic_attachment = self.client.UpsertAttachment(self.GetDocumentLink(db, collection, document, is_name_based),
                                               attachment_definition)
  
          # verify id property
          self.assertEqual(dynamic_attachment['id'],
                           attachment_definition['id'],
                           'id of attachment should be the same')
  
          # read all attachments and verify updated count
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments),
                           initial_count + 3,
                           'number of attachments should have increased by 3')
  
          dynamic_attachment['Author'] = 'new author'
          
          # replace the attachment using Upsert API
          upserted_attachment = self.client.UpsertAttachment(self.GetDocumentLink(db, collection, document, is_name_based), dynamic_attachment)
  
          # verify id property remains same
          self.assertEqual(dynamic_attachment['id'],
                           upserted_attachment['id'],
                           'id should stay the same')
  
          # verify author property gets updated
          self.assertEqual(upserted_attachment['Author'],
                           dynamic_attachment['Author'],
                           'invalid property value')
  
          # read all attachments and verify count doesn't increases again
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments),
                           initial_count + 3,
                           'number of attachments should remain same')
  
          dynamic_attachment['id'] = 'new dynamic attachment'
          
          # Upsert should create new attachment since id is different
          new_attachment = self.client.UpsertAttachment(self.GetDocumentLink(db, collection, document, is_name_based), dynamic_attachment)
  
          # verify id property remains same
          self.assertEqual(dynamic_attachment['id'],
                           new_attachment['id'],
                           'id should be same')
  
          # read all attachments and verify count increases
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments),
                           initial_count + 4,
                           'number of attachments should have increased')
  
          # deleting attachments
          self.client.DeleteAttachment(self.GetAttachmentLink(db, collection, document, valid_attachment, is_name_based))
          self.client.DeleteAttachment(self.GetAttachmentLink(db, collection, document, new_valid_attachment, is_name_based))
          self.client.DeleteAttachment(self.GetAttachmentLink(db, collection, document, upserted_attachment, is_name_based))
          self.client.DeleteAttachment(self.GetAttachmentLink(db, collection, document, new_attachment, is_name_based))
  
          # wait to ensure deletes are propagated for multimaster enabled accounts
          if self.configs.IS_MULTIMASTER_ENABLED:
              time.sleep(2)
  
          # read all attachments and verify count remains same
          attachments = list(self.client.ReadAttachments(self.GetDocumentLink(db, collection, document, is_name_based)))
          self.assertEqual(len(attachments),
                           initial_count,
                           'number of attachments should remain the same')
          
  
      def test_user_crud_self_link(self):
          self._test_user_crud(False)
  
      def test_user_crud_name_based(self):
          self._test_user_crud(True)
          
      def _test_user_crud(self, is_name_based):
          # Should do User CRUD operations successfully.
          # create database
          db = self.databseForTest
          # list users
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          before_create_count = len(users)
          # create user
          user_id = 'new user' + str(uuid.uuid4())
          user = self.client.CreateUser(self.GetDatabaseLink(db, is_name_based), { 'id': user_id })
          self.assertEqual(user['id'], user_id, 'user id error')
          # list users after creation
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          self.assertEqual(len(users), before_create_count + 1)
          # query users
          results = list(self.client.QueryUsers(
              self.GetDatabaseLink(db, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name':'@id', 'value':user_id }
                  ]
              }))
          self.assertTrue(results)
  
          # replace user
          change_user = user.copy()
          replaced_user_id = 'replaced user' + str(uuid.uuid4())
          user['id'] = replaced_user_id
          replaced_user = self.client.ReplaceUser(self.GetUserLink(db, change_user, is_name_based), user)
          self.assertEqual(replaced_user['id'],
                           replaced_user_id,
                           'user id should change')
          self.assertEqual(user['id'],
                           replaced_user['id'],
                           'user id should stay the same')
          # read user
          user = self.client.ReadUser(self.GetUserLink(db, replaced_user, is_name_based))
          self.assertEqual(replaced_user['id'], user['id'])
          # delete user
          self.client.DeleteUser(self.GetUserLink(db, user, is_name_based))
          # read user after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadUser,
                                             self.GetUserLink(db, user, is_name_based))
  
      
      # Upsert test for User resource - selflink version
      def test_user_upsert_self_link(self):
          self._test_user_upsert(False)
  
      # Upsert test for User resource - named based routing version
      def test_user_upsert_name_based(self):
          self._test_user_upsert(True)
          
      def _test_user_upsert(self, is_name_based):
          # create database
          db = self.databseForTest
          
          # read users and check count
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          before_create_count = len(users)
          
          # create user using Upsert API
          user_id = 'user' + str(uuid.uuid4())
          user = self.client.UpsertUser(self.GetDatabaseLink(db, is_name_based), { 'id': user_id })
  
          # verify id property
          self.assertEqual(user['id'], user_id, 'user id error')
          
          # read users after creation and verify updated count
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          self.assertEqual(len(users), before_create_count + 1)
          
          # Should replace the user since it already exists, there is no public property to change here
          upserted_user = self.client.UpsertUser(self.GetDatabaseLink(db, is_name_based), user)
          
          # verify id property
          self.assertEqual(upserted_user['id'],
                           user['id'],
                           'user id should remain same')
  
          # read users after upsert and verify count doesn't increases again
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          self.assertEqual(len(users), before_create_count + 1)
  
          user['id'] = 'new user' + str(uuid.uuid4())
  
          # Upsert should create new user since id is different
          new_user = self.client.UpsertUser(self.GetDatabaseLink(db, is_name_based), user)
  
          # verify id property
          self.assertEqual(new_user['id'], user['id'], 'user id error')
          
          # read users after upsert and verify count increases
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          self.assertEqual(len(users), before_create_count + 2)
  
          # delete users
          self.client.DeleteUser(self.GetUserLink(db, upserted_user, is_name_based))
          self.client.DeleteUser(self.GetUserLink(db, new_user, is_name_based))
  
          # read users after delete and verify count remains the same
          users = list(self.client.ReadUsers(self.GetDatabaseLink(db, is_name_based)))
          self.assertEqual(len(users), before_create_count)
  
  
      def test_permission_crud_self_link(self):
          self._test_permission_crud(False)
  
      def test_permission_crud_name_based(self):
          self._test_permission_crud(True)
          
      def _test_permission_crud(self, is_name_based):
          # Should do Permission CRUD operations successfully
          # create database
          db = self.databseForTest
          # create user
          user = self.client.CreateUser(self.GetDatabaseLink(db, is_name_based), { 'id': 'new user' + str(uuid.uuid4())})
          # list permissions
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          before_create_count = len(permissions)
          permission = {
              'id': 'new permission',
              'permissionMode': documents.PermissionMode.Read,
              'resource': 'dbs/AQAAAA==/colls/AQAAAJ0fgTc='  # A random one.
          }
          # create permission
          permission = self.client.CreatePermission(self.GetUserLink(db, user, is_name_based), permission)
          self.assertEqual(permission['id'],
                           'new permission',
                           'permission id error')
          # list permissions after creation
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          self.assertEqual(len(permissions), before_create_count + 1)
          # query permissions
          results = list(self.client.QueryPermissions(
              self.GetUserLink(db, user, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name':'@id', 'value':permission['id'] }
                  ]
              }))
          self.assert_(results)
  
          # replace permission
          change_permission = permission.copy()
          permission['id'] = 'replaced permission'
          replaced_permission = self.client.ReplacePermission(self.GetPermissionLink(db, user, change_permission, is_name_based),
                                                         permission)
          self.assertEqual(replaced_permission['id'],
                           'replaced permission',
                           'permission id should change')
          self.assertEqual(permission['id'],
                           replaced_permission['id'],
                           'permission id should stay the same')
          # read permission
          permission = self.client.ReadPermission(self.GetPermissionLink(db, user, replaced_permission, is_name_based))
          self.assertEqual(replaced_permission['id'], permission['id'])
          # delete permission
          self.client.DeletePermission(self.GetPermissionLink(db, user, replaced_permission, is_name_based))
          # read permission after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadPermission,
                                             self.GetPermissionLink(db, user, permission, is_name_based))
  
      # Upsert test for Permission resource - selflink version
      def test_permission_upsert_self_link(self):
          self._test_permission_upsert(False)
  
      # Upsert test for Permission resource - name based routing version
      def test_permission_upsert_name_based(self):
          self._test_permission_upsert(True)
          
      def _test_permission_upsert(self, is_name_based):
          # create database
          db = self.databseForTest
          
          # create user
          user = self.client.CreateUser(self.GetDatabaseLink(db, is_name_based), { 'id': 'new user' + str(uuid.uuid4())})
          
          # read permissions and check count
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          before_create_count = len(permissions)
          
          permission_definition = {
              'id': 'permission',
              'permissionMode': documents.PermissionMode.Read,
              'resource': 'dbs/AQAAAA==/colls/AQAAAJ0fgTc='  # A random one.
          }
          
          # create permission using Upsert API
          created_permission = self.client.UpsertPermission(self.GetUserLink(db, user, is_name_based), permission_definition)
          
          # verify id property
          self.assertEqual(created_permission['id'],
                           permission_definition['id'],
                           'permission id error')
          
          # read permissions after creation and verify updated count
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          self.assertEqual(len(permissions), before_create_count + 1)
          
          # update permission mode
          permission_definition['permissionMode'] = documents.PermissionMode.All
  
          # should repace the permission since it already exists
          upserted_permission = self.client.UpsertPermission(self.GetUserLink(db, user, is_name_based),
                                                         permission_definition)
          # verify id property
          self.assertEqual(upserted_permission['id'],
                           created_permission['id'],
                           'permission id should remain same')
          
          # verify changed property
          self.assertEqual(upserted_permission['permissionMode'],
                           permission_definition['permissionMode'],
                           'permissionMode should change')
          
          # read permissions and verify count doesn't increases again
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          self.assertEqual(len(permissions), before_create_count + 1)
  
          # update permission id
          created_permission['id'] = 'new permission'
          # resource needs to be changed along with the id in order to create a new permission
          created_permission['resource'] = 'dbs/N9EdAA==/colls/N9EdAIugXgA='
  
          # should create new permission since id has changed
          new_permission = self.client.UpsertPermission(self.GetUserLink(db, user, is_name_based),
                                                         created_permission)
          # verify id and resource property
          self.assertEqual(new_permission['id'],
                           created_permission['id'],
                           'permission id should be same')
  
          self.assertEqual(new_permission['resource'],
                           created_permission['resource'],
                           'permission resource should be same')
          
          # read permissions and verify count increases
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          self.assertEqual(len(permissions), before_create_count + 2)
  
          # delete permissions
          self.client.DeletePermission(self.GetPermissionLink(db, user, upserted_permission, is_name_based))
          self.client.DeletePermission(self.GetPermissionLink(db, user, new_permission, is_name_based))
  
          # read permissions and verify count remains the same
          permissions = list(self.client.ReadPermissions(self.GetUserLink(db, user, is_name_based)))
          self.assertEqual(len(permissions), before_create_count)
  
      def test_authorization(self):
          def __SetupEntities(client):
              """
              Sets up entities for this test.
  
              :Parameters:
                  - `client`: cosmos_client.CosmosClient
  
              :Returns:
                  dict
  
              """
              # create database
              db = self.databseForTest
              # create collection1
              collection1 = client.CreateContainer(
                  db['_self'], { 'id': 'test_authorization ' + str(uuid.uuid4()) })
              # create document1
              document1 = client.CreateItem(collection1['_self'],
                                                { 'id': 'coll1doc1',
                                                  'spam': 'eggs',
                                                  'key': 'value' })
              # create document 2
              document2 = client.CreateItem(
                  collection1['_self'],
                  { 'id': 'coll1doc2', 'spam': 'eggs2', 'key': 'value2' })
              # create attachment
              dynamic_attachment = {
                  'id': 'dynamic attachment',
                  'media': 'http://xstore.',
                  'MediaType': 'Book',
                  'Author': 'My Book Author',
                  'Title': 'My Book Title',
                  'contentType': 'application/text'
              }
              attachment = client.CreateAttachment(document1['_self'],
                                                   dynamic_attachment)
              # create collection 2
              collection2 = client.CreateContainer(
                  db['_self'],
                  { 'id': 'test_authorization2 ' + str(uuid.uuid4()) })
              # create user1
              user1 = client.CreateUser(db['_self'], { 'id': 'user1' })
              permission = {
                  'id': 'permission On Coll1',
                  'permissionMode': documents.PermissionMode.Read,
                  'resource': collection1['_self']
              }
              # create permission for collection1
              permission_on_coll1 = client.CreatePermission(user1['_self'],
                                                            permission)
              self.assertTrue(permission_on_coll1['_token'] != None,
                              'permission token is invalid')
              permission = {
                  'id': 'permission On Doc1',
                  'permissionMode': documents.PermissionMode.All,
                  'resource': document2['_self']
              }
              # create permission for document 2
              permission_on_doc2 = client.CreatePermission(user1['_self'],
                                                           permission)
              self.assertTrue(permission_on_doc2['_token'] != None,
                              'permission token is invalid')
              # create user 2
              user2 = client.CreateUser(db['_self'], { 'id': 'user2' })
              permission = {
                  'id': 'permission On coll2',
                  'permissionMode': documents.PermissionMode.All,
                  'resource': collection2['_self']
              }
              # create permission on collection 2
              permission_on_coll2 = client.CreatePermission(
                  user2['_self'], permission)
              entities = {
                  'db': db,
                  'coll1': collection1,
                  'coll2': collection2,
                  'doc1': document1,
                  'doc2': document2,
                  'user1': user1,
                  'user2': user2,
                  'attachment': attachment,
                  'permissionOnColl1': permission_on_coll1,
                  'permissionOnDoc2': permission_on_doc2,
                  'permissionOnColl2': permission_on_coll2
              }
              return entities
  
          # Client without any authorization will fail.
          client = cosmos_client.CosmosClient(CRUDTests.host, {}, CRUDTests.connectionPolicy)
          self.__AssertHTTPFailureWithStatus(StatusCodes.UNAUTHORIZED,
                                             list,
                                             client.ReadDatabases())
          # Client with master key.
          client = cosmos_client.CosmosClient(CRUDTests.host,
                                                  {'masterKey': CRUDTests.masterKey}, CRUDTests.connectionPolicy)
          # setup entities
          entities = __SetupEntities(client)
          resource_tokens = {}
          resource_tokens[entities['coll1']['_rid']] = (
              entities['permissionOnColl1']['_token'])
          resource_tokens[entities['doc1']['_rid']] = (
              entities['permissionOnColl1']['_token'])
          col1_client = cosmos_client.CosmosClient(
              CRUDTests.host, {'resourceTokens': resource_tokens}, CRUDTests.connectionPolicy)
          # 1. Success-- Use Col1 Permission to Read
          success_coll1 = col1_client.ReadContainer(
              entities['coll1']['_self'])
          # 2. Failure-- Use Col1 Permission to delete
          self.__AssertHTTPFailureWithStatus(StatusCodes.FORBIDDEN,
                                             col1_client.DeleteContainer,
                                             success_coll1['_self'])
          # 3. Success-- Use Col1 Permission to Read All Docs
          success_documents = list(col1_client.ReadItems(
              success_coll1['_self']))
          self.assertTrue(success_documents != None,
                          'error reading documents')
          self.assertEqual(len(success_documents),
                           2,
                           'Expected 2 Documents to be succesfully read')
          # 4. Success-- Use Col1 Permission to Read Col1Doc1
          success_doc = col1_client.ReadItem(entities['doc1']['_self'])
          self.assertTrue(success_doc != None, 'error reading document')
          self.assertEqual(
              success_doc['id'],
              entities['doc1']['id'],
              'Expected to read children using parent permissions')
          col2_client = cosmos_client.CosmosClient(
              CRUDTests.host,
              { 'permissionFeed': [ entities['permissionOnColl2'] ] }, CRUDTests.connectionPolicy)
          doc = {
              'CustomProperty1': 'BBBBBB',
              'customProperty2': 1000,
              'id': entities['doc2']['id']
          }
          success_doc = col2_client.CreateItem(
              entities['coll2']['_self'], doc)
          self.assertTrue(success_doc != None, 'error creating document')
          self.assertEqual(success_doc['CustomProperty1'],
                           doc['CustomProperty1'],
                           'document should have been created successfully')
  
          self.client.DeleteContainer(entities['coll1']['_self'])
          self.client.DeleteContainer(entities['coll2']['_self'])
  
      def test_trigger_crud_self_link(self):
          self._test_trigger_crud(False)
  
      def test_trigger_crud_name_based(self):
          self._test_trigger_crud(True)
          
      def _test_trigger_crud(self, is_name_based):
          # create database
          db = self.databseForTest
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          # read triggers
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          # create a trigger
          before_create_triggers_count = len(triggers)
          trigger_definition = {
              'id': 'sample trigger',
              'serverScript': 'function() {var x = 10;}',
              'triggerType': documents.TriggerType.Pre,
              'triggerOperation': documents.TriggerOperation.All
          }
          trigger = self.client.CreateTrigger(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                         trigger_definition)
          for property in trigger_definition:
              if property != "serverScript":
                  self.assertEqual(
                      trigger[property],
                      trigger_definition[property],
                      'property {property} should match'.format(property=property))
              else:
                      self.assertEqual(trigger['body'],
                                       'function() {var x = 10;}')
  
          # read triggers after creation
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(triggers),
                           before_create_triggers_count + 1,
                           'create should increase the number of triggers')
          # query triggers
          triggers = list(self.client.QueryTriggers(
              self.GetDocumentCollectionLink(db, collection, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name': '@id', 'value': trigger_definition['id']}
                  ]
              }))
          self.assert_(triggers)
  
          # replace trigger
          change_trigger = trigger.copy()
          trigger['body'] = 'function() {var x = 20;}'
          replaced_trigger = self.client.ReplaceTrigger(self.GetTriggerLink(db, collection, change_trigger, is_name_based), trigger)
          for property in trigger_definition:
              if property != "serverScript":
                  self.assertEqual(
                      replaced_trigger[property],
                      trigger[property],
                      'property {property} should match'.format(property=property))
              else:
                  self.assertEqual(replaced_trigger['body'],
                                   'function() {var x = 20;}')
  
          # read trigger
          trigger = self.client.ReadTrigger(self.GetTriggerLink(db, collection, replaced_trigger, is_name_based))
          self.assertEqual(replaced_trigger['id'], trigger['id'])
          # delete trigger
          self.client.DeleteTrigger(self.GetTriggerLink(db, collection, replaced_trigger, is_name_based))
          # read triggers after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadTrigger,
                                             self.GetTriggerLink(db, collection, replaced_trigger, is_name_based))
  
      # Upsert test for Trigger resource - selflink version
      def test_trigger_upsert_self_link(self):
          self._test_trigger_upsert(False)
  
      # Upsert test for Trigger resource - name based routing version
      def test_trigger_upsert_name_based(self):
          self._test_trigger_upsert(True)
          
      def _test_trigger_upsert(self, is_name_based):
          # create database
          db = self.databseForTest
          
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          
          # read triggers and check count
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          before_create_triggers_count = len(triggers)
  
          # create a trigger
          trigger_definition = {
              'id': 'sample trigger',
              'serverScript': 'function() {var x = 10;}',
              'triggerType': documents.TriggerType.Pre,
              'triggerOperation': documents.TriggerOperation.All
          }
  
          # create trigger using Upsert API
          created_trigger = self.client.UpsertTrigger(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                         trigger_definition)
  
          # verify id property
          self.assertEqual(created_trigger['id'],
                           trigger_definition['id'])
  
          # read triggers after creation and verify updated count
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(triggers),
                           before_create_triggers_count + 1,
                           'create should increase the number of triggers')
          
          # update trigger
          created_trigger['body'] = 'function() {var x = 20;}'
  
          # should replace trigger since it already exists
          upserted_trigger = self.client.UpsertTrigger(self.GetDocumentCollectionLink(db, collection, is_name_based), created_trigger)
  
          # verify id property
          self.assertEqual(created_trigger['id'],
                           upserted_trigger['id'])
  
          # verify changed properties
          self.assertEqual(upserted_trigger['body'],
                                   created_trigger['body'])
  
          # read triggers after upsert and verify count remains same
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(triggers),
                           before_create_triggers_count + 1,
                           'upsert should keep the number of triggers same')
  
          # update trigger
          created_trigger['id'] = 'new trigger'
  
          # should create new trigger since id is changed
          new_trigger = self.client.UpsertTrigger(self.GetDocumentCollectionLink(db, collection, is_name_based), created_trigger)
  
          # verify id property
          self.assertEqual(created_trigger['id'],
                           new_trigger['id'])
  
          # read triggers after upsert and verify count increases
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(triggers),
                           before_create_triggers_count + 2,
                           'upsert should increase the number of triggers')
          
          # delete triggers
          self.client.DeleteTrigger(self.GetTriggerLink(db, collection, upserted_trigger, is_name_based))
          self.client.DeleteTrigger(self.GetTriggerLink(db, collection, new_trigger, is_name_based))
  
          # read triggers after delete and verify count remains the same
          triggers = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(triggers),
                           before_create_triggers_count,
                           'delete should bring the number of triggers to original')
  
  
      def test_udf_crud_self_link(self):
          self._test_udf_crud(False)
  
      def test_udf_crud_name_based(self):
          self._test_udf_crud(True)
          
      def _test_udf_crud(self, is_name_based):
          # create database
          db = self.databseForTest
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          # read udfs
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          # create a udf
          before_create_udfs_count = len(udfs)
          udf_definition = {
              'id': 'sample udf',
              'body': 'function() {var x = 10;}'
          }
          udf = self.client.CreateUserDefinedFunction(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                 udf_definition)
          for property in udf_definition:
                  self.assertEqual(
                      udf[property],
                      udf_definition[property],
                      'property {property} should match'.format(property=property))
  
          # read udfs after creation
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(udfs),
                           before_create_udfs_count + 1,
                           'create should increase the number of udfs')
          # query udfs
          results = list(self.client.QueryUserDefinedFunctions(
              self.GetDocumentCollectionLink(db, collection, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      {'name':'@id', 'value':udf_definition['id']}
                  ]
              }))
          self.assert_(results)
          # replace udf
          change_udf = udf.copy()
          udf['body'] = 'function() {var x = 20;}'
          replaced_udf = self.client.ReplaceUserDefinedFunction(self.GetUserDefinedFunctionLink(db, collection, change_udf, is_name_based), udf)
          for property in udf_definition:
                  self.assertEqual(
                      replaced_udf[property],
                      udf[property],
                      'property {property} should match'.format(property=property))
          # read udf
          udf = self.client.ReadUserDefinedFunction(self.GetUserDefinedFunctionLink(db, collection, replaced_udf, is_name_based))
          self.assertEqual(replaced_udf['id'], udf['id'])
          # delete udf
          self.client.DeleteUserDefinedFunction(self.GetUserDefinedFunctionLink(db, collection, replaced_udf, is_name_based))
          # read udfs after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadUserDefinedFunction,
                                             self.GetUserDefinedFunctionLink(db, collection, replaced_udf, is_name_based))
  
  
      # Upsert test for User Defined Function resource - selflink version
      def test_udf_upsert_self_link(self):
          self._test_udf_upsert(False)
  
      # Upsert test for User Defined Function resource - name based routing version
      def test_udf_upsert_name_based(self):
          self._test_udf_upsert(True)
          
      def _test_udf_upsert(self, is_name_based):
          # create database
          db = self.databseForTest
          
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          
          # read udfs and check count
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          before_create_udfs_count = len(udfs)
          
          # create a udf definition
          udf_definition = {
              'id': 'sample udf',
              'body': 'function() {var x = 10;}'
          }
  
          # create udf using Upsert API
          created_udf = self.client.UpsertUserDefinedFunction(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                 udf_definition)
  
          # verify id property
          self.assertEqual(created_udf['id'],
                           udf_definition['id'])
  
          # read udfs after creation and verify updated count
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(udfs),
                           before_create_udfs_count + 1,
                           'create should increase the number of udfs')
  
          # update udf
          created_udf['body'] = 'function() {var x = 20;}'
          
          # should replace udf since it already exists
          upserted_udf = self.client.UpsertUserDefinedFunction(self.GetDocumentCollectionLink(db, collection, is_name_based), created_udf)
  
          # verify id property
          self.assertEqual(created_udf['id'],
                           upserted_udf['id'])
  
          # verify changed property
          self.assertEqual(upserted_udf['body'],
                                   created_udf['body'])
  
          # read udf and verify count doesn't increases again
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(udfs),
                           before_create_udfs_count + 1,
                           'upsert should keep the number of udfs same')
  
          created_udf['id'] = 'new udf'
          
          # should create new udf since the id is different
          new_udf = self.client.UpsertUserDefinedFunction(self.GetDocumentCollectionLink(db, collection, is_name_based), created_udf)
  
          # verify id property
          self.assertEqual(created_udf['id'],
                           new_udf['id'])
          
          # read udf and verify count increases
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(udfs),
                           before_create_udfs_count + 2,
                           'upsert should keep the number of udfs same')
          
          # delete udfs
          self.client.DeleteUserDefinedFunction(self.GetUserDefinedFunctionLink(db, collection, upserted_udf, is_name_based))
          self.client.DeleteUserDefinedFunction(self.GetUserDefinedFunctionLink(db, collection, new_udf, is_name_based))
  
          # read udf and verify count remains the same
          udfs = list(self.client.ReadUserDefinedFunctions(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(udfs),
                           before_create_udfs_count,
                           'delete should keep the number of udfs same')
  
  
      def test_sproc_crud_self_link(self):
          self._test_sproc_crud(False)
  
      def test_sproc_crud_name_based(self):
          self._test_sproc_crud(True)
          
      def _test_sproc_crud(self, is_name_based):
          # create database
          db = self.databseForTest
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          # read sprocs
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          # create a sproc
          before_create_sprocs_count = len(sprocs)
          sproc_definition = {
              'id': 'sample sproc',
              'serverScript': 'function() {var x = 10;}'
          }
          sproc = self.client.CreateStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                               sproc_definition)
          for property in sproc_definition:
              if property != "serverScript":
                  self.assertEqual(
                      sproc[property],
                      sproc_definition[property],
                      'property {property} should match'.format(property=property))
              else:
                  self.assertEqual(sproc['body'], 'function() {var x = 10;}')
  
          # read sprocs after creation
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(sprocs),
                           before_create_sprocs_count + 1,
                           'create should increase the number of sprocs')
          # query sprocs
          sprocs = list(self.client.QueryStoredProcedures(
              self.GetDocumentCollectionLink(db, collection, is_name_based),
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters':[
                      { 'name':'@id', 'value':sproc_definition['id'] }
                  ]
              }))
          self.assert_(sprocs)
          # replace sproc
          change_sproc = sproc.copy()
          sproc['body'] = 'function() {var x = 20;}'
          replaced_sproc = self.client.ReplaceStoredProcedure(self.GetStoredProcedureLink(db, collection, change_sproc, is_name_based),
                                                         sproc)
          for property in sproc_definition:
              if property != 'serverScript':
                  self.assertEqual(
                      replaced_sproc[property],
                      sproc[property],
                      'property {property} should match'.format(property=property))
              else:
                  self.assertEqual(replaced_sproc['body'],
                                   "function() {var x = 20;}")
          # read sproc
          sproc = self.client.ReadStoredProcedure(self.GetStoredProcedureLink(db, collection, replaced_sproc, is_name_based))
          self.assertEqual(replaced_sproc['id'], sproc['id'])
          # delete sproc
          self.client.DeleteStoredProcedure(self.GetStoredProcedureLink(db, collection, replaced_sproc, is_name_based))
          # read sprocs after deletion
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND,
                                             self.client.ReadStoredProcedure,
                                             self.GetStoredProcedureLink(db, collection, replaced_sproc, is_name_based))
  
      # Upsert test for sproc resource - selflink version
      def test_sproc_upsert_self_link(self):
          self._test_sproc_upsert(False)
  
      # Upsert test for sproc resource - name based routing version
      def test_sproc_upsert_name_based(self):
          self._test_sproc_upsert(True)
          
      def _test_sproc_upsert(self, is_name_based):
          # create database
          db = self.databseForTest
          
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
  
          # read sprocs and check count
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          before_create_sprocs_count = len(sprocs)
  
          # create a sproc definition
          sproc_definition = {
              'id': 'sample sproc',
              'serverScript': 'function() {var x = 10;}'
          }
  
          # create sproc using Upsert API
          created_sproc = self.client.UpsertStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                               sproc_definition)
  
          # verify id property
          self.assertEqual(created_sproc['id'],
                           sproc_definition['id'])
          
          # read sprocs after creation and verify updated count
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(sprocs),
                           before_create_sprocs_count + 1,
                           'create should increase the number of sprocs')
  
          # update sproc
          created_sproc['body'] = 'function() {var x = 20;}'
  
          # should replace sproc since it already exists
          upserted_sproc = self.client.UpsertStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                         created_sproc)
  
          # verify id property
          self.assertEqual(created_sproc['id'],
                           upserted_sproc['id'])
  
          # verify changed property
          self.assertEqual(upserted_sproc['body'],
                                   created_sproc['body'])
  
          # read sprocs after upsert and verify count remains the same
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(sprocs),
                           before_create_sprocs_count + 1,
                           'upsert should keep the number of sprocs same')
  
          # update sproc
          created_sproc['id'] = 'new sproc'
  
          # should create new sproc since id is different
          new_sproc = self.client.UpsertStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                         created_sproc)
  
          # verify id property
          self.assertEqual(created_sproc['id'],
                           new_sproc['id'])
          
          # read sprocs after upsert and verify count increases
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(sprocs),
                           before_create_sprocs_count + 2,
                           'upsert should keep the number of sprocs same')
          
          # delete sprocs
          self.client.DeleteStoredProcedure(self.GetStoredProcedureLink(db, collection, upserted_sproc, is_name_based))
          self.client.DeleteStoredProcedure(self.GetStoredProcedureLink(db, collection, new_sproc, is_name_based))
  
          # read sprocs after delete and verify count remains same
          sprocs = list(self.client.ReadStoredProcedures(self.GetDocumentCollectionLink(db, collection, is_name_based)))
          self.assertEqual(len(sprocs),
                           before_create_sprocs_count,
                           'delete should keep the number of sprocs same')
  
      def test_scipt_logging_execute_stored_procedure(self):
          created_db = self.databseForTest
  
          created_collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
  
          sproc = {
              'id': 'storedProcedure' + str(uuid.uuid4()),
              'body': (
                  'function () {' +
                  '   var mytext = \'x\';' +
                  '   var myval = 1;' +
                  '   try {' +
                  '       console.log(\'The value of %s is %s.\', mytext, myval);' +
                  '       getContext().getResponse().setBody(\'Success!\');' +
                  '   }' +
                  '   catch (err) {' +
                  '       getContext().getResponse().setBody(\'inline err: [\' + err.number + \'] \' + err);' +
                  '   }'
                  '}')
              }
  
          created_sproc = self.client.CreateStoredProcedure(self.GetDocumentCollectionLink(created_db, created_collection), sproc)
  
          result = self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(created_db, created_collection, created_sproc), None)
  
          self.assertEqual(result, 'Success!')
          self.assertFalse(HttpHeaders.ScriptLogResults in self.client.last_response_headers)
  
          options = { 'enableScriptLogging': True }
          result = self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(created_db, created_collection, created_sproc), None, options)
  
          self.assertEqual(result, 'Success!')
          self.assertEqual(urllib.quote('The value of x is 1.'), self.client.last_response_headers.get(HttpHeaders.ScriptLogResults))
  
          options = { 'enableScriptLogging': False }
          result = self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(created_db, created_collection, created_sproc), None, options)
  
          self.assertEqual(result, 'Success!')
          self.assertFalse(HttpHeaders.ScriptLogResults in self.client.last_response_headers)
  
      def test_collection_indexing_policy_self_link(self):
          self._test_collection_indexing_policy(False)
  
      def test_collection_indexing_policy_name_based(self):
          self._test_collection_indexing_policy(True)
  
      def _test_collection_indexing_policy(self, is_name_based):
          # create database
          db = self.databseForTest
          # create collection
          collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              { 'id': 'test_collection_indexing_policy default policy' + str(uuid.uuid4()) })
          self.assertEqual(collection['indexingPolicy']['indexingMode'],
                           documents.IndexingMode.Consistent,
                           'default indexing mode should be consistent')
          lazy_collection_definition = {
              'id': 'test_collection_indexing_policy lazy collection ' + str(uuid.uuid4()),
              'indexingPolicy': {
                  'indexingMode': documents.IndexingMode.Lazy
              }
          }
          self.client.DeleteContainer(self.GetDocumentCollectionLink(db, collection, is_name_based))
          lazy_collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              lazy_collection_definition)
          self.assertEqual(lazy_collection['indexingPolicy']['indexingMode'],
                           documents.IndexingMode.Lazy,
                           'indexing mode should be lazy')
  
          consistent_collection_definition = {
              'id': 'test_collection_indexing_policy consistent collection ' + str(uuid.uuid4()),
              'indexingPolicy': {
                  'indexingMode': documents.IndexingMode.Consistent
              }
          }
          self.client.DeleteContainer(self.GetDocumentCollectionLink(db, lazy_collection, is_name_based))
          consistent_collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based), consistent_collection_definition)
          self.assertEqual(collection['indexingPolicy']['indexingMode'],
                           documents.IndexingMode.Consistent,
                           'indexing mode should be consistent')
          collection_definition = {
              'id': 'CollectionWithIndexingPolicy',
              'indexingPolicy': {
                  'automatic': True,
                  'indexingMode': documents.IndexingMode.Lazy,
                  'includedPaths': [
                      {
                          'path': '/',
                          'indexes': [
                              {
                                  'kind': documents.IndexKind.Hash,
                                  'dataType': documents.DataType.Number,
                                  'precision': 2
                              }
                          ]
                      }
                  ],
                  'excludedPaths': [
                      {
                          'path': '/"systemMetadata"/*'
                      }
                  ]
              }
          }
          self.client.DeleteContainer(self.GetDocumentCollectionLink(db, consistent_collection, is_name_based))
          collection_with_indexing_policy = self.client.CreateContainer(self.GetDatabaseLink(db, is_name_based), collection_definition)
          self.assertEqual(1,
                           len(collection_with_indexing_policy['indexingPolicy']['includedPaths']),
                           'Unexpected includedPaths length')
          self.assertEqual(2,
                           len(collection_with_indexing_policy['indexingPolicy']['excludedPaths']),
                           'Unexpected excluded path count')
          self.client.DeleteContainer(self.GetDocumentCollectionLink(db, collection_with_indexing_policy, is_name_based))
  
      def test_create_default_indexing_policy_self_link(self):
          self._test_create_default_indexing_policy(False)
  
      def test_create_default_indexing_policy_name_based(self):
          self._test_create_default_indexing_policy(True)
          
      def _test_create_default_indexing_policy(self, is_name_based):
          # create database
          db = self.databseForTest
  
          # no indexing policy specified
          collection = self.client.CreateContainer(self.GetDatabaseLink(db, is_name_based),
                                                   {'id': 'test_create_default_indexing_policy TestCreateDefaultPolicy01' + str(uuid.uuid4())})
          self._check_default_indexing_policy_paths(collection['indexingPolicy'])
          self.client.DeleteContainer(collection['_self'])
  
          # partial policy specified
          collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              {
                  'id': 'test_create_default_indexing_policy TestCreateDefaultPolicy02' + str(uuid.uuid4()),
                  'indexingPolicy': {
                      'indexingMode': documents.IndexingMode.Lazy, 'automatic': True
                  }
              })
          self._check_default_indexing_policy_paths(collection['indexingPolicy'])
          self.client.DeleteContainer(collection['_self'])
  
          # default policy
          collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              {
                  'id': 'test_create_default_indexing_policy TestCreateDefaultPolicy03' + str(uuid.uuid4()),
                  'indexingPolicy': { }
              })
          self._check_default_indexing_policy_paths(collection['indexingPolicy'])
          self.client.DeleteContainer(collection['_self'])
  
          # missing indexes
          collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              {
                  'id': 'test_create_default_indexing_policy TestCreateDefaultPolicy04' + str(uuid.uuid4()),
                  'indexingPolicy': {
                      'includedPaths': [
                          {
                              'path': '/*'
                          }
                      ]
                  }
              })
          self._check_default_indexing_policy_paths(collection['indexingPolicy'])
          self.client.DeleteContainer(collection['_self'])
  
          # missing precision
          collection = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based),
              {
                  'id': 'test_create_default_indexing_policy TestCreateDefaultPolicy05' + str(uuid.uuid4()),
                  'indexingPolicy': {
                      'includedPaths': [
                          {
                              'path': '/*',
                              'indexes': [
                                  {
                                      'kind': documents.IndexKind.Hash,
                                      'dataType': documents.DataType.String
                                  },
                                  {
                                      'kind': documents.IndexKind.Range,
                                      'dataType': documents.DataType.Number
                                  }
                              ]
                          }
                      ]
                  }
              })
          self._check_default_indexing_policy_paths(collection['indexingPolicy'])
          self.client.DeleteContainer(collection['_self'])
  
      def test_create_indexing_policy_with_composite_and_spatial_indexes_self_link(self):
          self._test_create_indexing_policy_with_composite_and_spatial_indexes(False)
  
      def test_create_indexing_policy_with_composite_and_spatial_indexes_name_based(self):
          self._test_create_indexing_policy_with_composite_and_spatial_indexes(True)
  
      def _test_create_indexing_policy_with_composite_and_spatial_indexes(self, is_name_based):
          # create database
          db = self.databseForTest
  
          indexing_policy = {
              "spatialIndexes": [
                  {
                      "path": "/path0/*",
                      "types": [
                          "Point",
                          "LineString",
                          "Polygon"
                      ]
                  },
                  {
                      "path": "/path1/*",
                      "types": [
                          "LineString",
                          "Polygon",
                          "MultiPolygon"
                      ]
                  }
              ],
              "compositeIndexes": [
                  [
                      {
                          "path": "/path1",
                          "order": "ascending"
                      },
                      {
                          "path": "/path2",
                          "order": "descending"
                      },
                      {
                          "path": "/path3",
                          "order": "ascending"
                      }
                  ],
                  [
                      {
                          "path": "/path4",
                          "order": "ascending"
                      },
                      {
                          "path": "/path5",
                          "order": "descending"
                      },
                      {
                          "path": "/path6",
                          "order": "ascending"
                      }
                  ]
              ]
          }
  
          container_id = 'composite_index_spatial_index' + str(uuid.uuid4())
          container_definition = {'id': container_id, 'indexingPolicy': indexing_policy}
          created_container = self.client.CreateContainer(self.GetDatabaseLink(db, is_name_based), container_definition)
          read_indexing_policy = created_container['indexingPolicy']
          self.assertListEqual(indexing_policy['spatialIndexes'], read_indexing_policy['spatialIndexes'])
          self.assertListEqual(indexing_policy['compositeIndexes'], read_indexing_policy['compositeIndexes'])
          self.client.DeleteContainer(created_container['_self'])
  
      def _check_default_indexing_policy_paths(self, indexing_policy):
          def __get_first(array):
              if array:
                  return array[0]
              else:
                  return None
  
          # '/_etag' is present in excluded paths by default
          self.assertEqual(1, len(indexing_policy['excludedPaths']))
          # included paths should be 1: '/'.
          self.assertEqual(1, len(indexing_policy['includedPaths']))
  
          root_included_path = __get_first([included_path for included_path in indexing_policy['includedPaths']
                                if included_path['path'] == '/*'])
          self.assertEqual(0, len(root_included_path['indexes']))
          print(root_included_path['indexes'])
  
      def test_client_request_timeout(self):
          connection_policy = documents.ConnectionPolicy()
          # making timeout 0 ms to make sure it will throw
          connection_policy.RequestTimeout = 0
          with self.assertRaises(Exception):
              # client does a getDatabaseAccount on initialization, which will time out
              cosmos_client.CosmosClient(CRUDTests.host,
                                                  {'masterKey': CRUDTests.masterKey},
                                                  connection_policy)
  
      def test_query_iterable_functionality(self):
          def __CreateResources(client):
              """Creates resources for this test.
  
              :Parameters:
                  - `client`: cosmos_client.CosmosClient
  
              :Returns:
                  dict
  
              """
              db = self.databseForTest
              collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
              doc1 = client.CreateItem(
                  collection['_self'],
                  { 'id': 'doc1', 'prop1': 'value1'})
              doc2 = client.CreateItem(
                  collection['_self'],
                  { 'id': 'doc2', 'prop1': 'value2'})
              doc3 = client.CreateItem(
                  collection['_self'],
                  { 'id': 'doc3', 'prop1': 'value3'})
              resources = {
                  'coll': collection,
                  'doc1': doc1,
                  'doc2': doc2,
                  'doc3': doc3
              }
              return resources
  
          # Validate QueryIterable by converting it to a list.
          resources = __CreateResources(self.client)
          results = self.client.ReadItems(resources['coll']['_self'],
                                         {'maxItemCount':2})
          docs = list(iter(results))
          self.assertEqual(3,
                           len(docs),
                           'QueryIterable should return all documents' +
                           ' using continuation')
          self.assertEqual(resources['doc1']['id'], docs[0]['id'])
          self.assertEqual(resources['doc2']['id'], docs[1]['id'])
          self.assertEqual(resources['doc3']['id'], docs[2]['id'])
  
          # Validate QueryIterable iterator with 'for'.
          counter = 0
          # test QueryIterable with 'for'.
          for doc in iter(results):
              counter += 1
              if counter == 1:
                  self.assertEqual(resources['doc1']['id'],
                                   doc['id'],
                                   'first document should be doc1')
              elif counter == 2:
                  self.assertEqual(resources['doc2']['id'],
                                   doc['id'],
                                   'second document should be doc2')
              elif counter == 3:
                  self.assertEqual(resources['doc3']['id'],
                                   doc['id'],
                                   'third document should be doc3')
          self.assertEqual(counter, 3)
  
          # Get query results page by page.
          results = self.client.ReadItems(resources['coll']['_self'],
                                         {'maxItemCount':2})
          first_block = results.fetch_next_block()
          self.assertEqual(2,
                           len(first_block),
                           'First block should have 2 entries.')
          self.assertEqual(resources['doc1']['id'], first_block[0]['id'])
          self.assertEqual(resources['doc2']['id'], first_block[1]['id'])
          self.assertEqual(1,
                           len(results.fetch_next_block()),
                           'Second block should have 1 entry.')
          self.assertEqual(0,
                           len(results.fetch_next_block()),
                           'Then its empty.')
  
      def test_trigger_functionality_self_link(self):
          self._test_trigger_functionality(False)
  
      def test_trigger_functionality_name_based(self):
          self._test_trigger_functionality(True)
          
      def _test_trigger_functionality(self, is_name_based):
          triggers_in_collection1 = [
          {
              'id': 't1',
              'body': (
                  'function() {' +
                  '    var item = getContext().getRequest().getBody();' +
                  '    item.id = item.id.toUpperCase() + \'t1\';' +
                  '    getContext().getRequest().setBody(item);' +
                  '}'),
              'triggerType': documents.TriggerType.Pre,
              'triggerOperation': documents.TriggerOperation.All
          },
          {
              'id': 'response1',
              'body': (
                  'function() {' +
                  '    var prebody = getContext().getRequest().getBody();' +
                  '    if (prebody.id != \'TESTING POST TRIGGERt1\')'
                  '        throw \'id mismatch\';' +
                  '    var postbody = getContext().getResponse().getBody();' +
                  '    if (postbody.id != \'TESTING POST TRIGGERt1\')'
                  '        throw \'id mismatch\';'
                  '}'),
              'triggerType': documents.TriggerType.Post,
              'triggerOperation': documents.TriggerOperation.All
          },
          {
              'id': 'response2',
              # can't be used because setValue is currently disabled
              'body': (
                  'function() {' +
                  '    var predoc = getContext().getRequest().getBody();' +
                  '    var postdoc = getContext().getResponse().getBody();' +
                  '    getContext().getResponse().setValue(' +
                  '        \'predocname\', predoc.id + \'response2\');' +
                  '    getContext().getResponse().setValue(' +
                  '        \'postdocname\', postdoc.id + \'response2\');' +
                  '}'),
                  'triggerType': documents.TriggerType.Post,
                  'triggerOperation': documents.TriggerOperation.All,
          }]
          triggers_in_collection2 = [
          {
              'id': "t2",
              'body': "function() { }", # trigger already stringified
              'triggerType': documents.TriggerType.Pre,
              'triggerOperation': documents.TriggerOperation.All
          },
          {
              'id': "t3",
              'body': (
                  'function() {' +
                  '    var item = getContext().getRequest().getBody();' +
                  '    item.id = item.id.toLowerCase() + \'t3\';' +
                  '    getContext().getRequest().setBody(item);' +
                  '}'),
              'triggerType': documents.TriggerType.Pre,
              'triggerOperation': documents.TriggerOperation.All
          }]
          triggers_in_collection3 = [
          {
              'id': 'triggerOpType',
              'body': 'function() { }',
              'triggerType': documents.TriggerType.Post,
              'triggerOperation': documents.TriggerOperation.Delete,
          }]
  
          def __CreateTriggers(client, database, collection, triggers, is_name_based):
              """Creates triggers.
  
              :Parameters:
                  - `client`: cosmos_client.CosmosClient
                  - `collection`: dict
  
              """
              for trigger_i in triggers:
                  trigger = client.CreateTrigger(self.GetDocumentCollectionLink(database, collection, is_name_based),
                                                 trigger_i)
                  for property in trigger_i:
                      self.assertEqual(
                          trigger[property],
                          trigger_i[property],
                          'property {property} should match'.format(property=property))
  
          # create database
          db = self.databseForTest
          # create collections
          collection1 = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based), { 'id': 'test_trigger_functionality 1 ' + str(uuid.uuid4()) })
          collection2 = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based), { 'id': 'test_trigger_functionality 2 ' + str(uuid.uuid4()) })
          collection3 = self.client.CreateContainer(
              self.GetDatabaseLink(db, is_name_based), { 'id': 'test_trigger_functionality 3 ' + str(uuid.uuid4()) })
          # create triggers
          __CreateTriggers(self.client, db, collection1, triggers_in_collection1, is_name_based)
          __CreateTriggers(self.client, db, collection2, triggers_in_collection2, is_name_based)
          __CreateTriggers(self.client, db, collection3, triggers_in_collection3, is_name_based)
          # create document
          triggers_1 = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection1, is_name_based)))
          self.assertEqual(len(triggers_1), 3)
          document_1_1 = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection1, is_name_based),
                                               { 'id': 'doc1',
                                                 'key': 'value' },
                                               { 'preTriggerInclude': 't1' })
          self.assertEqual(document_1_1['id'],
                           'DOC1t1',
                           'id should be capitalized')
          document_1_2 = self.client.CreateItem(
              self.GetDocumentCollectionLink(db, collection1, is_name_based),
              { 'id': 'testing post trigger' },
              { 'postTriggerInclude': 'response1',
                'preTriggerInclude': 't1' })
          self.assertEqual(document_1_2['id'], 'TESTING POST TRIGGERt1')
          document_1_3 = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection1, is_name_based),
                                               { 'id': 'responseheaders' },
                                               { 'preTriggerInclude': 't1' })
          self.assertEqual(document_1_3['id'], "RESPONSEHEADERSt1")
  
          triggers_2 = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection2, is_name_based)))
          self.assertEqual(len(triggers_2), 2)
          document_2_1 = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection2, is_name_based),
                                               { 'id': 'doc2',
                                                 'key2': 'value2' },
                                               { 'preTriggerInclude': 't2' })
          self.assertEqual(document_2_1['id'],
                           'doc2',
                           'id shouldn\'t change')
          document_2_2 = self.client.CreateItem(self.GetDocumentCollectionLink(db, collection2, is_name_based),
                                               { 'id': 'Doc3',
                                                 'prop': 'empty' },
                                               { 'preTriggerInclude': 't3' })
          self.assertEqual(document_2_2['id'], 'doc3t3')
  
          triggers_3 = list(self.client.ReadTriggers(self.GetDocumentCollectionLink(db, collection3, is_name_based)))
          self.assertEqual(len(triggers_3), 1)
          with self.assertRaises(Exception):
              self.client.CreateItem(self.GetDocumentCollectionLink(db, collection3, is_name_based),
                                    { 'id': 'Docoptype' },
                                    { 'postTriggerInclude': 'triggerOpType' })
  
          self.client.DeleteContainer(collection1['_self'])
          self.client.DeleteContainer(collection2['_self'])
          self.client.DeleteContainer(collection3['_self'])
  
      def test_stored_procedure_functionality_self_link(self):
          self._test_stored_procedure_functionality(False)
  
      def test_stored_procedure_functionality_name_based(self):
          self._test_stored_procedure_functionality(True)
          
      def _test_stored_procedure_functionality(self, is_name_based):
          # create database
          db = self.databseForTest
          # create collection
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
  
          sproc1 = {
              'id': 'storedProcedure1' + str(uuid.uuid4()),
              'body': (
                  'function () {' +
                  '  for (var i = 0; i < 1000; i++) {' +
                  '    var item = getContext().getResponse().getBody();' +
                  '    if (i > 0 && item != i - 1) throw \'body mismatch\';' +
                  '    getContext().getResponse().setBody(i);' +
                  '  }' +
                  '}')
          }
  
          retrieved_sproc = self.client.CreateStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                         sproc1)
          result = self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(db, collection, retrieved_sproc, is_name_based),
                                                 None)
          self.assertEqual(result, 999)
          sproc2 = {
              'id': 'storedProcedure2' + str(uuid.uuid4()),
              'body': (
                  'function () {' +
                  '  for (var i = 0; i < 10; i++) {' +
                  '    getContext().getResponse().appendValue(\'Body\', i);' +
                  '  }' +
                  '}')
          }
          retrieved_sproc2 = self.client.CreateStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                          sproc2)
          result = self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(db, collection, retrieved_sproc2, is_name_based),
                                                 None)
          self.assertEqual(int(result), 123456789)
          sproc3 = {
              'id': 'storedProcedure3' + str(uuid.uuid4()),
              'body': (
                  'function (input) {' +
                      '  getContext().getResponse().setBody(' +
                      '      \'a\' + input.temp);' +
                  '}')
          }
          retrieved_sproc3 = self.client.CreateStoredProcedure(self.GetDocumentCollectionLink(db, collection, is_name_based),
                                                          sproc3)
          result = self.client.ExecuteStoredProcedure(self.GetStoredProcedureLink(db, collection, retrieved_sproc3, is_name_based),
                                                 {'temp': 'so'})
          self.assertEqual(result, 'aso')
  
      def __ValidateOfferResponseBody(self, offer, expected_coll_link, expected_offer_type):
          self.assert_(offer.get('id'), 'Id cannot be null.')
          self.assert_(offer.get('_rid'), 'Resource Id (Rid) cannot be null.')
          self.assert_(offer.get('_self'), 'Self Link cannot be null.')
          self.assert_(offer.get('resource'), 'Resource Link cannot be null.')
          self.assertTrue(offer['_self'].find(offer['id']) != -1,
                          'Offer id not contained in offer self link.')
          self.assertEqual(expected_coll_link.strip('/'), offer['resource'].strip('/'))
          if (expected_offer_type):
              self.assertEqual(expected_offer_type, offer.get('offerType'))
  
      def test_offer_read_and_query(self):
          # Create database.
          db = self.databseForTest
  
          offers = list(self.client.ReadOffers())
          initial_count = len(offers)
  
          # Create collection.
          collection = self.client.CreateContainer(db['_self'], { 'id': 'test_offer_read_and_query ' + str(uuid.uuid4()) })
          offers = list(self.client.ReadOffers())
          self.assertEqual(initial_count+1, len(offers))
  
          offers = self.GetCollectionOffers(self.client, collection['_rid'])
          
          self.assertEqual(1, len(offers))
          expected_offer = offers[0]
          self.__ValidateOfferResponseBody(expected_offer, collection.get('_self'), None)
          # Read the offer.
          read_offer = self.client.ReadOffer(expected_offer.get('_self'))
          self.__ValidateOfferResponseBody(read_offer, collection.get('_self'), expected_offer.get('offerType'))
          # Check if the read resource is what we expected.
          self.assertEqual(expected_offer.get('id'), read_offer.get('id'))
          self.assertEqual(expected_offer.get('_rid'), read_offer.get('_rid'))
          self.assertEqual(expected_offer.get('_self'), read_offer.get('_self'))
          self.assertEqual(expected_offer.get('resource'), read_offer.get('resource'))
          # Query for the offer.
  
          offers = list(self.client.QueryOffers(
              {
                  'query': 'SELECT * FROM root r WHERE r.id=@id',
                  'parameters': [
                      { 'name': '@id', 'value': expected_offer['id']}
                  ]
              }))
          
          self.assertEqual(1, len(offers))
          query_one_offer = offers[0]
          self.__ValidateOfferResponseBody(query_one_offer, collection.get('_self'), expected_offer.get('offerType'))
          # Check if the query result is what we expected.
          self.assertEqual(expected_offer.get('id'), query_one_offer.get('id'))
          self.assertEqual(expected_offer.get('_rid'), query_one_offer.get('_rid'))
          self.assertEqual(expected_offer.get('_self'), query_one_offer.get('_self'))
          self.assertEqual(expected_offer.get('resource'), query_one_offer.get('resource'))
          # Expects an exception when reading offer with bad offer link.
          self.__AssertHTTPFailureWithStatus(StatusCodes.BAD_REQUEST, self.client.ReadOffer, expected_offer.get('_self')[:-1] + 'x')
          # Now delete the collection.
          self.client.DeleteContainer(collection.get('_self'))
          # Reading fails.
          self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, self.client.ReadOffer, expected_offer.get('_self'))
          # Read feed now returns 0 results.
          offers = list(self.client.ReadOffers())
          self.assertEqual(initial_count, len(offers))
  
      def test_offer_replace(self):
          # Create database.
          db = self.databseForTest
          # Create collection.
          collection = self.configs.create_single_partition_collection_if_not_exist(self.client)
          offers = self.GetCollectionOffers(self.client, collection['_rid'])
          self.assertEqual(1, len(offers))
          expected_offer = offers[0]
          self.__ValidateOfferResponseBody(expected_offer, collection.get('_self'), None)
          # Replace the offer.
          offer_to_replace = dict(expected_offer)
          offer_to_replace['content']['offerThroughput'] += 100
          replaced_offer = self.client.ReplaceOffer(offer_to_replace['_self'], offer_to_replace)
          self.__ValidateOfferResponseBody(replaced_offer, collection.get('_self'), None)
          # Check if the replaced offer is what we expect.
          self.assertEqual(offer_to_replace.get('id'), replaced_offer.get('id'))
          self.assertEqual(offer_to_replace.get('_rid'), replaced_offer.get('_rid'))
          self.assertEqual(offer_to_replace.get('_self'), replaced_offer.get('_self'))
          self.assertEqual(offer_to_replace.get('resource'), replaced_offer.get('resource'))
          self.assertEqual(offer_to_replace.get('content').get('offerThroughput'), replaced_offer.get('content').get('offerThroughput'))
          # Expects an exception when replacing an offer with bad id.
          offer_to_replace_bad_id = dict(offer_to_replace)
          offer_to_replace_bad_id['_rid'] = 'NotAllowed'
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST, self.client.ReplaceOffer, offer_to_replace_bad_id['_self'], offer_to_replace_bad_id)
          # Expects an exception when replacing an offer with bad rid.
          offer_to_replace_bad_rid = dict(offer_to_replace)
          offer_to_replace_bad_rid['_rid'] = 'InvalidRid'
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST, self.client.ReplaceOffer, offer_to_replace_bad_rid['_self'], offer_to_replace_bad_rid)
          # Expects an exception when replaceing an offer with null id and rid.
          offer_to_replace_null_ids = dict(offer_to_replace)
          offer_to_replace_null_ids['id'] = None
          offer_to_replace_null_ids['_rid'] = None
          self.__AssertHTTPFailureWithStatus(
              StatusCodes.BAD_REQUEST, self.client.ReplaceOffer, offer_to_replace_null_ids['_self'], offer_to_replace_null_ids)
  
      def test_collection_with_offer_type(self):
          # create database
          created_db = self.databseForTest
  
          # create a collection
          offers = list(self.client.ReadOffers())
          before_offers_count = len(offers)
          
          collection_definition = { 'id': 'test_collection_with_offer_type ' + str(uuid.uuid4()) }
          collection = self.client.CreateContainer(created_db['_self'],
                                               collection_definition,
                                               {
                                                   'offerType': 'S2'
                                               })
          offers = list(self.client.ReadOffers())
          self.assertEqual(before_offers_count + 1, len(offers))
  
          offers = self.GetCollectionOffers(self.client, collection['_rid'])
          
          self.assertEqual(1, len(offers))
          expected_offer = offers[0]
  
          # We should have an offer of type S2.
          self.__ValidateOfferResponseBody(expected_offer, collection.get('_self'), 'S2')
          self.client.DeleteContainer(collection['_self'])
  
      def test_database_account_functionality(self):
          # Validate database account functionality.
          database_account = self.client.GetDatabaseAccount()
          self.assertEqual(database_account.DatabasesLink, '/dbs/')
          self.assertEqual(database_account.MediaLink, '/media/')
          if (HttpHeaders.MaxMediaStorageUsageInMB in
              self.client.last_response_headers):
              self.assertEqual(
                  database_account.MaxMediaStorageUsageInMB,
                  self.client.last_response_headers[
                      HttpHeaders.MaxMediaStorageUsageInMB])
          if (HttpHeaders.CurrentMediaStorageUsageInMB in
              self.client.last_response_headers):
              self.assertEqual(
                  database_account.CurrentMediaStorageUsageInMB,
                  self.client.last_response_headers[
                      HttpHeaders.
                      CurrentMediaStorageUsageInMB])
          self.assertTrue(
              database_account.ConsistencyPolicy['defaultConsistencyLevel']
              != None)
  
      def test_index_progress_headers_self_link(self):
          self._test_index_progress_headers(False)
  
      def test_index_progress_headers_name_based(self):
          self._test_index_progress_headers(True)
          
      def _test_index_progress_headers(self, is_name_based):
          created_db = self.databseForTest
          consistent_coll = self.client.CreateContainer(self.GetDatabaseLink(created_db, is_name_based), { 'id': 'test_index_progress_headers consistent_coll ' + str(uuid.uuid4()) })
          self.client.ReadContainer(self.GetDocumentCollectionLink(created_db, consistent_coll, is_name_based))
          self.assertFalse(HttpHeaders.LazyIndexingProgress in self.client.last_response_headers)
          self.assertTrue(HttpHeaders.IndexTransformationProgress in self.client.last_response_headers)
          lazy_coll = self.client.CreateContainer(self.GetDatabaseLink(created_db, is_name_based),
              {
                  'id': 'test_index_progress_headers lazy_coll ' + str(uuid.uuid4()),
                  'indexingPolicy': { 'indexingMode' : documents.IndexingMode.Lazy }
              })
          self.client.ReadContainer(self.GetDocumentCollectionLink(created_db, lazy_coll, is_name_based))
          self.assertTrue(HttpHeaders.LazyIndexingProgress in self.client.last_response_headers)
          self.assertTrue(HttpHeaders.IndexTransformationProgress in self.client.last_response_headers)
          none_coll = self.client.CreateContainer(self.GetDatabaseLink(created_db, is_name_based),
              {
                  'id': 'test_index_progress_headers none_coll ' + str(uuid.uuid4()),
                  'indexingPolicy': { 'indexingMode': documents.IndexingMode.NoIndex, 'automatic': False }
              })
          self.client.ReadContainer(self.GetDocumentCollectionLink(created_db, none_coll, is_name_based))
          self.assertFalse(HttpHeaders.LazyIndexingProgress in self.client.last_response_headers)
          self.assertTrue(HttpHeaders.IndexTransformationProgress in self.client.last_response_headers)
  
          self.client.DeleteContainer(consistent_coll['_self'])
          self.client.DeleteContainer(lazy_coll['_self'])
          self.client.DeleteContainer(none_coll['_self'])
  
      # To run this test, please provide your own CA certs file or download one from
      #     http://curl.haxx.se/docs/caextract.html
      #
      # def test_ssl_connection(self):
      #     connection_policy = documents.ConnectionPolicy()
      #     connection_policy.SSLConfiguration = documents.SSLConfiguration()
      #     connection_policy.SSLConfiguration.SSLCaCerts = './cacert.pem'
      #     client = cosmos_client.CosmosClient(CRUDTests.host, {'masterKey': CRUDTests.masterKey}, connection_policy)
      #     # Read databases after creation.
      #     databases = list(client.ReadDatabases())
  
      def test_id_validation(self):
          # Id shouldn't end with space.
          database_definition = { 'id': 'id_with_space ' }
          try:
              self.client.CreateDatabase(database_definition)
              self.assertFalse(True)
          except ValueError as e:
              self.assertEqual('Id ends with a space.', e.args[0])
          # Id shouldn't contain '/'.
          database_definition = { 'id': 'id_with_illegal/_char' }
          try:
              self.client.CreateDatabase(database_definition)
              self.assertFalse(True)
          except ValueError as e:
              self.assertEqual('Id contains illegal chars.', e.args[0])
          # Id shouldn't contain '\\'.
          database_definition = { 'id': 'id_with_illegal\\_char' }
          try:
              self.client.CreateDatabase(database_definition)
              self.assertFalse(True)
          except ValueError as e:
              self.assertEqual('Id contains illegal chars.', e.args[0])
          # Id shouldn't contain '?'.
          database_definition = { 'id': 'id_with_illegal?_char' }
          try:
              self.client.CreateDatabase(database_definition)
              self.assertFalse(True)
          except ValueError as e:
              self.assertEqual('Id contains illegal chars.', e.args[0])
          # Id shouldn't contain '#'.
          database_definition = { 'id': 'id_with_illegal#_char' }
          try:
              self.client.CreateDatabase(database_definition)
              self.assertFalse(True)
          except ValueError as e:
              self.assertEqual('Id contains illegal chars.', e.args[0])
  
          # Id can begin with space
          database_definition = { 'id': ' id_begin_space' }
          db = self.client.CreateDatabase(database_definition)
          self.assertTrue(True)
  
          self.client.DeleteDatabase(db['_self'])
  
      def test_id_case_validation(self):
          # create database
          created_db = self.databseForTest
  
          uuid_string = str(uuid.uuid4())
          # pascalCase
          collection_definition1 = { 'id': 'sampleCollection ' + uuid_string }
  
          # CamelCase
          collection_definition2 = { 'id': 'SampleCollection ' + uuid_string }
  
          # Verify that no collections exist
          collections = list(self.client.ReadContainers(self.GetDatabaseLink(created_db, True)))
          number_of_existing_collections = len(collections)
          
          # create 2 collections with different casing of IDs
          created_collection1 = self.client.CreateContainer(self.GetDatabaseLink(created_db, True),
                                                       collection_definition1)
  
          created_collection2 = self.client.CreateContainer(self.GetDatabaseLink(created_db, True),
                                                       collection_definition2)
  
          collections = list(self.client.ReadContainers(self.GetDatabaseLink(created_db, True)))
          
          # verify if a total of 2 collections got created
          self.assertEqual(len(collections), number_of_existing_collections + 2)
          
          # verify that collections are created with specified IDs
          self.assertEqual(collection_definition1['id'], created_collection1['id'])
          self.assertEqual(collection_definition2['id'], created_collection2['id'])
  
          self.client.DeleteContainer(created_collection1['_self'])
          self.client.DeleteContainer(created_collection2['_self'])
  
      def test_id_unicode_validation(self):
          # create database
          created_db = self.databseForTest
  
          # unicode chars in Hindi for Id which translates to: "Hindi is the national language of India"
          collection_definition1 = { 'id': u'हिन्दी भारत की राष्ट्रीय भाषा है' }
  
          # Special chars for Id
          collection_definition2 = { 'id': "!@$%^&*()-~`'_[]{}|;:,.<>" } 
  
          # verify that collections are created with specified IDs
          created_collection1 = self.client.CreateContainer(self.GetDatabaseLink(created_db, True),
                                                       collection_definition1)
  
          created_collection2 = self.client.CreateContainer(self.GetDatabaseLink(created_db, True),
                                                       collection_definition2)
          
          self.assertEqual(collection_definition1['id'], created_collection1['id'])
          self.assertEqual(collection_definition2['id'], created_collection2['id'])
  
          self.client.DeleteContainer(created_collection1['_self'])
          self.client.DeleteContainer(created_collection2['_self'])
  
      def GetDatabaseLink(self, database, is_name_based=True):
          if is_name_based:
              return 'dbs/' + database['id']
          else:
              return database['_self']
  
      def GetUserLink(self, database, user, is_name_based=True):
          if is_name_based:
              return self.GetDatabaseLink(database) + '/users/' + user['id']
          else:
              return user['_self']
  
      def GetPermissionLink(self, database, user, permission, is_name_based=True):
          if is_name_based:
              return self.GetUserLink(database, user) + '/permissions/' + permission['id']
          else:
              return permission['_self']
  
      def GetDocumentCollectionLink(self, database, document_collection, is_name_based=True):
          if is_name_based:
              return self.GetDatabaseLink(database) + '/colls/' + document_collection['id']
          else:
              return document_collection['_self']
  
      def GetDocumentLink(self, database, document_collection, document, is_name_based=True):
          if is_name_based:
              return self.GetDocumentCollectionLink(database, document_collection) + '/docs/' + document['id']
          else:
              return document['_self']
  
      def GetAttachmentLink(self, database, document_collection, document, attachment, is_name_based=True):
          if is_name_based:
              return self.GetDocumentLink(database, document_collection, document) + '/attachments/' + attachment['id']
          else:
              return attachment['_self']
  
      def GetTriggerLink(self, database, document_collection, trigger, is_name_based=True):
          if is_name_based:
              return self.GetDocumentCollectionLink(database, document_collection) + '/triggers/' + trigger['id']
          else:
              return trigger['_self']
  
      def GetUserDefinedFunctionLink(self, database, document_collection, user_defined_function, is_name_based=True):
          if is_name_based:
              return self.GetDocumentCollectionLink(database, document_collection) + '/udfs/' + user_defined_function['id']
          else:
              return user_defined_function['_self']
  
      def GetStoredProcedureLink(self, database, document_collection, stored_procedure, is_name_based=True):
          if is_name_based:
              return self.GetDocumentCollectionLink(database, document_collection) + '/sprocs/' + stored_procedure['id']
          else:
              return stored_procedure['_self']
  
      def GetConflictLink(self, database, document_collection, conflict, is_name_based=True):
          if is_name_based:
              return self.GetDocumentCollectionLink(database, document_collection) + '/conflicts/' + conflict['id']
          else:
              return conflict['_self']
  
      def GetCollectionOffers(self, client, collection_rid):
          return list(client.QueryOffers(
              {
                  'query': 'SELECT * FROM root r WHERE r.offerResourceId=@offerResourceId',
                  'parameters': [
                      { 'name': '@offerResourceId', 'value': collection_rid}
                  ]
              }))
  
  if __name__ == '__main__':
      try:
          unittest.main()
      except SystemExit as inst:
          if inst.args[0] is True:  # raised by sys.exit(True) when tests failed
              raise
  