Add tests of ZIP tool

Assisted-by: Codex

Author: datalogics-cgreen

tests/live/test_live_zip.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import pytest
+
+from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient
+from pdfrest.models import PdfRestFile
+
+from ..resources import get_test_resource_path
+
+
+@pytest.fixture(scope="module")
+def uploaded_zip_inputs(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+) -> list[PdfRestFile]:
+    paths = [
+        get_test_resource_path("report.pdf"),
+        get_test_resource_path("report.docx"),
+    ]
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        return client.files.create_from_paths(paths)
+
+
+def test_live_zip_files(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_zip_inputs: list[PdfRestFile],
+) -> None:
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = client.zip_files(
+            uploaded_zip_inputs,
+            output="live-zip",
+        )
+
+        assert response.output_file.name.startswith("live-zip")
+        assert response.output_file.name.endswith(".zip")
+        assert response.output_file.type == "application/zip"
+        assert response.output_file.size > 0
+        assert {str(file.id) for file in uploaded_zip_inputs} == {
+            str(file_id) for file_id in response.input_ids
+        }
+
+
+def test_live_zip_files_invalid_id_override(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_zip_inputs: list[PdfRestFile],
+) -> None:
+    with (
+        PdfRestClient(
+            api_key=pdfrest_api_key,
+            base_url=pdfrest_live_base_url,
+        ) as client,
+        pytest.raises(PdfRestApiError, match="id"),
+    ):
+        client.zip_files(
+            uploaded_zip_inputs,
+            extra_body={"id": "not-a-uuid"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_live_zip_files_async(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_zip_inputs: list[PdfRestFile],
+) -> None:
+    async with AsyncPdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = await client.zip_files(
+            uploaded_zip_inputs,
+            output="live-zip-async",
+            extra_query={"trace": "async"},
+        )
+
+        assert response.output_file.name.startswith("live-zip-async")
+        assert response.output_file.name.endswith(".zip")
+        assert response.output_file.type == "application/zip"
+        assert response.output_file.size > 0
+        assert {str(file.id) for file in uploaded_zip_inputs} == {
+            str(file_id) for file_id in response.input_ids
+        }

tests/test_zip_files.py

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+from pydantic import ValidationError
+
+from pdfrest import AsyncPdfRestClient, PdfRestClient
+from pdfrest.models import PdfRestFileBasedResponse, PdfRestFileID
+from pdfrest.models._internal import ZipPayload
+
+from .graphics_test_helpers import (
+    ASYNC_API_KEY,
+    VALID_API_KEY,
+    build_file_info_payload,
+    make_pdf_file,
+)
+
+
+def test_zip_files_success(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    first = make_pdf_file(PdfRestFileID.generate(1), name="first.pdf")
+    second = make_pdf_file(PdfRestFileID.generate(1), name="second.pdf")
+    output_id = str(PdfRestFileID.generate())
+
+    payload_dump = ZipPayload.model_validate(
+        {"files": [first, second], "output": "bundle"}
+    ).model_dump(mode="json", by_alias=True, exclude_none=True, exclude_unset=True)
+
+    seen: dict[str, int] = {"post": 0, "get": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and request.url.path == "/zip":
+            seen["post"] += 1
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload == payload_dump
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [first.id, second.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,
+                    "bundle.zip",
+                    "application/zip",
+                ),
+            )
+        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.zip_files([first, second], output="bundle")
+
+    assert seen == {"post": 1, "get": 1}
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "bundle.zip"
+    assert response.output_file.type == "application/zip"
+    assert {str(file_id) for file_id in response.input_ids} == {
+        str(first.id),
+        str(second.id),
+    }
+    assert response.warning is None
+
+
+def test_zip_files_request_customization(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    source = make_pdf_file(PdfRestFileID.generate(1))
+    output_id = str(PdfRestFileID.generate())
+    captured_timeout: dict[str, float | dict[str, float] | None] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and request.url.path == "/zip":
+            assert request.url.params["trace"] == "sync"
+            assert request.headers["X-Debug"] == "sync"
+            captured_timeout["value"] = request.extensions.get("timeout")
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload["id"] == [str(source.id)]
+            assert payload["output"] == "custom-zip"
+            assert payload["diagnostics"] == "on"
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [source.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"] == "sync"
+            assert request.headers["X-Debug"] == "sync"
+            return httpx.Response(
+                200,
+                json=build_file_info_payload(
+                    output_id,
+                    "custom-zip.zip",
+                    "application/zip",
+                ),
+            )
+        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.zip_files(
+            source,
+            output="custom-zip",
+            extra_query={"trace": "sync"},
+            extra_headers={"X-Debug": "sync"},
+            extra_body={"diagnostics": "on"},
+            timeout=0.5,
+        )
+
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "custom-zip.zip"
+    timeout_value = captured_timeout["value"]
+    assert timeout_value is not None
+    if isinstance(timeout_value, dict):
+        assert all(pytest.approx(0.5) == value for value in timeout_value.values())
+    else:
+        assert timeout_value == pytest.approx(0.5)
+
+
+def test_zip_files_requires_input(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    transport = httpx.MockTransport(lambda request: (_ for _ in ()).throw(RuntimeError))
+
+    with (
+        PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client,
+        pytest.raises(ValidationError, match="at least 1 item"),
+    ):
+        client.zip_files([])
+
+
+def test_zip_files_rejects_invalid_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    source = make_pdf_file(PdfRestFileID.generate(1))
+    transport = httpx.MockTransport(lambda request: (_ for _ in ()).throw(RuntimeError))
+
+    with (
+        PdfRestClient(api_key=VALID_API_KEY, transport=transport) as client,
+        pytest.raises(ValidationError, match="must not start with a `\\.`"),
+    ):
+        client.zip_files(source, output=".hidden")
+
+
+@pytest.mark.asyncio
+async def test_async_zip_files(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    source = make_pdf_file(PdfRestFileID.generate(1))
+    output_id = str(PdfRestFileID.generate())
+
+    payload_dump = ZipPayload.model_validate({"files": source}).model_dump(
+        mode="json", by_alias=True, exclude_none=True, exclude_unset=True
+    )
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and request.url.path == "/zip":
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload == payload_dump
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [source.id],
+                    "outputId": [output_id],
+                },
+            )
+        if request.method == "GET" and request.url.path == f"/resource/{output_id}":
+            return httpx.Response(
+                200,
+                json=build_file_info_payload(
+                    output_id,
+                    "async.zip",
+                    "application/zip",
+                ),
+            )
+        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.zip_files(source)
+
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "async.zip"
+    assert str(response.input_id) == str(source.id)