From aebabf261e872de24ade25c5c8efd0950ed5aa82 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 20 Oct 2026 13:45:11 -0500 Subject: [PATCH 01/15] models: Define ZUGFeRD PDF request contract - Add typed rendering options for invoice appearance and localization. - Validate uploaded resources, color channels, fonts, and regeneration. - Serialize the public input shape into the pdfRest wire contract. Assisted-by: Codex --- src/pdfrest/models/_internal.py | 288 ++++++++++++++++++++++++++++++++ src/pdfrest/types/__init__.py | 14 ++ src/pdfrest/types/public.py | 237 ++++++++++++++++++++++++++ 3 files changed, 539 insertions(+) diff --git a/src/pdfrest/models/_internal.py b/src/pdfrest/models/_internal.py index 1432caa..dfc0954 100644 --- a/src/pdfrest/models/_internal.py +++ b/src/pdfrest/models/_internal.py @@ -47,6 +47,11 @@ PdfStructuredTextPageSize, PdfStructuredTextTextAlignment, PdfXType, + PdfZugferdCurrencyDisplay, + PdfZugferdFontName, + PdfZugferdLabelLanguage, + PdfZugferdLocale, + PdfZugferdLogoCorner, PostScriptLevel, SummaryFormat, SummaryOutputFormat, @@ -1530,6 +1535,289 @@ class ExtractImagesPayload(BaseModel): CmykChannel = Annotated[int, Field(ge=0, le=100)] +def _validate_zugferd_label_language(value: str) -> str: + trimmed = value.strip() + if not re.fullmatch( + r"[A-Za-z]{2,3}(?:-(?:[A-Za-z]{4}|[A-Za-z]{2}|\d{3}))?", trimmed + ) or not tag_is_valid(trimmed): + msg = "label_language must be a valid BCP 47 language tag." + raise ValueError(msg) + if trimmed.split("-", maxsplit=1)[0].casefold() not in {"en", "de", "fr"}: + msg = "label_language must use English, German, or French." + raise ValueError(msg) + return trimmed + + +def _validate_zugferd_font(value: Any) -> Any: + if not isinstance(value, str): + return value + trimmed = value.strip() + if trimmed.casefold() == "webdings": + msg = "ZUGFeRD invoice fonts must be text-capable; Webdings is unsupported." + raise ValueError(msg) + return trimmed + + +_ZugferdFontName = Annotated[ + PdfZugferdFontName, + BeforeValidator(_validate_zugferd_font), + Field(min_length=1, max_length=200), +] + + +class _PdfZugferdRenderOptions(BaseModel): + model_config = ConfigDict(extra="forbid") + + logo_corner: PdfZugferdLogoCorner | None = None + font: _ZugferdFontName | None = None + bold_font: _ZugferdFontName | None = None + text_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="text_color", + serialization_alias="text_color_rgb", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="text_color", + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + ] = 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_color_by_channel_count( + color_name="text_color", + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + ] = None + muted_text_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="muted_text_color", + serialization_alias="muted_text_color_rgb", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="muted_text_color", + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + ] = None + muted_text_color_cmyk: Annotated[ + tuple[CmykChannel, CmykChannel, CmykChannel, CmykChannel] | None, + Field( + validation_alias="muted_text_color", + serialization_alias="muted_text_color_cmyk", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="muted_text_color", + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + ] = None + accent_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="accent_color", + serialization_alias="accent_color_rgb", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="accent_color", + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + ] = None + accent_color_cmyk: Annotated[ + tuple[CmykChannel, CmykChannel, CmykChannel, CmykChannel] | None, + Field( + validation_alias="accent_color", + serialization_alias="accent_color_cmyk", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="accent_color", + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + ] = None + border_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="border_color", + serialization_alias="border_color_rgb", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="border_color", + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + ] = None + border_color_cmyk: Annotated[ + tuple[CmykChannel, CmykChannel, CmykChannel, CmykChannel] | None, + Field( + validation_alias="border_color", + serialization_alias="border_color_cmyk", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="border_color", + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + ] = None + table_header_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="table_header_color", + serialization_alias="table_header_color_rgb", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="table_header_color", + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + ] = None + table_header_color_cmyk: Annotated[ + tuple[CmykChannel, CmykChannel, CmykChannel, CmykChannel] | None, + Field( + validation_alias="table_header_color", + serialization_alias="table_header_color_cmyk", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="table_header_color", + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + ] = None + table_alternate_color_rgb: Annotated[ + tuple[RgbChannel, RgbChannel, RgbChannel] | None, + Field( + validation_alias="table_alternate_color", + serialization_alias="table_alternate_color_rgb", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="table_alternate_color", + expected_channel_count=3, + alternate_channel_count=4, + ) + ), + ] = None + table_alternate_color_cmyk: Annotated[ + tuple[CmykChannel, CmykChannel, CmykChannel, CmykChannel] | None, + Field( + validation_alias="table_alternate_color", + serialization_alias="table_alternate_color_cmyk", + default=None, + ), + BeforeValidator( + _route_color_by_channel_count( + color_name="table_alternate_color", + expected_channel_count=4, + alternate_channel_count=3, + ) + ), + ] = None + footer_message: Annotated[str | None, Field(max_length=500)] = None + locale: PdfZugferdLocale | None = None + label_language: Annotated[ + PdfZugferdLabelLanguage | None, + AfterValidator(_validate_zugferd_label_language), + ] = None + currency_display: PdfZugferdCurrencyDisplay | None = None + date_format: str | None = None + decimal_precision: Annotated[int | None, Field(ge=0, le=4)] = None + + +class PdfZugferdPayload(BaseModel): + """Adapt uploaded invoice resources into a ZUGFeRD creation request.""" + + model_config = ConfigDict(extra="forbid") + + invoice: Annotated[ + list[PdfRestFile], + Field(min_length=1, max_length=1, serialization_alias="id"), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types( + "application/xml", + "text/xml", + error_msg="Invoice must be an XML file", + ) + ), + PlainSerializer(_serialize_as_first_file_id), + ] + pdf: Annotated[ + list[PdfRestFile] | None, + Field(min_length=1, max_length=1, serialization_alias="pdf_id"), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types("application/pdf", error_msg="Must be a PDF file") + ), + PlainSerializer(_serialize_as_first_file_id), + ] = None + logo: Annotated[ + list[PdfRestFile] | None, + Field(min_length=1, max_length=1, serialization_alias="logo_id"), + BeforeValidator(_ensure_list), + AfterValidator( + _allowed_mime_types( + "image/png", + "image/jpeg", + error_msg="Logo must be a PNG or JPEG file", + ) + ), + PlainSerializer(_serialize_as_first_file_id), + ] = None + render_options: _PdfZugferdRenderOptions | None = None + regenerate_pdf: bool = False + output: Annotated[ + str | None, + Field(min_length=1), + AfterValidator(_validate_output_prefix), + ] = None + + @model_validator(mode="after") + def _validate_regeneration_source(self) -> PdfZugferdPayload: + if self.regenerate_pdf and self.pdf is None: + msg = "regenerate_pdf=True requires a PDF resource." + raise ValueError(msg) + return self + + class PdfLiteralRedactionModel(BaseModel): type: Literal["literal"] value: Annotated[str, Field(min_length=1)] diff --git a/src/pdfrest/types/__init__.py b/src/pdfrest/types/__init__.py index 48b9ffa..62029d4 100644 --- a/src/pdfrest/types/__init__.py +++ b/src/pdfrest/types/__init__.py @@ -63,6 +63,13 @@ PdfStructuredTextTextAlignment, PdfTextColor, PdfXType, + PdfZugferdCurrencyDisplay, + PdfZugferdFontName, + PdfZugferdKnownFont, + PdfZugferdLabelLanguage, + PdfZugferdLocale, + PdfZugferdLogoCorner, + PdfZugferdRenderOptions, PngColorModel, PostScriptLevel, SummaryFormat, @@ -137,6 +144,13 @@ "PdfStructuredTextTextAlignment", "PdfTextColor", "PdfXType", + "PdfZugferdCurrencyDisplay", + "PdfZugferdFontName", + "PdfZugferdKnownFont", + "PdfZugferdLabelLanguage", + "PdfZugferdLocale", + "PdfZugferdLogoCorner", + "PdfZugferdRenderOptions", "PngColorModel", "PostScriptLevel", "SummaryFormat", diff --git a/src/pdfrest/types/public.py b/src/pdfrest/types/public.py index e74e433..7782633 100644 --- a/src/pdfrest/types/public.py +++ b/src/pdfrest/types/public.py @@ -75,6 +75,13 @@ "PdfStructuredTextTextAlignment", "PdfTextColor", "PdfXType", + "PdfZugferdCurrencyDisplay", + "PdfZugferdFontName", + "PdfZugferdKnownFont", + "PdfZugferdLabelLanguage", + "PdfZugferdLocale", + "PdfZugferdLogoCorner", + "PdfZugferdRenderOptions", "PngColorModel", "PostScriptLevel", "SummaryFormat", @@ -159,6 +166,236 @@ class PdfRedactionInstruction(TypedDict): PdfColor = PdfRGBColor | PdfCMYKColor PdfTextColor = PdfColor +PdfZugferdLogoCorner: TypeAlias = Literal[ + "top-left", "top-right", "bottom-left", "bottom-right" +] +"""Logo placement used while rendering a ZUGFeRD invoice. + +Accepted values: + +- `top-left`: Place the logo in the upper-left corner. +- `top-right`: Place the logo in the upper-right corner. +- `bottom-left`: Place the logo in the lower-left corner. +- `bottom-right`: Place the logo in the lower-right corner. +""" + +PdfZugferdLocale: TypeAlias = Literal["de-DE", "fr-FR", "nl-NL", "en-GB"] +"""Locale used to format values in a rendered ZUGFeRD invoice. + +Accepted values: + +- `de-DE`: Format values using German (Germany) conventions. +- `fr-FR`: Format values using French (France) conventions. +- `nl-NL`: Format values using Dutch (Netherlands) conventions. +- `en-GB`: Format values using English (United Kingdom) conventions. +""" + +PdfZugferdCurrencyDisplay: TypeAlias = Literal["code", "symbol", "name"] +"""Currency representation used in a rendered ZUGFeRD invoice. + +Accepted values: + +- `code`: Display the ISO currency code. +- `symbol`: Display the currency symbol, falling back to its code when unknown. +- `name`: Display the currency name. +""" + +PdfZugferdLabelLanguage: TypeAlias = Literal["en", "de", "fr"] | str +"""BCP 47 language tag used for generated ZUGFeRD invoice labels. + +The primary language must be English, German, or French. Region subtags such +as ``en-US`` are accepted and normalized by pdfRest to their primary language. + +Accepted values: + +- `en`: Render English labels. +- `de`: Render German labels. +- `fr`: Render French labels. +""" + +PdfZugferdKnownFont: TypeAlias = Literal[ + "andalemono", + "arial", + "arialitalic", + "arialbold", + "arialblack", + "calibri", + "calibriitalic", + "calibribold", + "calibribolditalic", + "cambriaitalic", + "cambriabold", + "cambriabolditalic", + "candara", + "candaraitalic", + "candarabold", + "candarabolditalic", + "cantarellregular", + "cantarelloblique", + "cantarellbold", + "cantarellboldoblique", + "comicsans", + "comicsansbold", + "constantia", + "consolas", + "consolasitalic", + "consolasbold", + "consolasbolditalic", + "constantiaitalic", + "constantiabold", + "constantiabolditalic", + "corbel", + "corbelitalic", + "corbelbold", + "corbelbolditalic", + "courier", + "courieritalic", + "courierbold", + "courierbolditalic", + "dejavu", + "dejavu sans mono oblique", + "dejavu sans mono bold", + "dejavu sans mono bold oblique", + "georgia", + "georgiaitalic", + "georgiabold", + "georgiabolditalic", + "impact", + "tahoma", + "timesnewroman", + "timesnewromanitalic", + "timesnewromanbold", + "timesnewromanbolditalic", + "trebuchet", + "trebuchetitalic", + "trebuchetbold", + "trebuchetbolditalic", + "verdana", + "verdanaitalic", + "verdanabold", + "verdanabolditalic", +] +"""Published text-capable pdfRest font names for ZUGFeRD rendering. + +These are suggestions, not a closed service catalog. Another installed, +text-capable font may be supplied through ``PdfZugferdFontName``. Symbol fonts +such as Webdings are not valid for readable invoice text. + +Accepted values: + +- `andalemono`: Andale Mono. +- `arial`: Arial. +- `arialitalic`: Arial Italic. +- `arialbold`: Arial Bold. +- `arialblack`: Arial Black. +- `calibri`: Calibri. +- `calibriitalic`: Calibri Italic. +- `calibribold`: Calibri Bold. +- `calibribolditalic`: Calibri Bold Italic. +- `cambriaitalic`: Cambria Italic. +- `cambriabold`: Cambria Bold. +- `cambriabolditalic`: Cambria Bold Italic. +- `candara`: Candara. +- `candaraitalic`: Candara Italic. +- `candarabold`: Candara Bold. +- `candarabolditalic`: Candara Bold Italic. +- `cantarellregular`: Cantarell. +- `cantarelloblique`: Cantarell Oblique. +- `cantarellbold`: Cantarell Bold. +- `cantarellboldoblique`: Cantarell Bold Oblique. +- `comicsans`: Comic Sans. +- `comicsansbold`: Comic Sans Bold. +- `constantia`: Constantia. +- `consolas`: Consolas. +- `consolasitalic`: Consolas Italic. +- `consolasbold`: Consolas Bold. +- `consolasbolditalic`: Consolas Bold Italic. +- `constantiaitalic`: Constantia Italic. +- `constantiabold`: Constantia Bold. +- `constantiabolditalic`: Constantia Bold Italic. +- `corbel`: Corbel. +- `corbelitalic`: Corbel Italic. +- `corbelbold`: Corbel Bold. +- `corbelbolditalic`: Corbel Bold Italic. +- `courier`: Courier New. +- `courieritalic`: Courier New Italic. +- `courierbold`: Courier New Bold. +- `courierbolditalic`: Courier New Bold Italic. +- `dejavu`: DejaVu Sans Mono. +- `dejavu sans mono oblique`: DejaVu Sans Mono Oblique. +- `dejavu sans mono bold`: DejaVu Sans Mono Bold. +- `dejavu sans mono bold oblique`: DejaVu Sans Mono Bold Oblique. +- `georgia`: Georgia. +- `georgiaitalic`: Georgia Italic. +- `georgiabold`: Georgia Bold. +- `georgiabolditalic`: Georgia Bold Italic. +- `impact`: Impact. +- `tahoma`: Tahoma. +- `timesnewroman`: Times New Roman. +- `timesnewromanitalic`: Times New Roman Italic. +- `timesnewromanbold`: Times New Roman Bold. +- `timesnewromanbolditalic`: Times New Roman Bold Italic. +- `trebuchet`: Trebuchet. +- `trebuchetitalic`: Trebuchet Italic. +- `trebuchetbold`: Trebuchet Bold. +- `trebuchetbolditalic`: Trebuchet Bold Italic. +- `verdana`: Verdana. +- `verdanaitalic`: Verdana Italic. +- `verdanabold`: Verdana Bold. +- `verdanabolditalic`: Verdana Bold Italic. +""" + +PdfZugferdFontName: TypeAlias = PdfZugferdKnownFont | str +"""A published pdfRest font token or another installed text-capable font.""" + + +class PdfZugferdRenderOptions(TypedDict, total=False): + """Appearance and regional options for generated ZUGFeRD invoice PDFs. + + Attributes: + logo_corner: Optional corner for an uploaded logo. Defaults to + ``top-left``. + font: Optional non-empty body-text font name, at most 200 characters. + bold_font: Optional non-empty emphasized-text font name, at most 200 + characters. + text_color: Optional primary text color as RGB (0-255 channels) or CMYK + (0-100 percentage channels). + muted_text_color: Optional secondary text color as RGB or CMYK. + accent_color: Optional accent color as RGB or CMYK. + border_color: Optional table/rule border color as RGB or CMYK. + table_header_color: Optional table-header fill color as RGB or CMYK. + table_alternate_color: Optional alternating-row fill color as RGB or + CMYK. + footer_message: Optional footer text, at most 500 characters. ``None`` + leaves the footer unset. + locale: Optional locale for value formatting. Defaults to ``en-GB``. + label_language: Optional BCP 47 label language whose primary subtag is + ``en``, ``de``, or ``fr``. Defaults to ``en``. + currency_display: Optional currency representation. Defaults to + ``code``. + date_format: Optional .NET date format that preserves the complete day, + month, and four-digit year. ``None`` uses the renderer default. + decimal_precision: Optional decimal places from 0 through 4. Defaults + to 2. + """ + + logo_corner: PdfZugferdLogoCorner + font: PdfZugferdFontName + bold_font: PdfZugferdFontName + text_color: PdfColor + muted_text_color: PdfColor + accent_color: PdfColor + border_color: PdfColor + table_header_color: PdfColor + table_alternate_color: PdfColor + footer_message: str | None + locale: PdfZugferdLocale + label_language: PdfZugferdLabelLanguage + currency_display: PdfZugferdCurrencyDisplay + date_format: str | None + decimal_precision: int + + PdfStructuredTextDataPresentation: TypeAlias = Literal["source", "hierarchy"] """JSON/XML presentation accepted by structured document conversion helpers. From 18eca2936b999b663cb5bb772e19b908c6dfe7ed Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 20 Oct 2026 13:45:45 -0500 Subject: [PATCH 02/15] client: Add ZUGFeRD PDF creation - Add synchronous and asynchronous helpers for uploaded invoice XML. - Support optional PDFs, logos, rendering controls, and request options. - Cover serialization, validation, response mapping, and both transports. Assisted-by: Codex --- src/pdfrest/client.py | 131 +++++++ tests/test_create_zugferd_pdf.py | 642 +++++++++++++++++++++++++++++++ 2 files changed, 773 insertions(+) create mode 100644 tests/test_create_zugferd_pdf.py diff --git a/src/pdfrest/client.py b/src/pdfrest/client.py index 7849669..94878ae 100644 --- a/src/pdfrest/client.py +++ b/src/pdfrest/client.py @@ -129,6 +129,7 @@ PdfToWordPayload, PdfUnrestrictPayload, PdfXfaToAcroformsPayload, + PdfZugferdPayload, PngPdfRestPayload, SummarizePdfTextPayload, TiffPdfRestPayload, @@ -178,6 +179,7 @@ PdfStructuredTextTableStyle, PdfTextColor, PdfXType, + PdfZugferdRenderOptions, PngColorModel, PostScriptLevel, SummaryFormat, @@ -4615,6 +4617,69 @@ def sign_pdf( timeout=timeout, ) + def create_zugferd_pdf( + self, + invoice: PdfRestFile | Sequence[PdfRestFile], + *, + pdf: PdfRestFile | Sequence[PdfRestFile] | None = None, + logo: PdfRestFile | Sequence[PdfRestFile] | None = None, + render_options: PdfZugferdRenderOptions | None = None, + regenerate_pdf: bool = False, + output: str | None = None, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Create a validated ZUGFeRD or Factur-X PDF/A-3 invoice. + + Args: + invoice: Uploaded CII invoice XML resource. + pdf: Optional uploaded PDF to preserve when its content matches the + invoice XML. + logo: Optional uploaded PNG or JPEG logo for generated or + regenerated visual PDFs. + render_options: Optional appearance and regional-formatting options. + regenerate_pdf: Allow replacement of the supplied PDF when its + content cannot be confirmed against the invoice XML. Requires + ``pdf`` when true. + output: Output filename prefix used by pdfRest when creating the + ZUGFeRD PDF. + extra_query: Additional query parameters merged into the request. + extra_headers: Additional HTTP headers merged into the request. + extra_body: Additional request body fields merged into the payload. + timeout: Request timeout override for this call. + + Returns: + Validated file-based response containing the generated PDF/A-3 file. + + Raises: + PdfRestError: If request execution fails at the client or API layer. + ValidationError: If local payload validation fails before sending. + """ + payload: dict[str, Any] = { + "invoice": invoice, + "regenerate_pdf": regenerate_pdf, + } + if pdf is not None: + payload["pdf"] = pdf + if logo is not None: + payload["logo"] = logo + if render_options is not None: + payload["render_options"] = render_options + if output is not None: + payload["output"] = output + + return self._post_file_operation( + endpoint="/zugferd-pdf", + payload=payload, + payload_model=PdfZugferdPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + def blank_pdf( self, *, @@ -8151,6 +8216,72 @@ async def sign_pdf( timeout=timeout, ) + async def create_zugferd_pdf( + self, + invoice: PdfRestFile | Sequence[PdfRestFile], + *, + pdf: PdfRestFile | Sequence[PdfRestFile] | None = None, + logo: PdfRestFile | Sequence[PdfRestFile] | None = None, + render_options: PdfZugferdRenderOptions | None = None, + regenerate_pdf: bool = False, + 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 create a ZUGFeRD or Factur-X PDF/A-3 invoice. + + This is the asynchronous variant of + [PdfRestClient.create_zugferd_pdf][pdfrest.PdfRestClient.create_zugferd_pdf]. + + Args: + invoice: Uploaded CII invoice XML resource. + pdf: Optional uploaded PDF to preserve when its content matches the + invoice XML. + logo: Optional uploaded PNG or JPEG logo for generated or + regenerated visual PDFs. + render_options: Optional appearance and regional-formatting options. + regenerate_pdf: Allow replacement of the supplied PDF when its + content cannot be confirmed against the invoice XML. Requires + ``pdf`` when true. + output: Output filename prefix used by pdfRest when creating the + ZUGFeRD PDF. + extra_query: Additional query parameters merged into the request. + extra_headers: Additional HTTP headers merged into the request. + extra_body: Additional request body fields merged into the payload. + timeout: Request timeout override for this call. + + Returns: + Validated file-based response containing the generated PDF/A-3 file. + + Raises: + PdfRestError: If request execution fails at the client or API layer. + ValidationError: If local payload validation fails before sending. + """ + payload: dict[str, Any] = { + "invoice": invoice, + "regenerate_pdf": regenerate_pdf, + } + if pdf is not None: + payload["pdf"] = pdf + if logo is not None: + payload["logo"] = logo + if render_options is not None: + payload["render_options"] = render_options + if output is not None: + payload["output"] = output + + return await self._post_file_operation( + endpoint="/zugferd-pdf", + payload=payload, + payload_model=PdfZugferdPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + async def blank_pdf( self, *, diff --git a/tests/test_create_zugferd_pdf.py b/tests/test_create_zugferd_pdf.py new file mode 100644 index 0000000..b877853 --- /dev/null +++ b/tests/test_create_zugferd_pdf.py @@ -0,0 +1,642 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any, cast, get_args + +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 PdfZugferdPayload +from pdfrest.types import ( + PdfZugferdCurrencyDisplay, + PdfZugferdKnownFont, + PdfZugferdLocale, + PdfZugferdLogoCorner, + PdfZugferdRenderOptions, +) + +from .graphics_test_helpers import ( + ASYNC_API_KEY, + VALID_API_KEY, + build_file_info_payload, +) + +KNOWN_FONTS = cast(tuple[PdfZugferdKnownFont, ...], get_args(PdfZugferdKnownFont)) +LOGO_CORNERS = cast(tuple[PdfZugferdLogoCorner, ...], get_args(PdfZugferdLogoCorner)) +LOCALES = cast(tuple[PdfZugferdLocale, ...], get_args(PdfZugferdLocale)) +CURRENCY_DISPLAYS = cast( + tuple[PdfZugferdCurrencyDisplay, ...], get_args(PdfZugferdCurrencyDisplay) +) + + +def _make_file(name: str, mime_type: str, *, prefix: int | None = None) -> PdfRestFile: + file_id = PdfRestFileID.generate(prefix) + return PdfRestFile.model_validate( + { + "id": file_id, + "name": name, + "url": f"https://api.pdfrest.com/resource/{file_id}", + "type": mime_type, + "size": 1024, + "modified": datetime(2024, 1, 1, tzinfo=timezone.utc) + .isoformat() + .replace("+00:00", "Z"), + "scheduledDeletionTimeUtc": None, + } + ) + + +def _dump(payload: dict[str, Any]) -> dict[str, Any]: + return PdfZugferdPayload.model_validate(payload).model_dump( + mode="json", by_alias=True, exclude_none=True, exclude_unset=True + ) + + +def _assert_timeout(value: float | dict[str, float] | None, expected: float) -> None: + assert value is not None + if isinstance(value, dict): + assert value + assert all(component == pytest.approx(expected) for component in value.values()) + else: + assert value == pytest.approx(expected) + + +def test_zugferd_payload_serializes_resources_and_render_options() -> None: + invoice = _make_file("invoice.xml", "application/xml", prefix=1) + pdf = _make_file("invoice.pdf", "application/pdf") + logo = _make_file("logo.png", "image/png", prefix=2) + + payload = _dump( + { + "invoice": invoice, + "pdf": pdf, + "logo": logo, + "regenerate_pdf": True, + "output": "customer-invoice", + "render_options": { + "logo_corner": "top-right", + "font": " arial ", + "bold_font": "arialbold", + "text_color": (26, 31, 41), + "muted_text_color": (10, 5, 0, 55), + "accent_color": (20, 64, 115), + "border_color": (12, 7, 0, 20), + "table_header_color": (20, 64, 115), + "table_alternate_color": (4, 2, 0, 2), + "footer_message": "Thank you for your business.", + "locale": "de-DE", + "label_language": "en-US", + "currency_display": "symbol", + "date_format": "dd.MM.yyyy", + "decimal_precision": 3, + }, + } + ) + + assert payload == { + "id": str(invoice.id), + "pdf_id": str(pdf.id), + "logo_id": str(logo.id), + "render_options": { + "logo_corner": "top-right", + "font": "arial", + "bold_font": "arialbold", + "text_color_rgb": [26, 31, 41], + "muted_text_color_cmyk": [10, 5, 0, 55], + "accent_color_rgb": [20, 64, 115], + "border_color_cmyk": [12, 7, 0, 20], + "table_header_color_rgb": [20, 64, 115], + "table_alternate_color_cmyk": [4, 2, 0, 2], + "footer_message": "Thank you for your business.", + "locale": "de-DE", + "label_language": "en-US", + "currency_display": "symbol", + "date_format": "dd.MM.yyyy", + "decimal_precision": 3, + }, + "regenerate_pdf": True, + "output": "customer-invoice", + } + + +@pytest.mark.parametrize( + "field", + [ + pytest.param("text_color", id="text"), + pytest.param("muted_text_color", id="muted-text"), + pytest.param("accent_color", id="accent"), + pytest.param("border_color", id="border"), + pytest.param("table_header_color", id="table-header"), + pytest.param("table_alternate_color", id="table-alternate"), + ], +) +@pytest.mark.parametrize( + ("color", "wire_suffix"), + [ + pytest.param((0, 127, 255), "rgb", id="rgb"), + pytest.param((0, 25, 50, 100), "cmyk", id="cmyk"), + ], +) +def test_zugferd_payload_routes_unified_colors( + field: str, + color: tuple[int, ...], + wire_suffix: str, +) -> None: + invoice = _make_file("invoice.xml", "text/xml") + + payload = _dump({"invoice": invoice, "render_options": {field: color}}) + + assert payload["render_options"] == {f"{field}_{wire_suffix}": list(color)} + + +@pytest.mark.parametrize("font", KNOWN_FONTS, ids=str) +def test_zugferd_payload_accepts_every_published_font( + font: PdfZugferdKnownFont, +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + payload = _dump({"invoice": invoice, "render_options": {"font": font}}) + + assert payload["render_options"] == {"font": font} + + +@pytest.mark.parametrize( + ("field", "value"), + [ + *( + pytest.param("logo_corner", value, id=f"logo-{value}") + for value in LOGO_CORNERS + ), + *(pytest.param("locale", value, id=f"locale-{value}") for value in LOCALES), + *( + pytest.param("currency_display", value, id=f"currency-{value}") + for value in CURRENCY_DISPLAYS + ), + pytest.param("label_language", "en", id="language-en"), + pytest.param("label_language", "de", id="language-de"), + pytest.param("label_language", "fr", id="language-fr"), + pytest.param("label_language", "fr-FR", id="language-region"), + ], +) +def test_zugferd_payload_accepts_render_option_literals(field: str, value: str) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + payload = _dump({"invoice": invoice, "render_options": {field: value}}) + + assert payload["render_options"] == {field: value} + + +@pytest.mark.parametrize( + "decimal_precision", + [pytest.param(0, id="minimum"), pytest.param(4, id="maximum")], +) +def test_zugferd_payload_accepts_decimal_precision_boundaries( + decimal_precision: int, +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + payload = _dump( + { + "invoice": invoice, + "render_options": {"decimal_precision": decimal_precision}, + } + ) + + assert payload["render_options"]["decimal_precision"] == decimal_precision + + +@pytest.mark.parametrize( + ("options", "error_match"), + [ + pytest.param( + {"decimal_precision": -1}, "greater than or equal to 0", id="precision-low" + ), + pytest.param( + {"decimal_precision": 5}, "less than or equal to 4", id="precision-high" + ), + pytest.param( + {"text_color": (0, 256, 0)}, "less than or equal to 255", id="rgb-high" + ), + pytest.param( + {"text_color": (0, 101, 0, 0)}, "less than or equal to 100", id="cmyk-high" + ), + pytest.param({"text_color": (0, 0)}, "exactly 3.*or 4", id="channel-count"), + pytest.param({"font": " "}, "at least 1 item", id="font-empty"), + pytest.param({"font": "x" * 201}, "at most 200 items", id="font-long"), + pytest.param({"font": "Webdings"}, "Webdings is unsupported", id="font-symbol"), + pytest.param( + {"footer_message": "x" * 501}, "at most 500 characters", id="footer-long" + ), + pytest.param( + {"label_language": "nl"}, + "English, German, or French", + id="language-primary", + ), + pytest.param( + {"label_language": "not_a_tag"}, "valid BCP 47", id="language-shape" + ), + pytest.param( + {"unsupported": True}, "Extra inputs are not permitted", id="unknown-field" + ), + ], +) +def test_zugferd_payload_rejects_invalid_render_options( + options: dict[str, Any], + error_match: str, +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + with pytest.raises(ValidationError, match=error_match): + PdfZugferdPayload.model_validate( + {"invoice": invoice, "render_options": options} + ) + + +def test_zugferd_payload_accepts_custom_installed_font() -> None: + invoice = _make_file("invoice.xml", "application/xml") + + payload = _dump({"invoice": invoice, "render_options": {"font": "Corporate Sans"}}) + + assert payload["render_options"] == {"font": "Corporate Sans"} + + +@pytest.mark.parametrize( + ("payload", "error_match"), + [ + pytest.param( + {"invoice": _make_file("invoice.pdf", "application/pdf")}, + "Invoice must be an XML file", + id="invoice-mime", + ), + pytest.param( + { + "invoice": _make_file("invoice.xml", "application/xml"), + "pdf": _make_file("source.png", "image/png"), + }, + "Must be a PDF file", + id="pdf-mime", + ), + pytest.param( + { + "invoice": _make_file("invoice.xml", "application/xml"), + "logo": _make_file("logo.gif", "image/gif"), + }, + "Logo must be a PNG or JPEG file", + id="logo-mime", + ), + pytest.param( + { + "invoice": [ + _make_file("one.xml", "application/xml"), + _make_file("two.xml", "text/xml"), + ] + }, + "at most 1 item", + id="invoice-cardinality", + ), + pytest.param( + { + "invoice": _make_file("invoice.xml", "application/xml"), + "regenerate_pdf": True, + }, + "requires a PDF resource", + id="regeneration-dependency", + ), + ], +) +def test_zugferd_payload_rejects_invalid_resources( + payload: dict[str, Any], error_match: str +) -> None: + with pytest.raises(ValidationError, match=error_match): + PdfZugferdPayload.model_validate(payload) + + +def test_create_zugferd_pdf_success_and_response_mapping() -> None: + invoice = _make_file("invoice.xml", "application/xml", prefix=1) + pdf = _make_file("invoice.pdf", "application/pdf") + logo = _make_file("logo.jpg", "image/jpeg", prefix=2) + output_id = PdfRestFileID.generate() + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + assert request.url.path == "/zugferd-pdf" + assert request.headers["Api-Key"] == VALID_API_KEY + assert json.loads(request.content) == { + "id": str(invoice.id), + "pdf_id": str(pdf.id), + "logo_id": str(logo.id), + "render_options": { + "logo_corner": "bottom-right", + "text_color_rgb": [1, 2, 3], + "accent_color_cmyk": [4, 3, 2, 1], + }, + "regenerate_pdf": True, + "output": "zugferd-result", + } + return httpx.Response( + 200, + json={ + "inputId": [str(invoice.id), str(pdf.id), str(logo.id)], + "outputId": str(output_id), + "outputUrl": f"https://api.pdfrest.com/resource/{output_id}", + }, + ) + assert request.method == "GET" + assert request.url.path == f"/resource/{output_id}" + return httpx.Response( + 200, + json=build_file_info_payload( + str(output_id), "zugferd-result.pdf", "application/pdf" + ), + ) + + options: PdfZugferdRenderOptions = { + "logo_corner": "bottom-right", + "text_color": (1, 2, 3), + "accent_color": (4, 3, 2, 1), + } + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = client.create_zugferd_pdf( + invoice, + pdf=pdf, + logo=logo, + render_options=options, + regenerate_pdf=True, + output="zugferd-result", + ) + + assert isinstance(response, PdfRestFileBasedResponse) + assert response.input_ids == [invoice.id, pdf.id, logo.id] + assert response.output_file.id == output_id + assert response.output_file.name == "zugferd-result.pdf" + assert response.output_file.type == "application/pdf" + + +@pytest.mark.asyncio +async def test_async_create_zugferd_pdf_success_and_response_mapping() -> None: + invoice = _make_file("invoice.xml", "text/xml", prefix=1) + pdf = _make_file("invoice.pdf", "application/pdf") + logo = _make_file("logo.png", "image/png", prefix=2) + output_id = PdfRestFileID.generate() + + async def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + assert request.url.path == "/zugferd-pdf" + assert request.headers["Api-Key"] == ASYNC_API_KEY + assert json.loads(request.content) == { + "id": str(invoice.id), + "pdf_id": str(pdf.id), + "logo_id": str(logo.id), + "render_options": { + "font": "Corporate Sans", + "bold_font": "arialbold", + "muted_text_color_cmyk": [0, 0, 0, 50], + "locale": "fr-FR", + "label_language": "fr-FR", + "currency_display": "name", + "decimal_precision": 4, + }, + "regenerate_pdf": True, + "output": "zugferd-async", + } + return httpx.Response( + 200, + json={ + "inputId": [str(invoice.id), str(pdf.id), str(logo.id)], + "outputId": str(output_id), + "outputUrl": f"https://api.pdfrest.com/resource/{output_id}", + }, + ) + assert request.method == "GET" + assert request.url.path == f"/resource/{output_id}" + return httpx.Response( + 200, + json=build_file_info_payload( + str(output_id), "zugferd-async.pdf", "application/pdf" + ), + ) + + options: PdfZugferdRenderOptions = { + "font": "Corporate Sans", + "bold_font": "arialbold", + "muted_text_color": (0, 0, 0, 50), + "locale": "fr-FR", + "label_language": "fr-FR", + "currency_display": "name", + "decimal_precision": 4, + } + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = await client.create_zugferd_pdf( + invoice, + pdf=pdf, + logo=logo, + render_options=options, + regenerate_pdf=True, + output="zugferd-async", + ) + + assert response.input_ids == [invoice.id, pdf.id, logo.id] + assert response.output_file.id == output_id + assert response.output_file.name == "zugferd-async.pdf" + + +def test_create_zugferd_pdf_request_customization() -> None: + invoice = _make_file("invoice.xml", "application/xml") + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.params["trace"] == "sync" + assert request.headers["X-Trace"] == "zugferd" + _assert_timeout(request.extensions.get("timeout"), 14.0) + assert json.loads(request.content) == { + "id": str(invoice.id), + "regenerate_pdf": False, + "server_debug": True, + } + return httpx.Response(200, json={"inputId": str(invoice.id)}) + + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = client.create_zugferd_pdf( + invoice, + extra_query={"trace": "sync"}, + extra_headers={"X-Trace": "zugferd"}, + extra_body={"server_debug": True}, + timeout=14.0, + ) + + assert response.input_id == invoice.id + assert response.output_files == [] + + +@pytest.mark.asyncio +async def test_async_create_zugferd_pdf_request_customization() -> None: + invoice = _make_file("invoice.xml", "application/xml") + + async def handler(request: httpx.Request) -> httpx.Response: + assert request.url.params["trace"] == "async" + assert request.headers["X-Trace"] == "zugferd-async" + _assert_timeout(request.extensions.get("timeout"), 21.0) + assert json.loads(request.content) == { + "id": str(invoice.id), + "regenerate_pdf": False, + "server_debug": True, + } + return httpx.Response(200, json={"inputId": str(invoice.id)}) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = await client.create_zugferd_pdf( + invoice, + extra_query={"trace": "async"}, + extra_headers={"X-Trace": "zugferd-async"}, + extra_body={"server_debug": True}, + timeout=21.0, + ) + + assert response.input_id == invoice.id + assert response.output_files == [] + + +@pytest.mark.parametrize("font", KNOWN_FONTS, ids=str) +def test_create_zugferd_pdf_accepts_every_published_font( + font: PdfZugferdKnownFont, +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + def handler(request: httpx.Request) -> httpx.Response: + assert json.loads(request.content)["render_options"] == {"font": font} + return httpx.Response(200, json={"inputId": str(invoice.id)}) + + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + client.create_zugferd_pdf(invoice, render_options={"font": font}) + + +@pytest.mark.parametrize("font", KNOWN_FONTS, ids=str) +@pytest.mark.asyncio +async def test_async_create_zugferd_pdf_accepts_every_published_font( + font: PdfZugferdKnownFont, +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + async def handler(request: httpx.Request) -> httpx.Response: + assert json.loads(request.content)["render_options"] == {"font": font} + return httpx.Response(200, json={"inputId": str(invoice.id)}) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + await client.create_zugferd_pdf(invoice, render_options={"font": font}) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + *( + pytest.param("logo_corner", value, id=f"logo-{value}") + for value in LOGO_CORNERS + ), + *(pytest.param("locale", value, id=f"locale-{value}") for value in LOCALES), + *( + pytest.param("currency_display", value, id=f"currency-{value}") + for value in CURRENCY_DISPLAYS + ), + pytest.param("label_language", "en", id="language-en"), + pytest.param("label_language", "de", id="language-de"), + pytest.param("label_language", "fr", id="language-fr"), + ], +) +def test_create_zugferd_pdf_accepts_every_render_literal( + field: str, value: str +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + def handler(request: httpx.Request) -> httpx.Response: + assert json.loads(request.content)["render_options"] == {field: value} + return httpx.Response(200, json={"inputId": str(invoice.id)}) + + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + client.create_zugferd_pdf( + invoice, + render_options=cast(PdfZugferdRenderOptions, cast(object, {field: value})), + ) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + *( + pytest.param("logo_corner", value, id=f"logo-{value}") + for value in LOGO_CORNERS + ), + *(pytest.param("locale", value, id=f"locale-{value}") for value in LOCALES), + *( + pytest.param("currency_display", value, id=f"currency-{value}") + for value in CURRENCY_DISPLAYS + ), + pytest.param("label_language", "en", id="language-en"), + pytest.param("label_language", "de", id="language-de"), + pytest.param("label_language", "fr", id="language-fr"), + ], +) +@pytest.mark.asyncio +async def test_async_create_zugferd_pdf_accepts_every_render_literal( + field: str, value: str +) -> None: + invoice = _make_file("invoice.xml", "application/xml") + + async def handler(request: httpx.Request) -> httpx.Response: + assert json.loads(request.content)["render_options"] == {field: value} + return httpx.Response(200, json={"inputId": str(invoice.id)}) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + await client.create_zugferd_pdf( + invoice, + render_options=cast(PdfZugferdRenderOptions, cast(object, {field: value})), + ) + + +def test_create_zugferd_pdf_rejects_invalid_resource_before_transport() -> None: + invoice = _make_file("invoice.pdf", "application/pdf") + + def handler(_: httpx.Request) -> httpx.Response: + pytest.fail("transport must not run for invalid local input") + + with ( + PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client, + pytest.raises(ValidationError, match="Invoice must be an XML file"), + ): + client.create_zugferd_pdf(invoice) + + +@pytest.mark.asyncio +async def test_async_create_zugferd_pdf_rejects_invalid_resource_before_transport() -> ( + None +): + invoice = _make_file("invoice.xml", "application/xml") + + async def handler(_: httpx.Request) -> httpx.Response: + pytest.fail("transport must not run for invalid local input") + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + with pytest.raises(ValidationError, match="requires a PDF resource"): + await client.create_zugferd_pdf(invoice, regenerate_pdf=True) From f16eae35ecbcb00c56e18b25e914f3ace3489c7a Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 20 Oct 2026 13:46:07 -0500 Subject: [PATCH 03/15] tests: Verify ZUGFeRD PDF creation live - Exercise every closed rendering option through both client transports. - Verify regenerated PDFs, response metadata, and server-side rejection. - Add a deterministic minimum-profile invoice fixture. Assisted-by: Codex --- tests/live/test_live_create_zugferd_pdf.py | 247 ++++++++++++++++++++ tests/resources/zugferd-minimum-invoice.xml | 51 ++++ 2 files changed, 298 insertions(+) create mode 100644 tests/live/test_live_create_zugferd_pdf.py create mode 100644 tests/resources/zugferd-minimum-invoice.xml diff --git a/tests/live/test_live_create_zugferd_pdf.py b/tests/live/test_live_create_zugferd_pdf.py new file mode 100644 index 0000000..aa1e835 --- /dev/null +++ b/tests/live/test_live_create_zugferd_pdf.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +from typing import TypedDict + +import pytest + +from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient +from pdfrest.models import PdfRestFile +from pdfrest.types import PdfZugferdRenderOptions + +from ..resources import get_test_resource_path + + +class ZugferdResources(TypedDict): + invoice: PdfRestFile + logo: PdfRestFile + pdf: PdfRestFile + + +RENDER_OPTION_CASES = [ + pytest.param( + PdfZugferdRenderOptions( + logo_corner="top-left", + font="courier", + text_color=(0, 0, 0), + locale="de-DE", + label_language="en", + currency_display="code", + decimal_precision=0, + ), + "top-left-de", + id="top-left-german-code", + ), + pytest.param( + PdfZugferdRenderOptions( + logo_corner="top-right", + font="DejaVu Sans", + muted_text_color=(10, 5, 0, 55), + locale="fr-FR", + label_language="de", + currency_display="symbol", + decimal_precision=1, + ), + "top-right-fr", + id="top-right-french-symbol", + ), + pytest.param( + PdfZugferdRenderOptions( + logo_corner="bottom-left", + accent_color=(20, 64, 115), + locale="nl-NL", + label_language="fr", + currency_display="name", + decimal_precision=4, + ), + "bottom-left-nl", + id="bottom-left-dutch-name", + ), + pytest.param( + PdfZugferdRenderOptions( + logo_corner="bottom-right", + table_header_color=(83, 44, 0, 55), + locale="en-GB", + label_language="en-US", + currency_display="code", + date_format="dd/MM/yyyy", + decimal_precision=2, + ), + "bottom-right-en", + id="bottom-right-english-region", + ), +] + + +@pytest.fixture(scope="module") +def uploaded_zugferd_resources( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> ZugferdResources: + paths = [ + get_test_resource_path("zugferd-minimum-invoice.xml"), + get_test_resource_path("signing_logo.png"), + get_test_resource_path("report.pdf"), + ] + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + invoice, logo, pdf = client.files.create_from_paths(paths) + return ZugferdResources(invoice=invoice, logo=logo, pdf=pdf) + + +def _assert_zugferd_output( + output: PdfRestFile, + output_bytes: bytes, + output_prefix: str, +) -> None: + assert output.name.startswith(output_prefix) + assert output.name.endswith(".pdf") + assert output.type == "application/pdf" + assert output.size == len(output_bytes) + assert output.size > 0 + assert f"/resource/{output.id}" in str(output.url) + assert output_bytes.startswith(b"%PDF-") + + +@pytest.mark.parametrize(("render_options", "output"), RENDER_OPTION_CASES) +def test_live_create_zugferd_pdf_render_options( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_zugferd_resources: ZugferdResources, + render_options: PdfZugferdRenderOptions, + output: str, +) -> None: + invoice = uploaded_zugferd_resources["invoice"] + logo = uploaded_zugferd_resources["logo"] + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.create_zugferd_pdf( + invoice, + logo=logo, + render_options=render_options, + output=output, + ) + output_bytes = client.files.read_bytes(response.output_file) + + assert response.input_ids == [invoice.id, logo.id] + assert response.warning is None + _assert_zugferd_output(response.output_file, output_bytes, output) + + +@pytest.mark.parametrize(("render_options", "output"), RENDER_OPTION_CASES) +@pytest.mark.asyncio +async def test_live_async_create_zugferd_pdf_render_options( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_zugferd_resources: ZugferdResources, + render_options: PdfZugferdRenderOptions, + output: str, +) -> None: + invoice = uploaded_zugferd_resources["invoice"] + logo = uploaded_zugferd_resources["logo"] + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.create_zugferd_pdf( + invoice, + logo=logo, + render_options=render_options, + output=f"async-{output}", + ) + output_bytes = await client.files.read_bytes(response.output_file) + + assert response.input_ids == [invoice.id, logo.id] + assert response.warning is None + _assert_zugferd_output(response.output_file, output_bytes, f"async-{output}") + + +def test_live_create_zugferd_pdf_regenerates_mismatched_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_zugferd_resources: ZugferdResources, +) -> None: + invoice = uploaded_zugferd_resources["invoice"] + pdf = uploaded_zugferd_resources["pdf"] + logo = uploaded_zugferd_resources["logo"] + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.create_zugferd_pdf( + invoice, + pdf=pdf, + logo=logo, + regenerate_pdf=True, + output="regenerated-zugferd", + ) + output_bytes = client.files.read_bytes(response.output_file) + + assert response.input_ids == [invoice.id, pdf.id, logo.id] + _assert_zugferd_output(response.output_file, output_bytes, "regenerated-zugferd") + + +@pytest.mark.asyncio +async def test_live_async_create_zugferd_pdf_regenerates_mismatched_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_zugferd_resources: ZugferdResources, +) -> None: + invoice = uploaded_zugferd_resources["invoice"] + pdf = uploaded_zugferd_resources["pdf"] + logo = uploaded_zugferd_resources["logo"] + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.create_zugferd_pdf( + invoice, + pdf=pdf, + logo=logo, + regenerate_pdf=True, + output="async-regenerated-zugferd", + ) + output_bytes = await client.files.read_bytes(response.output_file) + + assert response.input_ids == [invoice.id, pdf.id, logo.id] + _assert_zugferd_output( + response.output_file, output_bytes, "async-regenerated-zugferd" + ) + + +def test_live_create_zugferd_pdf_rejects_invalid_render_option( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_zugferd_resources: ZugferdResources, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)locale|invalid"), + ): + client.create_zugferd_pdf( + uploaded_zugferd_resources["invoice"], + extra_body={"render_options": {"locale": "xx-XX"}}, + ) + + +@pytest.mark.asyncio +async def test_live_async_create_zugferd_pdf_rejects_invalid_render_option( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_zugferd_resources: ZugferdResources, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + with pytest.raises(PdfRestApiError, match=r"(?i)locale|invalid"): + await client.create_zugferd_pdf( + uploaded_zugferd_resources["invoice"], + extra_body={"render_options": {"locale": "xx-XX"}}, + ) diff --git a/tests/resources/zugferd-minimum-invoice.xml b/tests/resources/zugferd-minimum-invoice.xml new file mode 100644 index 0000000..9fd3a95 --- /dev/null +++ b/tests/resources/zugferd-minimum-invoice.xml @@ -0,0 +1,51 @@ + + + + + + urn:factur-x.eu:1p0:minimum + + + + 471102 + 380 + + 20200305 + + + + + + Lieferant GmbH + + DE + + + 201/113/40209 + + + DE123456789 + + + + Kunden AG Frankreich + + + + + EUR + + 198.00 + 37.62 + 235.62 + 235.62 + + + + From e40eea24827941f21229016bff02addce6a29b7e Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 20 Oct 2026 13:46:42 -0500 Subject: [PATCH 04/15] examples: Demonstrate ZUGFeRD PDF creation - Add a standalone typed example with a deterministic invoice fixture. - Verify the generated PDF and print useful response metadata. - Link the helper from the API guide and example inventories. Assisted-by: Codex --- docs/api-guide.md | 3 + docs/llms-full.txt | 2 + examples/README.md | 2 + .../create_zugferd_pdf_example.py | 72 +++++++++++++++++++ .../resources/zugferd-minimum-invoice.xml | 51 +++++++++++++ 5 files changed, 130 insertions(+) create mode 100644 examples/create_zugferd_pdf/create_zugferd_pdf_example.py create mode 100644 examples/resources/zugferd-minimum-invoice.xml diff --git a/docs/api-guide.md b/docs/api-guide.md index b233d88..be447f2 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -107,6 +107,8 @@ Use this group for password protection, permissions, signatures, and standards. - Archival/print conformance: [convert_to_pdfa][pdfrest.PdfRestClient.convert_to_pdfa], [convert_to_pdfx][pdfrest.PdfRestClient.convert_to_pdfx] +- Electronic invoices: + [create_zugferd_pdf][pdfrest.PdfRestClient.create_zugferd_pdf] ## PDF cleanup and rendering normalization @@ -130,6 +132,7 @@ Use this group when the user starts with non-PDF content or needs downstream formats. - Into PDF: + [create_zugferd_pdf][pdfrest.PdfRestClient.create_zugferd_pdf], [convert_office_to_pdf][pdfrest.PdfRestClient.convert_office_to_pdf], [convert_postscript_to_pdf][pdfrest.PdfRestClient.convert_postscript_to_pdf], [convert_email_to_pdf][pdfrest.PdfRestClient.convert_email_to_pdf], diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 5b55afd..56da7ec 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -158,6 +158,8 @@ to the API guide and API reference for exact method names and options. - Apply an uploaded `.joboptions` Distiller profile when converting PostScript or EPS with `convert_postscript_to_pdf` - Convert to standards formats (`PDF/A`, `PDF/X`) +- Create validated ZUGFeRD or Factur-X PDF/A-3 invoices from uploaded CII XML + with optional source-PDF, logo, and render controls ### Extraction and Analysis diff --git a/examples/README.md b/examples/README.md index 39bf5ee..5fc6e5a 100644 --- a/examples/README.md +++ b/examples/README.md @@ -35,6 +35,8 @@ supported interpreter matrix. synchronously convert selected PDF pages and verify the PostScript output. - `examples/convert_pdf_to_postscript/async_convert_pdf_to_postscript_example.py` – asynchronously convert a PDF page and verify the PostScript output. +- `examples/create_zugferd_pdf/create_zugferd_pdf_example.py` – create a + formatted ZUGFeRD PDF/A-3 invoice from uploaded CII XML. - `examples/delete/delete_example.py` – demonstrate file deletion (sync + async variants). - `examples/extract_text/extract_pdf_text_example.py` – run `extract_pdf_text` diff --git a/examples/create_zugferd_pdf/create_zugferd_pdf_example.py b/examples/create_zugferd_pdf/create_zugferd_pdf_example.py new file mode 100644 index 0000000..1e9bb3a --- /dev/null +++ b/examples/create_zugferd_pdf/create_zugferd_pdf_example.py @@ -0,0 +1,72 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = ["pdfrest", "python-dotenv"] +# /// +"""Create a branded ZUGFeRD PDF/A-3 invoice from CII XML. + +This sample demonstrates how to: + +1. Upload the checked-in invoice XML resource. +2. Generate a ZUGFeRD PDF with typed appearance and locale options. +3. Download and verify the generated PDF. +4. Print the result metadata and input resource IDs. + +Set ``PDFREST_API_KEY`` and optionally ``PDFREST_BASE_URL`` for a self-hosted +deployment, then run from the repository root with +``uv run examples/create_zugferd_pdf/create_zugferd_pdf_example.py``. The +required XML input is included in the repository. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from dotenv import load_dotenv + +from pdfrest import PdfRestClient +from pdfrest.types import PdfZugferdRenderOptions + +RESOURCE_DIRECTORY = Path(__file__).resolve().parents[1] / "resources" +INVOICE_PATH = RESOURCE_DIRECTORY / "zugferd-minimum-invoice.xml" + + +def create_zugferd_pdf() -> None: + """Upload invoice resources, create the PDF/A-3 invoice, and verify it.""" + load_dotenv() + base_url = os.getenv("PDFREST_BASE_URL") + render_options = PdfZugferdRenderOptions( + font="courier", + bold_font="courierbold", + accent_color=(20, 64, 115), + locale="de-DE", + label_language="en", + currency_display="symbol", + date_format="dd.MM.yyyy", + decimal_precision=2, + footer_message="Thank you for your business.", + ) + + with PdfRestClient(base_url=base_url) as client: + invoice = client.files.create_from_paths(INVOICE_PATH)[0] + response = client.create_zugferd_pdf( + invoice, + render_options=render_options, + output="zugferd-invoice", + ) + pdf_bytes = client.files.read_bytes(response.output_file) + + if not pdf_bytes.startswith(b"%PDF-"): + msg = "The generated resource is not a PDF." + raise RuntimeError(msg) + + output = response.output_file + print(f"Created: {output.name}") + print(f"MIME type: {output.type}") + print(f"Size: {output.size} bytes") + print(f"Input IDs: {', '.join(response.input_ids)}") + print(f"Download URL: {output.url}") + + +if __name__ == "__main__": # pragma: no cover - manual example + create_zugferd_pdf() diff --git a/examples/resources/zugferd-minimum-invoice.xml b/examples/resources/zugferd-minimum-invoice.xml new file mode 100644 index 0000000..9fd3a95 --- /dev/null +++ b/examples/resources/zugferd-minimum-invoice.xml @@ -0,0 +1,51 @@ + + + + + + urn:factur-x.eu:1p0:minimum + + + + 471102 + 380 + + 20200305 + + + + + + Lieferant GmbH + + DE + + + 201/113/40209 + + + DE123456789 + + + + Kunden AG Frankreich + + + + + EUR + + 198.00 + 37.62 + 235.62 + 235.62 + + + + From 3b05f2ea881983b46c18e7c6d58dfa95437fcb0d Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Thu, 22 Oct 2026 11:19:25 -0500 Subject: [PATCH 05/15] models: Define ZUGFeRD validation contracts - Validate a single uploaded PDF resource before request serialization. - Model validation findings and response status fields returned by pdfRest. - Export the public response models for typed client consumers. Assisted-by: Codex --- src/pdfrest/models/__init__.py | 4 ++ src/pdfrest/models/_internal.py | 16 +++++++ src/pdfrest/models/public.py | 79 ++++++++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/src/pdfrest/models/__init__.py b/src/pdfrest/models/__init__.py index 2ed9563..62ab2fc 100644 --- a/src/pdfrest/models/__init__.py +++ b/src/pdfrest/models/__init__.py @@ -18,6 +18,8 @@ PdfRestFileBasedResponse, PdfRestFileID, PdfRestInfoResponse, + PdfZugferdFinding, + PdfZugferdValidationResponse, SummarizePdfTextResponse, TranslatePdfTextFileResponse, TranslatePdfTextResponse, @@ -42,6 +44,8 @@ "PdfRestFileBasedResponse", "PdfRestFileID", "PdfRestInfoResponse", + "PdfZugferdFinding", + "PdfZugferdValidationResponse", "SummarizePdfTextResponse", "TranslatePdfTextFileResponse", "TranslatePdfTextResponse", diff --git a/src/pdfrest/models/_internal.py b/src/pdfrest/models/_internal.py index 21173a6..ea2dc18 100644 --- a/src/pdfrest/models/_internal.py +++ b/src/pdfrest/models/_internal.py @@ -1833,6 +1833,22 @@ def _validate_regeneration_source(self) -> PdfZugferdPayload: return self +class PdfValidateZugferdPayload(BaseModel): + """Adapt an uploaded PDF resource into a ZUGFeRD validation request.""" + + model_config = ConfigDict(extra="forbid") + + file: Annotated[ + list[PdfRestFile], + Field(min_length=1, max_length=1, 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), + ] + + class PdfLiteralRedactionModel(BaseModel): type: Literal["literal"] value: Annotated[str, Field(min_length=1)] diff --git a/src/pdfrest/models/public.py b/src/pdfrest/models/public.py index 26ff77b..5fbc33b 100644 --- a/src/pdfrest/models/public.py +++ b/src/pdfrest/models/public.py @@ -5,7 +5,7 @@ import re import uuid as _uuid from datetime import date -from typing import Annotated, Any, ClassVar +from typing import Annotated, Any, ClassVar, Literal from pydantic import ( AliasChoices, @@ -45,6 +45,8 @@ "PdfRestFileBasedResponse", "PdfRestFileID", "PdfRestInfoResponse", + "PdfZugferdFinding", + "PdfZugferdValidationResponse", "SummarizePdfTextResponse", "TranslatePdfTextFileResponse", "TranslatePdfTextResponse", @@ -368,6 +370,81 @@ def output_file(self) -> PdfRestFile: raise ValueError(msg) +class PdfZugferdFinding(BaseModel): + """A finding reported while validating a ZUGFeRD or Factur-X PDF.""" + + code: str + """Stable finding code reported by the validation tooling.""" + + severity: Literal["INFO", "WARNING", "BLOCKING", "ERROR"] + """Severity assigned to the finding by the validation tooling.""" + + message: str + """Human-readable summary of the finding.""" + + field: str | None = None + """Invoice field associated with the finding, when available.""" + + status: str | None = None + """Status detail associated with the finding, when available.""" + + comparison_reason: Annotated[ + str | None, + Field( + validation_alias=AliasChoices("comparison_reason", "comparisonReason"), + pattern=r"^[A-Z][A-Z0-9_]{0,63}$", + ), + ] = None + """Machine-readable reason for a visual-comparison result, when available.""" + + model_config = ConfigDict(extra="allow", frozen=True) + + +class PdfZugferdValidationResponse(BaseModel): + """Result returned after validating a ZUGFeRD or Factur-X PDF.""" + + status: Literal["VALID", "INVALID"] + """Overall ZUGFeRD package, XML, business-rule, and PDF/A-3 result.""" + + pdfa: bool + """Whether the independent PDFChecker result confirms PDF/A-3 conformance.""" + + content_match_status: Annotated[ + Literal[ + "NOT_APPLICABLE", + "CONFIRMED", + "UNVERIFIED", + "FAILED", + "ERROR", + "DEMO_REDACTED", + ], + Field( + validation_alias=AliasChoices("content_match_status", "contentMatchStatus") + ), + ] + """Result of comparing visible PDF content with the embedded invoice XML.""" + + input_id: Annotated[ + PdfRestFileID, + Field(validation_alias=AliasChoices("input_id", "inputId")), + ] + """Identifier of the validated PDF resource.""" + + input_url: Annotated[ + str, + Field(validation_alias=AliasChoices("input_url", "inputUrl")), + ] + """Download URL for the validated PDF resource.""" + + findings: list[PdfZugferdFinding] | None = None + """Actionable validation findings and non-blocking warnings, when present.""" + + message: str | None = None + """Free-tier notice when detailed validation data has been redacted.""" + + model_config = ConfigDict(extra="allow", frozen=True) + + class PdfRestDeletionResponse(BaseModel): """Response returned by the delete tool.""" From c85ef95479ae6d46874d294d1bedc583db82fa99 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Thu, 22 Oct 2026 11:19:42 -0500 Subject: [PATCH 06/15] client: Add ZUGFeRD PDF validation helpers - Send uploaded PDF resource IDs to the validation endpoint. - Return typed results from synchronous and asynchronous clients. - Preserve request customization and timeout support. Assisted-by: Codex --- src/pdfrest/client.py | 91 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/src/pdfrest/client.py b/src/pdfrest/client.py index 94878ae..0e5e105 100644 --- a/src/pdfrest/client.py +++ b/src/pdfrest/client.py @@ -68,6 +68,7 @@ PdfRestFileBasedResponse, PdfRestFileID, PdfRestInfoResponse, + PdfZugferdValidationResponse, SummarizePdfTextResponse, TranslatePdfTextFileResponse, TranslatePdfTextResponse, @@ -128,6 +129,7 @@ PdfToPowerpointPayload, PdfToWordPayload, PdfUnrestrictPayload, + PdfValidateZugferdPayload, PdfXfaToAcroformsPayload, PdfZugferdPayload, PngPdfRestPayload, @@ -4617,6 +4619,51 @@ def sign_pdf( timeout=timeout, ) + def validate_zugferd_pdf( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfZugferdValidationResponse: + """Validate an existing ZUGFeRD or Factur-X PDF without modifying it. + + Checks the embedded invoice XML, attachment relationship, package + metadata, supported profile, visible-content match, and independent + PDF/A-3 conformance. + + Args: + file: Uploaded ZUGFeRD or Factur-X PDF resource to validate. + extra_query: Additional query parameters merged into the request. + extra_headers: Additional HTTP headers merged into the request. + extra_body: Additional request body fields merged into the payload. + timeout: Request timeout override for this call. + + Returns: + Typed validation status, PDF/A result, content-match status, and + any findings reported by pdfRest. + + Raises: + PdfRestError: If request execution fails at the client or API layer. + ValidationError: If local payload validation fails before sending. + """ + validated_payload = PdfValidateZugferdPayload.model_validate({"file": file}) + request = self.prepare_request( + "POST", + "/validated-zugferd", + json_body=validated_payload.model_dump( + mode="json", by_alias=True, exclude_none=True, exclude_unset=True + ), + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + raw_payload = self._send_request(request) + return PdfZugferdValidationResponse.model_validate(raw_payload) + def create_zugferd_pdf( self, invoice: PdfRestFile | Sequence[PdfRestFile], @@ -8216,6 +8263,50 @@ async def sign_pdf( timeout=timeout, ) + async def validate_zugferd_pdf( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfZugferdValidationResponse: + """Asynchronously validate an existing ZUGFeRD or Factur-X PDF. + + This is the asynchronous variant of + [PdfRestClient.validate_zugferd_pdf][pdfrest.PdfRestClient.validate_zugferd_pdf]. + + Args: + file: Uploaded ZUGFeRD or Factur-X PDF resource to validate. + extra_query: Additional query parameters merged into the request. + extra_headers: Additional HTTP headers merged into the request. + extra_body: Additional request body fields merged into the payload. + timeout: Request timeout override for this call. + + Returns: + Typed validation status, PDF/A result, content-match status, and + any findings reported by pdfRest. + + Raises: + PdfRestError: If request execution fails at the client or API layer. + ValidationError: If local payload validation fails before sending. + """ + validated_payload = PdfValidateZugferdPayload.model_validate({"file": file}) + request = self.prepare_request( + "POST", + "/validated-zugferd", + json_body=validated_payload.model_dump( + mode="json", by_alias=True, exclude_none=True, exclude_unset=True + ), + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + raw_payload = await self._send_request(request) + return PdfZugferdValidationResponse.model_validate(raw_payload) + async def create_zugferd_pdf( self, invoice: PdfRestFile | Sequence[PdfRestFile], From dca18895418395c5ff78b985ad46e99249fc7400 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Thu, 22 Oct 2026 11:19:54 -0500 Subject: [PATCH 07/15] tests: Cover ZUGFeRD PDF validation helpers - Verify payload serialization and local resource validation. - Exercise response literals, aliases, findings, and demo redactions. - Assert sync and async request customization and transport behavior. Assisted-by: Codex --- tests/test_validate_zugferd_pdf.py | 383 +++++++++++++++++++++++++++++ 1 file changed, 383 insertions(+) create mode 100644 tests/test_validate_zugferd_pdf.py diff --git a/tests/test_validate_zugferd_pdf.py b/tests/test_validate_zugferd_pdf.py new file mode 100644 index 0000000..8a69a06 --- /dev/null +++ b/tests/test_validate_zugferd_pdf.py @@ -0,0 +1,383 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any + +import httpx +import pytest +from pydantic import ValidationError + +from pdfrest import AsyncPdfRestClient, PdfRestClient +from pdfrest.models import ( + PdfRestFile, + PdfRestFileID, + PdfZugferdFinding, + PdfZugferdValidationResponse, +) +from pdfrest.models._internal import PdfValidateZugferdPayload + +from .graphics_test_helpers import ASYNC_API_KEY, VALID_API_KEY + + +def _make_file(name: str, mime_type: str) -> PdfRestFile: + file_id = PdfRestFileID.generate() + return PdfRestFile.model_validate( + { + "id": file_id, + "name": name, + "url": f"https://api.pdfrest.com/resource/{file_id}", + "type": mime_type, + "size": 1024, + "modified": datetime(2024, 1, 1, tzinfo=timezone.utc) + .isoformat() + .replace("+00:00", "Z"), + "scheduledDeletionTimeUtc": None, + } + ) + + +def _response_payload( + file_id: PdfRestFileID, + *, + status: str = "VALID", + content_match_status: str = "CONFIRMED", + severity: str = "WARNING", +) -> dict[str, Any]: + return { + "status": status, + "pdfa": True, + "contentMatchStatus": content_match_status, + "inputId": str(file_id), + "inputUrl": f"https://api.pdfrest.com/resource/{file_id}?format=file", + "findings": [ + { + "code": "PDF_XML_FIELD_RESULT", + "severity": severity, + "message": "Complete reverse validation is not available.", + "field": "visualConsistency", + "status": "UNVERIFIED", + "comparisonReason": "PARTIAL_REVERSE_VALIDATION", + } + ], + } + + +def _assert_timeout(value: float | dict[str, float] | None, expected: float) -> None: + assert value is not None + if isinstance(value, dict): + assert value + assert all(component == pytest.approx(expected) for component in value.values()) + else: + assert value == pytest.approx(expected) + + +def test_validate_zugferd_payload_serializes_uploaded_pdf_id() -> None: + pdf = _make_file("invoice.pdf", "application/pdf") + + payload = PdfValidateZugferdPayload.model_validate({"file": pdf}).model_dump( + mode="json", by_alias=True, exclude_none=True, exclude_unset=True + ) + + assert payload == {"id": str(pdf.id)} + + +@pytest.mark.parametrize( + ("file", "error_match"), + [ + pytest.param( + _make_file("invoice.xml", "application/xml"), + "Must be a PDF file", + id="mime", + ), + pytest.param( + [ + _make_file("invoice-one.pdf", "application/pdf"), + _make_file("invoice-two.pdf", "application/pdf"), + ], + "at most 1 item", + id="cardinality", + ), + ], +) +def test_validate_zugferd_payload_rejects_invalid_resources( + file: PdfRestFile | list[PdfRestFile], error_match: str +) -> None: + with pytest.raises(ValidationError, match=error_match): + PdfValidateZugferdPayload.model_validate({"file": file}) + + +@pytest.mark.parametrize("status", ["VALID", "INVALID"]) +def test_zugferd_validation_response_accepts_status_values(status: str) -> None: + file_id = PdfRestFileID.generate() + + response = PdfZugferdValidationResponse.model_validate( + _response_payload(file_id, status=status) + ) + + assert response.status == status + + +@pytest.mark.parametrize( + "content_match_status", + [ + "NOT_APPLICABLE", + "CONFIRMED", + "UNVERIFIED", + "FAILED", + "ERROR", + "DEMO_REDACTED", + ], +) +def test_zugferd_validation_response_accepts_content_match_values( + content_match_status: str, +) -> None: + file_id = PdfRestFileID.generate() + + response = PdfZugferdValidationResponse.model_validate( + _response_payload(file_id, content_match_status=content_match_status) + ) + + assert response.content_match_status == content_match_status + + +@pytest.mark.parametrize("severity", ["INFO", "WARNING", "BLOCKING", "ERROR"]) +def test_zugferd_validation_response_accepts_finding_severities( + severity: str, +) -> None: + file_id = PdfRestFileID.generate() + + response = PdfZugferdValidationResponse.model_validate( + _response_payload(file_id, severity=severity) + ) + + assert response.findings is not None + assert response.findings[0].severity == severity + + +def test_zugferd_validation_response_maps_camel_case_fields() -> None: + file_id = PdfRestFileID.generate() + + response = PdfZugferdValidationResponse.model_validate(_response_payload(file_id)) + + assert response.input_id == file_id + assert response.content_match_status == "CONFIRMED" + assert response.input_url.endswith("?format=file") + assert response.findings == [ + PdfZugferdFinding( + code="PDF_XML_FIELD_RESULT", + severity="WARNING", + message="Complete reverse validation is not available.", + field="visualConsistency", + status="UNVERIFIED", + comparison_reason="PARTIAL_REVERSE_VALIDATION", + ) + ] + + +def test_zugferd_validation_response_accepts_demo_redactions() -> None: + file_id = PdfRestFileID.generate(prefix=2) + + response = PdfZugferdValidationResponse.model_validate( + { + "status": "INVALID", + "pdfa": True, + "contentMatchStatus": "DEMO_REDACTED", + "inputId": str(file_id), + "inputUrl": f"https://api.pdfrest.com/resource/{file_id}?format=file", + "findings": [ + { + "code": "PD******************", + "severity": "BLOCKING", + "message": "PD* ***** ***** **** *** ***** ****", + "field": "BT****", + "status": "MISMATCH", + "comparisonReason": "DEMO_REDACTED", + } + ], + "message": "Validation details are restricted.", + } + ) + + assert response.findings is not None + assert response.findings[0].code == "PD******************" + assert response.findings[0].comparison_reason == "DEMO_REDACTED" + + +def test_zugferd_validation_response_rejects_invalid_comparison_reason() -> None: + file_id = PdfRestFileID.generate() + payload = _response_payload(file_id) + payload["findings"][0]["comparisonReason"] = "not-safe" + + with pytest.raises(ValidationError, match="comparisonReason"): + PdfZugferdValidationResponse.model_validate(payload) + + +def test_validate_zugferd_pdf_success_and_response_mapping() -> None: + pdf = _make_file("invoice.pdf", "application/pdf") + + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" + assert request.url.path == "/validated-zugferd" + assert request.headers["Api-Key"] == VALID_API_KEY + assert json.loads(request.content) == {"id": str(pdf.id)} + return httpx.Response(200, json=_response_payload(pdf.id)) + + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = client.validate_zugferd_pdf(pdf) + + assert isinstance(response, PdfZugferdValidationResponse) + assert response.status == "VALID" + assert response.pdfa is True + assert response.content_match_status == "CONFIRMED" + assert response.input_id == pdf.id + assert response.findings is not None + assert response.findings[0].comparison_reason == "PARTIAL_REVERSE_VALIDATION" + + +@pytest.mark.asyncio +async def test_async_validate_zugferd_pdf_success_and_response_mapping() -> None: + pdf = _make_file("invoice.pdf", "application/pdf") + + async def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" + assert request.url.path == "/validated-zugferd" + assert request.headers["Api-Key"] == ASYNC_API_KEY + assert json.loads(request.content) == {"id": str(pdf.id)} + payload = _response_payload( + pdf.id, status="INVALID", content_match_status="DEMO_REDACTED" + ) + payload["message"] = "Validation details are restricted." + return httpx.Response(200, json=payload) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = await client.validate_zugferd_pdf(pdf) + + assert response.status == "INVALID" + assert response.content_match_status == "DEMO_REDACTED" + assert response.input_id == pdf.id + assert response.message == "Validation details are restricted." + + +def test_validate_zugferd_pdf_request_customization() -> None: + pdf = _make_file("invoice.pdf", "application/pdf") + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.params["trace"] == "sync" + assert request.headers["X-Trace"] == "zugferd-validation" + _assert_timeout(request.extensions.get("timeout"), 13.0) + assert json.loads(request.content) == { + "id": str(pdf.id), + "server_debug": True, + } + return httpx.Response(200, json=_response_payload(pdf.id)) + + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = client.validate_zugferd_pdf( + pdf, + extra_query={"trace": "sync"}, + extra_headers={"X-Trace": "zugferd-validation"}, + extra_body={"server_debug": True}, + timeout=13.0, + ) + + assert response.input_id == pdf.id + + +@pytest.mark.asyncio +async def test_async_validate_zugferd_pdf_request_customization() -> None: + pdf = _make_file("invoice.pdf", "application/pdf") + + async def handler(request: httpx.Request) -> httpx.Response: + assert request.url.params["trace"] == "async" + assert request.headers["X-Trace"] == "zugferd-validation-async" + _assert_timeout(request.extensions.get("timeout"), 17.0) + assert json.loads(request.content) == { + "id": str(pdf.id), + "server_debug": True, + } + return httpx.Response(200, json=_response_payload(pdf.id)) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = await client.validate_zugferd_pdf( + pdf, + extra_query={"trace": "async"}, + extra_headers={"X-Trace": "zugferd-validation-async"}, + extra_body={"server_debug": True}, + timeout=17.0, + ) + + assert response.input_id == pdf.id + + +@pytest.mark.parametrize( + ("file", "error_match"), + [ + pytest.param( + _make_file("invoice.xml", "application/xml"), + "Must be a PDF file", + id="mime", + ), + pytest.param( + [ + _make_file("invoice-one.pdf", "application/pdf"), + _make_file("invoice-two.pdf", "application/pdf"), + ], + "at most 1 item", + id="cardinality", + ), + ], +) +def test_validate_zugferd_pdf_rejects_invalid_resource_before_transport( + file: PdfRestFile | list[PdfRestFile], error_match: str +) -> None: + def handler(_: httpx.Request) -> httpx.Response: + pytest.fail("transport must not run for invalid local input") + + with ( + PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client, + pytest.raises(ValidationError, match=error_match), + ): + client.validate_zugferd_pdf(file) + + +@pytest.mark.parametrize( + ("file", "error_match"), + [ + pytest.param( + _make_file("invoice.xml", "application/xml"), + "Must be a PDF file", + id="mime", + ), + pytest.param( + [ + _make_file("invoice-one.pdf", "application/pdf"), + _make_file("invoice-two.pdf", "application/pdf"), + ], + "at most 1 item", + id="cardinality", + ), + ], +) +@pytest.mark.asyncio +async def test_async_validate_zugferd_pdf_rejects_invalid_resource_before_transport( + file: PdfRestFile | list[PdfRestFile], error_match: str +) -> None: + async def handler(_: httpx.Request) -> httpx.Response: + pytest.fail("transport must not run for invalid local input") + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + with pytest.raises(ValidationError, match=error_match): + await client.validate_zugferd_pdf(file) From d508fffdc7d538891cd48dcae89b7bd5f48b5b60 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Thu, 22 Oct 2026 11:20:05 -0500 Subject: [PATCH 08/15] tests: Verify ZUGFeRD PDF validation live - Exercise valid and invalid validation requests through both clients. - Confirm created and regenerated invoices pass validation and PDF/A checks. - Accept confirmed and demo-redacted content comparison results. Assisted-by: Codex --- tests/live/test_live_create_zugferd_pdf.py | 21 +++- tests/live/test_live_validate_zugferd_pdf.py | 104 +++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 tests/live/test_live_validate_zugferd_pdf.py diff --git a/tests/live/test_live_create_zugferd_pdf.py b/tests/live/test_live_create_zugferd_pdf.py index aa1e835..b197532 100644 --- a/tests/live/test_live_create_zugferd_pdf.py +++ b/tests/live/test_live_create_zugferd_pdf.py @@ -5,7 +5,7 @@ import pytest from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient -from pdfrest.models import PdfRestFile +from pdfrest.models import PdfRestFile, PdfZugferdValidationResponse from pdfrest.types import PdfZugferdRenderOptions from ..resources import get_test_resource_path @@ -104,6 +104,17 @@ def _assert_zugferd_output( assert output_bytes.startswith(b"%PDF-") +def _assert_zugferd_validation( + validation: PdfZugferdValidationResponse, + output: PdfRestFile, +) -> None: + assert validation.status == "VALID" + assert validation.pdfa is True + assert validation.content_match_status in {"CONFIRMED", "DEMO_REDACTED"} + assert validation.input_id == output.id + assert f"/resource/{output.id}" in validation.input_url + + @pytest.mark.parametrize(("render_options", "output"), RENDER_OPTION_CASES) def test_live_create_zugferd_pdf_render_options( pdfrest_api_key: str, @@ -125,10 +136,12 @@ def test_live_create_zugferd_pdf_render_options( output=output, ) output_bytes = client.files.read_bytes(response.output_file) + validation = client.validate_zugferd_pdf(response.output_file) assert response.input_ids == [invoice.id, logo.id] assert response.warning is None _assert_zugferd_output(response.output_file, output_bytes, output) + _assert_zugferd_validation(validation, response.output_file) @pytest.mark.parametrize(("render_options", "output"), RENDER_OPTION_CASES) @@ -153,10 +166,12 @@ async def test_live_async_create_zugferd_pdf_render_options( output=f"async-{output}", ) output_bytes = await client.files.read_bytes(response.output_file) + validation = await client.validate_zugferd_pdf(response.output_file) assert response.input_ids == [invoice.id, logo.id] assert response.warning is None _assert_zugferd_output(response.output_file, output_bytes, f"async-{output}") + _assert_zugferd_validation(validation, response.output_file) def test_live_create_zugferd_pdf_regenerates_mismatched_pdf( @@ -179,9 +194,11 @@ def test_live_create_zugferd_pdf_regenerates_mismatched_pdf( output="regenerated-zugferd", ) output_bytes = client.files.read_bytes(response.output_file) + validation = client.validate_zugferd_pdf(response.output_file) assert response.input_ids == [invoice.id, pdf.id, logo.id] _assert_zugferd_output(response.output_file, output_bytes, "regenerated-zugferd") + _assert_zugferd_validation(validation, response.output_file) @pytest.mark.asyncio @@ -205,11 +222,13 @@ async def test_live_async_create_zugferd_pdf_regenerates_mismatched_pdf( output="async-regenerated-zugferd", ) output_bytes = await client.files.read_bytes(response.output_file) + validation = await client.validate_zugferd_pdf(response.output_file) assert response.input_ids == [invoice.id, pdf.id, logo.id] _assert_zugferd_output( response.output_file, output_bytes, "async-regenerated-zugferd" ) + _assert_zugferd_validation(validation, response.output_file) def test_live_create_zugferd_pdf_rejects_invalid_render_option( diff --git a/tests/live/test_live_validate_zugferd_pdf.py b/tests/live/test_live_validate_zugferd_pdf.py new file mode 100644 index 0000000..f4dee12 --- /dev/null +++ b/tests/live/test_live_validate_zugferd_pdf.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import pytest + +from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient +from pdfrest.models import PdfRestFile, PdfZugferdValidationResponse + +from ..resources import get_test_resource_path + + +@pytest.fixture(scope="module") +def generated_zugferd_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> PdfRestFile: + invoice_path = get_test_resource_path("zugferd-minimum-invoice.xml") + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + invoice = client.files.create_from_paths(invoice_path)[0] + response = client.create_zugferd_pdf( + invoice, + output="validation-source-zugferd", + ) + return response.output_file + + +def _assert_valid_response( + response: PdfZugferdValidationResponse, + source: PdfRestFile, +) -> None: + assert response.status == "VALID" + assert response.pdfa is True + assert response.content_match_status in {"CONFIRMED", "DEMO_REDACTED"} + assert response.input_id == source.id + assert f"/resource/{source.id}" in response.input_url + if response.findings is not None: + assert all(finding.code for finding in response.findings) + assert all(finding.message for finding in response.findings) + + +def test_live_validate_zugferd_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + generated_zugferd_pdf: PdfRestFile, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.validate_zugferd_pdf(generated_zugferd_pdf) + + _assert_valid_response(response, generated_zugferd_pdf) + + +@pytest.mark.asyncio +async def test_live_async_validate_zugferd_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + generated_zugferd_pdf: PdfRestFile, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.validate_zugferd_pdf(generated_zugferd_pdf) + + _assert_valid_response(response, generated_zugferd_pdf) + + +def test_live_validate_zugferd_pdf_rejects_invalid_resource_id( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + generated_zugferd_pdf: PdfRestFile, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)invalid|resource|processing"), + ): + client.validate_zugferd_pdf( + generated_zugferd_pdf, + extra_body={"id": "not-a-resource-id"}, + ) + + +@pytest.mark.asyncio +async def test_live_async_validate_zugferd_pdf_rejects_invalid_resource_id( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + generated_zugferd_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)invalid|resource|processing"): + await client.validate_zugferd_pdf( + generated_zugferd_pdf, + extra_body={"id": "not-a-resource-id"}, + ) From b95f4325d648abdce071235689d6dea66dc96faa Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Thu, 22 Oct 2026 11:20:18 -0500 Subject: [PATCH 09/15] examples: Demonstrate ZUGFeRD PDF validation - Create a deterministic invoice before validating its conformance. - Display typed status, PDF/A, content-match, and finding details. - Add the workflow to the runnable example inventory. Assisted-by: Codex --- examples/README.md | 2 + .../validate_zugferd_pdf_example.py | 59 +++++++++++++++++++ 2 files changed, 61 insertions(+) create mode 100644 examples/validate_zugferd_pdf/validate_zugferd_pdf_example.py diff --git a/examples/README.md b/examples/README.md index 5fc6e5a..3a76e55 100644 --- a/examples/README.md +++ b/examples/README.md @@ -37,6 +37,8 @@ supported interpreter matrix. – asynchronously convert a PDF page and verify the PostScript output. - `examples/create_zugferd_pdf/create_zugferd_pdf_example.py` – create a formatted ZUGFeRD PDF/A-3 invoice from uploaded CII XML. +- `examples/validate_zugferd_pdf/validate_zugferd_pdf_example.py` – create and + validate a ZUGFeRD PDF/A-3 invoice, including content-match findings. - `examples/delete/delete_example.py` – demonstrate file deletion (sync + async variants). - `examples/extract_text/extract_pdf_text_example.py` – run `extract_pdf_text` diff --git a/examples/validate_zugferd_pdf/validate_zugferd_pdf_example.py b/examples/validate_zugferd_pdf/validate_zugferd_pdf_example.py new file mode 100644 index 0000000..0885c46 --- /dev/null +++ b/examples/validate_zugferd_pdf/validate_zugferd_pdf_example.py @@ -0,0 +1,59 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = ["pdfrest", "python-dotenv"] +# /// +"""Create and validate a ZUGFeRD PDF/A-3 invoice. + +This sample demonstrates how to: + +1. Upload the checked-in CII invoice XML resource. +2. Create a ZUGFeRD PDF/A-3 invoice to use as deterministic validation input. +3. Validate its embedded XML, package metadata, visible content, and PDF/A-3 + conformance. +4. Print the typed validation status and any findings. + +Set ``PDFREST_API_KEY`` and optionally ``PDFREST_BASE_URL`` for a self-hosted +deployment, then run from the repository root with +``uv run examples/validate_zugferd_pdf/validate_zugferd_pdf_example.py``. The +required XML input is included in the repository. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from dotenv import load_dotenv + +from pdfrest import PdfRestClient + +RESOURCE_DIRECTORY = Path(__file__).resolve().parents[1] / "resources" +INVOICE_PATH = RESOURCE_DIRECTORY / "zugferd-minimum-invoice.xml" + + +def validate_zugferd_pdf() -> None: + """Create a deterministic ZUGFeRD PDF and validate its conformance.""" + load_dotenv() + base_url = os.getenv("PDFREST_BASE_URL") + + with PdfRestClient(base_url=base_url) as client: + invoice = client.files.create_from_paths(INVOICE_PATH)[0] + created = client.create_zugferd_pdf( + invoice, + output="zugferd-validation-example", + ) + validation = client.validate_zugferd_pdf(created.output_file) + + print(f"Status: {validation.status}") + print(f"PDF/A-3 conformant: {validation.pdfa}") + print(f"Content match: {validation.content_match_status}") + print(f"Input ID: {validation.input_id}") + print(f"Input URL: {validation.input_url}") + if validation.message: + print(f"Notice: {validation.message}") + for finding in validation.findings or []: + print(f"{finding.severity}: {finding.code} — {finding.message}") + + +if __name__ == "__main__": # pragma: no cover - manual example + validate_zugferd_pdf() From 402959cba10584710b421dad7efdca9efa955b7e Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Thu, 22 Oct 2026 11:20:32 -0500 Subject: [PATCH 10/15] docs: Document ZUGFeRD PDF validation - Link the validation helper from the electronic invoice API guide. - Describe package, content, and PDF/A conformance validation. Assisted-by: Codex --- docs/api-guide.md | 3 ++- docs/llms-full.txt | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/api-guide.md b/docs/api-guide.md index be447f2..30a5b3b 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -108,7 +108,8 @@ Use this group for password protection, permissions, signatures, and standards. [convert_to_pdfa][pdfrest.PdfRestClient.convert_to_pdfa], [convert_to_pdfx][pdfrest.PdfRestClient.convert_to_pdfx] - Electronic invoices: - [create_zugferd_pdf][pdfrest.PdfRestClient.create_zugferd_pdf] + [create_zugferd_pdf][pdfrest.PdfRestClient.create_zugferd_pdf], + [validate_zugferd_pdf][pdfrest.PdfRestClient.validate_zugferd_pdf] ## PDF cleanup and rendering normalization diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 56da7ec..8053417 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -160,6 +160,8 @@ to the API guide and API reference for exact method names and options. - Convert to standards formats (`PDF/A`, `PDF/X`) - Create validated ZUGFeRD or Factur-X PDF/A-3 invoices from uploaded CII XML with optional source-PDF, logo, and render controls +- Validate existing ZUGFeRD or Factur-X PDFs for package, XML, visible-content, + and independent PDF/A-3 conformance ### Extraction and Analysis From 8b5cda0d0baf7174944239d7cd6f42f8dad4f177 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Mon, 2 Nov 2026 14:58:11 -0600 Subject: [PATCH 11/15] tests: Strengthen ZUGFeRD validation coverage - Verify INVALID report.pdf responses live through both clients. - Match valid and demo-redacted response assertions across transports. - Cover empty resource-list rejection before request execution. Assisted-by: Codex --- tests/live/test_live_validate_zugferd_pdf.py | 58 ++++++ tests/test_validate_zugferd_pdf.py | 186 +++++++++++++++---- 2 files changed, 209 insertions(+), 35 deletions(-) diff --git a/tests/live/test_live_validate_zugferd_pdf.py b/tests/live/test_live_validate_zugferd_pdf.py index f4dee12..d498e90 100644 --- a/tests/live/test_live_validate_zugferd_pdf.py +++ b/tests/live/test_live_validate_zugferd_pdf.py @@ -26,6 +26,19 @@ def generated_zugferd_pdf( return response.output_file +@pytest.fixture(scope="module") +def uploaded_non_zugferd_pdf( + pdfrest_api_key: str, + pdfrest_live_base_url: str, +) -> PdfRestFile: + pdf_path = 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(pdf_path)[0] + + def _assert_valid_response( response: PdfZugferdValidationResponse, source: PdfRestFile, @@ -40,6 +53,22 @@ def _assert_valid_response( assert all(finding.message for finding in response.findings) +def _assert_invalid_response( + response: PdfZugferdValidationResponse, + source: PdfRestFile, +) -> None: + assert response.status == "INVALID" + assert response.pdfa is False + assert response.input_id == source.id + assert f"/resource/{source.id}" in response.input_url + assert response.findings + assert all(finding.code for finding in response.findings) + assert all(finding.message for finding in response.findings) + assert any( + finding.severity in {"BLOCKING", "ERROR"} for finding in response.findings + ) + + def test_live_validate_zugferd_pdf( pdfrest_api_key: str, pdfrest_live_base_url: str, @@ -69,6 +98,35 @@ async def test_live_async_validate_zugferd_pdf( _assert_valid_response(response, generated_zugferd_pdf) +def test_live_validate_zugferd_pdf_reports_invalid_document( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_non_zugferd_pdf: PdfRestFile, +) -> None: + with PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = client.validate_zugferd_pdf(uploaded_non_zugferd_pdf) + + _assert_invalid_response(response, uploaded_non_zugferd_pdf) + + +@pytest.mark.asyncio +async def test_live_async_validate_zugferd_pdf_reports_invalid_document( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_non_zugferd_pdf: PdfRestFile, +) -> None: + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.validate_zugferd_pdf(uploaded_non_zugferd_pdf) + + _assert_invalid_response(response, uploaded_non_zugferd_pdf) + + def test_live_validate_zugferd_pdf_rejects_invalid_resource_id( pdfrest_api_key: str, pdfrest_live_base_url: str, diff --git a/tests/test_validate_zugferd_pdf.py b/tests/test_validate_zugferd_pdf.py index 8a69a06..41ce35d 100644 --- a/tests/test_validate_zugferd_pdf.py +++ b/tests/test_validate_zugferd_pdf.py @@ -41,12 +41,13 @@ def _response_payload( file_id: PdfRestFileID, *, status: str = "VALID", + pdfa: bool = True, content_match_status: str = "CONFIRMED", severity: str = "WARNING", ) -> dict[str, Any]: return { "status": status, - "pdfa": True, + "pdfa": pdfa, "contentMatchStatus": content_match_status, "inputId": str(file_id), "inputUrl": f"https://api.pdfrest.com/resource/{file_id}?format=file", @@ -63,6 +64,50 @@ def _response_payload( } +def _demo_response_payload(file_id: PdfRestFileID) -> dict[str, Any]: + return { + "status": "INVALID", + "pdfa": False, + "contentMatchStatus": "DEMO_REDACTED", + "inputId": str(file_id), + "inputUrl": f"https://api.pdfrest.com/resource/{file_id}?format=file", + "findings": [ + { + "code": "PD******************", + "severity": "BLOCKING", + "message": "PD* ***** ***** **** *** ***** ****", + "field": "BT****", + "status": "MISMATCH", + "comparisonReason": "DEMO_REDACTED", + } + ], + "message": "Validation details are restricted.", + } + + +def _assert_validation_response( + response: PdfZugferdValidationResponse, + source: PdfRestFile, + *, + status: str, + pdfa: bool, + content_match_status: str, + finding: PdfZugferdFinding, + message: str | None, +) -> None: + assert isinstance(response, PdfZugferdValidationResponse) + assert response.status == status + assert response.pdfa is pdfa + assert response.content_match_status == content_match_status + assert response.input_id == source.id + assert ( + response.input_url + == f"https://api.pdfrest.com/resource/{source.id}?format=file" + ) + assert response.findings == [finding] + assert response.message == message + + def _assert_timeout(value: float | dict[str, float] | None, expected: float) -> None: assert value is not None if isinstance(value, dict): @@ -85,6 +130,7 @@ def test_validate_zugferd_payload_serializes_uploaded_pdf_id() -> None: @pytest.mark.parametrize( ("file", "error_match"), [ + pytest.param([], "at least 1 item", id="empty"), pytest.param( _make_file("invoice.xml", "application/xml"), "Must be a PDF file", @@ -179,26 +225,10 @@ def test_zugferd_validation_response_accepts_demo_redactions() -> None: file_id = PdfRestFileID.generate(prefix=2) response = PdfZugferdValidationResponse.model_validate( - { - "status": "INVALID", - "pdfa": True, - "contentMatchStatus": "DEMO_REDACTED", - "inputId": str(file_id), - "inputUrl": f"https://api.pdfrest.com/resource/{file_id}?format=file", - "findings": [ - { - "code": "PD******************", - "severity": "BLOCKING", - "message": "PD* ***** ***** **** *** ***** ****", - "field": "BT****", - "status": "MISMATCH", - "comparisonReason": "DEMO_REDACTED", - } - ], - "message": "Validation details are restricted.", - } + _demo_response_payload(file_id) ) + assert response.pdfa is False assert response.findings is not None assert response.findings[0].code == "PD******************" assert response.findings[0].comparison_reason == "DEMO_REDACTED" @@ -228,13 +258,22 @@ def handler(request: httpx.Request) -> httpx.Response: ) as client: response = client.validate_zugferd_pdf(pdf) - assert isinstance(response, PdfZugferdValidationResponse) - assert response.status == "VALID" - assert response.pdfa is True - assert response.content_match_status == "CONFIRMED" - assert response.input_id == pdf.id - assert response.findings is not None - assert response.findings[0].comparison_reason == "PARTIAL_REVERSE_VALIDATION" + _assert_validation_response( + response, + pdf, + status="VALID", + pdfa=True, + content_match_status="CONFIRMED", + finding=PdfZugferdFinding( + code="PDF_XML_FIELD_RESULT", + severity="WARNING", + message="Complete reverse validation is not available.", + field="visualConsistency", + status="UNVERIFIED", + comparison_reason="PARTIAL_REVERSE_VALIDATION", + ), + message=None, + ) @pytest.mark.asyncio @@ -246,21 +285,96 @@ async def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/validated-zugferd" assert request.headers["Api-Key"] == ASYNC_API_KEY assert json.loads(request.content) == {"id": str(pdf.id)} - payload = _response_payload( - pdf.id, status="INVALID", content_match_status="DEMO_REDACTED" - ) - payload["message"] = "Validation details are restricted." - return httpx.Response(200, json=payload) + return httpx.Response(200, json=_response_payload(pdf.id)) async with AsyncPdfRestClient( api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) ) as client: response = await client.validate_zugferd_pdf(pdf) - assert response.status == "INVALID" - assert response.content_match_status == "DEMO_REDACTED" - assert response.input_id == pdf.id - assert response.message == "Validation details are restricted." + _assert_validation_response( + response, + pdf, + status="VALID", + pdfa=True, + content_match_status="CONFIRMED", + finding=PdfZugferdFinding( + code="PDF_XML_FIELD_RESULT", + severity="WARNING", + message="Complete reverse validation is not available.", + field="visualConsistency", + status="UNVERIFIED", + comparison_reason="PARTIAL_REVERSE_VALIDATION", + ), + message=None, + ) + + +def test_validate_zugferd_pdf_invalid_demo_response_mapping() -> None: + pdf = _make_file("invalid-invoice.pdf", "application/pdf") + + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" + assert request.url.path == "/validated-zugferd" + assert request.headers["Api-Key"] == VALID_API_KEY + assert json.loads(request.content) == {"id": str(pdf.id)} + return httpx.Response(200, json=_demo_response_payload(pdf.id)) + + with PdfRestClient( + api_key=VALID_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = client.validate_zugferd_pdf(pdf) + + _assert_validation_response( + response, + pdf, + status="INVALID", + pdfa=False, + content_match_status="DEMO_REDACTED", + finding=PdfZugferdFinding( + code="PD******************", + severity="BLOCKING", + message="PD* ***** ***** **** *** ***** ****", + field="BT****", + status="MISMATCH", + comparison_reason="DEMO_REDACTED", + ), + message="Validation details are restricted.", + ) + + +@pytest.mark.asyncio +async def test_async_validate_zugferd_pdf_invalid_demo_response_mapping() -> None: + pdf = _make_file("invalid-invoice.pdf", "application/pdf") + + async def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" + assert request.url.path == "/validated-zugferd" + assert request.headers["Api-Key"] == ASYNC_API_KEY + assert json.loads(request.content) == {"id": str(pdf.id)} + return httpx.Response(200, json=_demo_response_payload(pdf.id)) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, transport=httpx.MockTransport(handler) + ) as client: + response = await client.validate_zugferd_pdf(pdf) + + _assert_validation_response( + response, + pdf, + status="INVALID", + pdfa=False, + content_match_status="DEMO_REDACTED", + finding=PdfZugferdFinding( + code="PD******************", + severity="BLOCKING", + message="PD* ***** ***** **** *** ***** ****", + field="BT****", + status="MISMATCH", + comparison_reason="DEMO_REDACTED", + ), + message="Validation details are restricted.", + ) def test_validate_zugferd_pdf_request_customization() -> None: @@ -321,6 +435,7 @@ async def handler(request: httpx.Request) -> httpx.Response: @pytest.mark.parametrize( ("file", "error_match"), [ + pytest.param([], "at least 1 item", id="empty"), pytest.param( _make_file("invoice.xml", "application/xml"), "Must be a PDF file", @@ -354,6 +469,7 @@ def handler(_: httpx.Request) -> httpx.Response: @pytest.mark.parametrize( ("file", "error_match"), [ + pytest.param([], "at least 1 item", id="empty"), pytest.param( _make_file("invoice.xml", "application/xml"), "Must be a PDF file", From 3ed8a99f0732ed5aa1d6b30823525b923b67f104 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 3 Nov 2026 14:38:11 -0600 Subject: [PATCH 12/15] client: Add TDM reservation policy support - Add synchronous and asynchronous helpers for uploaded PDF resources. - Validate HTTP and HTTPS policy URLs without changing valid values. - Cover serialization, customization, validation, and response mapping. Assisted-by: Codex --- src/pdfrest/client.py | 94 ++++++ src/pdfrest/models/_internal.py | 47 +++ tests/test_apply_tdm_reservation.py | 468 ++++++++++++++++++++++++++++ 3 files changed, 609 insertions(+) create mode 100644 tests/test_apply_tdm_reservation.py diff --git a/src/pdfrest/client.py b/src/pdfrest/client.py index 0e5e105..c8a411f 100644 --- a/src/pdfrest/client.py +++ b/src/pdfrest/client.py @@ -122,6 +122,7 @@ PdfRestrictPayload, PdfSignPayload, PdfSplitPayload, + PdfTdmReservationPayload, PdfTextWatermarkPayload, PdfToExcelPayload, PdfToPdfaPayload, @@ -4619,6 +4620,53 @@ def sign_pdf( timeout=timeout, ) + def apply_tdm_reservation( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + policy: str, + output: str | None = None, + extra_query: Query | None = None, + extra_headers: AnyMapping | None = None, + extra_body: Body | None = None, + timeout: TimeoutTypes | None = None, + ) -> PdfRestFileBasedResponse: + """Apply a text-and-data-mining reservation policy to a PDF. + + Embeds a URL that points to hosted terms of use or a machine-readable + TDM policy document. The URL is stored as metadata without fetching or + inspecting its destination. + + Args: + file: Uploaded PDF resource to update. + policy: HTTP or HTTPS URL for the hosted TDM policy or terms of use. + output: Generated output filename without its extension. + extra_query: Additional query parameters merged into the request. + extra_headers: Additional HTTP headers merged into the request. + extra_body: Additional request body fields merged into the payload. + timeout: Request timeout override for this call. + + Returns: + Validated file-based response containing the TDM-reserved PDF. + + Raises: + PdfRestError: If request execution fails at the client or API layer. + ValidationError: If local payload validation fails before sending. + """ + payload: dict[str, Any] = {"files": file, "policy": policy} + if output is not None: + payload["output"] = output + + return self._post_file_operation( + endpoint="/tdm-reserved-pdf", + payload=payload, + payload_model=PdfTdmReservationPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + def validate_zugferd_pdf( self, file: PdfRestFile | Sequence[PdfRestFile], @@ -8263,6 +8311,52 @@ async def sign_pdf( timeout=timeout, ) + async def apply_tdm_reservation( + self, + file: PdfRestFile | Sequence[PdfRestFile], + *, + policy: str, + 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 apply a TDM reservation policy to a PDF. + + This is the asynchronous variant of + [PdfRestClient.apply_tdm_reservation][pdfrest.PdfRestClient.apply_tdm_reservation]. + + Args: + file: Uploaded PDF resource to update. + policy: HTTP or HTTPS URL for the hosted TDM policy or terms of use. + output: Generated output filename without its extension. + extra_query: Additional query parameters merged into the request. + extra_headers: Additional HTTP headers merged into the request. + extra_body: Additional request body fields merged into the payload. + timeout: Request timeout override for this call. + + Returns: + Validated file-based response containing the TDM-reserved PDF. + + Raises: + PdfRestError: If request execution fails at the client or API layer. + ValidationError: If local payload validation fails before sending. + """ + payload: dict[str, Any] = {"files": file, "policy": policy} + if output is not None: + payload["output"] = output + + return await self._post_file_operation( + endpoint="/tdm-reserved-pdf", + payload=payload, + payload_model=PdfTdmReservationPayload, + extra_query=extra_query, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=timeout, + ) + async def validate_zugferd_pdf( self, file: PdfRestFile | Sequence[PdfRestFile], diff --git a/src/pdfrest/models/_internal.py b/src/pdfrest/models/_internal.py index ea2dc18..70b15e7 100644 --- a/src/pdfrest/models/_internal.py +++ b/src/pdfrest/models/_internal.py @@ -16,6 +16,7 @@ HttpUrl, PlainSerializer, StringConstraints, + TypeAdapter, ValidatorFunctionWrapHandler, WrapValidator, model_serializer, @@ -70,6 +71,7 @@ PDFA_OUTPUT_TYPE_MAP: dict[str, PdfAType] = { output_type.casefold(): output_type for output_type in PDFA_OUTPUT_TYPES } +_HTTP_URL_ADAPTER = TypeAdapter(HttpUrl) def _ensure_list(value: Any) -> Any: @@ -130,6 +132,22 @@ def _validate_output_prefix(value: str | None) -> str | None: return value +def _validate_http_url(value: str) -> str: + """Validate an HTTP(S) URL without normalizing the caller's value.""" + message = "Must be a valid HTTP or HTTPS URL." + if ( + any(character.isspace() for character in value) + or "\\" in value + or re.search(r"%(?![0-9A-Fa-f]{2})", value) + ): + raise ValueError(message) + try: + _ = _HTTP_URL_ADAPTER.validate_python(value) + except ValueError: + raise ValueError(message) from None + return value + + def _split_comma_list(value: Any) -> Any: if isinstance(value, str): return value.split(",") @@ -3641,6 +3659,35 @@ class PdfDecryptPayload(BaseModel): ] = None +class PdfTdmReservationPayload(BaseModel): + """Adapt caller options into a pdfRest-ready TDM reservation 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), + ] + policy: Annotated[ + str, + Field(min_length=1, serialization_alias="policy"), + AfterValidator(_validate_http_url), + ] + output: Annotated[ + str | None, + Field(serialization_alias="output", min_length=1, default=None), + AfterValidator(_validate_output_prefix), + ] = None + + class BmpPdfRestPayload(BasePdfRestGraphicPayload[Literal["rgb", "gray"]]): """Adapt caller options into a pdfRest-ready BMP request payload.""" diff --git a/tests/test_apply_tdm_reservation.py b/tests/test_apply_tdm_reservation.py new file mode 100644 index 0000000..cbf03b6 --- /dev/null +++ b/tests/test_apply_tdm_reservation.py @@ -0,0 +1,468 @@ +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 PdfTdmReservationPayload + +from .graphics_test_helpers import ( + ASYNC_API_KEY, + VALID_API_KEY, + build_file_info_payload, + make_pdf_file, +) + +POLICY_URL = "https://rights.example/tdm-policy?version=2#terms" + + +@pytest.mark.parametrize( + "policy", + [ + pytest.param("https://example.com", id="origin"), + pytest.param( + "https://example.com/terms%20and%20conditions", + id="percent-encoded-path", + ), + ], +) +def test_tdm_reservation_payload_preserves_policy_url(policy: str) -> None: + input_file = make_pdf_file(PdfRestFileID.generate(1)) + + payload = PdfTdmReservationPayload.model_validate( + { + "file": input_file, + "policy": policy, + "output": "tdm-reserved", + } + ) + + assert payload.model_dump( + mode="json", + by_alias=True, + exclude_none=True, + exclude_unset=True, + ) == { + "id": str(input_file.id), + "policy": policy, + "output": "tdm-reserved", + } + + +def test_tdm_reservation_payload_requires_policy() -> None: + input_file = make_pdf_file(PdfRestFileID.generate(1)) + + with pytest.raises(ValidationError, match=r"policy\s+Field required"): + PdfTdmReservationPayload.model_validate({"file": input_file}) + + +def test_apply_tdm_reservation_success(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 = PdfTdmReservationPayload.model_validate( + {"file": input_file, "policy": POLICY_URL} + ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True) + seen = {"post": 0, "get": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/tdm-reserved-pdf": + seen["post"] += 1 + assert json.loads(request.content) == 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, + "report_pdfrest_tdm_reserved_pdf.pdf", + "application/pdf", + ), + ) + message = f"Unexpected request {request.method} {request.url}" + raise AssertionError(message) + + with PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = client.apply_tdm_reservation(input_file, policy=POLICY_URL) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert str(response.input_id) == str(input_file.id) + assert response.output_file.name == "report_pdfrest_tdm_reserved_pdf.pdf" + assert response.output_file.type == "application/pdf" + assert response.output_file.size > 0 + assert response.warning is None + + +def test_apply_tdm_reservation_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: + assert request.url.params["trace"] == "sync" + assert request.headers["X-Debug"] == "sync" + if request.method == "POST" and request.url.path == "/tdm-reserved-pdf": + captured_timeout["value"] = request.extensions.get("timeout") + assert json.loads(request.content) == { + "id": str(input_file.id), + "policy": POLICY_URL, + "output": "reserved-sync", + "debug": "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" + return httpx.Response( + 200, + json=build_file_info_payload( + output_id, + "reserved-sync.pdf", + "application/pdf", + ), + ) + message = f"Unexpected request {request.method} {request.url}" + raise AssertionError(message) + + with PdfRestClient( + api_key=VALID_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = client.apply_tdm_reservation( + input_file, + policy=POLICY_URL, + output="reserved-sync", + extra_query={"trace": "sync"}, + extra_headers={"X-Debug": "sync"}, + extra_body={"debug": "yes"}, + timeout=0.61, + ) + + assert response.output_file.name == "reserved-sync.pdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.61) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.61) + + +@pytest.mark.asyncio +async def test_async_apply_tdm_reservation_success( + 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 = PdfTdmReservationPayload.model_validate( + {"file": input_file, "policy": POLICY_URL} + ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True) + seen = {"post": 0, "get": 0} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST" and request.url.path == "/tdm-reserved-pdf": + seen["post"] += 1 + assert json.loads(request.content) == 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, + "async-report_pdfrest_tdm_reserved_pdf.pdf", + "application/pdf", + ), + ) + message = f"Unexpected request {request.method} {request.url}" + raise AssertionError(message) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = await client.apply_tdm_reservation(input_file, policy=POLICY_URL) + + assert seen == {"post": 1, "get": 1} + assert isinstance(response, PdfRestFileBasedResponse) + assert str(response.input_id) == str(input_file.id) + assert response.output_file.name == "async-report_pdfrest_tdm_reserved_pdf.pdf" + assert response.output_file.type == "application/pdf" + assert response.output_file.size > 0 + assert response.warning is None + + +@pytest.mark.asyncio +async def test_async_apply_tdm_reservation_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: + assert request.url.params["trace"] == "async" + assert request.headers["X-Debug"] == "async" + if request.method == "POST" and request.url.path == "/tdm-reserved-pdf": + captured_timeout["value"] = request.extensions.get("timeout") + assert json.loads(request.content) == { + "id": str(input_file.id), + "policy": POLICY_URL, + "output": "reserved-async", + "flags": ["a", "b"], + } + 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, + "reserved-async.pdf", + "application/pdf", + ), + ) + message = f"Unexpected request {request.method} {request.url}" + raise AssertionError(message) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=httpx.MockTransport(handler), + ) as client: + response = await client.apply_tdm_reservation( + input_file, + policy=POLICY_URL, + output="reserved-async", + extra_query={"trace": "async"}, + extra_headers={"X-Debug": "async"}, + extra_body={"flags": ["a", "b"]}, + timeout=0.83, + ) + + assert response.output_file.name == "reserved-async.pdf" + timeout_value = captured_timeout["value"] + assert timeout_value is not None + if isinstance(timeout_value, dict): + assert all( + component == pytest.approx(0.83) for component in timeout_value.values() + ) + else: + assert timeout_value == pytest.approx(0.83) + + +@pytest.mark.parametrize( + ("policy", "match"), + [ + pytest.param("", "String should have at least 1 character", id="empty"), + pytest.param("my-policy", "Must be a valid HTTP or HTTPS URL", id="bare-name"), + pytest.param( + "No AI training allowed", + "Must be a valid HTTP or HTTPS URL", + id="policy-text", + ), + pytest.param( + "example.com/tdm-policy", + "Must be a valid HTTP or HTTPS URL", + id="missing-scheme", + ), + pytest.param( + "ftp://example.com/tdm-policy", + "Must be a valid HTTP or HTTPS URL", + id="unsupported-scheme", + ), + pytest.param( + " https://example.com/tdm-policy ", + "Must be a valid HTTP or HTTPS URL", + id="surrounding-whitespace", + ), + pytest.param( + "https://example.com/%ZZ", + "Must be a valid HTTP or HTTPS URL", + id="invalid-percent-escape", + ), + pytest.param( + "https://example.com\\tdm-policy", + "Must be a valid HTTP or HTTPS URL", + id="backslash", + ), + ], +) +def test_apply_tdm_reservation_rejects_invalid_policy( + monkeypatch: pytest.MonkeyPatch, + policy: str, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(1)) + transport = httpx.MockTransport(lambda request: pytest.fail("request was sent")) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises(ValidationError, match=match), + ): + client.apply_tdm_reservation(input_file, policy=policy) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("policy", "match"), + [ + pytest.param("", "String should have at least 1 character", id="empty"), + pytest.param("my-policy", "Must be a valid HTTP or HTTPS URL", id="bare-name"), + pytest.param( + "No AI training allowed", + "Must be a valid HTTP or HTTPS URL", + id="policy-text", + ), + pytest.param( + "example.com/tdm-policy", + "Must be a valid HTTP or HTTPS URL", + id="missing-scheme", + ), + pytest.param( + "ftp://example.com/tdm-policy", + "Must be a valid HTTP or HTTPS URL", + id="unsupported-scheme", + ), + pytest.param( + "https://example.com/tdm policy", + "Must be a valid HTTP or HTTPS URL", + id="embedded-whitespace", + ), + pytest.param( + "https://example.com/%ZZ", + "Must be a valid HTTP or HTTPS URL", + id="invalid-percent-escape", + ), + pytest.param( + "https://example.com\\tdm-policy", + "Must be a valid HTTP or HTTPS URL", + id="backslash", + ), + ], +) +async def test_async_apply_tdm_reservation_rejects_invalid_policy( + monkeypatch: pytest.MonkeyPatch, + policy: str, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + input_file = make_pdf_file(PdfRestFileID.generate(2)) + transport = httpx.MockTransport(lambda request: pytest.fail("request was sent")) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=transport, + ) as client: + with pytest.raises(ValidationError, match=match): + await client.apply_tdm_reservation(input_file, policy=policy) + + +@pytest.mark.parametrize( + ("files", "match"), + [ + pytest.param("empty", "at least 1 item", id="empty"), + pytest.param("png", "Must be a PDF file", id="non-pdf-file"), + pytest.param( + "multiple", + "List should have at most 1 item after validation", + id="multiple-files", + ), + ], +) +def test_apply_tdm_reservation_rejects_incompatible_resources( + monkeypatch: pytest.MonkeyPatch, + files: str, + match: str, +) -> 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", + ) + ) + files_argument = { + "empty": [], + "png": png_file, + "multiple": [pdf_file, make_pdf_file(PdfRestFileID.generate())], + }[files] + transport = httpx.MockTransport(lambda request: pytest.fail("request was sent")) + + with ( + PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client, + pytest.raises(ValidationError, match=match), + ): + client.apply_tdm_reservation(files_argument, policy=POLICY_URL) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("files", "match"), + [ + pytest.param("empty", "at least 1 item", id="empty"), + pytest.param("png", "Must be a PDF file", id="non-pdf-file"), + pytest.param( + "multiple", + "List should have at most 1 item after validation", + id="multiple-files", + ), + ], +) +async def test_async_apply_tdm_reservation_rejects_incompatible_resources( + monkeypatch: pytest.MonkeyPatch, + files: str, + match: str, +) -> None: + monkeypatch.delenv("PDFREST_API_KEY", raising=False) + pdf_file = make_pdf_file(PdfRestFileID.generate(2)) + png_file = PdfRestFile.model_validate( + build_file_info_payload( + PdfRestFileID.generate(), + "example.png", + "image/png", + ) + ) + files_argument = { + "empty": [], + "png": png_file, + "multiple": [pdf_file, make_pdf_file(PdfRestFileID.generate())], + }[files] + transport = httpx.MockTransport(lambda request: pytest.fail("request was sent")) + + async with AsyncPdfRestClient( + api_key=ASYNC_API_KEY, + transport=transport, + ) as client: + with pytest.raises(ValidationError, match=match): + await client.apply_tdm_reservation(files_argument, policy=POLICY_URL) From 1aec3819e1e9d8abe0b09b1da2b4317cbf38e535 Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 3 Nov 2026 14:38:47 -0600 Subject: [PATCH 13/15] tests: Verify TDM reservation live - Exercise default and custom output names through both client transports. - Inspect generated PDF metadata and server-side policy rejection. Assisted-by: Codex --- tests/live/test_live_apply_tdm_reservation.py | 150 ++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 tests/live/test_live_apply_tdm_reservation.py diff --git a/tests/live/test_live_apply_tdm_reservation.py b/tests/live/test_live_apply_tdm_reservation.py new file mode 100644 index 0000000..106a229 --- /dev/null +++ b/tests/live/test_live_apply_tdm_reservation.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import pytest + +from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient +from pdfrest.models import PdfRestFile + +from ..resources import get_test_resource_path + +POLICY_URL = "https://example.com/terms/tdm-policy" + + +@pytest.fixture(scope="module") +def uploaded_pdf_for_tdm_reservation( + 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.mark.parametrize( + "output_name", + [ + pytest.param(None, id="default-output"), + pytest.param("tdm-reserved-live", id="custom-output"), + ], +) +def test_live_apply_tdm_reservation( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_tdm_reservation: 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.apply_tdm_reservation( + uploaded_pdf_for_tdm_reservation, + policy=POLICY_URL, + **kwargs, + ) + output_bytes = client.files.read_bytes(response.output_file) + + assert response.output_files + output_file = response.output_file + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert output_file.url is not None + assert response.warning is None + assert str(response.input_id) == str(uploaded_pdf_for_tdm_reservation.id) + assert POLICY_URL.encode() in output_bytes + assert b"tdm:policy" in output_bytes + if output_name is not None: + assert output_file.name.startswith(output_name) + else: + assert output_file.name.endswith(".pdf") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "output_name", + [ + pytest.param(None, id="default-output"), + pytest.param("tdm-reserved-live-async", id="custom-output"), + ], +) +async def test_live_async_apply_tdm_reservation( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_tdm_reservation: PdfRestFile, + output_name: str | None, +) -> None: + kwargs: dict[str, str] = {} + if output_name is not None: + kwargs["output"] = output_name + + async with AsyncPdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client: + response = await client.apply_tdm_reservation( + uploaded_pdf_for_tdm_reservation, + policy=POLICY_URL, + **kwargs, + ) + output_bytes = await client.files.read_bytes(response.output_file) + + assert response.output_files + output_file = response.output_file + assert output_file.type == "application/pdf" + assert output_file.size > 0 + assert output_file.url is not None + assert response.warning is None + assert str(response.input_id) == str(uploaded_pdf_for_tdm_reservation.id) + assert POLICY_URL.encode() in output_bytes + assert b"tdm:policy" in output_bytes + if output_name is not None: + assert output_file.name.startswith(output_name) + else: + assert output_file.name.endswith(".pdf") + + +def test_live_apply_tdm_reservation_rejects_missing_policy( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_tdm_reservation: PdfRestFile, +) -> None: + with ( + PdfRestClient( + api_key=pdfrest_api_key, + base_url=pdfrest_live_base_url, + ) as client, + pytest.raises(PdfRestApiError, match=r"(?i)TDM policy setting is required"), + ): + client.apply_tdm_reservation( + uploaded_pdf_for_tdm_reservation, + policy=POLICY_URL, + extra_body={"policy": ""}, + ) + + +@pytest.mark.asyncio +async def test_live_async_apply_tdm_reservation_rejects_missing_policy( + pdfrest_api_key: str, + pdfrest_live_base_url: str, + uploaded_pdf_for_tdm_reservation: 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)TDM policy setting is required", + ): + await client.apply_tdm_reservation( + uploaded_pdf_for_tdm_reservation, + policy=POLICY_URL, + extra_body={"policy": ""}, + ) From 4cea3398b99051e454b830837ee2cee9bcfdc91e Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 3 Nov 2026 14:39:16 -0600 Subject: [PATCH 14/15] docs: Document TDM reservation workflow - Link the helper and explain URL-based reservation metadata behavior. - Add a runnable upload example and list it in the example inventory. Assisted-by: Codex --- docs/api-guide.md | 3 ++ docs/llms-full.txt | 1 + examples/README.md | 2 + .../apply_tdm_reservation_example.py | 54 +++++++++++++++++++ 4 files changed, 60 insertions(+) create mode 100644 examples/apply_tdm_reservation/apply_tdm_reservation_example.py diff --git a/docs/api-guide.md b/docs/api-guide.md index 30a5b3b..e1df8e7 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -104,6 +104,9 @@ Use this group for password protection, permissions, signatures, and standards. [remove_open_password][pdfrest.PdfRestClient.remove_open_password] - Digital signatures: [sign_pdf][pdfrest.PdfRestClient.sign_pdf] +- Text-and-data-mining reservation: + [apply_tdm_reservation][pdfrest.PdfRestClient.apply_tdm_reservation] embeds + the URL of hosted terms or a machine-readable policy without fetching it - Archival/print conformance: [convert_to_pdfa][pdfrest.PdfRestClient.convert_to_pdfa], [convert_to_pdfx][pdfrest.PdfRestClient.convert_to_pdfx] diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 8053417..aa4e0a1 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -183,6 +183,7 @@ to the API guide and API reference for exact method names and options. - Redaction preview/apply - Passwords and permissions - Encryption/decryption/restrictions +- URL-based text-and-data-mining reservation metadata - Watermarking (text/image) - Digital signing diff --git a/examples/README.md b/examples/README.md index 3a76e55..d9be39b 100644 --- a/examples/README.md +++ b/examples/README.md @@ -21,6 +21,8 @@ supported interpreter matrix. ## Available Examples +- `examples/apply_tdm_reservation/apply_tdm_reservation_example.py` – embed a + hosted TDM policy URL in an uploaded PDF. - `examples/add_shapes/add_shapes_to_pdf_example.py` – add a styled rectangle and divider line to a PDF with accessibility tagging enabled. - `examples/convert_structured_documents/convert_structured_documents_to_pdf_example.py` diff --git a/examples/apply_tdm_reservation/apply_tdm_reservation_example.py b/examples/apply_tdm_reservation/apply_tdm_reservation_example.py new file mode 100644 index 0000000..1a7de82 --- /dev/null +++ b/examples/apply_tdm_reservation/apply_tdm_reservation_example.py @@ -0,0 +1,54 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = ["pdfrest", "python-dotenv"] +# /// +"""Apply a text-and-data-mining reservation policy to a PDF. + +This sample demonstrates how to: + +1. Load ``PDFREST_API_KEY`` from the environment or a local ``.env`` file. +2. Upload the checked-in ``examples/resources/report.pdf`` input. +3. Embed a URL for hosted TDM terms or a machine-readable policy document. +4. Print the generated PDF metadata. + +Replace ``POLICY_URL`` with the URL of your hosted policy in a real integration. +Set ``PDFREST_BASE_URL`` only for a self-hosted deployment. pdfRest stores the +URL without fetching or inspecting its destination. Run from the repository root with +``uv run examples/apply_tdm_reservation/apply_tdm_reservation_example.py``. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from dotenv import load_dotenv + +from pdfrest import PdfRestClient + +RESOURCE_DIRECTORY = Path(__file__).resolve().parents[1] / "resources" +INPUT_PATH = RESOURCE_DIRECTORY / "report.pdf" +POLICY_URL = "https://example.com/terms/tdm-policy" + + +def apply_tdm_reservation() -> None: + """Upload a PDF and add a URL-based TDM reservation policy.""" + load_dotenv() + base_url = os.getenv("PDFREST_BASE_URL") + + with PdfRestClient(base_url=base_url) as client: + uploaded_pdf = client.files.create_from_paths(INPUT_PATH)[0] + response = client.apply_tdm_reservation( + uploaded_pdf, + policy=POLICY_URL, + output="report-with-tdm-reservation", + ) + + print(f"Input ID: {response.input_id}") + print(f"Output file: {response.output_file.name}") + print(f"Output MIME type: {response.output_file.type}") + print(f"Output URL: {response.output_file.url}") + + +if __name__ == "__main__": # pragma: no cover - manual example + apply_tdm_reservation() From a92ce614f783d599027a52936a3c6ed2a03f321e Mon Sep 17 00:00:00 2001 From: Taylor Smith Date: Tue, 3 Nov 2026 14:39:40 -0600 Subject: [PATCH 15/15] pyproject: Bump version to 1.5.0 - Mark the additive TDM helper for a minor feature release. - Keep the project lock metadata synchronized. Assisted-by: Codex --- pyproject.toml | 2 +- uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c4e489a..d830161 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pdfrest" -version = "1.4.0" +version = "1.5.0" description = "Python client library for interacting with the pdfRest API" readme = { file = "README.md", content-type = "text/markdown" } authors = [ diff --git a/uv.lock b/uv.lock index a085a9f..ed26310 100644 --- a/uv.lock +++ b/uv.lock @@ -964,7 +964,7 @@ wheels = [ [[package]] name = "pdfrest" -version = "1.4.0" +version = "1.5.0" source = { editable = "." } dependencies = [ { name = "exceptiongroup" },