watermark-pdf: Fix incomplete unit test coverage

Assisted-by: Codex

Author: datalogics-cgreen

tests/test_watermark_pdf.py

@@ -226,6 +226,81 @@ def handler(request: httpx.Request) -> httpx.Response:
         assert timeout_value == pytest.approx(0.31)
 
 
+def test_watermark_pdf_with_image_request_customization(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    input_file = make_pdf_file(PdfRestFileID.generate(1))
+    watermark_file = make_pdf_file(PdfRestFileID.generate(1), name="sync-stamp.pdf")
+    output_id = str(PdfRestFileID.generate())
+    captured_timeout: dict[str, float | dict[str, float] | None] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and request.url.path == "/watermarked-pdf":
+            assert request.url.params["trace"] == "img-sync"
+            assert request.headers["X-Debug"] == "img-sync"
+            captured_timeout["value"] = request.extensions.get("timeout")
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload["watermark_file_id"] == str(watermark_file.id)
+            assert payload["watermark_file_scale"] == pytest.approx(0.65)
+            assert payload["horizontal_alignment"] == "left"
+            assert payload["vertical_alignment"] == "top"
+            assert payload["x"] == -24
+            assert payload["y"] == 48
+            assert payload["rotation"] == 15
+            assert payload["debug"] == "img-sync"
+            assert payload["id"] == str(input_file.id)
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [input_file.id, watermark_file.id],
+                    "outputId": [output_id],
+                },
+            )
+        if request.method == "GET" and request.url.path == f"/resource/{output_id}":
+            assert request.url.params["format"] == "info"
+            assert request.url.params["trace"] == "img-sync"
+            assert request.headers["X-Debug"] == "img-sync"
+            return httpx.Response(
+                200,
+                json=build_file_info_payload(
+                    output_id,
+                    "img-sync.pdf",
+                    "application/pdf",
+                ),
+            )
+        msg = f"Unexpected request {request.method} {request.url}"
+        raise AssertionError(msg)
+
+    transport = httpx.MockTransport(handler)
+    with PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client:
+        response = client.watermark_pdf_with_image(
+            input_file,
+            watermark_file=watermark_file,
+            watermark_file_scale=0.65,
+            horizontal_alignment="left",
+            vertical_alignment="top",
+            x=-24,
+            y=48,
+            rotation=15,
+            extra_query={"trace": "img-sync"},
+            extra_headers={"X-Debug": "img-sync"},
+            extra_body={"debug": "img-sync"},
+            timeout=0.57,
+        )
+
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "img-sync.pdf"
+    timeout_value = captured_timeout["value"]
+    assert timeout_value is not None
+    if isinstance(timeout_value, dict):
+        assert all(
+            component == pytest.approx(0.57) for component in timeout_value.values()
+        )
+    else:
+        assert timeout_value == pytest.approx(0.57)
+
+
 def test_watermark_pdf_with_text_validation_rejects_invalid_text_color(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -303,6 +378,49 @@ def test_watermark_pdf_with_image_validation_rejects_non_pdf_watermark(
         client.watermark_pdf_with_image(input_file, watermark_file=bad_watermark)
 
 
+def test_watermark_pdf_with_text_validation_rejects_multiple_input_files(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    input_file_one = make_pdf_file(PdfRestFileID.generate(1))
+    input_file_two = make_pdf_file(PdfRestFileID.generate(1), name="second.pdf")
+    with (
+        PdfRestClient(
+            api_key=VALID_API_KEY,
+            transport=httpx.MockTransport(
+                lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called"))
+            ),
+        ) as client,
+        pytest.raises(ValidationError, match="at most 1 item"),
+    ):
+        client.watermark_pdf_with_text(
+            [input_file_one, input_file_two],
+            watermark_text="TooManyInputs",
+        )
+
+
+def test_watermark_pdf_with_image_validation_rejects_multiple_watermark_files(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    input_file = make_pdf_file(PdfRestFileID.generate(1))
+    watermark_file_one = make_pdf_file(PdfRestFileID.generate(1), name="first.pdf")
+    watermark_file_two = make_pdf_file(PdfRestFileID.generate(1), name="second.pdf")
+    with (
+        PdfRestClient(
+            api_key=VALID_API_KEY,
+            transport=httpx.MockTransport(
+                lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called"))
+            ),
+        ) as client,
+        pytest.raises(ValidationError, match="at most 1 item"),
+    ):
+        client.watermark_pdf_with_image(
+            input_file,
+            watermark_file=[watermark_file_one, watermark_file_two],
+        )
+
+
 def test_watermark_pdf_with_text_validation_rejects_short_text_size(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -526,3 +644,118 @@ def handler(request: httpx.Request) -> httpx.Response:
         )
     else:
         assert timeout_value == pytest.approx(0.42)
+
+
+@pytest.mark.asyncio
+async def test_async_watermark_pdf_with_image_request_customization(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    input_file = make_pdf_file(PdfRestFileID.generate(2))
+    watermark_file = make_pdf_file(PdfRestFileID.generate(2), name="async-stamp.pdf")
+    output_id = str(PdfRestFileID.generate())
+    captured_timeout: dict[str, float | dict[str, float] | None] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and request.url.path == "/watermarked-pdf":
+            assert request.url.params["trace"] == "img-async"
+            assert request.headers["X-Debug"] == "img-async"
+            captured_timeout["value"] = request.extensions.get("timeout")
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload["watermark_file_id"] == str(watermark_file.id)
+            assert payload["watermark_file_scale"] == pytest.approx(0.25)
+            assert payload["opacity"] == pytest.approx(0.85)
+            assert payload["horizontal_alignment"] == "right"
+            assert payload["vertical_alignment"] == "bottom"
+            assert payload["x"] == 12
+            assert payload["y"] == -18
+            assert payload["rotation"] == 330
+            assert payload["debug"] == "img-async"
+            assert payload["id"] == str(input_file.id)
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [input_file.id, watermark_file.id],
+                    "outputId": [output_id],
+                },
+            )
+        if request.method == "GET" and request.url.path == f"/resource/{output_id}":
+            assert request.url.params["format"] == "info"
+            assert request.url.params["trace"] == "img-async"
+            assert request.headers["X-Debug"] == "img-async"
+            return httpx.Response(
+                200,
+                json=build_file_info_payload(
+                    output_id,
+                    "img-async.pdf",
+                    "application/pdf",
+                ),
+            )
+        msg = f"Unexpected request {request.method} {request.url}"
+        raise AssertionError(msg)
+
+    transport = httpx.MockTransport(handler)
+    async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client:
+        response = await client.watermark_pdf_with_image(
+            input_file,
+            watermark_file=watermark_file,
+            watermark_file_scale=0.25,
+            opacity=0.85,
+            horizontal_alignment="right",
+            vertical_alignment="bottom",
+            x=12,
+            y=-18,
+            rotation=330,
+            extra_query={"trace": "img-async"},
+            extra_headers={"X-Debug": "img-async"},
+            extra_body={"debug": "img-async"},
+            timeout=0.29,
+        )
+
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "img-async.pdf"
+    timeout_value = captured_timeout["value"]
+    assert timeout_value is not None
+    if isinstance(timeout_value, dict):
+        assert all(
+            component == pytest.approx(0.29) for component in timeout_value.values()
+        )
+    else:
+        assert timeout_value == pytest.approx(0.29)
+
+
+@pytest.mark.asyncio
+async def test_async_watermark_pdf_with_text_validation_rejects_multiple_input_files(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    input_file_one = make_pdf_file(PdfRestFileID.generate(2))
+    input_file_two = make_pdf_file(PdfRestFileID.generate(2), name="second.pdf")
+    transport = httpx.MockTransport(
+        lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called"))
+    )
+    async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client:
+        with pytest.raises(ValidationError, match="at most 1 item"):
+            await client.watermark_pdf_with_text(
+                [input_file_one, input_file_two],
+                watermark_text="TooManyInputs",
+            )
+
+
+@pytest.mark.asyncio
+async def test_async_watermark_pdf_with_image_validation_rejects_multiple_watermark_files(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    input_file = make_pdf_file(PdfRestFileID.generate(2))
+    watermark_file_one = make_pdf_file(PdfRestFileID.generate(2), name="first.pdf")
+    watermark_file_two = make_pdf_file(PdfRestFileID.generate(2), name="second.pdf")
+    transport = httpx.MockTransport(
+        lambda _: (_ for _ in ()).throw(RuntimeError("Should not be called"))
+    )
+    async with AsyncPdfRestClient(api_key=ASYNC_API_KEY, transport=transport) as client:
+        with pytest.raises(ValidationError, match="at most 1 item"):
+            await client.watermark_pdf_with_image(
+                input_file,
+                watermark_file=[watermark_file_one, watermark_file_two],
+            )