# coding=utf-8 # ------------------------------------ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. # ------------------------------------ import pytest import platform import functools from azure.core.exceptions import HttpResponseError, ClientAuthenticationError from azure.core.pipeline.transport import AioHttpTransport from azure.core.credentials import AzureKeyCredential from multidict import CIMultiDict, CIMultiDictProxy from azure.ai.textanalytics.aio import TextAnalyticsClient from azure.ai.textanalytics import ( VERSION, DetectLanguageInput, DetectLanguageInput, TextAnalyticsApiVersion, ) from testcase import GlobalTextAnalyticsAccountPreparer from testcase import TextAnalyticsClientPreparer as _TextAnalyticsClientPreparer from asynctestcase import AsyncTextAnalyticsTest # pre-apply the client_cls positional argument so it needn't be explicitly passed below TextAnalyticsClientPreparer = functools.partial(_TextAnalyticsClientPreparer, TextAnalyticsClient) class AiohttpTestTransport(AioHttpTransport): """Workaround to vcrpy bug: https://github.com/kevin1024/vcrpy/pull/461 """ async def send(self, request, **config): response = await super(AiohttpTestTransport, self).send(request, **config) if not isinstance(response.headers, CIMultiDictProxy): response.headers = CIMultiDictProxy(CIMultiDict(response.internal_response.headers)) response.content_type = response.headers.get("content-type") return response class TestDetectLanguage(AsyncTextAnalyticsTest): @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_no_single_input(self, client): with self.assertRaises(TypeError): response = await client.detect_language("hello world") @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_all_successful_passing_dict(self, client): docs = [{"id": "1", "text": "I should take my cat to the veterinarian."}, {"id": "2", "text": "Este es un document escrito en Español."}, {"id": "3", "text": "猫は幸せ"}, {"id": "4", "text": "Fahrt nach Stuttgart und dann zum Hotel zu Fu."}] response = await client.detect_language(docs, show_stats=True) self.assertEqual(response[0].primary_language.name, "English") self.assertEqual(response[1].primary_language.name, "Spanish") self.assertEqual(response[2].primary_language.name, "Japanese") self.assertEqual(response[3].primary_language.name, "German") self.assertEqual(response[0].primary_language.iso6391_name, "en") self.assertEqual(response[1].primary_language.iso6391_name, "es") self.assertEqual(response[2].primary_language.iso6391_name, "ja") self.assertEqual(response[3].primary_language.iso6391_name, "de") for doc in response: self.assertIsNotNone(doc.id) self.assertIsNotNone(doc.statistics) self.assertIsNotNone(doc.primary_language.confidence_score) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_all_successful_passing_text_document_input(self, client): docs = [ DetectLanguageInput(id="1", text="I should take my cat to the veterinarian"), DetectLanguageInput(id="2", text="Este es un document escrito en Español."), DetectLanguageInput(id="3", text="猫は幸せ"), DetectLanguageInput(id="4", text="Fahrt nach Stuttgart und dann zum Hotel zu Fu.") ] response = await client.detect_language(docs) self.assertEqual(response[0].primary_language.name, "English") self.assertEqual(response[1].primary_language.name, "Spanish") self.assertEqual(response[2].primary_language.name, "Japanese") self.assertEqual(response[3].primary_language.name, "German") self.assertEqual(response[0].primary_language.iso6391_name, "en") self.assertEqual(response[1].primary_language.iso6391_name, "es") self.assertEqual(response[2].primary_language.iso6391_name, "ja") self.assertEqual(response[3].primary_language.iso6391_name, "de") for doc in response: self.assertIsNotNone(doc.primary_language.confidence_score) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_passing_only_string(self, client): docs = [ u"I should take my cat to the veterinarian.", u"Este es un document escrito en Español.", u"猫は幸せ", u"Fahrt nach Stuttgart und dann zum Hotel zu Fu.", u"" ] response = await client.detect_language(docs) self.assertEqual(response[0].primary_language.name, "English") self.assertEqual(response[1].primary_language.name, "Spanish") self.assertEqual(response[2].primary_language.name, "Japanese") self.assertEqual(response[3].primary_language.name, "German") self.assertTrue(response[4].is_error) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_input_with_some_errors(self, client): docs = [{"id": "1", "country_hint": "United States", "text": "I should take my cat to the veterinarian."}, {"id": "2", "text": "Este es un document escrito en Español."}, {"id": "3", "text": ""}, {"id": "4", "text": "Fahrt nach Stuttgart und dann zum Hotel zu Fu."}] response = await client.detect_language(docs) self.assertTrue(response[0].is_error) self.assertFalse(response[1].is_error) self.assertTrue(response[2].is_error) self.assertFalse(response[3].is_error) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_input_with_all_errors(self, client): text = "" for _ in range(5121): text += "x" docs = [{"id": "1", "text": ""}, {"id": "2", "text": ""}, {"id": "3", "text": ""}, {"id": "4", "text": text}] response = await client.detect_language(docs) for resp in response: self.assertTrue(resp.is_error) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_output_same_order_as_input(self, client): docs = [ DetectLanguageInput(id="1", text="one"), DetectLanguageInput(id="2", text="two"), DetectLanguageInput(id="3", text="three"), DetectLanguageInput(id="4", text="four"), DetectLanguageInput(id="5", text="five") ] response = await client.detect_language(docs) for idx, doc in enumerate(response): self.assertEqual(str(idx + 1), doc.id) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"text_analytics_account_key": ""}) async def test_empty_credential_class(self, client): with self.assertRaises(ClientAuthenticationError): response = await client.detect_language( ["This is written in English."] ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"text_analytics_account_key": "xxxxxxxxxxxx"}) async def test_bad_credentials(self, client): with self.assertRaises(ClientAuthenticationError): response = await client.detect_language( ["This is written in English."] ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_bad_document_input(self, client): docs = "This is the wrong type" with self.assertRaises(TypeError): response = await client.detect_language(docs) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_mixing_inputs(self, client): docs = [ {"id": "1", "text": "Microsoft was founded by Bill Gates and Paul Allen."}, DetectLanguageInput(id="2", text="I did not like the hotel we stayed at. It was too expensive."), u"You cannot mix string input with the above documents" ] with self.assertRaises(TypeError): response = await client.detect_language(docs) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_out_of_order_ids(self, client): docs = [{"id": "56", "text": ":)"}, {"id": "0", "text": ":("}, {"id": "22", "text": ""}, {"id": "19", "text": ":P"}, {"id": "1", "text": ":D"}] response = await client.detect_language(docs) in_order = ["56", "0", "22", "19", "1"] for idx, resp in enumerate(response): self.assertEqual(resp.id, in_order[idx]) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_show_stats_and_model_version(self, client): def callback(response): self.assertIsNotNone(response) self.assertIsNotNone(response.model_version, msg=response.raw_response) self.assertIsNotNone(response.raw_response) self.assertEqual(response.statistics.document_count, 5) self.assertEqual(response.statistics.transaction_count, 4) self.assertEqual(response.statistics.valid_document_count, 4) self.assertEqual(response.statistics.erroneous_document_count, 1) docs = [{"id": "56", "text": ":)"}, {"id": "0", "text": ":("}, {"id": "22", "text": ""}, {"id": "19", "text": ":P"}, {"id": "1", "text": ":D"}] response = await client.detect_language( docs, show_stats=True, model_version="latest", raw_response_hook=callback ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_batch_size_over_limit(self, client): docs = [u"hello world"] * 1050 with self.assertRaises(HttpResponseError): response = await client.detect_language(docs) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_whole_batch_country_hint(self, client): def callback(resp): country_str = "\"countryHint\": \"CA\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 3) docs = [ u"This was the best day of my life.", u"I did not like the hotel we stayed at. It was too expensive.", u"The restaurant was not as good as I hoped." ] response = await client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_whole_batch_dont_use_country_hint(self, client): def callback(resp): country_str = "\"countryHint\": \"\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 3) docs = [ u"This was the best day of my life.", u"I did not like the hotel we stayed at. It was too expensive.", u"The restaurant was not as good as I hoped." ] response = await client.detect_language(docs, country_hint="", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_per_item_dont_use_country_hint(self, client): def callback(resp): country_str = "\"countryHint\": \"\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 2) country_str = "\"countryHint\": \"US\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 1) docs = [{"id": "1", "country_hint": "", "text": "I will go to the park."}, {"id": "2", "country_hint": "", "text": "I did not like the hotel we stayed at."}, {"id": "3", "text": "The restaurant had really good food."}] response = await client.detect_language(docs, raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_whole_batch_country_hint_and_obj_input(self, client): def callback(resp): country_str = "\"countryHint\": \"CA\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 3) docs = [ DetectLanguageInput(id="1", text="I should take my cat to the veterinarian."), DetectLanguageInput(id="2", text="Este es un document escrito en Español."), DetectLanguageInput(id="3", text="猫は幸せ"), ] response = await client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_whole_batch_country_hint_and_dict_input(self, client): def callback(resp): country_str = "\"countryHint\": \"CA\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 3) docs = [{"id": "1", "text": "I will go to the park."}, {"id": "2", "text": "I did not like the hotel we stayed at."}, {"id": "3", "text": "The restaurant had really good food."}] response = await client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_whole_batch_country_hint_and_obj_per_item_hints(self, client): def callback(resp): country_str = "\"countryHint\": \"CA\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 2) country_str = "\"countryHint\": \"US\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 1) docs = [ DetectLanguageInput(id="1", text="I should take my cat to the veterinarian.", country_hint="CA"), DetectLanguageInput(id="4", text="Este es un document escrito en Español.", country_hint="CA"), DetectLanguageInput(id="3", text="猫は幸せ"), ] response = await client.detect_language(docs, country_hint="US", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_whole_batch_country_hint_and_dict_per_item_hints(self, client): def callback(resp): country_str = "\"countryHint\": \"CA\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 1) country_str = "\"countryHint\": \"US\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 2) docs = [{"id": "1", "country_hint": "US", "text": "I will go to the park."}, {"id": "2", "country_hint": "US", "text": "I did not like the hotel we stayed at."}, {"id": "3", "text": "The restaurant had really good food."}] response = await client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"default_country_hint": "CA"}) async def test_client_passed_default_country_hint(self, client): def callback(resp): country_str = "\"countryHint\": \"CA\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 3) def callback_2(resp): country_str = "\"countryHint\": \"DE\"" country = resp.http_request.body.count(country_str) self.assertEqual(country, 3) docs = [{"id": "1", "text": "I will go to the park."}, {"id": "2", "text": "I did not like the hotel we stayed at."}, {"id": "3", "text": "The restaurant had really good food."}] response = await client.detect_language(docs, raw_response_hook=callback) response = await client.detect_language(docs, country_hint="DE", raw_response_hook=callback_2) response = await client.detect_language(docs, raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() async def test_rotate_subscription_key(self, resource_group, location, text_analytics_account, text_analytics_account_key): credential = AzureKeyCredential(text_analytics_account_key) client = TextAnalyticsClient(text_analytics_account, credential) docs = [{"id": "1", "text": "I will go to the park."}, {"id": "2", "text": "I did not like the hotel we stayed at."}, {"id": "3", "text": "The restaurant had really good food."}] response = await client.detect_language(docs) self.assertIsNotNone(response) credential.update("xxx") # Make authentication fail with self.assertRaises(ClientAuthenticationError): response = await client.detect_language(docs) credential.update(text_analytics_account_key) # Authenticate successfully again response = await client.detect_language(docs) self.assertIsNotNone(response) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_user_agent(self, client): def callback(resp): self.assertIn("azsdk-python-ai-textanalytics/{} Python/{} ({})".format( VERSION, platform.python_version(), platform.platform()), resp.http_request.headers["User-Agent"] ) docs = [{"id": "1", "text": "I will go to the park."}, {"id": "2", "text": "I did not like the hotel we stayed at."}, {"id": "3", "text": "The restaurant had really good food."}] response = await client.detect_language(docs, raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_document_attribute_error_no_result_attribute(self, client): docs = [{"id": "1", "text": ""}] response = await client.detect_language(docs) # Attributes on DocumentError self.assertTrue(response[0].is_error) self.assertEqual(response[0].id, "1") self.assertIsNotNone(response[0].error) # Result attribute not on DocumentError, custom error message try: primary_language = response[0].primary_language except AttributeError as custom_error: self.assertEqual( custom_error.args[0], '\'DocumentError\' object has no attribute \'primary_language\'. ' 'The service was unable to process this document:\nDocument Id: 1\nError: ' 'InvalidDocument - Document text is empty.\n' ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_document_attribute_error_nonexistent_attribute(self, client): docs = [{"id": "1", "text": ""}] response = await client.detect_language(docs) # Attribute not found on DocumentError or result obj, default behavior/message try: primary_language = response[0].attribute_not_on_result_or_error except AttributeError as default_behavior: self.assertEqual( default_behavior.args[0], '\'DocumentError\' object has no attribute \'attribute_not_on_result_or_error\'' ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_bad_model_version_error(self, client): docs = [{"id": "1", "language": "english", "text": "I did not like the hotel we stayed at."}] try: result = await client.detect_language(docs, model_version="bad") except HttpResponseError as err: self.assertEqual(err.error.code, "ModelVersionIncorrect") self.assertIsNotNone(err.error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_document_errors(self, client): text = "" for _ in range(5121): text += "x" docs = [{"id": "1", "text": ""}, {"id": "2", "text": text}] doc_errors = await client.detect_language(docs) self.assertEqual(doc_errors[0].error.code, "InvalidDocument") self.assertIsNotNone(doc_errors[0].error.message) self.assertEqual(doc_errors[1].error.code, "InvalidDocument") self.assertIsNotNone(doc_errors[1].error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_document_warnings(self, client): # No warnings actually returned for detect_language. Will update when they add docs = [ {"id": "1", "text": "This won't actually create a warning :'("}, ] result = await client.detect_language(docs) for doc in result: doc_warnings = doc.warnings self.assertEqual(len(doc_warnings), 0) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_not_passing_list_for_docs(self, client): docs = {"id": "1", "text": "hello world"} with pytest.raises(TypeError) as excinfo: await client.detect_language(docs) assert "Input documents cannot be a dict" in str(excinfo.value) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_missing_input_records_error(self, client): docs = [] with pytest.raises(ValueError) as excinfo: await client.detect_language(docs) assert "Input documents can not be empty or None" in str(excinfo.value) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_passing_none_docs(self, client): with pytest.raises(ValueError) as excinfo: await client.detect_language(None) assert "Input documents can not be empty or None" in str(excinfo.value) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_duplicate_ids_error(self, client): # Duplicate Ids docs = [{"id": "1", "text": "hello world"}, {"id": "1", "text": "I did not like the hotel we stayed at."}] try: result = await client.detect_language(docs) except HttpResponseError as err: self.assertEqual(err.error.code, "InvalidDocument") self.assertIsNotNone(err.error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_batch_size_over_limit_error(self, client): # Batch size over limit docs = [u"hello world"] * 1001 try: response = await client.detect_language(docs) except HttpResponseError as err: self.assertEqual(err.error.code, "InvalidDocumentBatch") self.assertIsNotNone(err.error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_invalid_country_hint_method(self, client): docs = [{"id": "1", "text": "hello world"}] response = await client.detect_language(docs, country_hint="United States") self.assertEqual(response[0].error.code, "InvalidCountryHint") self.assertIsNotNone(response[0].error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_invalid_country_hint_docs(self, client): docs = [{"id": "1", "country_hint": "United States", "text": "hello world"}] response = await client.detect_language(docs) self.assertEqual(response[0].error.code, "InvalidCountryHint") self.assertIsNotNone(response[0].error.message) @GlobalTextAnalyticsAccountPreparer() async def test_country_hint_none(self, resource_group, location, text_analytics_account, text_analytics_account_key): client = TextAnalyticsClient(text_analytics_account, AzureKeyCredential(text_analytics_account_key)) # service will eventually support this and we will not need to send "" for input == "none" documents = [{"id": "0", "country_hint": "none", "text": "This is written in English."}] documents2 = [DetectLanguageInput(id="1", country_hint="none", text="This is written in English.")] def callback(response): country_str = "\"countryHint\": \"\"" country = response.http_request.body.count(country_str) self.assertEqual(country, 1) # test dict result = await client.detect_language(documents, raw_response_hook=callback) # test DetectLanguageInput result2 = await client.detect_language(documents2, raw_response_hook=callback) # test per-operation result3 = await client.detect_language(documents=["this is written in english"], country_hint="none", raw_response_hook=callback) # test client default new_client = TextAnalyticsClient(text_analytics_account, AzureKeyCredential(text_analytics_account_key), default_country_hint="none") result4 = await new_client.detect_language(documents=["this is written in english"], raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() async def test_country_hint_kwarg(self, client): def callback(response): country_str = "\"countryHint\": \"ES\"" self.assertEqual(response.http_request.body.count(country_str), 1) self.assertIsNotNone(response.model_version) self.assertIsNotNone(response.statistics) res = await client.detect_language( documents=["this is written in english"], model_version="latest", show_stats=True, country_hint="ES", raw_response_hook=callback ) @GlobalTextAnalyticsAccountPreparer() async def test_pass_cls(self, resource_group, location, text_analytics_account, text_analytics_account_key): def callback(pipeline_response, deserialized, _): return "cls result" text_analytics = TextAnalyticsClient(text_analytics_account, AzureKeyCredential(text_analytics_account_key)) res = await text_analytics.detect_language( documents=["Test passing cls to endpoint"], cls=callback ) assert res == "cls result" @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"api_version": TextAnalyticsApiVersion.V3_0}) async def test_string_index_type_not_fail_v3(self, client): # make sure that the addition of the string_index_type kwarg for v3.1-preview.1 doesn't # cause v3.0 calls to fail await client.detect_language(["please don't fail"])