From 10e95a6867c6fb923492a5711b2d16e4b6655f65 Mon Sep 17 00:00:00 2001 From: Trenton H <797416+stumpylog@users.noreply.github.com> Date: Sat, 3 Oct 2026 14:47:53 -0700 Subject: [PATCH] Refactor: add pure pdf_ops module with real-PDF tests --- .mypy-baseline.txt | 1 - .pyrefly-baseline.json | 7 - src/documents/bulk_edit.py | 404 ++++++++++------------ src/documents/pdf_ops.py | 184 ++++++++++ src/documents/serialisers.py | 2 + src/documents/tests/test_api_bulk_edit.py | 30 ++ src/documents/tests/test_pdf_ops.py | 386 +++++++++++++++++++++ 7 files changed, 787 insertions(+), 227 deletions(-) create mode 100644 src/documents/pdf_ops.py create mode 100644 src/documents/tests/test_pdf_ops.py diff --git a/.mypy-baseline.txt b/.mypy-baseline.txt index ef9043fd8..853090d3b 100644 --- a/.mypy-baseline.txt +++ b/.mypy-baseline.txt @@ -38,7 +38,6 @@ src/documents/bulk_edit.py:0: error: Incompatible types in assignment (expressio src/documents/bulk_edit.py:0: error: Invalid index type "str" for "dict[FieldDataType, str]"; expected type "FieldDataType" [index] src/documents/bulk_edit.py:0: error: List comprehension has incompatible type List[tuple[int, Any]]; expected List[int] [misc] src/documents/bulk_edit.py:0: error: List comprehension has incompatible type List[tuple[int, None]]; expected List[int] [misc] -src/documents/bulk_edit.py:0: error: Missing named argument "p" for "remove" of "PageList" [call-arg] src/documents/bulk_edit.py:0: error: Missing type arguments for generic type "dict" [type-arg] src/documents/bulk_edit.py:0: error: Missing type arguments for generic type "dict" [type-arg] src/documents/bulk_edit.py:0: error: Need type annotation for "to_create" (hint: "to_create: list[] = ...") [var-annotated] diff --git a/.pyrefly-baseline.json b/.pyrefly-baseline.json index 9305cfe00..3c09a1099 100644 --- a/.pyrefly-baseline.json +++ b/.pyrefly-baseline.json @@ -91,13 +91,6 @@ "concise_description": "Argument `list[int]` is not assignable to parameter `args` with type `tuple[Any, ...] | None` in function `celery.app.task.Task.apply_async`", "severity": "error" }, - { - "column": 33, - "path": "src/documents/bulk_edit.py", - "name": "missing-argument", - "concise_description": "Missing argument `p` in function `pikepdf._core.PageList.remove`", - "severity": "error" - }, { "column": 25, "path": "src/documents/caching.py", diff --git a/src/documents/bulk_edit.py b/src/documents/bulk_edit.py index 399282dc3..1ad228cc9 100644 --- a/src/documents/bulk_edit.py +++ b/src/documents/bulk_edit.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging import tempfile import uuid +from functools import partial from pathlib import Path from typing import TYPE_CHECKING from typing import Literal @@ -17,6 +18,7 @@ from django.db.models import Max from django.db.models import Q from django.utils import timezone +from documents import pdf_ops from documents.data_models import ConsumableDocument from documents.data_models import DocumentMetadataOverrides from documents.data_models import DocumentSource @@ -116,6 +118,11 @@ def _resolve_root_and_source_doc( ) +def _scratch_path(name: str) -> Path: + """A path inside a fresh directory under SCRATCH_DIR.""" + return Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / name + + def set_correspondent( doc_ids: list[int], correspondent: Correspondent, @@ -474,8 +481,6 @@ def rotate( pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) docs_by_root_id.setdefault(pair.root_doc.id, pair) - import pikepdf - for pair in docs_by_root_id.values(): if pair.source_doc.mime_type != "application/pdf": logger.warning( @@ -488,11 +493,7 @@ def rotate( Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / f"{pair.root_doc.id}_rotated.pdf" ) - with pikepdf.open(pair.source_doc.source_path) as pdf: - for page in pdf.pages: - page.rotate(degrees, relative=True) - pdf.remove_unreferenced_resources() - pdf.save(filepath) + pdf_ops.rotate_pdf(pair.source_doc.source_path, filepath, degrees) # Preserve metadata/permissions via overrides; mark as new version overrides = DocumentMetadataOverrides().from_document(pair.root_doc) @@ -535,48 +536,45 @@ def merge( qs = Document.objects.select_related("root_document").filter(id__in=doc_ids) docs_by_id = {doc.id: doc for doc in qs} affected_docs: list[int] = [] - import pikepdf - - merged_pdf = pikepdf.new() - version: str = merged_pdf.pdf_version handoff_asn: int | None = None - # use doc_ids to preserve order - for doc_id in doc_ids: - doc = docs_by_id.get(doc_id) - if doc is None: - continue - pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) - try: - doc_path = ( - pair.source_doc.archive_path - if archive_fallback - and pair.source_doc.mime_type != "application/pdf" - and pair.source_doc.has_archive_version - else pair.source_doc.source_path - ) - with pikepdf.open(str(doc_path)) as pdf: - version = max(version, pdf.pdf_version) - merged_pdf.pages.extend(pdf.pages) - affected_docs.append(doc.id) - if handoff_asn is None and doc.archive_serial_number is not None: - handoff_asn = doc.archive_serial_number - except Exception as e: - logger.exception( - f"Error merging document {doc.id}, it will not be included in the merge: {e}", - ) - if len(affected_docs) == 0: - logger.warning("No documents were merged") - return "OK" + with pdf_ops.PdfMerger() as merger: + # use doc_ids to preserve order + for doc_id in doc_ids: + doc = docs_by_id.get(doc_id) + if doc is None: + continue + pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) + try: + # archive_path is None when there is no archive version + archive_path = ( + pair.source_doc.archive_path + if archive_fallback + and pair.source_doc.mime_type != "application/pdf" + else None + ) + merger.add( + archive_path + if archive_path is not None + else pair.source_doc.source_path, + ) + affected_docs.append(doc.id) + if handoff_asn is None and doc.archive_serial_number is not None: + handoff_asn = doc.archive_serial_number + except Exception as e: + logger.exception( + f"Error merging document {doc.id}, it will not be included in the merge: {e}", + ) + if len(affected_docs) == 0: + logger.warning("No documents were merged") + return "OK" - filepath = ( - Path( - tempfile.mkdtemp(dir=settings.SCRATCH_DIR), + filepath = ( + Path( + tempfile.mkdtemp(dir=settings.SCRATCH_DIR), + ) + / f"{'_'.join([str(doc_id) for doc_id in affected_docs])[:100]}_merged.pdf" ) - / f"{'_'.join([str(doc_id) for doc_id in affected_docs])[:100]}_merged.pdf" - ) - merged_pdf.remove_unreferenced_resources() - merged_pdf.save(filepath, min_version=version) - merged_pdf.close() + merger.save(filepath) if metadata_document_id: metadata_document = qs.get(id=metadata_document_id) @@ -752,64 +750,60 @@ def split( ) doc = Document.objects.select_related("root_document").get(id=doc_ids[0]) pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) - import pikepdf consume_tasks = [] try: - with pikepdf.open(pair.source_doc.source_path) as pdf: - for idx, split_doc in enumerate(pages): - dst: pikepdf.Pdf = pikepdf.new() - for page in split_doc: - dst.pages.append(pdf.pages[page - 1]) - filepath: Path = ( - Path( - tempfile.mkdtemp(dir=settings.SCRATCH_DIR), - ) - / f"{doc.id}_{split_doc[0]}-{split_doc[-1]}.pdf" - ) - dst.remove_unreferenced_resources() - dst.save(filepath) - dst.close() + outputs = [ + ( + [pdf_ops.PageSpec(page) for page in split_doc], + partial(_scratch_path, f"{doc.id}_{split_doc[0]}-{split_doc[-1]}.pdf"), + ) + for split_doc in pages + ] + filepaths = pdf_ops.build_pdfs(pair.source_doc.source_path, outputs) - overrides: DocumentMetadataOverrides = ( - DocumentMetadataOverrides().from_document(doc) - ) - overrides.title = f"{doc.title} (split {idx + 1})" - if user is not None: - overrides.owner_id = user.id - if not delete_originals: - overrides.skip_asn_if_exists = True - logger.info( - f"Adding split document with pages {split_doc} to the task queue.", - ) - consume_tasks.append( - consume_file.s( - input_doc=ConsumableDocument( - source=DocumentSource.ConsumeFolder, - original_file=filepath, - ), - overrides=overrides, - ).set(headers={"trigger_source": trigger_source}), - ) + for idx, (split_doc, filepath) in enumerate( + zip(pages, filepaths, strict=True), + ): + overrides: DocumentMetadataOverrides = ( + DocumentMetadataOverrides().from_document(doc) + ) + overrides.title = f"{doc.title} (split {idx + 1})" + if user is not None: + overrides.owner_id = user.id + if not delete_originals: + overrides.skip_asn_if_exists = True + logger.info( + f"Adding split document with pages {split_doc} to the task queue.", + ) + consume_tasks.append( + consume_file.s( + input_doc=ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + ), + overrides=overrides, + ).set(headers={"trigger_source": trigger_source}), + ) - if delete_originals: - backup = release_archive_serial_numbers([doc.id]) - logger.info( - "Queueing removal of original document after consumption of the split documents", - ) - try: - chord( - header=consume_tasks, - body=delete.si([doc.id]), - ).on_error( - restore_archive_serial_numbers_task.s(backup), - ).apply_async() - except Exception: - restore_archive_serial_numbers(backup) - raise - else: - group(consume_tasks).delay() + if delete_originals: + backup = release_archive_serial_numbers([doc.id]) + logger.info( + "Queueing removal of original document after consumption of the split documents", + ) + try: + chord( + header=consume_tasks, + body=delete.si([doc.id]), + ).on_error( + restore_archive_serial_numbers_task.s(backup), + ).apply_async() + except Exception: + restore_archive_serial_numbers(backup) + raise + else: + group(consume_tasks).delay() except Exception as e: logger.exception(f"Error splitting document {doc.id}: {e}") @@ -830,8 +824,7 @@ def delete_pages( ) doc = Document.objects.select_related("root_document").get(id=doc_ids[0]) pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) - pages = sorted(pages) # sort pages to avoid index issues - import pikepdf + pages = sorted(set(pages)) try: # Produce edited PDF to a temp file and create a new version @@ -839,13 +832,7 @@ def delete_pages( Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / f"{pair.root_doc.id}_pages_deleted.pdf" ) - with pikepdf.open(pair.source_doc.source_path) as pdf: - offset = 1 # pages are 1-indexed - for page_num in pages: - pdf.pages.remove(pdf.pages[page_num - offset]) - offset += 1 # remove() changes the index of the pages - pdf.remove_unreferenced_resources() - pdf.save(filepath) + pdf_ops.remove_pages(pair.source_doc.source_path, filepath, pages) overrides = DocumentMetadataOverrides().from_document(pair.root_doc) if user is not None: @@ -894,47 +881,28 @@ def edit_pdf( ) doc = Document.objects.select_related("root_document").get(id=doc_ids[0]) pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) - import pikepdf - - pdf_docs: list[pikepdf.Pdf] = [] - try: - if not operations: - raise ValueError("Output document index is out of bounds") - - max_idx = max(op.get("doc", 0) for op in operations) - if update_document and max_idx > 0: - logger.error( - "Update requested but multiple output documents specified", + output_count = pdf_ops.validate_page_operations( + operations, + single_output=update_document, + ) + page_specs: list[list[pdf_ops.PageSpec]] = [[] for _ in range(output_count)] + for op in operations: + page_specs[op.get("doc", 0)].append( + pdf_ops.PageSpec(op["page"], op.get("rotate", 0)), ) - raise ValueError("Multiple output documents specified") - - if any( - op.get("doc", 0) < 0 or op.get("doc", 0) >= len(operations) - for op in operations - ): - raise ValueError("Output document index is out of bounds") - - with pikepdf.open(pair.source_doc.source_path) as src: - # prepare output documents - pdf_docs = [pikepdf.new() for _ in range(max_idx + 1)] - - for op in operations: - dst = pdf_docs[op.get("doc", 0)] - page = src.pages[op["page"] - 1] - dst.pages.append(page) - if op.get("rotate"): - dst.pages[-1].rotate(op["rotate"], relative=True) if update_document: # Create a new version from the edited PDF rather than replacing in-place - pdf = pdf_docs[0] - pdf.remove_unreferenced_resources() - filepath: Path = ( - Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) - / f"{pair.root_doc.id}_edited.pdf" + (filepath,) = pdf_ops.build_pdfs( + pair.source_doc.source_path, + [ + ( + page_specs[0], + partial(_scratch_path, f"{pair.root_doc.id}_edited.pdf"), + ), + ], ) - pdf.save(filepath) overrides = ( DocumentMetadataOverrides().from_document(pair.root_doc) if include_metadata @@ -955,6 +923,19 @@ def edit_pdf( headers={"trigger_source": trigger_source}, ) else: + version_filepaths = pdf_ops.build_pdfs( + pair.source_doc.source_path, + [ + ( + specs, + partial( + _scratch_path, + f"{pair.root_doc.id}_edit_{idx}.pdf", + ), + ) + for idx, specs in enumerate(page_specs, start=1) + ], + ) consume_tasks = [] overrides = ( DocumentMetadataOverrides().from_document(pair.root_doc) @@ -966,15 +947,9 @@ def edit_pdf( overrides.actor_id = user.id if not delete_original: overrides.skip_asn_if_exists = True - if delete_original and len(pdf_docs) == 1: + if delete_original and output_count == 1: overrides.asn = pair.root_doc.archive_serial_number - for idx, pdf in enumerate(pdf_docs, start=1): - version_filepath: Path = ( - Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) - / f"{pair.root_doc.id}_edit_{idx}.pdf" - ) - pdf.remove_unreferenced_resources() - pdf.save(version_filepath) + for version_filepath in version_filepaths: consume_tasks.append( consume_file.s( input_doc=ConsumableDocument( @@ -1024,8 +999,6 @@ def remove_password( """ Remove password protection from PDF documents. """ - import pikepdf - for doc_id in doc_ids: doc = Document.objects.select_related("root_document").get(id=doc_id) pair = _resolve_root_and_source_doc(doc, source_mode=source_mode) @@ -1039,76 +1012,69 @@ def remove_password( doc.id, pair.source_doc.source_path, ) - try: - with pikepdf.open(source_path) as pdf: - if not pdf.is_encrypted: - logger.info( - "Skipping password removal for document %s because the " - "source PDF is not encrypted", - pair.root_doc.id, - ) - continue - except pikepdf.PasswordError: - # Password-protected PDFs need the supplied password below. - pass - - with pikepdf.open(source_path, password=password) as pdf: - filepath: Path = ( - Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) - / f"{pair.root_doc.id}_unprotected.pdf" + if not pdf_ops.needs_decrypt(source_path): + logger.info( + "Skipping password removal for document %s because the " + "source PDF is not encrypted", + pair.root_doc.id, ) - pdf.remove_unreferenced_resources() - pdf.save(filepath) + continue - if update_document: - # Create a new version rather than modifying the root/original in place. - overrides = ( - DocumentMetadataOverrides().from_document(pair.root_doc) - if include_metadata - else DocumentMetadataOverrides() - ) - if user is not None: - overrides.owner_id = user.id - overrides.actor_id = user.id - consume_file.apply_async( - kwargs={ - "input_doc": ConsumableDocument( - source=DocumentSource.ConsumeFolder, - original_file=filepath, - root_document_id=pair.root_doc.id, - ), - "overrides": overrides, - }, - headers={"trigger_source": trigger_source}, - ) + filepath = pdf_ops.decrypt_pdf( + source_path, + partial(_scratch_path, f"{pair.root_doc.id}_unprotected.pdf"), + password, + ) + + if update_document: + # Create a new version rather than modifying the root/original in place. + overrides = ( + DocumentMetadataOverrides().from_document(pair.root_doc) + if include_metadata + else DocumentMetadataOverrides() + ) + if user is not None: + overrides.owner_id = user.id + overrides.actor_id = user.id + consume_file.apply_async( + kwargs={ + "input_doc": ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + root_document_id=pair.root_doc.id, + ), + "overrides": overrides, + }, + headers={"trigger_source": trigger_source}, + ) + else: + consume_tasks = [] + overrides = ( + DocumentMetadataOverrides().from_document(pair.root_doc) + if include_metadata + else DocumentMetadataOverrides() + ) + if user is not None: + overrides.owner_id = user.id + overrides.actor_id = user.id + + consume_tasks.append( + consume_file.s( + input_doc=ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + ), + overrides=overrides, + ).set(headers={"trigger_source": trigger_source}), + ) + + if delete_original: + chord( + header=consume_tasks, + body=delete.si([doc.id]), + ).delay() else: - consume_tasks = [] - overrides = ( - DocumentMetadataOverrides().from_document(pair.root_doc) - if include_metadata - else DocumentMetadataOverrides() - ) - if user is not None: - overrides.owner_id = user.id - overrides.actor_id = user.id - - consume_tasks.append( - consume_file.s( - input_doc=ConsumableDocument( - source=DocumentSource.ConsumeFolder, - original_file=filepath, - ), - overrides=overrides, - ).set(headers={"trigger_source": trigger_source}), - ) - - if delete_original: - chord( - header=consume_tasks, - body=delete.si([doc.id]), - ).delay() - else: - group(consume_tasks).delay() + group(consume_tasks).delay() except Exception as e: logger.exception( diff --git a/src/documents/pdf_ops.py b/src/documents/pdf_ops.py new file mode 100644 index 000000000..c35a9b2f4 --- /dev/null +++ b/src/documents/pdf_ops.py @@ -0,0 +1,184 @@ +""" +Pure PDF page operations used by documents.bulk_edit. + +This module deliberately knows nothing about Django, Celery or the documents +app: callers resolve documents, choose output paths and queue work. Every +function that writes a PDF removes unreferenced resources before saving. + +pikepdf is always called as ``pikepdf.open(...)`` / ``pikepdf.new()`` (never +``from pikepdf import open``) so tests can patch those module attributes. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from typing import NamedTuple + +import pikepdf + +if TYPE_CHECKING: + from collections.abc import Callable + from collections.abc import Iterable + from collections.abc import Mapping + from collections.abc import Sequence + from pathlib import Path + from types import TracebackType + + +class PageSpec(NamedTuple): + """One page of an output PDF: a 1-indexed source page, optionally rotated.""" + + page: int + rotate: int = 0 # relative degrees, 0 leaves the page alone + + +def _require_positive(pages: Iterable[int]) -> None: + for page in pages: + if page < 1: + raise ValueError(f"Page numbers start at 1, got {page}") + + +def rotate_pdf(src: Path, dst: Path, degrees: int) -> None: + """ + Rotate every page relatively on the opened document, not a rebuild, so Info, + XMP and outlines are kept. ``src`` is not modified. + """ + with pikepdf.open(src) as pdf: + for page in pdf.pages: + page.rotate(degrees, relative=True) + pdf.remove_unreferenced_resources() + pdf.save(dst) + + +def remove_pages(src: Path, dst: Path, pages: Iterable[int]) -> None: + """ + Remove 1-indexed pages from the opened document, not a rebuild, so Info, XMP + and outlines are kept. ``src`` is not modified. + + Duplicates are ignored. Pages are removed highest first so earlier removals + never shift the index of later ones. + """ + unique = sorted(set(pages)) + _require_positive(unique) + with pikepdf.open(src) as pdf: + for page_num in reversed(unique): + del pdf.pages[page_num - 1] + pdf.remove_unreferenced_resources() + pdf.save(dst) + + +def build_pdfs( + src: Path, + outputs: Sequence[tuple[Sequence[PageSpec], Callable[[], Path]]], +) -> list[Path]: + """ + Build one new PDF per output from pages of ``src``, opening ``src`` once. + + Each output is ``(page_specs, make_dst)``. ``make_dst`` is called after that + output's pages are copied and immediately before it is saved, so a bad page + number never leaves a destination behind. Document-level data (Info, XMP, + outlines) is not carried over. Returns the written paths in output order. + """ + for specs, _ in outputs: + _require_positive(spec.page for spec in specs) + + written: list[Path] = [] + with pikepdf.open(src) as source: + for specs, make_dst in outputs: + dst = pikepdf.new() + for spec in specs: + dst.pages.append(source.pages[spec.page - 1]) + if spec.rotate: + dst.pages[-1].rotate(spec.rotate, relative=True) + dst.remove_unreferenced_resources() + path = make_dst() + dst.save(path) + dst.close() + written.append(path) + return written + + +def validate_page_operations( + operations: Sequence[Mapping[str, int]], + *, + single_output: bool, +) -> int: + """ + Validate ``edit_pdf`` style operations and return the output document count. + + Each operation has ``page`` and optionally ``rotate`` and ``doc`` (the output + document index, default 0). The bounds rule is kept as it was: a ``doc`` index + must be below the number of operations. + """ + if not operations: + raise ValueError("Output document index is out of bounds") + + max_idx = max(op.get("doc", 0) for op in operations) + if single_output and max_idx > 0: + raise ValueError("Multiple output documents specified") + + if any( + op.get("doc", 0) < 0 or op.get("doc", 0) >= len(operations) for op in operations + ): + raise ValueError("Output document index is out of bounds") + + return max_idx + 1 + + +def needs_decrypt(src: Path) -> bool: + """ + True if ``src`` is encrypted. A PDF that needs a password to open at all + counts as encrypted. + """ + try: + with pikepdf.open(src) as pdf: + return bool(pdf.is_encrypted) + except pikepdf.PasswordError: + return True + + +def decrypt_pdf(src: Path, make_dst: Callable[[], Path], password: str) -> Path: + """ + Write an unencrypted copy of ``src`` and return its path. + + ``make_dst`` is only called once the password has been accepted, so a wrong + password never leaves a destination behind. + """ + with pikepdf.open(src, password=password) as pdf: + pdf.remove_unreferenced_resources() + dst = make_dst() + pdf.save(dst) + return dst + + +class PdfMerger: + """ + Accumulates the pages of several PDFs into one new PDF. + + ``add`` raises if a source cannot be read; deciding whether to skip it is the + caller's policy. Use as a context manager so the merged PDF is closed. + """ + + def __init__(self) -> None: + self._pdf = pikepdf.new() + self._version: str = self._pdf.pdf_version + + def __enter__(self) -> PdfMerger: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + self._pdf.close() + + def add(self, path: Path) -> None: + with pikepdf.open(str(path)) as pdf: + self._version = max(self._version, pdf.pdf_version) + self._pdf.pages.extend(pdf.pages) + + def save(self, dst: Path) -> None: + self._pdf.remove_unreferenced_resources() + self._pdf.save(dst, min_version=self._version) diff --git a/src/documents/serialisers.py b/src/documents/serialisers.py index c2c875029..1547a81bb 100644 --- a/src/documents/serialisers.py +++ b/src/documents/serialisers.py @@ -2144,6 +2144,8 @@ class BulkEditSerializer( raise serializers.ValidationError("pages must be a list") if not all(isinstance(i, int) for i in parameters["pages"]): raise serializers.ValidationError("pages must be a list of integers") + if any(i < 1 for i in parameters["pages"]): + raise serializers.ValidationError("pages must be positive integers") def _validate_parameters_merge(self, parameters) -> None: if "delete_originals" in parameters: diff --git a/src/documents/tests/test_api_bulk_edit.py b/src/documents/tests/test_api_bulk_edit.py index 9c7ba3db8..539cd68d8 100644 --- a/src/documents/tests/test_api_bulk_edit.py +++ b/src/documents/tests/test_api_bulk_edit.py @@ -1843,6 +1843,36 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase): m.assert_called_once() self.assertEqual(m.call_args.kwargs["pages"], [[1], [2, 3, 4], [5]]) + @mock.patch("documents.serialisers.bulk_edit.delete_pages") + def test_bulk_edit_delete_pages_rejects_pages_below_one(self, m) -> None: + """ + GIVEN: + - A legacy delete_pages bulk edit + WHEN: + - API to bulk edit is called with a page number below 1 + THEN: + - API returns HTTP 400 + - delete_pages is not called + """ + self.setup_mock(m, "delete_pages") + + for pages in ([0], [-1], [1, 0]): + with self.subTest(pages=pages): + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "delete_pages", + "parameters": {"pages": pages}, + }, + ), + content_type="application/json", + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn(b"pages must be positive integers", response.content) + m.assert_not_called() + @mock.patch("documents.views.bulk_edit.rotate") def test_rotate_insufficient_permissions(self, m) -> None: self.doc1.owner = User.objects.get(username="temp_admin") diff --git a/src/documents/tests/test_pdf_ops.py b/src/documents/tests/test_pdf_ops.py new file mode 100644 index 000000000..2fdc54466 --- /dev/null +++ b/src/documents/tests/test_pdf_ops.py @@ -0,0 +1,386 @@ +""" +Tests for documents.pdf_ops. + +These use real PDFs from the sample directories. No database, Celery or mocks. +Pages are compared by a hash of their content stream, so page identity and order +are easy to assert. +""" + +import ast +import hashlib +from collections.abc import Callable +from pathlib import Path + +import pikepdf +import pytest + +from documents import pdf_ops +from documents.pdf_ops import PageSpec + +SRC_ROOT = Path(__file__).parents[2] +SAMPLES = Path(__file__).parent / "samples" +THREE_PAGES = SAMPLES / "documents" / "originals" / "0000002.pdf" +TWELVE_PAGES = SAMPLES / "barcodes" / "split-by-asn-2.pdf" +ENCRYPTED = SAMPLES / "password-is-test.pdf" +SIGNED = SRC_ROOT / "paperless" / "tests" / "samples" / "tesseract" / "signed.pdf" + + +def _page_fingerprint(page: pikepdf.Page) -> str: + contents = page.obj.get("/Contents") + assert contents is not None, "sample page has no /Contents" + streams = list(contents) if isinstance(contents, pikepdf.Array) else [contents] + return hashlib.sha1(b"".join(s.read_bytes() for s in streams)).hexdigest() + + +def fingerprints(path: Path) -> list[str]: + with pikepdf.open(path) as pdf: + return [_page_fingerprint(page) for page in pdf.pages] + + +def rotations(path: Path) -> list[int]: + with pikepdf.open(path) as pdf: + return [int(page.obj.get("/Rotate", 0)) for page in pdf.pages] + + +def docinfo_keys(path: Path) -> set[str]: + with pikepdf.open(path) as pdf: + return set(pdf.docinfo.keys()) + + +def constant(path: Path) -> Callable[[], Path]: + return lambda: path + + +@pytest.fixture +def source_fingerprints() -> list[str]: + fps = fingerprints(THREE_PAGES) + assert len(set(fps)) == 3, "sample must have three distinct pages" + return fps + + +class TestRotatePdf: + def test_rotation_is_relative_and_applies_to_every_page(self, tmp_path: Path): + once = tmp_path / "once.pdf" + twice = tmp_path / "twice.pdf" + + pdf_ops.rotate_pdf(THREE_PAGES, once, 90) + pdf_ops.rotate_pdf(once, twice, 90) + + assert rotations(once) == [90, 90, 90] + assert rotations(twice) == [180, 180, 180] + assert fingerprints(twice) == fingerprints(THREE_PAGES) + + def test_keeps_document_info(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + pdf_ops.rotate_pdf(THREE_PAGES, dst, 90) + + assert "/Creator" in docinfo_keys(dst) + + +class TestRemovePages: + def test_removes_selected_pages( + self, + tmp_path: Path, + source_fingerprints: list[str], + ): + dst = tmp_path / "out.pdf" + + pdf_ops.remove_pages(THREE_PAGES, dst, [2]) + + assert fingerprints(dst) == [source_fingerprints[0], source_fingerprints[2]] + + def test_duplicate_page_numbers_remove_the_page_once( + self, + tmp_path: Path, + source_fingerprints: list[str], + ): + dst = tmp_path / "out.pdf" + + pdf_ops.remove_pages(THREE_PAGES, dst, [2, 2]) + + assert fingerprints(dst) == [source_fingerprints[0], source_fingerprints[2]] + + def test_unordered_pages(self, tmp_path: Path, source_fingerprints: list[str]): + dst = tmp_path / "out.pdf" + + pdf_ops.remove_pages(THREE_PAGES, dst, [3, 1]) + + assert fingerprints(dst) == [source_fingerprints[1]] + + def test_empty_list_keeps_every_page( + self, + tmp_path: Path, + source_fingerprints: list[str], + ): + dst = tmp_path / "out.pdf" + + pdf_ops.remove_pages(THREE_PAGES, dst, []) + + assert fingerprints(dst) == source_fingerprints + + def test_removing_every_page_writes_an_empty_pdf(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + pdf_ops.remove_pages(THREE_PAGES, dst, [1, 2, 3]) + + assert fingerprints(dst) == [] + + def test_keeps_document_info(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + pdf_ops.remove_pages(THREE_PAGES, dst, [1]) + + assert "/Creator" in docinfo_keys(dst) + + @pytest.mark.parametrize("bad_page", [0, -1]) + def test_rejects_pages_below_one(self, tmp_path: Path, bad_page: int): + dst = tmp_path / "out.pdf" + + with pytest.raises(ValueError, match="start at 1"): + pdf_ops.remove_pages(THREE_PAGES, dst, [1, bad_page]) + + assert not dst.exists() + + def test_page_past_the_end_raises_and_writes_nothing(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + with pytest.raises(IndexError): + pdf_ops.remove_pages(THREE_PAGES, dst, [99]) + + assert not dst.exists() + + +class TestBuildPdfs: + def test_selects_and_orders_pages( + self, + tmp_path: Path, + source_fingerprints: list[str], + ): + dst = tmp_path / "out.pdf" + + written = pdf_ops.build_pdfs( + THREE_PAGES, + [([PageSpec(3), PageSpec(1)], constant(dst))], + ) + + assert written == [dst] + assert fingerprints(dst) == [source_fingerprints[2], source_fingerprints[0]] + + def test_rotates_only_the_requested_pages(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + pdf_ops.build_pdfs( + THREE_PAGES, + [([PageSpec(1), PageSpec(2, 90), PageSpec(3, 180)], constant(dst))], + ) + + assert rotations(dst) == [0, 90, 180] + + def test_writes_one_file_per_output_in_order(self, tmp_path: Path): + first = tmp_path / "first.pdf" + second = tmp_path / "second.pdf" + source = fingerprints(TWELVE_PAGES) + + written = pdf_ops.build_pdfs( + TWELVE_PAGES, + [ + ([PageSpec(p) for p in (1, 2, 3)], constant(first)), + ([PageSpec(p) for p in range(4, 13)], constant(second)), + ], + ) + + assert written == [first, second] + assert fingerprints(first) == source[:3] + assert fingerprints(second) == source[3:] + + def test_empty_page_list_writes_a_zero_page_file(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + pdf_ops.build_pdfs(THREE_PAGES, [([], constant(dst))]) + + assert fingerprints(dst) == [] + + def test_does_not_carry_over_document_info(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + assert "/Creator" in docinfo_keys(THREE_PAGES) + + pdf_ops.build_pdfs(THREE_PAGES, [([PageSpec(1)], constant(dst))]) + + assert "/Creator" not in docinfo_keys(dst) + + def test_destination_is_not_requested_when_a_page_is_out_of_range( + self, + tmp_path: Path, + ): + requested: list[Path] = [] + + def make_dst() -> Path: + requested.append(tmp_path / "out.pdf") + return requested[-1] + + with pytest.raises(IndexError): + pdf_ops.build_pdfs(THREE_PAGES, [([PageSpec(99)], make_dst)]) + + assert requested == [] + + @pytest.mark.parametrize("bad_page", [0, -1]) + def test_rejects_pages_below_one_before_opening_anything( + self, + tmp_path: Path, + bad_page: int, + ): + requested: list[Path] = [] + + def make_dst() -> Path: + requested.append(tmp_path / "out.pdf") + return requested[-1] + + with pytest.raises(ValueError, match="start at 1"): + pdf_ops.build_pdfs( + THREE_PAGES, + [([PageSpec(1)], make_dst), ([PageSpec(bad_page)], make_dst)], + ) + + assert requested == [] + + +class TestValidatePageOperations: + def test_returns_the_output_count(self): + operations = [{"page": 1}, {"page": 2}, {"page": 3}] + + assert pdf_ops.validate_page_operations(operations, single_output=True) == 1 + + def test_gap_in_output_indices_counts_up_to_the_highest(self): + operations = [ + {"page": 1, "doc": 0}, + {"page": 2, "doc": 2}, + {"page": 3, "doc": 0}, + ] + + count = pdf_ops.validate_page_operations(operations, single_output=False) + + assert count == 3 + + def test_empty_operations_are_rejected(self): + with pytest.raises(ValueError, match="index is out of bounds"): + pdf_ops.validate_page_operations([], single_output=False) + + def test_multiple_outputs_rejected_when_single_output_required(self): + operations = [{"page": 1, "doc": 0}, {"page": 2, "doc": 1}] + + with pytest.raises(ValueError, match="Multiple output documents"): + pdf_ops.validate_page_operations(operations, single_output=True) + + @pytest.mark.parametrize("doc", [-1, 2, 2**32]) + def test_output_index_out_of_bounds(self, doc: int): + operations = [{"page": 1, "doc": 0}, {"page": 2, "doc": doc}] + + with pytest.raises(ValueError, match="index is out of bounds"): + pdf_ops.validate_page_operations(operations, single_output=False) + + +class TestDecrypt: + def test_needs_decrypt(self): + assert pdf_ops.needs_decrypt(ENCRYPTED) is True + assert pdf_ops.needs_decrypt(THREE_PAGES) is False + + def test_pdf_that_opens_without_a_password_but_is_flagged_encrypted(self): + assert pdf_ops.needs_decrypt(SIGNED) is True + + def test_decrypt_writes_an_unencrypted_copy(self, tmp_path: Path): + dst = tmp_path / "out.pdf" + + result = pdf_ops.decrypt_pdf(ENCRYPTED, constant(dst), "test") + + assert result == dst + assert pdf_ops.needs_decrypt(dst) is False + + def test_wrong_password_raises_and_never_requests_a_destination( + self, + tmp_path: Path, + ): + requested: list[Path] = [] + + def make_dst() -> Path: + requested.append(tmp_path / "out.pdf") + return requested[-1] + + with pytest.raises(pikepdf.PasswordError): + pdf_ops.decrypt_pdf(ENCRYPTED, make_dst, "wrong") + + assert requested == [] + + +class TestPdfMerger: + def test_pages_are_appended_in_the_order_added( + self, + tmp_path: Path, + source_fingerprints: list[str], + ): + reordered = tmp_path / "reordered.pdf" + merged = tmp_path / "merged.pdf" + pdf_ops.build_pdfs( + THREE_PAGES, + [([PageSpec(3), PageSpec(1)], constant(reordered))], + ) + + with pdf_ops.PdfMerger() as merger: + merger.add(reordered) + merger.add(THREE_PAGES) + merger.save(merged) + + assert fingerprints(merged) == [ + source_fingerprints[2], + source_fingerprints[0], + *source_fingerprints, + ] + + def test_output_version_is_at_least_the_highest_source_version( + self, + tmp_path: Path, + ): + merged = tmp_path / "merged.pdf" + with pikepdf.open(TWELVE_PAGES) as pdf: + source_versions = [pdf.pdf_version] + with pikepdf.open(THREE_PAGES) as pdf: + source_versions.append(pdf.pdf_version) + + with pdf_ops.PdfMerger() as merger: + merger.add(TWELVE_PAGES) + merger.add(THREE_PAGES) + merger.save(merged) + + with pikepdf.open(merged) as pdf: + assert pdf.pdf_version >= max(source_versions) + + def test_unreadable_source_raises_so_the_caller_can_skip_it( + self, + tmp_path: Path, + ): + garbage = tmp_path / "garbage.pdf" + garbage.write_bytes(b"not a pdf") + + with pdf_ops.PdfMerger() as merger: + with pytest.raises(pikepdf.PdfError): + merger.add(garbage) + + +def test_pdf_ops_imports_only_the_standard_library_and_pikepdf(): + tree = ast.parse(Path(pdf_ops.__file__).read_text()) + imported: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + imported.add(node.module.split(".")[0]) + + coupled = imported & { + "django", + "celery", + "documents", + "paperless", + "paperless_mail", + "paperless_ai", + } + assert not coupled