1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51
|
# -*- coding: utf-8 -*-
import unittest
import uuid
import pytest
import azure.cosmos.cosmos_client as cosmos_client
import azure.cosmos.documents as documents
import test.test_config as test_config
@pytest.mark.usefixtures("teardown")
class EncodingTest(unittest.TestCase):
"""Test to ensure escaping of non-ascii characters from partition key"""
host = test_config._test_config.host
masterKey = test_config._test_config.masterKey
connectionPolicy = test_config._test_config.connectionPolicy
client = cosmos_client.CosmosClient(host, {'masterKey': masterKey}, connectionPolicy)
created_collection = test_config._test_config.create_multi_partition_collection_with_custom_pk_if_not_exist(client)
def test_unicode_characters_in_partition_key (self):
test_string = u'€€ کلید پارتیشن विभाजन कुंजी 123'
document_definition = {'pk': test_string, 'id':'myid' + str(uuid.uuid4())}
created_doc = self.client.CreateItem(self.created_collection['_self'], document_definition)
read_options = {'partitionKey': test_string }
read_doc = self.client.ReadItem(created_doc['_self'], read_options)
self.assertEqual(read_doc['pk'], test_string)
def test_create_document_with_line_separator_para_seperator_next_line_unicodes (self):
test_string = u'Line Separator (
) & Paragraph Separator (
) & Next Line (
) & نیمفاصله'
document_definition = {'pk': 'pk', 'id':'myid' + str(uuid.uuid4()), 'unicode_content':test_string }
created_doc = self.client.CreateItem(self.created_collection['_self'], document_definition)
read_options = {'partitionKey': 'pk' }
read_doc = self.client.ReadItem(created_doc['_self'], read_options)
self.assertEqual(read_doc['unicode_content'], test_string)
def test_create_stored_procedure_with_line_separator_para_seperator_next_line_unicodes (self):
test_string = 'Line Separator (
) & Paragraph Separator (
) & Next Line (
) & نیمفاصله'
test_string_unicode = u'Line Separator (
) & Paragraph Separator (
) & Next Line (
) & نیمفاصله'
stored_proc_definition = {'id':'myid' + str(uuid.uuid4()), 'body': test_string}
created_sp = self.client.CreateStoredProcedure(self.created_collection['_self'], stored_proc_definition)
read_sp = self.client.ReadStoredProcedure(created_sp['_self'], dict())
self.assertEqual(read_sp['body'], test_string_unicode)
if __name__ == '__main__':
unittest.main()
|