diff --git a/src/pdfrest/client.py b/src/pdfrest/client.py index 7855deb7..05976b6a 100644 --- a/src/pdfrest/client.py +++ b/src/pdfrest/client.py @@ -101,6 +101,7 @@ PdfFlattenFormsPayload, PdfFlattenLayersPayload, PdfFlattenTransparenciesPayload, + PdfImageWatermarkPayload, PdfImportFormDataPayload, PdfInfoPayload, PdfLinearizePayload, @@ -111,6 +112,7 @@ PdfRestRawFileResponse, PdfRestrictPayload, PdfSplitPayload, + PdfTextWatermarkPayload, PdfToExcelPayload, PdfToPdfaPayload, PdfToPdfxPayload, @@ -154,12 +156,15 @@ PdfRedactionInstruction, PdfRestriction, PdfRGBColor, + PdfTextColor, PdfXType, PngColorModel, SummaryFormat, SummaryOutputFormat, TiffColorModel, TranslateOutputFormat, + WatermarkHorizontalAlignment, + WatermarkVerticalAlignment, ) __all__ = ("AsyncPdfRestClient", "PdfRestClient") @@ -3615,6 +3620,106 @@ def convert_url_to_pdf( timeout=timeout, ) + def watermark_pdf_with_text( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + watermark_text: str, + output: str | None = None, + font: str | None = None, + text_size: int = 72, + text_color: PdfTextColor = (0, 0, 0), + opacity: float = 0.5, + horizontal_alignment: WatermarkHorizontalAlignment = "center", + vertical_alignment: WatermarkVerticalAlignment = "center", + x: int = 0, + y: int = 0, + rotation: int = 0, + pages: PdfPageSelection | None = None, + behind_page: bool = False, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Apply a text watermark to a PDF.""" + + payload: dict[str, Any] = { + "files": file, + "watermark_text": watermark_text, + "output": output, + "font": font, + "text_size": text_size, + "text_color": text_color, + "opacity": opacity, + "horizontal_alignment": horizontal_alignment, + "vertical_alignment": vertical_alignment, + "x": x, + "y": y, + "rotation": rotation, + "pages": pages, + "behind_page": behind_page, + } + payload = {key: value for key, value in payload.items() if value is not None} + + return self._post_file_operation( + endpoint="/watermarked-pdf", + payload=payload, + payload_model=PdfTextWatermarkPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + + def watermark_pdf_with_image( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + watermark_file: PdfRestFile | Sequence[PdfRestFile], + output: str | None = None, + watermark_file_scale: float = 0.5, + opacity: float = 0.5, + horizontal_alignment: WatermarkHorizontalAlignment = "center", + vertical_alignment: WatermarkVerticalAlignment = "center", + x: int = 0, + y: int = 0, + rotation: int = 0, + pages: PdfPageSelection | None = None, + behind_page: bool = False, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Apply an image watermark to a PDF.""" + + payload: dict[str, Any] = { + "files": file, + "watermark_file": watermark_file, + "output": output, + "watermark_file_scale": watermark_file_scale, + "opacity": opacity, + "horizontal_alignment": horizontal_alignment, + "vertical_alignment": vertical_alignment, + "x": x, + "y": y, + "rotation": rotation, + "pages": pages, + "behind_page": behind_page, + } + payload = {key: value for key, value in payload.items() if value is not None} + + return self._post_file_operation( + endpoint="/watermarked-pdf", + payload=payload, + payload_model=PdfImageWatermarkPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + def convert_to_pdfa( self, file: PdfRestFile | Sequence[PdfRestFile], @@ -5412,6 +5517,106 @@ async def convert_url_to_pdf( timeout=timeout, ) + async def watermark_pdf_with_text( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + watermark_text: str, + output: str | None = None, + font: str | None = None, + text_size: int = 72, + text_color: PdfTextColor = (0, 0, 0), + opacity: float = 0.5, + horizontal_alignment: WatermarkHorizontalAlignment = "center", + vertical_alignment: WatermarkVerticalAlignment = "center", + x: int = 0, + y: int = 0, + rotation: int = 0, + pages: PdfPageSelection | None = None, + behind_page: bool = False, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Asynchronously apply a text watermark to a PDF.""" + + payload: dict[str, Any] = { + "files": file, + "watermark_text": watermark_text, + "output": output, + "font": font, + "text_size": text_size, + "text_color": text_color, + "opacity": opacity, + "horizontal_alignment": horizontal_alignment, + "vertical_alignment": vertical_alignment, + "x": x, + "y": y, + "rotation": rotation, + "pages": pages, + "behind_page": behind_page, + } + payload = {key: value for key, value in payload.items() if value is not None} + + return await self._post_file_operation( + endpoint="/watermarked-pdf", + payload=payload, + payload_model=PdfTextWatermarkPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + + async def watermark_pdf_with_image( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + watermark_file: PdfRestFile | Sequence[PdfRestFile], + output: str | None = None, + watermark_file_scale: float = 0.5, + opacity: float = 0.5, + horizontal_alignment: WatermarkHorizontalAlignment = "center", + vertical_alignment: WatermarkVerticalAlignment = "center", + x: int = 0, + y: int = 0, + rotation: int = 0, + pages: PdfPageSelection | None = None, + behind_page: bool = False, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Asynchronously apply an image watermark to a PDF.""" + + payload: dict[str, Any] = { + "files": file, + "watermark_file": watermark_file, + "output": output, + "watermark_file_scale": watermark_file_scale, + "opacity": opacity, + "horizontal_alignment": horizontal_alignment, + "vertical_alignment": vertical_alignment, + "x": x, + "y": y, + "rotation": rotation, + "pages": pages, + "behind_page": behind_page, + } + payload = {key: value for key, value in payload.items() if value is not None} + + return await self._post_file_operation( + endpoint="/watermarked-pdf", + payload=payload, + payload_model=PdfImageWatermarkPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + async def convert_to_pdfa( self, file: PdfRestFile | Sequence[PdfRestFile], diff --git a/src/pdfrest/models/_internal.py b/src/pdfrest/models/_internal.py index 6a354a77..3e5f7f02 100644 --- a/src/pdfrest/models/_internal.py +++ b/src/pdfrest/models/_internal.py @@ -42,6 +42,8 @@ SummaryOutputFormat, SummaryOutputType, TranslateOutputFormat, + WatermarkHorizontalAlignment, + WatermarkVerticalAlignment, ) from .public import PdfRestFile, PdfRestFileID @@ -128,6 +130,25 @@ def _split_comma_string(value: Any) -> list[Any] | None: raise ValueError(msg) +def _route_text_color_by_channel_count( + *, + expected_channel_count: int, + alternate_channel_count: int, +) -> Callable[[Any], list[Any] | None]: + def _validator(value: Any) -> list[Any] | None: + channels = _split_comma_string(value) + if channels is None: + return None + if len(channels) == expected_channel_count: + return channels + if len(channels) == alternate_channel_count: + return None + msg = "text_color must include exactly 3 (RGB) or 4 (CMYK) values." + raise ValueError(msg) + + return _validator + + def _serialize_as_first_file_id(value: list[PdfRestFile]) -> str: return str(value[0].id) @@ -142,6 +163,10 @@ def _serialize_as_comma_separated_string(value: list[Any] | None) -> str | None: return ",".join(str(element) for element in value) +def _serialize_as_string(value: Any) -> str: + return str(value) + + def _serialize_file_ids(value: list[PdfRestFile]) -> str: return ",".join(str(file.id) for file in value) @@ -1604,6 +1629,148 @@ class PdfAddImagePayload(BaseModel): ] = None +class _BasePdfWatermarkPayload(BaseModel): + """Shared fields for watermark request payloads.""" + + 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), + ] + output: Annotated[ + str | None, + Field(serialization_alias="output", min_length=1, default=None), + AfterValidator(_validate_output_prefix), + ] = None + opacity: Annotated[ + float, + Field(serialization_alias="opacity", ge=0, le=1, default=0.5), + PlainSerializer(_serialize_as_string), + ] = 0.5 + horizontal_alignment: Annotated[ + WatermarkHorizontalAlignment, + Field(serialization_alias="horizontal_alignment", default="center"), + ] = "center" + vertical_alignment: Annotated[ + WatermarkVerticalAlignment, + Field(serialization_alias="vertical_alignment", default="center"), + ] = "center" + x: Annotated[ + int, + Field(serialization_alias="x", default=0), + PlainSerializer(_serialize_as_string), + ] = 0 + y: Annotated[ + int, + Field(serialization_alias="y", default=0), + PlainSerializer(_serialize_as_string), + ] = 0 + rotation: Annotated[ + int, + Field(serialization_alias="rotation", default=0), + PlainSerializer(_serialize_as_string), + ] = 0 + pages: Annotated[ + list[AscendingPageRange] | None, + Field(serialization_alias="pages", min_length=1, default=None), + BeforeValidator(_ensure_list), + BeforeValidator(_split_comma_list), + BeforeValidator(_int_to_string), + PlainSerializer(_serialize_page_ranges), + ] = None + behind_page: Annotated[ + bool, + Field(serialization_alias="behind_page", default=False), + PlainSerializer(_bool_to_true_false), + ] = False + + +class PdfTextWatermarkPayload(_BasePdfWatermarkPayload): + """Adapt caller options into a text watermark request payload.""" + + watermark_text: Annotated[ + str, + Field(serialization_alias="watermark_text", min_length=1), + ] + font: Annotated[ + str | None, Field(serialization_alias="font", min_length=1, default=None) + ] = None + text_size: Annotated[ + int, + Field(serialization_alias="text_size", ge=5, le=100, default=72), + PlainSerializer(_serialize_as_string), + ] = 72 + text_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="text_color", + serialization_alias="text_color_rgb", + default=None, + ), + BeforeValidator( + _route_text_color_by_channel_count( + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + PlainSerializer(_serialize_as_comma_separated_string), + ] = None + text_color_cmyk: Annotated[ + tuple[CmykChannel, CmykChannel, CmykChannel, CmykChannel] | None, + Field( + validation_alias="text_color", + serialization_alias="text_color_cmyk", + default=None, + ), + BeforeValidator( + _route_text_color_by_channel_count( + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + PlainSerializer(_serialize_as_comma_separated_string), + ] = None + + @model_validator(mode="after") + def _validate_text_colors(self) -> PdfTextWatermarkPayload: + if self.text_color_rgb is not None and self.text_color_cmyk is not None: + msg = "Specify only one of text_color_rgb or text_color_cmyk." + raise ValueError(msg) + return self + + +class PdfImageWatermarkPayload(_BasePdfWatermarkPayload): + """Adapt caller options into an image watermark request payload.""" + + watermark_file: Annotated[ + list[PdfRestFile], + Field( + min_length=1, + max_length=1, + serialization_alias="watermark_file_id", + ), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types("application/pdf", error_msg="Must be a PDF file") + ), + PlainSerializer(_serialize_as_first_file_id), + ] + watermark_file_scale: Annotated[ + float, + Field(serialization_alias="watermark_file_scale", ge=0, default=0.5), + PlainSerializer(_serialize_as_string), + ] = 0.5 + + class PdfXfaToAcroformsPayload(BaseModel): """Adapt caller options into a pdfRest-ready XFA-to-AcroForms request payload.""" diff --git a/src/pdfrest/types/__init__.py b/src/pdfrest/types/__init__.py index 3ed13b5b..aafe2453 100644 --- a/src/pdfrest/types/__init__.py +++ b/src/pdfrest/types/__init__.py @@ -18,7 +18,7 @@ OcrLanguage, PdfAddTextObject, PdfAType, - PdfCmykColor, + PdfCMYKColor, PdfColorProfile, PdfConversionCompression, PdfConversionDownsample, @@ -36,6 +36,7 @@ PdfRedactionType, PdfRestriction, PdfRGBColor, + PdfTextColor, PdfXType, PngColorModel, SummaryFormat, @@ -43,6 +44,8 @@ SummaryOutputType, TiffColorModel, TranslateOutputFormat, + WatermarkHorizontalAlignment, + WatermarkVerticalAlignment, ) __all__ = [ @@ -63,7 +66,7 @@ "OcrLanguage", "PdfAType", "PdfAddTextObject", - "PdfCmykColor", + "PdfCMYKColor", "PdfColorProfile", "PdfConversionCompression", "PdfConversionDownsample", @@ -81,6 +84,7 @@ "PdfRedactionPreset", "PdfRedactionType", "PdfRestriction", + "PdfTextColor", "PdfXType", "PngColorModel", "SummaryFormat", @@ -88,4 +92,6 @@ "SummaryOutputType", "TiffColorModel", "TranslateOutputFormat", + "WatermarkHorizontalAlignment", + "WatermarkVerticalAlignment", ] diff --git a/src/pdfrest/types/public.py b/src/pdfrest/types/public.py index 322d5837..09683eb6 100644 --- a/src/pdfrest/types/public.py +++ b/src/pdfrest/types/public.py @@ -30,7 +30,7 @@ "OcrLanguage", "PdfAType", "PdfAddTextObject", - "PdfCmykColor", + "PdfCMYKColor", "PdfColorProfile", "PdfConversionCompression", "PdfConversionDownsample", @@ -48,6 +48,7 @@ "PdfRedactionPreset", "PdfRedactionType", "PdfRestriction", + "PdfTextColor", "PdfXType", "PngColorModel", "SummaryFormat", @@ -55,6 +56,8 @@ "SummaryOutputType", "TiffColorModel", "TranslateOutputFormat", + "WatermarkHorizontalAlignment", + "WatermarkVerticalAlignment", ) PdfInfoQuery = Literal[ @@ -118,9 +121,9 @@ class PdfRedactionInstruction(TypedDict): value: PdfRedactionPreset | str +PdfCMYKColor = tuple[int, int, int, int] PdfRGBColor = tuple[int, int, int] - -PdfCmykColor = tuple[int, int, int, int] +PdfTextColor = PdfRGBColor | PdfCMYKColor class PdfAddTextObject(TypedDict, total=False): @@ -131,7 +134,7 @@ class PdfAddTextObject(TypedDict, total=False): rotation: Required[float] text: Required[str] text_color_rgb: PdfRGBColor - text_color_cmyk: PdfCmykColor + text_color_cmyk: PdfCMYKColor text_size: Required[float] x: Required[float] y: Required[float] @@ -223,6 +226,7 @@ class PdfMergeSource(TypedDict, total=False): ALL_PDF_RESTRICTIONS: tuple[PdfRestriction, ...] = cast( tuple[PdfRestriction, ...], get_args(PdfRestriction) ) + PdfPageSize = Literal["letter", "legal", "ledger", "A3", "A4", "A5"] | PdfCustomPageSize PdfPageOrientation = Literal["portrait", "landscape"] PdfPresetColorProfile = Literal[ @@ -243,3 +247,5 @@ class PdfMergeSource(TypedDict, total=False): ] PdfColorProfile = PdfPresetColorProfile +WatermarkHorizontalAlignment = Literal["left", "center", "right"] +WatermarkVerticalAlignment = Literal["top", "center", "bottom"] diff --git a/tests/live/test_live_watermark_pdf.py b/tests/live/test_live_watermark_pdf.py new file mode 100644 index 00000000..df5f6bde --- /dev/null +++ b/tests/live/test_live_watermark_pdf.py @@ -0,0 +1,457 @@ +from __future__ import annotations + +import pytest + +from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient +from pdfrest.models import PdfRestFile + +from ..resources import get_test_resource_path + + +@pytest.fixture(scope="module") +def uploaded_pdf_for_watermark( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> PdfRestFile: + resource = get_test_resource_path("report.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.fixture(scope="module") +def uploaded_watermark_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> PdfRestFile: + resource = get_test_resource_path("duckhat.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( + "output_name", + [ + pytest.param(None, id="default-output"), + pytest.param("watermark-text", id="custom-output"), + ], +) +def test_live_watermark_pdf_text_success( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + output_name: str | None, +) -> None: + kwargs: dict[str, str] = {} + if output_name is not None: + kwargs["output"] = output_name + + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="CONFIDENTIAL", + opacity=0.6, + pages=["1", "last"], + **kwargs, + ) + + output_file = response.output_file + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert str(response.input_id) == str(uploaded_pdf_for_watermark.id) + if output_name is not None: + assert output_file.name.startswith(output_name) + else: + assert output_file.name.endswith(".pdf") + + +def test_live_watermark_pdf_image_success( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + uploaded_watermark_pdf: PdfRestFile, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.watermark_pdf_with_image( + uploaded_pdf_for_watermark, + watermark_file=uploaded_watermark_pdf, + watermark_file_scale=0.75, + behind_page=True, + output="watermark-file", + ) + + output_file = response.output_file + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert output_file.name.startswith("watermark-file") + assert [str(value) for value in response.input_ids] == [ + str(uploaded_pdf_for_watermark.id), + str(uploaded_watermark_pdf.id), + ] + + +@pytest.mark.parametrize( + "horizontal_alignment", + [ + pytest.param("left", id="left"), + pytest.param("center", id="center"), + pytest.param("right", id="right"), + ], +) +def test_live_watermark_pdf_text_horizontal_alignment_literals( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + horizontal_alignment: str, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="HORIZONTAL", + horizontal_alignment=horizontal_alignment, + output=f"align-h-{horizontal_alignment}", + ) + + output_file = response.output_file + assert output_file.name.startswith(f"align-h-{horizontal_alignment}") + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert str(response.input_id) == str(uploaded_pdf_for_watermark.id) + + +@pytest.mark.parametrize( + "vertical_alignment", + [ + pytest.param("top", id="top"), + pytest.param("center", id="center"), + pytest.param("bottom", id="bottom"), + ], +) +def test_live_watermark_pdf_text_vertical_alignment_literals( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + vertical_alignment: str, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="VERTICAL", + vertical_alignment=vertical_alignment, + output=f"align-v-{vertical_alignment}", + ) + + output_file = response.output_file + assert output_file.name.startswith(f"align-v-{vertical_alignment}") + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert str(response.input_id) == str(uploaded_pdf_for_watermark.id) + + +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_text_success( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="ASYNC", + horizontal_alignment="right", + vertical_alignment="top", + x=-36, + y=36, + output="async-watermark", + ) + + output_file = response.output_file + assert output_file.name.startswith("async-watermark") + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert str(response.input_id) == str(uploaded_pdf_for_watermark.id) + + +@pytest.mark.parametrize( + "horizontal_alignment", + [ + pytest.param("left", id="left"), + pytest.param("center", id="center"), + pytest.param("right", id="right"), + ], +) +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_text_horizontal_alignment_literals( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + horizontal_alignment: str, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="ASYNC-HORIZONTAL", + horizontal_alignment=horizontal_alignment, + output=f"async-align-h-{horizontal_alignment}", + ) + + output_file = response.output_file + assert output_file.name.startswith(f"async-align-h-{horizontal_alignment}") + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert str(response.input_id) == str(uploaded_pdf_for_watermark.id) + + +@pytest.mark.parametrize( + "vertical_alignment", + [ + pytest.param("top", id="top"), + pytest.param("center", id="center"), + pytest.param("bottom", id="bottom"), + ], +) +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_text_vertical_alignment_literals( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + vertical_alignment: str, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="ASYNC-VERTICAL", + vertical_alignment=vertical_alignment, + output=f"async-align-v-{vertical_alignment}", + ) + + output_file = response.output_file + assert output_file.name.startswith(f"async-align-v-{vertical_alignment}") + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert str(response.input_id) == str(uploaded_pdf_for_watermark.id) + + +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_image_success( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + uploaded_watermark_pdf: PdfRestFile, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.watermark_pdf_with_image( + uploaded_pdf_for_watermark, + watermark_file=uploaded_watermark_pdf, + watermark_file_scale=0.75, + behind_page=True, + output="async-watermark-file", + ) + + output_file = response.output_file + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert output_file.name.startswith("async-watermark-file") + assert [str(value) for value in response.input_ids] == [ + str(uploaded_pdf_for_watermark.id), + str(uploaded_watermark_pdf.id), + ] + + +def test_live_watermark_pdf_invalid_alignment( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)alignment"), + ): + client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="BadAlignment", + extra_body={"horizontal_alignment": "diagonal"}, + ) + + +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_invalid_file_id( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises(PdfRestApiError, match=r"(?i)(id|file)"): + await client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="AsyncInvalid", + extra_body={"id": "00000000-0000-0000-0000-000000000000"}, + ) + + +@pytest.mark.parametrize( + ("extra_body", "error_match"), + [ + pytest.param( + {"opacity": "1.01"}, + r"(?i)(opacity|range)", + id="opacity-above-max", + ), + pytest.param( + {"opacity": "-0.01"}, + r"(?i)(opacity|range)", + id="opacity-below-min", + ), + pytest.param( + {"text_size": "101"}, + r"(?i)(text_size|size|range)", + id="text-size-above-max", + ), + pytest.param( + {"text_size": "4"}, + r"(?i)(text_size|size|range)", + id="text-size-below-min", + ), + ], +) +def test_live_watermark_pdf_text_invalid_numeric_bounds( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + extra_body: dict[str, str], + error_match: str, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=error_match), + ): + client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="InvalidNumericText", + extra_body=extra_body, + ) + + +def test_live_watermark_pdf_image_invalid_scale_bounds( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + uploaded_watermark_pdf: PdfRestFile, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)(watermark_file_scale|scale|range)"), + ): + client.watermark_pdf_with_image( + uploaded_pdf_for_watermark, + watermark_file=uploaded_watermark_pdf, + extra_body={"watermark_file_scale": "-0.01"}, + ) + + +@pytest.mark.parametrize( + ("extra_body", "error_match"), + [ + pytest.param( + {"opacity": "1.01"}, + r"(?i)(opacity|range)", + id="opacity-above-max", + ), + pytest.param( + {"opacity": "-0.01"}, + r"(?i)(opacity|range)", + id="opacity-below-min", + ), + pytest.param( + {"text_size": "101"}, + r"(?i)(text_size|size|range)", + id="text-size-above-max", + ), + pytest.param( + {"text_size": "4"}, + r"(?i)(text_size|size|range)", + id="text-size-below-min", + ), + ], +) +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_text_invalid_numeric_bounds( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + extra_body: dict[str, str], + error_match: str, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises(PdfRestApiError, match=error_match): + await client.watermark_pdf_with_text( + uploaded_pdf_for_watermark, + watermark_text="AsyncInvalidNumericText", + extra_body=extra_body, + ) + + +@pytest.mark.asyncio +async def test_live_async_watermark_pdf_image_invalid_scale_bounds( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_watermark: PdfRestFile, + uploaded_watermark_pdf: PdfRestFile, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises( + PdfRestApiError, + match=r"(?i)(watermark_file_scale|scale|range)", + ): + await client.watermark_pdf_with_image( + uploaded_pdf_for_watermark, + watermark_file=uploaded_watermark_pdf, + extra_body={"watermark_file_scale": "-0.01"}, + ) diff --git a/tests/test_watermark_pdf.py b/tests/test_watermark_pdf.py new file mode 100644 index 00000000..882c01ea --- /dev/null +++ b/tests/test_watermark_pdf.py @@ -0,0 +1,1231 @@ +from __future__ import annotations + +import json +import re + +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 PdfImageWatermarkPayload, PdfTextWatermarkPayload + +from .graphics_test_helpers import ( + ASYNC_API_KEY, + VALID_API_KEY, + build_file_info_payload, + make_pdf_file, +) + + +def test_watermark_pdf_with_text(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfTextWatermarkPayload.model_validate( + { + "files": [input_file], + "watermark_text": "Confidential", + "text_size": 72, + "text_color": (255, 0, 0), + "opacity": 0.5, + "horizontal_alignment": "center", + "vertical_alignment": "center", + "x": 0, + "y": 0, + "rotation": 0, + "pages": ["1", "3-5"], + "behind_page": False, + "output": "watermarked", + } + ).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 == "/watermarked-pdf": + 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, + "watermarked.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.watermark_pdf_with_text( + input_file, + watermark_text="Confidential", + text_color=(255, 0, 0), + pages=["1", "3-5"], + output="watermarked", + ) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "watermarked.pdf" + assert response.output_file.type == "application/pdf" + assert str(response.input_id) == str(input_file.id) + + +def test_watermark_pdf_with_text_defaults_text_color_to_rgb_black( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["text_color_rgb"] == "0,0,0" + assert "text_color_cmyk" not in payload + 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" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "default-color.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.watermark_pdf_with_text( + input_file, + watermark_text="DefaultColor", + ) + + assert response.output_file.name == "default-color.pdf" + + +def test_watermark_pdf_with_image(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + watermark_file = make_pdf_file(PdfRestFileID.generate(1), name="stamp.pdf") + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfImageWatermarkPayload.model_validate( + { + "files": [input_file], + "watermark_file": [watermark_file], + "watermark_file_scale": 0.8, + "opacity": 0.5, + "horizontal_alignment": "center", + "vertical_alignment": "center", + "x": 0, + "behind_page": True, + "rotation": 45, + "y": 25, + } + ).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 == "/watermarked-pdf": + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + assert payload["behind_page"] == "true" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, watermark_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, + "stamped.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.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=0.8, + behind_page=True, + rotation=45, + y=25, + ) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "stamped.pdf" + assert response.output_file.type == "application/pdf" + assert response.warning is None + assert [str(value) for value in response.input_ids] == [ + str(input_file.id), + str(watermark_file.id), + ] + + +def test_watermark_pdf_with_text_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 == "/watermarked-pdf": + 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["watermark_text"] == "Draft" + assert payload["text_color_cmyk"] == "0,0,0,50" + assert payload["opacity"] == "0.25" + assert payload["behind_page"] == "false" + assert payload["output"] == "custom" + assert payload["debug"] == "yes" + assert payload["id"] == str(input_file.id) + 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.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.watermark_pdf_with_text( + input_file, + watermark_text="Draft", + text_color=(0, 0, 0, 50), + opacity=0.25, + output="custom", + extra_query={"trace": "true"}, + extra_headers={"X-Debug": "sync"}, + extra_body={"debug": "yes"}, + timeout=0.31, + ) + + 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.31) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.31) + + +def test_watermark_pdf_with_image_request_customization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + watermark_file = make_pdf_file(PdfRestFileID.generate(1), name="custom-stamp.pdf") + 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 == "/watermarked-pdf": + assert request.url.params["trace"] == "true" + assert request.headers["X-Debug"] == "sync-image" + captured_timeout["value"] = request.extensions.get("timeout") + payload = json.loads(request.content.decode("utf-8")) + assert payload["id"] == str(input_file.id) + assert payload["watermark_file_id"] == str(watermark_file.id) + assert payload["watermark_file_scale"] == "0.75" + assert payload["opacity"] == "0.2" + assert payload["behind_page"] == "false" + assert payload["output"] == "custom-image" + assert payload["debug"] == "yes" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, watermark_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-image" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "custom-image.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.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=0.75, + opacity=0.2, + output="custom-image", + extra_query={"trace": "true"}, + extra_headers={"X-Debug": "sync-image"}, + extra_body={"debug": "yes"}, + timeout=0.33, + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "custom-image.pdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.33) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.33) + + +def test_watermark_pdf_with_text_validation_rejects_invalid_text_color_channel_count( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises( + ValidationError, + match=re.escape( + "text_color must include exactly 3 (RGB) or 4 (CMYK) values." + ), + ), + ): + client.watermark_pdf_with_text( + input_file, + watermark_text="Confidential", + text_color=(0, 0), + ) + + +def test_watermark_pdf_with_text_validation_rejects_non_pdf( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + bad_file = PdfRestFile.model_validate( + { + "id": str(PdfRestFileID.generate()), + "name": "image.png", + "type": "image/png", + "url": "https://example.com/image.png", + "size": 12, + "modified": "2024-01-01T00:00:00Z", + } + ) + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises(ValidationError, match="Must be a PDF file"), + ): + client.watermark_pdf_with_text(bad_file, watermark_text="Hi") + + +def test_watermark_pdf_with_image_validation_rejects_non_pdf_watermark( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + bad_watermark = PdfRestFile.model_validate( + { + "id": str(PdfRestFileID.generate()), + "name": "overlay.png", + "type": "image/png", + "url": "https://example.com/overlay.png", + "size": 12, + "modified": "2024-01-01T00:00:00Z", + } + ) + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises(ValidationError, match="Must be a PDF file"), + ): + client.watermark_pdf_with_image(input_file, watermark_file=bad_watermark) + + +def test_watermark_pdf_with_text_validation_rejects_short_text_size( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises( + ValidationError, match="Input should be greater than or equal to 5" + ), + ): + client.watermark_pdf_with_text(input_file, watermark_text="Hi", text_size=4) + + +@pytest.mark.parametrize( + ("text_size", "expected_text_size"), + [ + pytest.param(5, "5", id="min"), + pytest.param(100, "100", id="max"), + ], +) +def test_watermark_pdf_with_text_text_size_boundary_values( + monkeypatch: pytest.MonkeyPatch, + text_size: int, + expected_text_size: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["text_size"] == expected_text_size + 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" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "text-size.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + with PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = client.watermark_pdf_with_text( + input_file, + watermark_text="SizeBoundary", + text_size=text_size, + ) + + assert response.output_file.name == "text-size.pdf" + + +@pytest.mark.parametrize( + ("text_size", "match"), + [ + pytest.param(4, "Input should be greater than or equal to 5", id="below-min"), + pytest.param(101, "Input should be less than or equal to 100", id="above-max"), + ], +) +def test_watermark_pdf_with_text_validation_rejects_out_of_range_text_size( + monkeypatch: pytest.MonkeyPatch, + text_size: int, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises(ValidationError, match=match), + ): + client.watermark_pdf_with_text( + input_file, + watermark_text="SizeBoundary", + text_size=text_size, + ) + + +@pytest.mark.parametrize( + ("opacity", "expected_opacity"), + [ + pytest.param(0.0, "0.0", id="min"), + pytest.param(1.0, "1.0", id="max"), + ], +) +def test_watermark_pdf_with_text_opacity_boundary_values( + monkeypatch: pytest.MonkeyPatch, + opacity: float, + expected_opacity: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["opacity"] == expected_opacity + 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" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "opacity-text.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + with PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = client.watermark_pdf_with_text( + input_file, + watermark_text="OpacityBoundary", + opacity=opacity, + ) + + assert response.output_file.name == "opacity-text.pdf" + + +@pytest.mark.parametrize( + ("opacity", "match"), + [ + pytest.param( + -0.01, "Input should be greater than or equal to 0", id="below-min" + ), + pytest.param(1.01, "Input should be less than or equal to 1", id="above-max"), + ], +) +def test_watermark_pdf_with_text_validation_rejects_out_of_range_opacity( + monkeypatch: pytest.MonkeyPatch, + opacity: float, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises(ValidationError, match=match), + ): + client.watermark_pdf_with_text( + input_file, + watermark_text="OpacityBoundary", + opacity=opacity, + ) + + +@pytest.mark.parametrize( + ("watermark_file_scale", "expected_scale"), + [ + pytest.param(0.0, "0.0", id="min"), + pytest.param(0.01, "0.01", id="inside"), + ], +) +def test_watermark_pdf_with_image_scale_boundary_values( + monkeypatch: pytest.MonkeyPatch, + watermark_file_scale: float, + expected_scale: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + watermark_file = make_pdf_file(PdfRestFileID.generate(1), name="boundary-stamp.pdf") + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["watermark_file_scale"] == expected_scale + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, watermark_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + assert request.url.params["format"] == "info" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "scale-image.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + with PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = client.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=watermark_file_scale, + ) + + assert response.output_file.name == "scale-image.pdf" + + +def test_watermark_pdf_with_image_validation_rejects_negative_watermark_file_scale( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + watermark_file = make_pdf_file(PdfRestFileID.generate(1), name="boundary-stamp.pdf") + with ( + PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ), + ) as client, + pytest.raises( + ValidationError, match="Input should be greater than or equal to 0" + ), + ): + client.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=-0.01, + ) + + +@pytest.mark.parametrize( + ("text_size", "expected_text_size"), + [ + pytest.param(5, "5", id="min"), + pytest.param(100, "100", id="max"), + ], +) +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text_text_size_boundary_values( + monkeypatch: pytest.MonkeyPatch, + text_size: int, + expected_text_size: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["text_size"] == expected_text_size + 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" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async-text-size.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = await client.watermark_pdf_with_text( + input_file, + watermark_text="AsyncSizeBoundary", + text_size=text_size, + ) + + assert response.output_file.name == "async-text-size.pdf" + + +@pytest.mark.parametrize( + ("text_size", "match"), + [ + pytest.param(4, "Input should be greater than or equal to 5", id="below-min"), + pytest.param(101, "Input should be less than or equal to 100", id="above-max"), + ], +) +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text_validation_rejects_out_of_range_text_size( + monkeypatch: pytest.MonkeyPatch, + text_size: int, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + transport = httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises(ValidationError, match=match): + await client.watermark_pdf_with_text( + input_file, + watermark_text="AsyncSizeBoundary", + text_size=text_size, + ) + + +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text_validation_rejects_invalid_text_color_channel_count( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + transport = httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises( + ValidationError, + match=re.escape( + "text_color must include exactly 3 (RGB) or 4 (CMYK) values." + ), + ): + await client.watermark_pdf_with_text( + input_file, + watermark_text="AsyncColorValidation", + text_color=(0, 0), + ) + + +@pytest.mark.parametrize( + ("opacity", "expected_opacity"), + [ + pytest.param(0.0, "0.0", id="min"), + pytest.param(1.0, "1.0", id="max"), + ], +) +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text_opacity_boundary_values( + monkeypatch: pytest.MonkeyPatch, + opacity: float, + expected_opacity: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["opacity"] == expected_opacity + 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" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async-opacity-text.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = await client.watermark_pdf_with_text( + input_file, + watermark_text="AsyncOpacityBoundary", + opacity=opacity, + ) + + assert response.output_file.name == "async-opacity-text.pdf" + + +@pytest.mark.parametrize( + ("opacity", "match"), + [ + pytest.param( + -0.01, "Input should be greater than or equal to 0", id="below-min" + ), + pytest.param(1.01, "Input should be less than or equal to 1", id="above-max"), + ], +) +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text_validation_rejects_out_of_range_opacity( + monkeypatch: pytest.MonkeyPatch, + opacity: float, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + transport = httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises(ValidationError, match=match): + await client.watermark_pdf_with_text( + input_file, + watermark_text="AsyncOpacityBoundary", + opacity=opacity, + ) + + +@pytest.mark.parametrize( + ("watermark_file_scale", "expected_scale"), + [ + pytest.param(0.0, "0.0", id="min"), + pytest.param(0.01, "0.01", id="inside"), + ], +) +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_image_scale_boundary_values( + monkeypatch: pytest.MonkeyPatch, + watermark_file_scale: float, + expected_scale: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + watermark_file = make_pdf_file(PdfRestFileID.generate(2), name="boundary-stamp.pdf") + output_id = str(PdfRestFileID.generate()) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/watermarked-pdf": + payload = json.loads(request.content.decode("utf-8")) + assert payload["watermark_file_scale"] == expected_scale + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, watermark_file.id], + "outputId": [output_id], + }, + ) + if request.method == "GET" and request.url.path == f"/resource/{output_id}": + assert request.url.params["format"] == "info" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async-scale-image.pdf", + "application/pdf", + ), + ) + msg = f"Unexpected request {request.method} {request.url}" + raise AssertionError(msg) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = await client.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=watermark_file_scale, + ) + + assert response.output_file.name == "async-scale-image.pdf" + + +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_image_validation_rejects_negative_watermark_file_scale( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + watermark_file = make_pdf_file(PdfRestFileID.generate(2), name="boundary-stamp.pdf") + transport = httpx.MockTransport( + lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called")) + ) + async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client: + with pytest.raises( + ValidationError, match="Input should be greater than or equal to 0" + ): + await client.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=-0.01, + ) + + +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfTextWatermarkPayload.model_validate( + { + "files": [input_file], + "watermark_text": "Async", + "text_size": 72, + "text_color": (0, 0, 0), + "opacity": 0.6, + "horizontal_alignment": "center", + "vertical_alignment": "center", + "x": 0, + "y": 0, + "rotation": 0, + "behind_page": False, + } + ).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 == "/watermarked-pdf": + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + assert payload["behind_page"] == "false" + 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, + "async-watermarked.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.watermark_pdf_with_text( + input_file, + watermark_text="Async", + opacity=0.6, + ) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "async-watermarked.pdf" + assert response.output_file.type == "application/pdf" + assert str(response.input_id) == str(input_file.id) + + +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_image(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + watermark_file = make_pdf_file(PdfRestFileID.generate(2), name="async-stamp.pdf") + output_id = str(PdfRestFileID.generate()) + + payload_dump = PdfImageWatermarkPayload.model_validate( + { + "files": [input_file], + "watermark_file": [watermark_file], + "watermark_file_scale": 0.5, + "opacity": 0.2, + "horizontal_alignment": "center", + "vertical_alignment": "center", + "x": 0, + "y": 0, + "rotation": 0, + "pages": ["2-last"], + "behind_page": False, + } + ).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 == "/watermarked-pdf": + seen["post"] += 1 + payload = json.loads(request.content.decode("utf-8")) + assert payload == payload_dump + assert payload["behind_page"] == "false" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, watermark_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-stamped.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.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + opacity=0.2, + pages=["2-last"], + ) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "async-stamped.pdf" + assert response.output_file.type == "application/pdf" + assert [str(value) for value in response.input_ids] == [ + str(input_file.id), + str(watermark_file.id), + ] + + +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_text_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 == "/watermarked-pdf": + 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["watermark_text"] == "AsyncDraft" + assert payload["text_color_rgb"] == "12,34,56" + assert payload["horizontal_alignment"] == "left" + assert payload["vertical_alignment"] == "bottom" + assert payload["x"] == "-72" + assert payload["y"] == "144" + assert payload["rotation"] == "30" + assert payload["behind_page"] == "false" + 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.watermark_pdf_with_text( + input_file, + watermark_text="AsyncDraft", + text_color=(12, 34, 56), + horizontal_alignment="left", + vertical_alignment="bottom", + x=-72, + y=144, + rotation=30, + extra_query={"trace": "async"}, + extra_headers={"X-Debug": "async"}, + extra_body={"debug": "async"}, + timeout=0.42, + ) + + 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.42) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.42) + + +@pytest.mark.asyncio +async def test_async_watermark_pdf_with_image_request_customization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + watermark_file = make_pdf_file( + PdfRestFileID.generate(2), name="async-custom-stamp.pdf" + ) + 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 == "/watermarked-pdf": + assert request.url.params["trace"] == "async" + assert request.headers["X-Debug"] == "async-image" + captured_timeout["value"] = request.extensions.get("timeout") + payload = json.loads(request.content.decode("utf-8")) + assert payload["id"] == str(input_file.id) + assert payload["watermark_file_id"] == str(watermark_file.id) + assert payload["watermark_file_scale"] == "0.6" + assert payload["opacity"] == "0.25" + assert payload["behind_page"] == "false" + assert payload["output"] == "async-custom-image" + assert payload["debug"] == "async" + return httpx.Response( + 200, + json={ + "inputId": [input_file.id, watermark_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-image" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "async-custom-image.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.watermark_pdf_with_image( + input_file, + watermark_file=watermark_file, + watermark_file_scale=0.6, + opacity=0.25, + output="async-custom-image", + extra_query={"trace": "async"}, + extra_headers={"X-Debug": "async-image"}, + extra_body={"debug": "async"}, + timeout=0.52, + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.output_file.name == "async-custom-image.pdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.52) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.52)