diff --git a/src/pdfrest/client.py b/src/pdfrest/client.py index 505f1288..7fd45ac6 100644 --- a/src/pdfrest/client.py +++ b/src/pdfrest/client.py @@ -95,10 +95,12 @@ PdfCompressPayload, PdfDecryptPayload, PdfEncryptPayload, + PdfExportFormDataPayload, PdfFlattenAnnotationsPayload, PdfFlattenFormsPayload, PdfFlattenLayersPayload, PdfFlattenTransparenciesPayload, + PdfImportFormDataPayload, PdfInfoPayload, PdfLinearizePayload, PdfMergePayload, @@ -127,6 +129,7 @@ ALL_PDF_INFO_QUERIES, BmpColorModel, CompressionLevel, + ExportDataFormat, ExtractTextGranularity, FlattenQuality, GifColorModel, @@ -2846,6 +2849,68 @@ def convert_to_word( timeout=timeout, ) + def import_form_data( + self, + file: PdfRestFile | Sequence[PdfRestFile], + data_file: PdfRestFile | Sequence[PdfRestFile], + *, + output: str | None = None, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Import form data from a data file into an existing PDF with form fields.""" + + payload: dict[str, Any] = {"files": file, "data_file": data_file} + if output is not None: + payload["output"] = output + + return self._post_file_operation( + endpoint="/pdf-with-imported-form-data", + payload=payload, + payload_model=PdfImportFormDataPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + + def export_form_data( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + data_format: ExportDataFormat, + output: str | None = None, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Export form data from a PDF into an external data file. + + `data_format` support depends on detected form type: + - AcroForm PDFs: `xfdf`, `fdf`, `xml` + - XFA PDFs: `xfd`, `xdp`, `xml` + """ + + payload: dict[str, Any] = { + "files": file, + "data_format": data_format, + } + if output is not None: + payload["output"] = output + + return self._post_file_operation( + endpoint="/exported-form-data", + payload=payload, + payload_model=PdfExportFormDataPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + def flatten_pdf_forms( self, file: PdfRestFile | Sequence[PdfRestFile], @@ -4549,6 +4614,68 @@ async def convert_to_word( timeout=timeout, ) + async def import_form_data( + self, + file: PdfRestFile | Sequence[PdfRestFile], + data_file: PdfRestFile | Sequence[PdfRestFile], + *, + output: str | None = None, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Asynchronously import form data from a data file into a PDF.""" + + payload: dict[str, Any] = {"files": file, "data_file": data_file} + if output is not None: + payload["output"] = output + + return await self._post_file_operation( + endpoint="/pdf-with-imported-form-data", + payload=payload, + payload_model=PdfImportFormDataPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + + async def export_form_data( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + data_format: ExportDataFormat, + output: str | None = None, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Asynchronously export form data from a PDF into a data file. + + `data_format` support depends on detected form type: + - AcroForm PDFs: `xfdf`, `fdf`, `xml` + - XFA PDFs: `xfd`, `xdp`, `xml` + """ + + payload: dict[str, Any] = { + "files": file, + "data_format": data_format, + } + if output is not None: + payload["output"] = output + + return await self._post_file_operation( + endpoint="/exported-form-data", + payload=payload, + payload_model=PdfExportFormDataPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + async def flatten_pdf_forms( self, file: PdfRestFile | Sequence[PdfRestFile], diff --git a/src/pdfrest/models/_internal.py b/src/pdfrest/models/_internal.py index 65f65197..8e31dba5 100644 --- a/src/pdfrest/models/_internal.py +++ b/src/pdfrest/models/_internal.py @@ -23,6 +23,7 @@ from pdfrest.types.public import PdfRedactionPreset from ..types import ( + ExportDataFormat, HtmlPageOrientation, HtmlPageSize, HtmlWebLayout, @@ -1318,6 +1319,80 @@ class PdfFlattenFormsPayload(BaseModel): ] = None +class PdfImportFormDataPayload(BaseModel): + """Adapt caller options into a pdfRest-ready import-form-data request payload.""" + + files: Annotated[ + list[PdfRestFile], + Field( + min_length=1, + max_length=1, + validation_alias=AliasChoices("file", "files"), + serialization_alias="id", + ), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types("application/pdf", error_msg="Must be a PDF file") + ), + PlainSerializer(_serialize_as_first_file_id), + ] + data_file: Annotated[ + list[PdfRestFile], + Field( + min_length=1, + max_length=1, + validation_alias=AliasChoices("data_file", "data_files"), + serialization_alias="data_file_id", + ), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types( + "application/xml", + "text/xml", + "application/vnd.fdf", + "application/vnd.adobe.xfdf", + "application/vnd.adobe.xdp+xml", + "application/vnd.adobe.xfd+xml", + error_msg="Data file must be an XFDF, XDP, XFD, FDF, or XML file", + ) + ), + PlainSerializer(_serialize_as_first_file_id), + ] + output: Annotated[ + str | None, + Field(serialization_alias="output", min_length=1, default=None), + AfterValidator(_validate_output_prefix), + ] = None + + +class PdfExportFormDataPayload(BaseModel): + """Adapt caller options into a pdfRest-ready export-form-data request payload.""" + + files: Annotated[ + list[PdfRestFile], + Field( + min_length=1, + max_length=1, + validation_alias=AliasChoices("file", "files"), + serialization_alias="id", + ), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types("application/pdf", error_msg="Must be a PDF file") + ), + PlainSerializer(_serialize_as_first_file_id), + ] + data_format: Annotated[ + ExportDataFormat, + Field(serialization_alias="data_format"), + ] + output: Annotated[ + str | None, + Field(serialization_alias="output", min_length=1, default=None), + AfterValidator(_validate_output_prefix), + ] = None + + class PdfCompressPayload(BaseModel): """Adapt caller options into a pdfRest-ready compress request payload.""" diff --git a/src/pdfrest/types/__init__.py b/src/pdfrest/types/__init__.py index b7db11ae..61360937 100644 --- a/src/pdfrest/types/__init__.py +++ b/src/pdfrest/types/__init__.py @@ -6,6 +6,7 @@ ALL_PDF_RESTRICTIONS, BmpColorModel, CompressionLevel, + ExportDataFormat, ExtractTextGranularity, FlattenQuality, GifColorModel, @@ -48,6 +49,7 @@ "ALL_PDF_RESTRICTIONS", "BmpColorModel", "CompressionLevel", + "ExportDataFormat", "ExtractTextGranularity", "FlattenQuality", "GifColorModel", diff --git a/src/pdfrest/types/public.py b/src/pdfrest/types/public.py index 20c73d97..55497645 100644 --- a/src/pdfrest/types/public.py +++ b/src/pdfrest/types/public.py @@ -18,6 +18,7 @@ "ALL_PDF_RESTRICTIONS", "BmpColorModel", "CompressionLevel", + "ExportDataFormat", "ExtractTextGranularity", "FlattenQuality", "GifColorModel", @@ -167,6 +168,9 @@ class PdfMergeSource(TypedDict, total=False): JpegColorModel = Literal["rgb", "cmyk", "gray"] TiffColorModel = Literal["rgb", "rgba", "cmyk", "lab", "gray"] GraphicSmoothing = Literal["none", "all", "text", "line", "image"] +# Server accepts all values here, but enforces form-type subsets at runtime: +# AcroForm -> xfdf/fdf/xml, XFA -> xfd/xdp/xml. +ExportDataFormat = Literal["fdf", "xfdf", "xml", "xdp", "xfd"] SummaryFormat = Literal[ "overview", diff --git a/tests/live/test_live_export_form_data.py b/tests/live/test_live_export_form_data.py new file mode 100644 index 00000000..881c2771 --- /dev/null +++ b/tests/live/test_live_export_form_data.py @@ -0,0 +1,275 @@ +from __future__ import annotations + +import pytest + +from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient +from pdfrest.models import PdfRestFile +from pdfrest.types import ExportDataFormat + +from ..resources import get_test_resource_path + + +@pytest.fixture(scope="module") +def uploaded_pdf_with_forms( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> PdfRestFile: + resource = get_test_resource_path("form_with_data.pdf") + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + return client.files.create_from_paths([resource])[0] + + +@pytest.mark.parametrize( + "data_format", + [ + pytest.param("fdf", id="fdf"), + pytest.param("xfdf", id="xfdf"), + pytest.param("xml", id="xml"), + ], +) +def test_live_export_form_data_acroform( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_with_forms: PdfRestFile, + data_format: ExportDataFormat, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.export_form_data( + uploaded_pdf_with_forms, + data_format=data_format, + output=f"exported-{data_format}", + ) + + assert response.output_files + output_file = response.output_file + assert output_file.name.startswith(f"exported-{data_format}") + assert output_file.type + assert output_file.size > 0 + assert response.warning is None + assert str(uploaded_pdf_with_forms.id) in { + str(file_id) for file_id in response.input_ids + } + + +@pytest.fixture(scope="module") +def uploaded_xfa_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> PdfRestFile: + resource = get_test_resource_path("xfa.pdf") + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + return client.files.create_from_paths([resource])[0] + + +@pytest.mark.parametrize( + "data_format", + [ + pytest.param("xfd", id="xfd"), + pytest.param("xdp", id="xdp"), + pytest.param("xml", id="xml"), + ], +) +def test_live_export_form_data_xfa( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_xfa_pdf: PdfRestFile, + data_format: ExportDataFormat, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.export_form_data( + uploaded_xfa_pdf, + data_format=data_format, + output=f"exported-xfa-{data_format}", + ) + + assert response.output_files + output_file = response.output_file + assert output_file.name.startswith(f"exported-xfa-{data_format}") + assert output_file.type + assert output_file.size > 0 + assert response.warning is None + assert str(uploaded_xfa_pdf.id) in {str(file_id) for file_id in response.input_ids} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "data_format", + [ + pytest.param("fdf", id="fdf"), + pytest.param("xfdf", id="xfdf"), + pytest.param("xml", id="xml"), + ], +) +async def test_live_async_export_form_data_acroform( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_with_forms: PdfRestFile, + data_format: ExportDataFormat, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.export_form_data( + uploaded_pdf_with_forms, + data_format=data_format, + output=f"async-acro-{data_format}", + ) + + assert response.output_files + output_file = response.output_file + assert output_file.name.startswith(f"async-acro-{data_format}") + assert output_file.type + assert output_file.size > 0 + assert response.warning is None + assert str(uploaded_pdf_with_forms.id) in { + str(file_id) for file_id in response.input_ids + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "data_format", + [ + pytest.param("xfd", id="xfd"), + pytest.param("xdp", id="xdp"), + pytest.param("xml", id="xml"), + ], +) +async def test_live_async_export_form_data_xfa( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_xfa_pdf: PdfRestFile, + data_format: ExportDataFormat, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.export_form_data( + uploaded_xfa_pdf, + data_format=data_format, + output=f"async-xfa-{data_format}", + ) + + assert response.output_files + output_file = response.output_file + assert output_file.name.startswith(f"async-xfa-{data_format}") + assert output_file.type + assert output_file.size > 0 + assert response.warning is None + assert str(uploaded_xfa_pdf.id) in {str(file_id) for file_id in response.input_ids} + + +@pytest.mark.parametrize( + "invalid_format", + [ + pytest.param("xdp", id="xdp"), + pytest.param("xfd", id="xfd"), + ], +) +def test_live_export_form_data_invalid_format_for_pdf_type( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_with_forms: PdfRestFile, + invalid_format: ExportDataFormat, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)(acroform|data_format)"), + ): + client.export_form_data( + uploaded_pdf_with_forms, + data_format=invalid_format, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "invalid_format", + [ + pytest.param("xdp", id="xdp"), + pytest.param("xfd", id="xfd"), + ], +) +async def test_live_async_export_form_data_invalid_format_for_pdf_type( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_with_forms: PdfRestFile, + invalid_format: ExportDataFormat, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises(PdfRestApiError, match=r"(?i)(acroform|data_format)"): + await client.export_form_data( + uploaded_pdf_with_forms, + data_format=invalid_format, + ) + + +@pytest.mark.parametrize( + "invalid_format", + [ + pytest.param("xfdf", id="xfdf"), + pytest.param("fdf", id="fdf"), + ], +) +def test_live_export_form_data_invalid_format_for_xfa( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_xfa_pdf: PdfRestFile, + invalid_format: ExportDataFormat, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)(xfa|data_format)"), + ): + client.export_form_data( + uploaded_xfa_pdf, + data_format=invalid_format, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "invalid_format", + [ + pytest.param("xfdf", id="xfdf"), + pytest.param("fdf", id="fdf"), + ], +) +async def test_live_async_export_form_data_invalid_format_for_xfa( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_xfa_pdf: PdfRestFile, + invalid_format: ExportDataFormat, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises(PdfRestApiError, match=r"(?i)(xfa|data_format)"): + await client.export_form_data( + uploaded_xfa_pdf, + data_format=invalid_format, + ) diff --git a/tests/live/test_live_import_form_data.py b/tests/live/test_live_import_form_data.py new file mode 100644 index 00000000..f20ca1b2 --- /dev/null +++ b/tests/live/test_live_import_form_data.py @@ -0,0 +1,260 @@ +from __future__ import annotations + +import pytest + +from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient +from pdfrest.models import PdfRestFile + +from ..resources import get_test_resource_path + +IMPORT_FORM_DATA_SUCCESS_CASES = ( + pytest.param( + ("acroform.pdf", "test-data-acro.xml", None), + id="acro-xml", + ), + pytest.param( + ("acroform.pdf", "test-data-acro.xfdf", None), + id="acro-xfdf", + ), + pytest.param( + ("xfa.pdf", "test-data-xfa.xml", None), + id="xfa-xml", + ), + pytest.param( + ("xfa.pdf", "test-data-xfa.xdp", None), + id="xfa-xdp", + ), + pytest.param( + ( + "xfa.pdf", + "test-data-xfa.xfd", + "application/vnd.adobe.xfd+xml", + ), + id="xfa-xfd", + ), +) + + +@pytest.fixture(scope="module", params=IMPORT_FORM_DATA_SUCCESS_CASES) +def uploaded_success_import_form_data_case( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + request: pytest.FixtureRequest, +) -> tuple[PdfRestFile, PdfRestFile]: + pdf_resource_name, data_resource_name, forced_data_mime = request.param + pdf_resource = get_test_resource_path(pdf_resource_name) + data_resource = get_test_resource_path(data_resource_name) + + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + input_file = client.files.create_from_paths([pdf_resource])[0] + data_file = client.files.create_from_paths([data_resource])[0] + + # pdfRest currently reports .xfd uploads as application/octet-stream. + # Override the local MIME metadata so the request can exercise xfd imports. + if forced_data_mime is not None: + data_file = data_file.model_copy(update={"type": forced_data_mime}) + + return input_file, data_file + + +@pytest.fixture(scope="module") +def uploaded_acro_import_pair( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> tuple[PdfRestFile, PdfRestFile]: + acro_pdf = get_test_resource_path("acroform.pdf") + acro_data = get_test_resource_path("test-data-acro.xml") + + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + input_file = client.files.create_from_paths([acro_pdf])[0] + data_file = client.files.create_from_paths([acro_data])[0] + + return input_file, data_file + + +@pytest.fixture(scope="module") +def uploaded_acro_fdf_import_pair( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> tuple[PdfRestFile, PdfRestFile]: + acro_pdf = get_test_resource_path("acroform.pdf") + acro_fdf = get_test_resource_path("test-data-acro.fdf") + + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + input_file = client.files.create_from_paths([acro_pdf])[0] + data_file = client.files.create_from_paths([acro_fdf])[0] + + return input_file, data_file + + +@pytest.mark.parametrize( + "output_name", + [ + pytest.param(None, id="default-output"), + pytest.param("imported-form", id="custom-output"), + ], +) +def test_live_import_form_data( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_success_import_form_data_case: tuple[PdfRestFile, PdfRestFile], + output_name: str | None, +) -> None: + kwargs: dict[str, str] = {} + if output_name is not None: + kwargs["output"] = output_name + + uploaded_pdf_with_forms, uploaded_form_data_file = ( + uploaded_success_import_form_data_case + ) + + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.import_form_data( + uploaded_pdf_with_forms, + uploaded_form_data_file, + **kwargs, + ) + + assert response.output_files + output_file = response.output_file + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert response.warning is None + assert str(uploaded_pdf_with_forms.id) in { + str(file_id) for file_id in response.input_ids + } + assert str(uploaded_form_data_file.id) in { + str(file_id) for file_id in response.input_ids + } + if output_name is not None: + assert output_file.name.startswith(output_name) + else: + assert output_file.name.endswith(".pdf") + + +@pytest.mark.asyncio +async def test_live_async_import_form_data_success( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_success_import_form_data_case: tuple[PdfRestFile, PdfRestFile], +) -> None: + uploaded_pdf_with_forms, uploaded_form_data_file = ( + uploaded_success_import_form_data_case + ) + + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.import_form_data( + uploaded_pdf_with_forms, + uploaded_form_data_file, + output="async-imported", + ) + + assert response.output_files + output_file = response.output_file + assert output_file.name.startswith("async-imported") + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert response.warning is None + assert str(uploaded_pdf_with_forms.id) in { + str(file_id) for file_id in response.input_ids + } + assert str(uploaded_form_data_file.id) in { + str(file_id) for file_id in response.input_ids + } + + +def test_live_import_form_data_fdf_server_error( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_acro_fdf_import_pair: tuple[PdfRestFile, PdfRestFile], +) -> None: + uploaded_pdf_with_forms, uploaded_form_data_file = uploaded_acro_fdf_import_pair + + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises( + PdfRestApiError, + match=r"(?i)(issue processing|filled correctly|corrupted)", + ), + ): + client.import_form_data(uploaded_pdf_with_forms, uploaded_form_data_file) + + +@pytest.mark.asyncio +async def test_live_async_import_form_data_fdf_server_error( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_acro_fdf_import_pair: tuple[PdfRestFile, PdfRestFile], +) -> None: + uploaded_pdf_with_forms, uploaded_form_data_file = uploaded_acro_fdf_import_pair + + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises( + PdfRestApiError, + match=r"(?i)(issue processing|filled correctly|corrupted)", + ): + await client.import_form_data( + uploaded_pdf_with_forms, uploaded_form_data_file + ) + + +def test_live_import_form_data_invalid_data_file_id( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_acro_import_pair: tuple[PdfRestFile, PdfRestFile], +) -> None: + uploaded_pdf_with_forms, uploaded_form_data_file = uploaded_acro_import_pair + + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)(data|id)"), + ): + client.import_form_data( + uploaded_pdf_with_forms, + uploaded_form_data_file, + extra_body={"data_file_id": "ffffffff-ffff-ffff-ffff-ffffffffffff"}, + ) + + +@pytest.mark.asyncio +async def test_live_async_import_form_data_invalid_data_file_id( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_acro_import_pair: tuple[PdfRestFile, PdfRestFile], +) -> None: + uploaded_pdf_with_forms, uploaded_form_data_file = uploaded_acro_import_pair + + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises(PdfRestApiError, match=r"(?i)(data|id)"): + await client.import_form_data( + uploaded_pdf_with_forms, + uploaded_form_data_file, + extra_body={"data_file_id": "00000000-0000-0000-0000-000000000000"}, + ) diff --git a/tests/resources/acroform.pdf b/tests/resources/acroform.pdf new file mode 100644 index 00000000..f8a37c6b Binary files /dev/null and b/tests/resources/acroform.pdf differ diff --git a/tests/resources/form_data.xml b/tests/resources/form_data.xml new file mode 100644 index 00000000..9aa268dd --- /dev/null +++ b/tests/resources/form_data.xml @@ -0,0 +1,14 @@ + + + + + Jamie + + + Appleseed + + + Green + + + diff --git a/tests/resources/test-data-acro.fdf b/tests/resources/test-data-acro.fdf new file mode 100644 index 00000000..68a80087 --- /dev/null +++ b/tests/resources/test-data-acro.fdf @@ -0,0 +1,19 @@ +%FDF-1.2 +%âãÏÓ +1 0 obj +<< +/F (minimal_contact_form_acroform.pdf) +/Fields [ + << /T (Name) /V (Alice Example) >> + << /T (Email) /V (alice@example.com) >> + << /T (Phone) /V (+1 312 555 0100) >> + << /T (Message) /V (Hello!\nThis is a sample message populated via FDF.) >> + << /T (Subscribe) /V /Yes >> +] +>> +endobj +trailer +<< +/Root 1 0 R +>> +%%EOF diff --git a/tests/resources/test-data-acro.xfdf b/tests/resources/test-data-acro.xfdf new file mode 100644 index 00000000..8d2a7d76 --- /dev/null +++ b/tests/resources/test-data-acro.xfdf @@ -0,0 +1,11 @@ + + + + Alice Example + alice@example.com + +1 312 555 0100 + Hello! +This is a sample message populated via XFDF. + Yes + + diff --git a/tests/resources/test-data-acro.xml b/tests/resources/test-data-acro.xml new file mode 100644 index 00000000..32ebb856 --- /dev/null +++ b/tests/resources/test-data-acro.xml @@ -0,0 +1,9 @@ + + + Alice Example + alice@example.com + +1 312 555 0100 + Hello! +This is a sample message populated via XML (convert to FDF/XFDF as needed). + true + diff --git a/tests/resources/test-data-xfa.xdp b/tests/resources/test-data-xfa.xdp new file mode 100644 index 00000000..39dcfbcb --- /dev/null +++ b/tests/resources/test-data-xfa.xdp @@ -0,0 +1,292 @@ + + + +JohnDoexxxxxxxxxJaneDoexxxxxxxxx
123 Oak StAnytown, USA 00000
HoH + diff --git a/tests/resources/test-data-xfa.xfd b/tests/resources/test-data-xfa.xfd new file mode 100644 index 00000000..5a3f984f --- /dev/null +++ b/tests/resources/test-data-xfa.xfd @@ -0,0 +1,300 @@ + + +JohnDoexxxxxxxxxJaneDoexxxxxxxxx
123 Oak StAnytown, USA 00000
HoH diff --git a/tests/resources/test-data-xfa.xml b/tests/resources/test-data-xfa.xml new file mode 100644 index 00000000..3319f551 --- /dev/null +++ b/tests/resources/test-data-xfa.xml @@ -0,0 +1,284 @@ + +JohnDoexxxxxxxxxJaneDoexxxxxxxxx
123 Oak StAnytown, USA 00000
HoH diff --git a/tests/test_export_form_data.py b/tests/test_export_form_data.py new file mode 100644 index 00000000..f590e504 --- /dev/null +++ b/tests/test_export_form_data.py @@ -0,0 +1,379 @@ +from __future__ import annotations + +import json + +import httpx +import pytest +from pydantic import ValidationError + +from pdfrest import AsyncPdfRestClient, PdfRestClient +from pdfrest.models import PdfRestFile, PdfRestFileBasedResponse, PdfRestFileID +from pdfrest.models._internal import PdfExportFormDataPayload +from pdfrest.types import ExportDataFormat + +from .graphics_test_helpers import ( + ASYNC_API_KEY, + VALID_API_KEY, + build_file_info_payload, + make_pdf_file, +) + + +@pytest.mark.parametrize( + ("data_format", "output_name", "mime_type"), + [ + pytest.param("fdf", "exported-data.fdf", "application/vnd.fdf", id="fdf"), + pytest.param( + "xfdf", + "exported-data.xfdf", + "application/vnd.adobe.xfdf", + id="xfdf", + ), + pytest.param("xml", "exported-data.xml", "application/xml", id="xml"), + pytest.param( + "xdp", + "exported-data.xdp", + "application/vnd.adobe.xdp+xml", + id="xdp", + ), + pytest.param( + "xfd", + "exported-data.xfd", + "application/vnd.adobe.xfd+xml", + id="xfd", + ), + ], +) +def test_export_form_data_success( + monkeypatch: pytest.MonkeyPatch, + data_format: ExportDataFormat, + output_name: str, + mime_type: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfExportFormDataPayload.model_validate( + {"files": [input_file], "data_format": data_format, "output": "exported-data"} + ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True) + + seen: dict[str, int] = {"post": 0, "get": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/exported-form-data": + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + return httpx.Response( + 200, + json={ + "inputId": [input_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + seen["get"] += 1 + assert request.url.params["format"] == "info" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + output_name, + mime_type, + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + with PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client: + response = client.export_form_data( + input_file, + data_format=data_format, + output="exported-data", + ) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == output_name + assert response.output_file.type == mime_type + assert str(response.input_id) == str(input_file.id) + assert response.warning is None + + +def test_export_form_data_request_customization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + captured_timeout: dict[str, float | dict[str, float] | None] = {} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/exported-form-data": + assert request.url.params["trace"] == "true" + assert request.headers["X-Debug"] == "sync" + captured_timeout["value"] = request.extensions.get("timeout") + payload = json.loads(request.content.decode("utf-8")) + assert payload["id"] == str(input_file.id) + assert payload["data_format"] == "fdf" + assert payload["output"] == "custom-data" + assert payload["flag"] == "yes" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + assert request.url.params["format"] == "info" + assert request.url.params["trace"] == "true" + assert request.headers["X-Debug"] == "sync" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "custom-data.fdf", + "application/vnd.fdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + with PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client: + response = client.export_form_data( + input_file, + data_format="fdf", + output="custom-data", + extra_query={"trace": "true"}, + extra_headers={"X-Debug": "sync"}, + extra_body={"flag": "yes"}, + timeout=0.37, + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "custom-data.fdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.37) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.37) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("data_format", "output_name", "mime_type"), + [ + pytest.param("fdf", "async-data.fdf", "application/vnd.fdf", id="fdf"), + pytest.param( + "xfdf", + "async-data.xfdf", + "application/vnd.adobe.xfdf", + id="xfdf", + ), + pytest.param("xml", "async-data.xml", "application/xml", id="xml"), + pytest.param( + "xdp", + "async-data.xdp", + "application/vnd.adobe.xdp+xml", + id="xdp", + ), + pytest.param( + "xfd", + "async-data.xfd", + "application/vnd.adobe.xfd+xml", + id="xfd", + ), + ], +) +async def test_async_export_form_data_success( + monkeypatch: pytest.MonkeyPatch, + data_format: ExportDataFormat, + output_name: str, + mime_type: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfExportFormDataPayload.model_validate( + {"files": [input_file], "data_format": data_format} + ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True) + + seen: dict[str, int] = {"post": 0, "get": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/exported-form-data": + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + return httpx.Response( + 200, + json={ + "inputId": [input_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + seen["get"] += 1 + assert request.url.params["format"] == "info" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + output_name, + mime_type, + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + response = await client.export_form_data(input_file, data_format=data_format) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == output_name + assert response.output_file.type == mime_type + assert str(response.input_id) == str(input_file.id) + + +@pytest.mark.asyncio +async def test_async_export_form_data_request_customization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + output_id = str(PdfRestFileID.generate()) + captured_timeout: dict[str, float | dict[str, float] | None] = {} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/exported-form-data": + assert request.url.params["trace"] == "async" + assert request.headers["X-Debug"] == "async" + captured_timeout["value"] = request.extensions.get("timeout") + payload = json.loads(request.content.decode("utf-8")) + assert payload["id"] == str(input_file.id) + assert payload["data_format"] == "xml" + assert payload["note"] == "details" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + assert request.url.params["format"] == "info" + assert request.url.params["trace"] == "async" + assert request.headers["X-Debug"] == "async" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async-custom.xml", + "application/xml", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + response = await client.export_form_data( + input_file, + data_format="xml", + extra_query={"trace": "async"}, + extra_headers={"X-Debug": "async"}, + extra_body={"note": "details"}, + timeout=0.62, + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "async-custom.xml" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.62) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.62) + + +def test_export_form_data_validation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + pdf_file = make_pdf_file(PdfRestFileID.generate(1)) + png_file = PdfRestFile.model_validate( + build_file_info_payload( + PdfRestFileID.generate(), + "example.png", + "image/png", + ) + ) + transport = httpx.MockTransport(lambda request: (_ for _ in ()).throw(RuntimeError)) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises(ValidationError, match="Must be a PDF file"), + ): + client.export_form_data(png_file, data_format="xml") + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises(ValidationError, match="Input should be 'fdf'"), + ): + client.export_form_data(pdf_file, data_format="yaml") # type: ignore[arg-type] + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises( + ValidationError, match="List should have at most 1 item after validation" + ), + ): + client.export_form_data( + [pdf_file, make_pdf_file(PdfRestFileID.generate())], data_format="xml" + ) + + +@pytest.mark.asyncio +async def test_async_export_form_data_validation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + pdf_file = make_pdf_file(PdfRestFileID.generate(1)) + png_file = PdfRestFile.model_validate( + build_file_info_payload( + PdfRestFileID.generate(), + "example.png", + "image/png", + ) + ) + transport = httpx.MockTransport(lambda request: (_ for _ in ()).throw(RuntimeError)) + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises(ValidationError, match="Must be a PDF file"): + await client.export_form_data(png_file, data_format="xml") + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises(ValidationError, match="Input should be 'fdf'"): + await client.export_form_data( + pdf_file, + data_format="yaml", # type: ignore[arg-type] + ) + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises( + ValidationError, match="List should have at most 1 item after validation" + ): + await client.export_form_data( + [pdf_file, make_pdf_file(PdfRestFileID.generate())], + data_format="xml", + ) diff --git a/tests/test_import_form_data.py b/tests/test_import_form_data.py new file mode 100644 index 00000000..10519bb3 --- /dev/null +++ b/tests/test_import_form_data.py @@ -0,0 +1,406 @@ +from __future__ import annotations + +import json + +import httpx +import pytest +from pydantic import ValidationError + +from pdfrest import AsyncPdfRestClient, PdfRestClient +from pdfrest.models import PdfRestFile, PdfRestFileBasedResponse, PdfRestFileID +from pdfrest.models._internal import PdfImportFormDataPayload + +from .graphics_test_helpers import ( + ASYNC_API_KEY, + VALID_API_KEY, + build_file_info_payload, + make_pdf_file, +) + +ACCEPTED_IMPORT_DATA_FILE_MIME_TYPES = ( + pytest.param("application/xml", id="application-xml"), + pytest.param("text/xml", id="text-xml"), + pytest.param("application/vnd.fdf", id="application-vnd-fdf"), + pytest.param( + "application/vnd.adobe.xfdf", + id="application-vnd-adobe-xfdf", + ), + pytest.param( + "application/vnd.adobe.xdp+xml", + id="application-vnd-adobe-xdp+xml", + ), + pytest.param( + "application/vnd.adobe.xfd+xml", + id="application-vnd-adobe-xfd+xml", + ), +) + + +def _make_data_file( + file_id: PdfRestFileID, *, mime_type: str = "application/xml" +) -> PdfRestFile: + file_name = "form-data.xml" + if mime_type == "application/vnd.fdf": + file_name = "form-data.fdf" + elif mime_type == "application/vnd.adobe.xfdf": + file_name = "form-data.xfdf" + elif mime_type == "application/vnd.adobe.xdp+xml": + file_name = "form-data.xdp" + elif mime_type == "application/vnd.adobe.xfd+xml": + file_name = "form-data.xfd" + + return PdfRestFile.model_validate( + build_file_info_payload(file_id, file_name, mime_type) + ) + + +@pytest.mark.parametrize("data_file_mime", ACCEPTED_IMPORT_DATA_FILE_MIME_TYPES) +def test_import_form_data_success( + monkeypatch: pytest.MonkeyPatch, + data_file_mime: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + data_file = _make_data_file(PdfRestFileID.generate(2), mime_type=data_file_mime) + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfImportFormDataPayload.model_validate( + {"files": [input_file], "data_file": [data_file], "output": "filled-form"} + ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True) + + seen: dict[str, int] = {"post": 0, "get": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if ( + request.method == "POST" + and request.url.path == "/pdf-with-imported-form-data" + ): + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, data_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + seen["get"] += 1 + assert request.url.params["format"] == "info" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "filled-form.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + with PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client: + response = client.import_form_data( + input_file, + data_file, + output="filled-form", + ) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "filled-form.pdf" + assert response.output_file.type == "application/pdf" + assert str(input_file.id) in {str(file_id) for file_id in response.input_ids} + assert str(data_file.id) in {str(file_id) for file_id in response.input_ids} + assert response.warning is None + + +def test_import_form_data_request_customization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + data_file = _make_data_file(PdfRestFileID.generate(2)) + output_id = str(PdfRestFileID.generate()) + captured_timeout: dict[str, float | dict[str, float] | None] = {} + + def handler(request: httpx.Request) -> httpx.Response: + if ( + request.method == "POST" + and request.url.path == "/pdf-with-imported-form-data" + ): + assert request.url.params["trace"] == "true" + assert request.headers["X-Debug"] == "sync" + captured_timeout["value"] = request.extensions.get("timeout") + payload = json.loads(request.content.decode("utf-8")) + assert payload["id"] == str(input_file.id) + assert payload["data_file_id"] == str(data_file.id) + assert payload["output"] == "custom-output" + assert payload["debug"] == "flag" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, data_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + assert request.url.params["format"] == "info" + assert request.url.params["trace"] == "true" + assert request.headers["X-Debug"] == "sync" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "custom.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + with PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client: + response = client.import_form_data( + input_file, + data_file, + output="custom-output", + extra_query={"trace": "true"}, + extra_headers={"X-Debug": "sync"}, + extra_body={"debug": "flag"}, + timeout=0.41, + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "custom.pdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.41) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.41) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("data_file_mime", ACCEPTED_IMPORT_DATA_FILE_MIME_TYPES) +async def test_async_import_form_data_success( + monkeypatch: pytest.MonkeyPatch, + data_file_mime: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + data_file = _make_data_file(PdfRestFileID.generate(1), mime_type=data_file_mime) + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfImportFormDataPayload.model_validate( + {"files": [input_file], "data_file": [data_file]} + ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True) + + seen: dict[str, int] = {"post": 0, "get": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if ( + request.method == "POST" + and request.url.path == "/pdf-with-imported-form-data" + ): + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, data_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + seen["get"] += 1 + assert request.url.params["format"] == "info" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + response = await client.import_form_data(input_file, data_file) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "async.pdf" + assert response.output_file.type == "application/pdf" + assert str(input_file.id) in {str(file_id) for file_id in response.input_ids} + assert str(data_file.id) in {str(file_id) for file_id in response.input_ids} + + +@pytest.mark.asyncio +async def test_async_import_form_data_request_customization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + data_file = _make_data_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + captured_timeout: dict[str, float | dict[str, float] | None] = {} + + def handler(request: httpx.Request) -> httpx.Response: + if ( + request.method == "POST" + and request.url.path == "/pdf-with-imported-form-data" + ): + assert request.url.params["trace"] == "async" + assert request.headers["X-Debug"] == "async" + captured_timeout["value"] = request.extensions.get("timeout") + payload = json.loads(request.content.decode("utf-8")) + assert payload["id"] == str(input_file.id) + assert payload["data_file_id"] == str(data_file.id) + assert payload["note"] == "details" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + assert request.url.params["format"] == "info" + assert request.url.params["trace"] == "async" + assert request.headers["X-Debug"] == "async" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async-custom.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + transport = httpx.MockTransport(handler) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + response = await client.import_form_data( + input_file, + data_file, + extra_query={"trace": "async"}, + extra_headers={"X-Debug": "async"}, + extra_body={"note": "details"}, + timeout=0.73, + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "async-custom.pdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.73) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.73) + + +def test_import_form_data_validation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + pdf_file = make_pdf_file(PdfRestFileID.generate(1)) + data_file = _make_data_file(PdfRestFileID.generate(2)) + png_file = PdfRestFile.model_validate( + build_file_info_payload( + PdfRestFileID.generate(), + "example.png", + "image/png", + ) + ) + transport = httpx.MockTransport(lambda request: (_ for _ in ()).throw(RuntimeError)) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises(ValidationError, match="Must be a PDF file"), + ): + client.import_form_data(png_file, data_file) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises( + ValidationError, + match="Data file must be an XFDF, XDP, XFD, FDF, or XML file", + ), + ): + client.import_form_data(pdf_file, png_file) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises( + ValidationError, match="List should have at most 1 item after validation" + ), + ): + client.import_form_data( + [pdf_file, make_pdf_file(PdfRestFileID.generate())], + data_file, + ) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises( + ValidationError, match="List should have at most 1 item after validation" + ), + ): + client.import_form_data( + pdf_file, + [data_file, _make_data_file(PdfRestFileID.generate())], + ) + + +@pytest.mark.asyncio +async def test_async_import_form_data_validation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + pdf_file = make_pdf_file(PdfRestFileID.generate(1)) + data_file = _make_data_file(PdfRestFileID.generate(2)) + png_file = PdfRestFile.model_validate( + build_file_info_payload( + PdfRestFileID.generate(), + "example.png", + "image/png", + ) + ) + transport = httpx.MockTransport(lambda request: (_ for _ in ()).throw(RuntimeError)) + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises(ValidationError, match="Must be a PDF file"): + await client.import_form_data(png_file, data_file) + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises( + ValidationError, + match="Data file must be an XFDF, XDP, XFD, FDF, or XML file", + ): + await client.import_form_data(pdf_file, png_file) + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises( + ValidationError, match="List should have at most 1 item after validation" + ): + await client.import_form_data( + [pdf_file, make_pdf_file(PdfRestFileID.generate())], + data_file, + ) + + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises( + ValidationError, match="List should have at most 1 item after validation" + ): + await client.import_form_data( + pdf_file, + [data_file, _make_data_file(PdfRestFileID.generate())], + )