diff --git a/docs/api-guide.md b/docs/api-guide.md index 7a27f8c..8ad4823 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -113,7 +113,10 @@ Use this group for password protection, permissions, signatures, and standards. [convert_pdf_to_zugferd_pdf][pdfrest.PdfRestClient.convert_pdf_to_zugferd_pdf] embeds CII XML into an existing matching PDF. The latter preserves the PDF by default; set `allow_regeneration=True` only when the service may replace a - mismatched PDF with a newly rendered invoice. + mismatched PDF with a newly rendered invoice. Use + [validate_zugferd_pdf][pdfrest.PdfRestClient.validate_zugferd_pdf] to check an + existing ZUGFeRD or Factur-X package, visible-content match, and independent + PDF/A-3 conformance without modifying the file. ## PDF cleanup and rendering normalization diff --git a/docs/llms-full.txt b/docs/llms-full.txt index d46ac4e..86484f1 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -163,6 +163,8 @@ to the API guide and API reference for exact method names and options. - Embed CII XML into a matching PDF with ``convert_pdf_to_zugferd_pdf``; preserve the supplied PDF by default or explicitly permit regeneration when the PDF and invoice do not match +- Validate existing ZUGFeRD or Factur-X PDFs for package, XML, visible-content, + and independent PDF/A-3 conformance with ``validate_zugferd_pdf`` ### Extraction and Analysis diff --git a/examples/README.md b/examples/README.md index 12a6744..97ba697 100644 --- a/examples/README.md +++ b/examples/README.md @@ -40,6 +40,8 @@ supported interpreter matrix. - `examples/convert_pdf_to_zugferd_pdf/convert_pdf_to_zugferd_pdf_example.py` – combine an uploaded PDF with CII XML and explicitly permit replacement when the visual PDF does not match the invoice. +- `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..95d83af --- /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.convert_invoice_xml_to_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() diff --git a/pyproject.toml b/pyproject.toml index 07937e2..f70e8aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pdfrest" -version = "1.5.0" +version = "1.6.0" description = "Python client library for interacting with the pdfRest API" readme = { file = "README.md", content-type = "text/markdown" } license = "Apache-2.0" diff --git a/src/pdfrest/client.py b/src/pdfrest/client.py index f75e2c8..9e8ef03 100644 --- a/src/pdfrest/client.py +++ b/src/pdfrest/client.py @@ -68,6 +68,7 @@ PdfRestFileBasedResponse, PdfRestFileID, PdfRestInfoResponse, + PdfZugferdValidationResponse, SummarizePdfTextResponse, TranslatePdfTextFileResponse, TranslatePdfTextResponse, @@ -130,6 +131,7 @@ PdfToWordPayload, PdfToZugferdPayload, PdfUnrestrictPayload, + PdfValidateZugferdPayload, PdfXfaToAcroformsPayload, PngPdfRestPayload, SummarizePdfTextPayload, @@ -4618,6 +4620,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 convert_invoice_xml_to_zugferd_pdf( self, invoice: PdfRestFile | Sequence[PdfRestFile], @@ -8268,6 +8315,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 convert_invoice_xml_to_zugferd_pdf( self, invoice: PdfRestFile | Sequence[PdfRestFile], 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 11c0703..c140a27 100644 --- a/src/pdfrest/models/_internal.py +++ b/src/pdfrest/models/_internal.py @@ -1848,6 +1848,22 @@ def _validate_regeneration_options(self) -> PdfToZugferdPayload: 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.""" 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..b8a8631 --- /dev/null +++ b/tests/live/test_live_validate_zugferd_pdf.py @@ -0,0 +1,162 @@ +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.convert_invoice_xml_to_zugferd_pdf( + invoice, + output="validation-source-zugferd", + ) + 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, +) -> 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 _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, + 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_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, + 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"}, + ) diff --git a/tests/test_validate_zugferd_pdf.py b/tests/test_validate_zugferd_pdf.py new file mode 100644 index 0000000..41ce35d --- /dev/null +++ b/tests/test_validate_zugferd_pdf.py @@ -0,0 +1,499 @@ +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", + pdfa: bool = True, + content_match_status: str = "CONFIRMED", + severity: str = "WARNING", +) -> dict[str, Any]: + return { + "status": status, + "pdfa": pdfa, + "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 _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): + 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([], "at least 1 item", id="empty"), + 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( + _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" + + +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_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 +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)} + 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_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: + 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([], "at least 1 item", id="empty"), + 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([], "at least 1 item", id="empty"), + 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) diff --git a/uv.lock b/uv.lock index e261891..926ab8d 100644 --- a/uv.lock +++ b/uv.lock @@ -964,7 +964,7 @@ wheels = [ [[package]] name = "pdfrest" -version = "1.5.0" +version = "1.6.0" source = { editable = "." } dependencies = [ { name = "exceptiongroup" },