# coding=utf-8 # ------------------------------------ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. # ------------------------------------ import pytest import functools from io import BytesIO from azure.core.exceptions import ServiceRequestError, ClientAuthenticationError, HttpResponseError from azure.core.credentials import AzureKeyCredential from azure.ai.formrecognizer._generated.models import AnalyzeOperationResult from azure.ai.formrecognizer._response_handlers import prepare_content_result from azure.ai.formrecognizer import FormRecognizerClient, FormContentType, FormRecognizerApiVersion from testcase import FormRecognizerTest, GlobalFormRecognizerAccountPreparer from testcase import GlobalClientPreparer as _GlobalClientPreparer GlobalClientPreparer = functools.partial(_GlobalClientPreparer, FormRecognizerClient) class TestContentFromStream(FormRecognizerTest): @GlobalFormRecognizerAccountPreparer() def test_content_bad_endpoint(self, resource_group, location, form_recognizer_account, form_recognizer_account_key): with open(self.invoice_pdf, "rb") as fd: myfile = fd.read() with self.assertRaises(ServiceRequestError): client = FormRecognizerClient("http://notreal.azure.com", AzureKeyCredential(form_recognizer_account_key)) poller = client.begin_recognize_content(myfile) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_authentication_successful_key(self, client): with open(self.invoice_pdf, "rb") as fd: myfile = fd.read() poller = client.begin_recognize_content(myfile) result = poller.result() @GlobalFormRecognizerAccountPreparer() def test_content_authentication_bad_key(self, resource_group, location, form_recognizer_account, form_recognizer_account_key): client = FormRecognizerClient(form_recognizer_account, AzureKeyCredential("xxxx")) with self.assertRaises(ClientAuthenticationError): poller = client.begin_recognize_content(b"xx", content_type="application/pdf") @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_passing_enum_content_type(self, client): with open(self.invoice_pdf, "rb") as fd: myfile = fd.read() poller = client.begin_recognize_content( myfile, content_type=FormContentType.APPLICATION_PDF ) result = poller.result() self.assertIsNotNone(result) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_damaged_file_passed_as_bytes(self, client): damaged_pdf = b"\x25\x50\x44\x46\x55\x55\x55" # still has correct bytes to be recognized as PDF with self.assertRaises(HttpResponseError): poller = client.begin_recognize_content( damaged_pdf, ) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_damaged_file_bytes_fails_autodetect_content_type(self, client): damaged_pdf = b"\x50\x44\x46\x55\x55\x55" # doesn't match any magic file numbers with self.assertRaises(ValueError): poller = client.begin_recognize_content( damaged_pdf, ) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_damaged_file_passed_as_bytes_io(self, client): damaged_pdf = BytesIO(b"\x25\x50\x44\x46\x55\x55\x55") # still has correct bytes to be recognized as PDF with self.assertRaises(HttpResponseError): poller = client.begin_recognize_content( damaged_pdf, ) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_damaged_file_bytes_io_fails_autodetect(self, client): damaged_pdf = BytesIO(b"\x50\x44\x46\x55\x55\x55") # doesn't match any magic file numbers with self.assertRaises(ValueError): poller = client.begin_recognize_content( damaged_pdf, ) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_blank_page(self, client): with open(self.blank_pdf, "rb") as stream: poller = client.begin_recognize_content( stream, ) result = poller.result() self.assertIsNotNone(result) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_passing_bad_content_type_param_passed(self, client): with open(self.invoice_pdf, "rb") as fd: myfile = fd.read() with self.assertRaises(ValueError): poller = client.begin_recognize_content( myfile, content_type="application/jpeg" ) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_stream_passing_url(self, client): with self.assertRaises(TypeError): poller = client.begin_recognize_content("https://badurl.jpg", content_type="application/json") @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_auto_detect_unsupported_stream_content(self, client): with open(self.unsupported_content_py, "rb") as fd: myfile = fd.read() with self.assertRaises(ValueError): poller = client.begin_recognize_content( myfile ) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_stream_transform_pdf(self, client): with open(self.invoice_pdf, "rb") as fd: myform = fd.read() responses = [] def callback(raw_response, _, headers): analyze_result = client._deserialize(AnalyzeOperationResult, raw_response) extracted_layout = prepare_content_result(analyze_result) responses.append(analyze_result) responses.append(extracted_layout) poller = client.begin_recognize_content(myform, cls=callback) result = poller.result() raw_response = responses[0] layout = responses[1] page_results = raw_response.analyze_result.page_results read_results = raw_response.analyze_result.read_results # Check form pages self.assertFormPagesTransformCorrect(layout, read_results, page_results) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_stream_pdf(self, client): with open(self.invoice_pdf, "rb") as fd: myform = fd.read() poller = client.begin_recognize_content(myform) result = poller.result() self.assertEqual(len(result), 1) layout = result[0] self.assertEqual(layout.page_number, 1) self.assertFormPagesHasValues(result) self.assertEqual(layout.tables[0].row_count, 3) self.assertEqual(layout.tables[0].column_count, 6) self.assertEqual(layout.tables[0].page_number, 1) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_stream_transform_jpg(self, client): with open(self.form_jpg, "rb") as fd: myform = fd.read() responses = [] def callback(raw_response, _, headers): analyze_result = client._deserialize(AnalyzeOperationResult, raw_response) extracted_layout = prepare_content_result(analyze_result) responses.append(analyze_result) responses.append(extracted_layout) poller = client.begin_recognize_content(myform, cls=callback) result = poller.result() raw_response = responses[0] layout = responses[1] page_results = raw_response.analyze_result.page_results read_results = raw_response.analyze_result.read_results # Check form pages self.assertFormPagesTransformCorrect(layout, read_results, page_results) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_stream_jpg(self, client): with open(self.form_jpg, "rb") as stream: poller = client.begin_recognize_content(stream) result = poller.result() self.assertEqual(len(result), 1) layout = result[0] self.assertEqual(layout.page_number, 1) self.assertFormPagesHasValues(result) self.assertEqual(layout.tables[0].row_count, 5) self.assertEqual(layout.tables[0].column_count, 5) self.assertEqual(layout.tables[1].row_count, 4) self.assertEqual(layout.tables[1].column_count, 2) self.assertEqual(layout.tables[0].page_number, 1) self.assertEqual(layout.tables[1].page_number, 1) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_multipage(self, client): with open(self.multipage_invoice_pdf, "rb") as fd: invoice = fd.read() poller = client.begin_recognize_content(invoice) result = poller.result() self.assertEqual(len(result), 3) self.assertFormPagesHasValues(result) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_multipage_transform(self, client): with open(self.multipage_invoice_pdf, "rb") as fd: myform = fd.read() responses = [] def callback(raw_response, _, headers): analyze_result = client._deserialize(AnalyzeOperationResult, raw_response) extracted_layout = prepare_content_result(analyze_result) responses.append(analyze_result) responses.append(extracted_layout) poller = client.begin_recognize_content(myform, cls=callback) result = poller.result() raw_response = responses[0] layout = responses[1] page_results = raw_response.analyze_result.page_results read_results = raw_response.analyze_result.read_results # Check form pages self.assertFormPagesTransformCorrect(layout, read_results, page_results) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() @pytest.mark.live_test_only def test_content_continuation_token(self, client): with open(self.form_jpg, "rb") as fd: myfile = fd.read() initial_poller = client.begin_recognize_content(myfile) cont_token = initial_poller.continuation_token() poller = client.begin_recognize_content(None, continuation_token=cont_token) result = poller.result() self.assertIsNotNone(result) initial_poller.wait() # necessary so azure-devtools doesn't throw assertion error @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_multipage_table_span_pdf(self, client): with open(self.multipage_table_pdf, "rb") as stream: poller = client.begin_recognize_content(stream) result = poller.result() self.assertEqual(len(result), 2) layout = result[0] self.assertEqual(layout.page_number, 1) self.assertEqual(len(layout.tables), 2) self.assertEqual(layout.tables[0].row_count, 29) self.assertEqual(layout.tables[0].column_count, 4) self.assertEqual(layout.tables[0].page_number, 1) self.assertEqual(layout.tables[1].row_count, 6) self.assertEqual(layout.tables[1].column_count, 5) self.assertEqual(layout.tables[1].page_number, 1) layout = result[1] self.assertEqual(len(layout.tables), 1) self.assertEqual(layout.page_number, 2) self.assertEqual(layout.tables[0].row_count, 23) self.assertEqual(layout.tables[0].column_count, 5) self.assertEqual(layout.tables[0].page_number, 2) self.assertFormPagesHasValues(result) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_multipage_table_span_transform(self, client): with open(self.multipage_table_pdf, "rb") as fd: myform = fd.read() responses = [] def callback(raw_response, _, headers): analyze_result = client._deserialize(AnalyzeOperationResult, raw_response) extracted_layout = prepare_content_result(analyze_result) responses.append(analyze_result) responses.append(extracted_layout) poller = client.begin_recognize_content(myform, cls=callback) result = poller.result() raw_response = responses[0] layout = responses[1] page_results = raw_response.analyze_result.page_results read_results = raw_response.analyze_result.read_results # Check form pages self.assertFormPagesTransformCorrect(layout, read_results, page_results) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_selection_marks(self, client): with open(self.selection_form_pdf, "rb") as fd: myform = fd.read() poller = client.begin_recognize_content(myform) result = poller.result() self.assertEqual(len(result), 1) layout = result[0] self.assertEqual(layout.page_number, 1) self.assertFormPagesHasValues(result) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer(client_kwargs={"api_version": FormRecognizerApiVersion.V2_0}) def test_content_selection_marks_v2(self, client): with open(self.selection_form_pdf, "rb") as fd: myform = fd.read() poller = client.begin_recognize_content(myform) result = poller.result() self.assertEqual(len(result), 1) layout = result[0] self.assertEqual(layout.page_number, 1) self.assertFormPagesHasValues(result) @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_specify_pages(self, client): with open(self.multipage_invoice_pdf, "rb") as fd: myform = fd.read() poller = client.begin_recognize_content(myform, pages=["1"]) result = poller.result() assert len(result) == 1 poller = client.begin_recognize_content(myform, pages=["1", "3"]) result = poller.result() assert len(result) == 2 poller = client.begin_recognize_content(myform, pages=["1-2"]) result = poller.result() assert len(result) == 2 poller = client.begin_recognize_content(myform, pages=["1-2", "3"]) result = poller.result() assert len(result) == 3 @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_language_specified(self, client): with open(self.form_jpg, "rb") as fd: myfile = fd.read() poller = client.begin_recognize_content(myfile, language="de") assert 'de' == poller._polling_method._initial_response.http_response.request.query['language'] poller.wait() @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer() def test_content_language_error(self, client): with open(self.form_jpg, "rb") as fd: myfile = fd.read() with pytest.raises(HttpResponseError) as e: client.begin_recognize_content(myfile, language="not a language") assert "NotSupportedLanguage" == e.value.error.code @GlobalFormRecognizerAccountPreparer() @GlobalClientPreparer(client_kwargs={"api_version": FormRecognizerApiVersion.V2_0}) def test_content_language_v2(self, client): with open(self.form_jpg, "rb") as fd: myfile = fd.read() with pytest.raises(ValueError) as e: client.begin_recognize_content(myfile, language="en") assert "'language' is only available for API version V2_1_PREVIEW and up" in str(e.value)