# -*- coding: utf-8 -*- # The MIT License (MIT) # Copyright (c) Microsoft Corporation. All rights reserved. """End-to-end test. """ import json import os.path import time import unittest import urllib.parse as urllib import uuid import pytest import requests from azure.core import MatchConditions from azure.core.exceptions import AzureError, ServiceResponseError from azure.core.pipeline.transport import RequestsTransport, RequestsTransportResponse from urllib3.util.retry import Retry import azure.cosmos._base as base import azure.cosmos.cosmos_client as cosmos_client import azure.cosmos.documents as documents import azure.cosmos.exceptions as exceptions import test_config from azure.cosmos import _retry_utility from azure.cosmos.http_constants import HttpHeaders, StatusCodes from azure.cosmos.partition_key import PartitionKey class TimeoutTransport(RequestsTransport): def __init__(self, response): self._response = response super(TimeoutTransport, self).__init__() def send(self, *args, **kwargs): if kwargs.pop("passthrough", False): return super(TimeoutTransport, self).send(*args, **kwargs) time.sleep(5) if isinstance(self._response, Exception): raise self._response output = requests.Response() output.status_code = self._response response = RequestsTransportResponse(None, output) return response @pytest.mark.cosmosLong class TestCRUDContainerOperations(unittest.TestCase): """Python CRUD Tests. """ configs = test_config.TestConfig host = configs.host masterKey = configs.masterKey connectionPolicy = configs.connectionPolicy last_headers = [] client: cosmos_client.CosmosClient = None 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 exceptions.CosmosHttpResponseError 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, cls.masterKey) cls.databaseForTest = cls.client.get_database_client(cls.configs.TEST_DATABASE_ID) def test_collection_crud(self): created_db = self.databaseForTest collections = list(created_db.list_containers()) # create a collection before_create_collections_count = len(collections) collection_id = 'test_collection_crud ' + str(uuid.uuid4()) collection_indexing_policy = {'indexingMode': 'consistent'} created_collection = created_db.create_container(id=collection_id, indexing_policy=collection_indexing_policy, partition_key=PartitionKey(path="/pk", kind="Hash")) self.assertEqual(collection_id, created_collection.id) created_properties = created_collection.read() self.assertEqual('consistent', created_properties['indexingPolicy']['indexingMode']) self.assertDictEqual(PartitionKey(path='/pk', kind='Hash'), created_properties['partitionKey']) # read collections after creation collections = list(created_db.list_containers()) self.assertEqual(len(collections), before_create_collections_count + 1, 'create should increase the number of collections') # query collections collections = list(created_db.query_containers( { 'query': 'SELECT * FROM root r WHERE r.id=@id', 'parameters': [ {'name': '@id', 'value': collection_id} ] })) self.assertTrue(collections) # delete collection created_db.delete_container(created_collection.id) # read collection after deletion created_container = created_db.get_container_client(created_collection.id) self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, created_container.read) def test_partitioned_collection(self): created_db = self.databaseForTest collection_definition = {'id': 'test_partitioned_collection ' + str(uuid.uuid4()), 'partitionKey': { 'paths': ['/id'], 'kind': documents.PartitionKind.Hash } } offer_throughput = 10100 created_collection = created_db.create_container(id=collection_definition['id'], partition_key=collection_definition['partitionKey'], offer_throughput=offer_throughput) self.assertEqual(collection_definition.get('id'), created_collection.id) created_collection_properties = created_collection.read( populate_partition_key_range_statistics=True, populate_quota_info=True) self.assertEqual(collection_definition.get('partitionKey').get('paths')[0], created_collection_properties['partitionKey']['paths'][0]) self.assertEqual(collection_definition.get('partitionKey').get('kind'), created_collection_properties['partitionKey']['kind']) self.assertIsNotNone(created_collection_properties.get("statistics")) self.assertIsNotNone(created_db.client_connection.last_response_headers.get("x-ms-resource-usage")) expected_offer = created_collection.get_throughput() self.assertIsNotNone(expected_offer) self.assertEqual(expected_offer.offer_throughput, offer_throughput) created_db.delete_container(created_collection.id) def test_partitioned_collection_partition_key_extraction(self): created_db = self.databaseForTest collection_id = 'test_partitioned_collection_partition_key_extraction ' + str(uuid.uuid4()) created_collection = created_db.create_container( id=collection_id, partition_key=PartitionKey(path='/address/state', kind=documents.PartitionKind.Hash) ) document_definition = {'id': 'document1', 'address': {'street': '1 Microsoft Way', 'city': 'Redmond', 'state': 'WA', 'zip code': 98052 } } self.OriginalExecuteFunction = _retry_utility.ExecuteFunction _retry_utility.ExecuteFunction = self._MockExecuteFunction # create document without partition key being specified created_document = created_collection.create_item(body=document_definition) _retry_utility.ExecuteFunction = self.OriginalExecuteFunction self.assertEqual(self.last_headers[0], '["WA"]') del self.last_headers[:] self.assertEqual(created_document.get('id'), document_definition.get('id')) self.assertEqual(created_document.get('address').get('state'), document_definition.get('address').get('state')) collection_id = 'test_partitioned_collection_partition_key_extraction1 ' + str(uuid.uuid4()) created_collection1 = created_db.create_container( id=collection_id, partition_key=PartitionKey(path='/address', kind=documents.PartitionKind.Hash) ) self.OriginalExecuteFunction = _retry_utility.ExecuteFunction _retry_utility.ExecuteFunction = self._MockExecuteFunction # Create document with partitionkey not present as a leaf level property but a dict created_document = created_collection1.create_item(document_definition) _retry_utility.ExecuteFunction = self.OriginalExecuteFunction self.assertEqual(self.last_headers[0], [{}]) del self.last_headers[:] # self.assertEqual(options['partitionKey'], documents.Undefined) collection_id = 'test_partitioned_collection_partition_key_extraction2 ' + str(uuid.uuid4()) created_collection2 = created_db.create_container( id=collection_id, partition_key=PartitionKey(path='/address/state/city', kind=documents.PartitionKind.Hash) ) self.OriginalExecuteFunction = _retry_utility.ExecuteFunction _retry_utility.ExecuteFunction = self._MockExecuteFunction # Create document with partitionkey not present in the document created_document = created_collection2.create_item(document_definition) _retry_utility.ExecuteFunction = self.OriginalExecuteFunction self.assertEqual(self.last_headers[0], [{}]) del self.last_headers[:] # self.assertEqual(options['partitionKey'], documents.Undefined) created_db.delete_container(created_collection.id) created_db.delete_container(created_collection1.id) created_db.delete_container(created_collection2.id) def test_partitioned_collection_partition_key_extraction_special_chars(self): created_db = self.databaseForTest collection_id = 'test_partitioned_collection_partition_key_extraction_special_chars1 ' + str(uuid.uuid4()) created_collection1 = created_db.create_container( id=collection_id, partition_key=PartitionKey(path='/\"level\' 1*()\"/\"le/vel2\"', kind=documents.PartitionKind.Hash) ) document_definition = {'id': 'document1', "level' 1*()": {"le/vel2": 'val1'} } self.OriginalExecuteFunction = _retry_utility.ExecuteFunction _retry_utility.ExecuteFunction = self._MockExecuteFunction created_document = created_collection1.create_item(body=document_definition) _retry_utility.ExecuteFunction = self.OriginalExecuteFunction self.assertEqual(self.last_headers[0], '["val1"]') del self.last_headers[:] collection_definition2 = { 'id': 'test_partitioned_collection_partition_key_extraction_special_chars2 ' + str(uuid.uuid4()), 'partitionKey': { 'paths': ['/\'level\" 1*()\'/\'le/vel2\''], 'kind': documents.PartitionKind.Hash } } collection_id = 'test_partitioned_collection_partition_key_extraction_special_chars2 ' + str(uuid.uuid4()) created_collection2 = created_db.create_container( id=collection_id, partition_key=PartitionKey(path='/\'level\" 1*()\'/\'le/vel2\'', kind=documents.PartitionKind.Hash) ) document_definition = {'id': 'document2', 'level\" 1*()': {'le/vel2': 'val2'} } self.OriginalExecuteFunction = _retry_utility.ExecuteFunction _retry_utility.ExecuteFunction = self._MockExecuteFunction # create document without partition key being specified created_document = created_collection2.create_item(body=document_definition) _retry_utility.ExecuteFunction = self.OriginalExecuteFunction self.assertEqual(self.last_headers[0], '["val2"]') del self.last_headers[:] created_db.delete_container(created_collection1.id) created_db.delete_container(created_collection2.id) 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_conflict_crud_and_query(self): created_db = self.databaseForTest created_collection = self.databaseForTest.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) conflict_definition = {'id': 'new conflict', 'resourceId': 'doc1', 'operationType': 'create', 'resourceType': 'document' } # read conflict here will return resource not found(404) since there is no conflict here self.__AssertHTTPFailureWithStatus( StatusCodes.NOT_FOUND, created_collection.get_conflict, conflict_definition['id'], conflict_definition['id'] ) # Read conflict feed doesn't require partitionKey to be specified as it's a cross partition thing conflict_list = list(created_collection.list_conflicts()) self.assertEqual(0, len(conflict_list)) # delete conflict here will return resource not found(404) since there is no conflict here self.__AssertHTTPFailureWithStatus( StatusCodes.NOT_FOUND, created_collection.delete_conflict, conflict_definition['id'], conflict_definition['id'] ) # query conflicts on any property other than partitionKey will fail without setting enableCrossPartitionQuery or passing in the partitionKey value try: list(created_collection.query_conflicts( query='SELECT * FROM root r WHERE r.resourceType=\'' + conflict_definition.get( # nosec 'resourceType') + '\'' )) except Exception: pass conflict_list = list(created_collection.query_conflicts( query='SELECT * FROM root r WHERE r.resourceType=\'' + conflict_definition.get('resourceType') + '\'', # nosec enable_cross_partition_query=True )) self.assertEqual(0, len(conflict_list)) # query conflicts by providing the partitionKey value options = {'partitionKey': conflict_definition.get('id')} conflict_list = list(created_collection.query_conflicts( query='SELECT * FROM root r WHERE r.resourceType=\'' + conflict_definition.get('resourceType') + '\'', # nosec partition_key=conflict_definition['id'] )) self.assertEqual(0, len(conflict_list)) def test_trigger_crud(self): # create database db = self.databaseForTest # create collection collection = self.databaseForTest.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) # read triggers triggers = list(collection.scripts.list_triggers()) # create a trigger before_create_triggers_count = len(triggers) trigger_id = 'sample trigger-' + str(uuid.uuid4()) trigger_definition = { 'id': trigger_id, 'serverScript': 'function() {var x = 10;}', 'triggerType': documents.TriggerType.Pre, 'triggerOperation': documents.TriggerOperation.All } trigger = collection.scripts.create_trigger(body=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(collection.scripts.list_triggers()) self.assertEqual(len(triggers), before_create_triggers_count + 1, 'create should increase the number of triggers') # query triggers triggers = list(collection.scripts.query_triggers( query='SELECT * FROM root r WHERE r.id=@id', parameters=[ {'name': '@id', 'value': trigger_definition['id']} ] )) self.assertTrue(triggers) # replace trigger change_trigger = trigger.copy() trigger['body'] = 'function() {var x = 20;}' replaced_trigger = collection.scripts.replace_trigger(change_trigger['id'], 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 = collection.scripts.get_trigger(replaced_trigger['id']) self.assertEqual(replaced_trigger['id'], trigger['id']) # delete trigger collection.scripts.delete_trigger(replaced_trigger['id']) # read triggers after deletion self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, collection.scripts.delete_trigger, replaced_trigger['id']) def test_udf_crud(self): # create database db = self.databaseForTest # create collection collection = self.databaseForTest.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) # read udfs udfs = list(collection.scripts.list_user_defined_functions()) # create a udf before_create_udfs_count = len(udfs) udf_definition = { 'id': 'sample udf', 'body': 'function() {var x = 10;}' } udf = collection.scripts.create_user_defined_function(body=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(collection.scripts.list_user_defined_functions()) self.assertEqual(len(udfs), before_create_udfs_count + 1, 'create should increase the number of udfs') # query udfs results = list(collection.scripts.query_user_defined_functions( query='SELECT * FROM root r WHERE r.id=@id', parameters=[ {'name': '@id', 'value': udf_definition['id']} ] )) self.assertTrue(results) # replace udf change_udf = udf.copy() udf['body'] = 'function() {var x = 20;}' replaced_udf = collection.scripts.replace_user_defined_function(udf=udf['id'], body=udf) for property in udf_definition: self.assertEqual( replaced_udf[property], udf[property], 'property {property} should match'.format(property=property)) # read udf udf = collection.scripts.get_user_defined_function(replaced_udf['id']) self.assertEqual(replaced_udf['id'], udf['id']) # delete udf collection.scripts.delete_user_defined_function(replaced_udf['id']) # read udfs after deletion self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, collection.scripts.get_user_defined_function, replaced_udf['id']) def test_sproc_crud(self): # create database db = self.databaseForTest # create collection collection = self.databaseForTest.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) # read sprocs sprocs = list(collection.scripts.list_stored_procedures()) # create a sproc before_create_sprocs_count = len(sprocs) sproc_id = 'sample sproc-' + str(uuid.uuid4()) sproc_definition = { 'id': sproc_id, 'serverScript': 'function() {var x = 10;}' } sproc = collection.scripts.create_stored_procedure(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(collection.scripts.list_stored_procedures()) self.assertEqual(len(sprocs), before_create_sprocs_count + 1, 'create should increase the number of sprocs') # query sprocs sprocs = list(collection.scripts.query_stored_procedures( query='SELECT * FROM root r WHERE r.id=@id', parameters=[ {'name': '@id', 'value': sproc_definition['id']} ] )) self.assertIsNotNone(sprocs) # replace sproc change_sproc = sproc.copy() sproc['body'] = 'function() {var x = 20;}' replaced_sproc = collection.scripts.replace_stored_procedure(sproc=change_sproc['id'], body=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 = collection.scripts.get_stored_procedure(replaced_sproc['id']) self.assertEqual(replaced_sproc['id'], sproc['id']) # delete sproc collection.scripts.delete_stored_procedure(replaced_sproc['id']) # read sprocs after deletion self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, collection.scripts.get_stored_procedure, replaced_sproc['id']) def test_collection_indexing_policy(self): # create database db = self.databaseForTest # create collection collection = db.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) collection_properties = collection.read() self.assertEqual(collection_properties['indexingPolicy']['indexingMode'], documents.IndexingMode.Consistent, 'default indexing mode should be consistent') collection_with_indexing_policy = db.create_container( id='CollectionWithIndexingPolicy ' + str(uuid.uuid4()), indexing_policy={ 'automatic': True, 'indexingMode': documents.IndexingMode.Consistent, 'includedPaths': [ { 'path': '/', 'indexes': [ { 'kind': documents.IndexKind.Hash, 'dataType': documents.DataType.Number, 'precision': 2 } ] } ], 'excludedPaths': [ { 'path': '/"systemMetadata"/*' } ] }, partition_key=PartitionKey(path='/id', kind='Hash') ) collection_with_indexing_policy_properties = collection_with_indexing_policy.read() self.assertEqual(1, len(collection_with_indexing_policy_properties['indexingPolicy']['includedPaths']), 'Unexpected includedPaths length') self.assertEqual(2, len(collection_with_indexing_policy_properties['indexingPolicy']['excludedPaths']), 'Unexpected excluded path count') db.delete_container(collection_with_indexing_policy.id) def test_create_default_indexing_policy(self): # create database db = self.databaseForTest # no indexing policy specified collection = db.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) collection_properties = collection.read() self._check_default_indexing_policy_paths(collection_properties['indexingPolicy']) # partial policy specified collection = db.create_container( id='test_create_default_indexing_policy TestCreateDefaultPolicy01' + str(uuid.uuid4()), indexing_policy={ 'indexingMode': documents.IndexingMode.Consistent, 'automatic': True }, partition_key=PartitionKey(path='/id', kind='Hash') ) collection_properties = collection.read() self._check_default_indexing_policy_paths(collection_properties['indexingPolicy']) db.delete_container(container=collection) # default policy collection = db.create_container( id='test_create_default_indexing_policy TestCreateDefaultPolicy03' + str(uuid.uuid4()), indexing_policy={}, partition_key=PartitionKey(path='/id', kind='Hash') ) collection_properties = collection.read() self._check_default_indexing_policy_paths(collection_properties['indexingPolicy']) db.delete_container(container=collection) # missing indexes collection = db.create_container( id='test_create_default_indexing_policy TestCreateDefaultPolicy04' + str(uuid.uuid4()), indexing_policy={ 'includedPaths': [ { 'path': '/*' } ] }, partition_key=PartitionKey(path='/id', kind='Hash') ) collection_properties = collection.read() self._check_default_indexing_policy_paths(collection_properties['indexingPolicy']) db.delete_container(container=collection) # missing precision collection = db.create_container( id='test_create_default_indexing_policy TestCreateDefaultPolicy05' + str(uuid.uuid4()), indexing_policy={ 'includedPaths': [ { 'path': '/*', 'indexes': [ { 'kind': documents.IndexKind.Hash, 'dataType': documents.DataType.String }, { 'kind': documents.IndexKind.Range, 'dataType': documents.DataType.Number } ] } ] }, partition_key=PartitionKey(path='/id', kind='Hash') ) collection_properties = collection.read() self._check_default_indexing_policy_paths(collection_properties['indexingPolicy']) db.delete_container(container=collection) def test_create_indexing_policy_with_composite_and_spatial_indexes(self): # create database db = self.databaseForTest indexing_policy = { "spatialIndexes": [ { "path": "/path0/*", "types": [ "Point", "LineString", "Polygon", "MultiPolygon" ] }, { "path": "/path1/*", "types": [ "Point", "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" } ] ] } # TODO: custom_logger = logging.getLogger("CustomLogger") was in old code, check on later created_container = db.create_container( id='composite_index_spatial_index' + str(uuid.uuid4()), indexing_policy=indexing_policy, partition_key=PartitionKey(path='/id', kind='Hash'), headers={"Foo": "bar"}, user_agent="blah", user_agent_overwrite=True, logging_enable=True, ) # TODO: logger was passed into read previously created_properties = created_container.read() read_indexing_policy = created_properties['indexingPolicy'] if 'localhost' in self.host or '127.0.0.1' in self.host: # TODO: Differing result between live and emulator self.assertListEqual(indexing_policy['spatialIndexes'], read_indexing_policy['spatialIndexes']) else: # All types are returned for spatial Indexes self.assertListEqual(indexing_policy['spatialIndexes'], read_indexing_policy['spatialIndexes']) self.assertListEqual(indexing_policy['compositeIndexes'], read_indexing_policy['compositeIndexes']) db.delete_container(container=created_container) 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.assertFalse(root_included_path.get('indexes')) def test_trigger_functionality(self): 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(collection, triggers): """Creates triggers. :Parameters: - `client`: cosmos_client_connection.CosmosClientConnection - `collection`: dict """ for trigger_i in triggers: trigger = collection.scripts.create_trigger(body=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.databaseForTest # create collections pkd = PartitionKey(path='/id', kind='Hash') collection1 = db.create_container(id='test_trigger_functionality 1 ' + str(uuid.uuid4()), partition_key=PartitionKey(path='/key', kind='Hash')) collection2 = db.create_container(id='test_trigger_functionality 2 ' + str(uuid.uuid4()), partition_key=PartitionKey(path='/key', kind='Hash')) collection3 = db.create_container(id='test_trigger_functionality 3 ' + str(uuid.uuid4()), partition_key=PartitionKey(path='/key', kind='Hash')) # create triggers __CreateTriggers(collection1, triggers_in_collection1) __CreateTriggers(collection2, triggers_in_collection2) __CreateTriggers(collection3, triggers_in_collection3) # create document triggers_1 = list(collection1.scripts.list_triggers()) self.assertEqual(len(triggers_1), 3) document_1_1 = collection1.create_item( body={'id': 'doc1', 'key': 'value'}, pre_trigger_include='t1' ) self.assertEqual(document_1_1['id'], 'DOC1t1', 'id should be capitalized') document_1_2 = collection1.create_item( body={'id': 'testing post trigger', 'key': 'value'}, pre_trigger_include='t1', post_trigger_include='response1', ) self.assertEqual(document_1_2['id'], 'TESTING POST TRIGGERt1') document_1_3 = collection1.create_item( body={'id': 'responseheaders', 'key': 'value'}, pre_trigger_include='t1' ) self.assertEqual(document_1_3['id'], "RESPONSEHEADERSt1") triggers_2 = list(collection2.scripts.list_triggers()) self.assertEqual(len(triggers_2), 2) document_2_1 = collection2.create_item( body={'id': 'doc2', 'key': 'value2'}, pre_trigger_include='t2' ) self.assertEqual(document_2_1['id'], 'doc2', 'id shouldn\'t change') document_2_2 = collection2.create_item( body={'id': 'Doc3', 'prop': 'empty', 'key': 'value2'}, pre_trigger_include='t3') self.assertEqual(document_2_2['id'], 'doc3t3') triggers_3 = list(collection3.scripts.list_triggers()) self.assertEqual(len(triggers_3), 1) with self.assertRaises(Exception): collection3.create_item( body={'id': 'Docoptype', 'key': 'value2'}, post_trigger_include='triggerOpType' ) db.delete_container(collection1) db.delete_container(collection2) db.delete_container(collection3) def __ValidateOfferResponseBody(self, offer, expected_coll_link, expected_offer_type): # type: (Offer, str, Any) -> None self.assertIsNotNone(offer.properties['id'], 'Id cannot be null.') self.assertIsNotNone(offer.properties.get('_rid'), 'Resource Id (Rid) cannot be null.') self.assertIsNotNone(offer.properties.get('_self'), 'Self Link cannot be null.') self.assertIsNotNone(offer.properties.get('resource'), 'Resource Link cannot be null.') self.assertTrue(offer.properties['_self'].find(offer.properties['id']) != -1, 'Offer id not contained in offer self link.') self.assertEqual(expected_coll_link.strip('/'), offer.properties['resource'].strip('/')) if (expected_offer_type): self.assertEqual(expected_offer_type, offer.properties.get('offerType')) def test_offer_read_and_query(self): # Create database. db = self.databaseForTest collection = db.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) # Read the offer. expected_offer = collection.get_throughput() collection_properties = collection.read() self.__ValidateOfferResponseBody(expected_offer, collection_properties.get('_self'), None) def test_offer_replace(self): # Create database. db = self.databaseForTest # Create collection. collection = db.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) # Read Offer expected_offer = collection.get_throughput() collection_properties = collection.read() self.__ValidateOfferResponseBody(expected_offer, collection_properties.get('_self'), None) # Replace the offer. replaced_offer = collection.replace_throughput(expected_offer.offer_throughput + 100) collection_properties = collection.read() self.__ValidateOfferResponseBody(replaced_offer, collection_properties.get('_self'), None) # Check if the replaced offer is what we expect. self.assertEqual(expected_offer.properties.get('content').get('offerThroughput') + 100, replaced_offer.properties.get('content').get('offerThroughput')) self.assertEqual(expected_offer.offer_throughput + 100, replaced_offer.offer_throughput) def test_index_progress_headers(self): created_db = self.databaseForTest created_container = created_db.get_container_client(self.configs.TEST_MULTI_PARTITION_CONTAINER_ID) created_container.read(populate_quota_info=True) self.assertFalse(HttpHeaders.LazyIndexingProgress in created_db.client_connection.last_response_headers) self.assertTrue(HttpHeaders.IndexTransformationProgress in created_db.client_connection.last_response_headers) none_coll = created_db.create_container( id='test_index_progress_headers none_coll ' + str(uuid.uuid4()), indexing_policy={ 'indexingMode': documents.IndexingMode.NoIndex, 'automatic': False }, partition_key=PartitionKey(path="/id", kind='Hash') ) created_container = created_db.get_container_client(container=none_coll) created_container.read(populate_quota_info=True) self.assertFalse(HttpHeaders.LazyIndexingProgress in created_db.client_connection.last_response_headers) self.assertTrue(HttpHeaders.IndexTransformationProgress in created_db.client_connection.last_response_headers) created_db.delete_container(none_coll) def _MockExecuteFunction(self, function, *args, **kwargs): if HttpHeaders.PartitionKey in args[4].headers: self.last_headers.append(args[4].headers[HttpHeaders.PartitionKey]) return self.OriginalExecuteFunction(function, *args, **kwargs) 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