# 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.credentials import AzureKeyCredential from testcase import TextAnalyticsTest, GlobalTextAnalyticsAccountPreparer from testcase import TextAnalyticsClientPreparer as _TextAnalyticsClientPreparer from azure.ai.textanalytics import ( DetectLanguageInput, TextAnalyticsClient, DetectLanguageInput, VERSION, TextAnalyticsApiVersion, ) # pre-apply the client_cls positional argument so it needn't be explicitly passed below TextAnalyticsClientPreparer = functools.partial(_TextAnalyticsClientPreparer, TextAnalyticsClient) class TestDetectLanguage(TextAnalyticsTest): @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_no_single_input(self, client): with self.assertRaises(TypeError): response = client.detect_language("hello world") @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = 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() 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 = 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() 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 = 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() 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 = 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() 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 = client.detect_language(docs) for resp in response: self.assertTrue(resp.is_error) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = 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": ""}) def test_empty_credential_class(self, client): with self.assertRaises(ClientAuthenticationError): response = client.detect_language( ["This is written in English."] ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"text_analytics_account_key": "xxxxxxxxxxxx"}) def test_bad_credentials(self, client): with self.assertRaises(ClientAuthenticationError): response = client.detect_language( ["This is written in English."] ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_bad_document_input(self, client): docs = "This is the wrong type" with self.assertRaises(TypeError): response = client.detect_language(docs) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 inputs" ] with self.assertRaises(TypeError): response = client.detect_language(docs) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = 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() 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 = client.detect_language( docs, show_stats=True, model_version="latest", raw_response_hook=callback ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_batch_size_over_limit(self, client): docs = [u"hello world"] * 1050 with self.assertRaises(HttpResponseError): response = client.detect_language(docs) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, country_hint="", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, country_hint="US", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, country_hint="CA", raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"default_country_hint": "CA"}) 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 = client.detect_language(docs, raw_response_hook=callback) response = client.detect_language(docs, country_hint="DE", raw_response_hook=callback_2) response = client.detect_language(docs, raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() 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 = client.detect_language(docs) self.assertIsNotNone(response) credential.update("xxx") # Make authentication fail with self.assertRaises(ClientAuthenticationError): response = client.detect_language(docs) credential.update(text_analytics_account_key) # Authenticate successfully again response = client.detect_language(docs) self.assertIsNotNone(response) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs, raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_document_attribute_error_no_result_attribute(self, client): docs = [{"id": "1", "text": ""}] response = 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() def test_document_attribute_error_nonexistent_attribute(self, client): docs = [{"id": "1", "text": ""}] response = 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() 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 = client.detect_language(docs, model_version="bad") except HttpResponseError as err: self.assertEqual(err.error.code, "ModelVersionIncorrect") self.assertIsNotNone(err.error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_document_errors(self, client): text = "" for _ in range(5121): text += "x" docs = [{"id": "1", "text": ""}, {"id": "2", "text": text}] doc_errors = 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() 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 = client.detect_language(docs) for doc in result: doc_warnings = doc.warnings self.assertEqual(len(doc_warnings), 0) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_not_passing_list_for_docs(self, client): docs = {"id": "1", "text": "hello world"} with pytest.raises(TypeError) as excinfo: client.detect_language(docs) assert "Input documents cannot be a dict" in str(excinfo.value) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_missing_input_records_error(self, client): docs = [] with pytest.raises(ValueError) as excinfo: client.detect_language(docs) assert "Input documents can not be empty or None" in str(excinfo.value) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_passing_none_docs(self, client): with pytest.raises(ValueError) as excinfo: client.detect_language(None) assert "Input documents can not be empty or None" in str(excinfo.value) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language(docs) except HttpResponseError as err: self.assertEqual(err.error.code, "InvalidDocument") self.assertIsNotNone(err.error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_batch_size_over_limit_error(self, client): # Batch size over limit docs = [u"hello world"] * 1001 try: response = client.detect_language(docs) except HttpResponseError as err: self.assertEqual(err.error.code, "InvalidDocumentBatch") self.assertIsNotNone(err.error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_invalid_country_hint_method(self, client): docs = [{"id": "1", "text": "hello world"}] response = client.detect_language(docs, country_hint="United States") self.assertEqual(response[0].error.code, "InvalidCountryHint") self.assertIsNotNone(response[0].error.message) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_invalid_country_hint_docs(self, client): docs = [{"id": "1", "country_hint": "United States", "text": "hello world"}] response = client.detect_language(docs) self.assertEqual(response[0].error.code, "InvalidCountryHint") self.assertIsNotNone(response[0].error.message) @GlobalTextAnalyticsAccountPreparer() 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 = client.detect_language(documents, raw_response_hook=callback) # test DetectLanguageInput result2 = client.detect_language(documents2, raw_response_hook=callback) # test per-operation result3 = 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 = new_client.detect_language(documents=["this is written in english"], raw_response_hook=callback) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() 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 = client.detect_language( documents=["this is written in english"], model_version="latest", show_stats=True, country_hint="ES", raw_response_hook=callback ) @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer() def test_pass_cls(self, client): def callback(pipeline_response, deserialized, _): return "cls result" res = client.detect_language( documents=["Test passing cls to endpoint"], cls=callback ) assert res == "cls result" @GlobalTextAnalyticsAccountPreparer() @TextAnalyticsClientPreparer(client_kwargs={"api_version": TextAnalyticsApiVersion.V3_0}) 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 client.detect_language(["please don't fail"])