Add tests of Unzip tool

Note use of `noqa: S105` in test_unzip_file.py. This is to silence warnings over a test password value used only in testing.

Assisted-by: Codex

Author: datalogics-cgreen

tests/live/test_live_unzip.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import zipfile
+from pathlib import Path
+
+import pytest
+
+from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient
+from pdfrest.models import PdfRestFile
+
+from ..resources import get_test_resource_path
+
+
+def _build_zip_payload(tmp_path: Path) -> Path:
+    zip_path = tmp_path / "sample.zip"
+    with zipfile.ZipFile(zip_path, "w") as bundle:
+        report_path = get_test_resource_path("report.pdf")
+        bundle.write(report_path, arcname="report.pdf")
+    return zip_path
+
+
+@pytest.fixture(scope="module")
+def uploaded_zip_file(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> PdfRestFile:
+    zip_path = _build_zip_payload(tmp_path_factory.mktemp("zip-input"))
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        return client.files.create_from_paths([zip_path])[0]
+
+
+def test_live_unzip_file(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_zip_file: PdfRestFile,
+) -> None:
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = client.unzip_file(uploaded_zip_file)
+
+        assert response.output_files
+        assert all(file.size > 0 for file in response.output_files)
+        assert all(file.name for file in response.output_files)
+        assert str(response.input_id) == str(uploaded_zip_file.id)
+
+
+def test_live_unzip_invalid_override(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_zip_file: PdfRestFile,
+) -> None:
+    with (
+        PdfRestClient(
+            api_key=pdfrest_api_key,
+            base_url=pdfrest_live_base_url,
+        ) as client,
+        pytest.raises(PdfRestApiError, match="id"),
+    ):
+        client.unzip_file(
+            uploaded_zip_file,
+            extra_body={"id": "not-a-uuid"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_live_unzip_file_async(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_zip_file: PdfRestFile,
+) -> None:
+    async with AsyncPdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = await client.unzip_file(
+            uploaded_zip_file,
+            extra_query={"trace": "async"},
+        )
+
+        assert response.output_files
+        assert all(file.size > 0 for file in response.output_files)
+        assert all(file.name for file in response.output_files)
+        assert str(response.input_id) == str(uploaded_zip_file.id)

tests/test_unzip_file.py

@@ -0,0 +1,215 @@
+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 UnzipPayload
+
+from .graphics_test_helpers import (
+    ASYNC_API_KEY,
+    VALID_API_KEY,
+    build_file_info_payload,
+)
+
+
+def make_zip_file(file_id: str, name: str = "archive.zip") -> PdfRestFile:
+    return PdfRestFile.model_validate(
+        build_file_info_payload(file_id, name, "application/zip")
+    )
+
+
+def test_unzip_file_success(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    zip_file = make_zip_file(str(PdfRestFileID.generate()))
+    output_id = str(PdfRestFileID.generate())
+
+    payload_dump = UnzipPayload.model_validate({"files": zip_file}).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 == "/unzip":
+            seen["post"] += 1
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload == payload_dump
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [zip_file.id],
+                    "files": [
+                        {
+                            "name": "inner.txt",
+                            "id": output_id,
+                            "outputUrl": f"https://api.pdfrest.com/resource/{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,
+                    "inner.txt",
+                    "text/plain",
+                ),
+            )
+        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.unzip_file(zip_file)
+
+    assert seen == {"post": 1, "get": 1}
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "inner.txt"
+    assert response.output_file.type == "text/plain"
+    assert str(response.input_id) == str(zip_file.id)
+
+
+def test_unzip_file_with_password_and_customization(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    zip_file = make_zip_file(str(PdfRestFileID.generate()))
+    output_id = str(PdfRestFileID.generate())
+    captured_timeout: dict[str, float | dict[str, float] | None] = {}
+
+    password = "Secret123"  # noqa: S105 - test-only password fixture
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and request.url.path == "/unzip":
+            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(zip_file.id)
+            assert payload["password"] == password
+            assert payload["diagnostics"] == "on"
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [zip_file.id],
+                    "outputId": [output_id],
+                },
+            )
+        if request.method == "GET" and request.url.path == f"/resource/{output_id}":
+            assert request.url.params["trace"] == "sync"
+            assert request.headers["X-Debug"] == "sync"
+            return httpx.Response(
+                200,
+                json=build_file_info_payload(
+                    output_id,
+                    "custom-unzip.txt",
+                    "text/plain",
+                ),
+            )
+        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.unzip_file(
+            zip_file,
+            password=password,
+            extra_query={"trace": "sync"},
+            extra_headers={"X-Debug": "sync"},
+            extra_body={"diagnostics": "on"},
+            timeout=0.25,
+        )
+
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "custom-unzip.txt"
+    timeout_value = captured_timeout["value"]
+    assert timeout_value is not None
+    if isinstance(timeout_value, dict):
+        assert all(pytest.approx(0.25) == value for value in timeout_value.values())
+    else:
+        assert timeout_value == pytest.approx(0.25)
+
+
+def test_unzip_file_requires_zip(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    non_zip = PdfRestFile.model_validate(
+        build_file_info_payload(
+            str(PdfRestFileID.generate()), "document.pdf", "application/pdf"
+        )
+    )
+    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 be a ZIP file"),
+    ):
+        client.unzip_file(non_zip)
+
+
+def test_unzip_file_single_input(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    first = make_zip_file(str(PdfRestFileID.generate()))
+    second = make_zip_file(str(PdfRestFileID.generate()))
+    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 most 1 item"),
+    ):
+        client.unzip_file([first, second])
+
+
+@pytest.mark.asyncio
+async def test_async_unzip_file(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PDFREST_API_KEY", raising=False)
+    zip_file = make_zip_file(str(PdfRestFileID.generate()))
+    output_id = str(PdfRestFileID.generate())
+
+    payload_dump = UnzipPayload.model_validate({"files": zip_file}).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 == "/unzip":
+            payload = json.loads(request.content.decode("utf-8"))
+            assert payload == payload_dump
+            return httpx.Response(
+                200,
+                json={
+                    "inputId": [zip_file.id],
+                    "files": [
+                        {
+                            "name": "async.txt",
+                            "id": output_id,
+                            "outputUrl": f"https://api.pdfrest.com/resource/{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.txt",
+                    "text/plain",
+                ),
+            )
+        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.unzip_file(zip_file)
+
+    assert isinstance(response, PdfRestFileBasedResponse)
+    assert response.output_file.name == "async.txt"
+    assert str(response.input_id) == str(zip_file.id)