diff --git a/src/documents/bulk_edit.py b/src/documents/bulk_edit.py index 41bc5931ce..11edd5b3e4 100644 --- a/src/documents/bulk_edit.py +++ b/src/documents/bulk_edit.py @@ -39,6 +39,8 @@ from documents.models import Tag from documents.permissions import PermissionsSpec from documents.permissions import set_permissions_for_objects from documents.plugins.helpers import DocumentsStatusManager +from documents.task_tracking import queue_consumption +from documents.task_tracking import queue_processing_tasks from documents.tasks import bulk_update_documents from documents.tasks import consume_file from documents.tasks import remove_document_from_index @@ -536,6 +538,7 @@ def rotate( source_mode: SourceMode = SourceModeChoices.LATEST_VERSION, user: User | None = None, trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI, + parent_task_id: str | None = None, ) -> Literal["OK"]: logger.info( f"Attempting to rotate {len(doc_ids)} documents by {degrees} degrees.", @@ -573,16 +576,15 @@ def rotate( if user is not None: 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}, + queue_consumption( + input_doc=ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + root_document_id=pair.root_doc.id, + ), + overrides=overrides, + trigger_source=trigger_source, + parent_task_id=parent_task_id, ) logger.info( f"Queued new rotated version for document {pair.root_doc.id} by {degrees} degrees", @@ -603,13 +605,15 @@ def merge( source_mode: SourceMode = SourceModeChoices.LATEST_VERSION, user: User | None = None, trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI, -) -> Literal["OK"]: + parent_task_id: str | None = None, +) -> Literal["OK"] | dict: logger.info( f"Attempting to merge {len(doc_ids)} documents into a single document.", ) 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] = [] + failed_ids: list[int] = [] handoff_asn: int | None = None with pdf_ops.PdfMerger() as merger: # use doc_ids to preserve order @@ -635,11 +639,14 @@ def merge( if handoff_asn is None and doc.archive_serial_number is not None: handoff_asn = doc.archive_serial_number except Exception as e: + failed_ids.append(doc.id) 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") + if parent_task_id is not None: + raise ValueError("None of the selected documents could be merged.") return "OK" filepath = ( @@ -682,6 +689,23 @@ def merge( overrides=overrides, ).set(headers={"trigger_source": trigger_source}) + if parent_task_id is not None: + backup = ( + release_archive_serial_numbers(affected_docs) if delete_originals else {} + ) + queue_processing_tasks( + parent_task_id, + [consume_task], + delete_ids=affected_docs if delete_originals else None, + asn_backup=backup, + ) + if failed_ids: + return { + "failed_document_ids": failed_ids, + "error_message": f"Documents {failed_ids} were excluded from the merge.", + } + return "OK" + if delete_originals: backup = release_archive_serial_numbers(affected_docs) logger.info( @@ -826,6 +850,7 @@ def split( source_mode: SourceMode = SourceModeChoices.LATEST_VERSION, user: User | None = None, trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI, + parent_task_id: str | None = None, ) -> Literal["OK"]: logger.info( f"Attempting to split document {doc_ids[0]} into {len(pages)} documents", @@ -869,7 +894,17 @@ def split( ).set(headers={"trigger_source": trigger_source}), ) - if delete_originals: + if parent_task_id is not None: + backup = ( + release_archive_serial_numbers([doc.id]) if delete_originals else {} + ) + queue_processing_tasks( + parent_task_id, + consume_tasks, + delete_ids=[doc.id] if delete_originals else None, + asn_backup=backup, + ) + elif delete_originals: backup = release_archive_serial_numbers([doc.id]) logger.info( "Queueing removal of original document after consumption of the split documents", @@ -901,6 +936,7 @@ def delete_pages( source_mode: SourceMode = SourceModeChoices.LATEST_VERSION, user: User | None = None, trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI, + parent_task_id: str | None = None, ) -> Literal["OK"]: logger.info( f"Attempting to delete pages {pages} from {len(doc_ids)} documents", @@ -919,16 +955,15 @@ def delete_pages( overrides = DocumentMetadataOverrides().from_document(pair.root_doc) if user is not None: 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}, + queue_consumption( + input_doc=ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + root_document_id=pair.root_doc.id, + ), + overrides=overrides, + trigger_source=trigger_source, + parent_task_id=parent_task_id, ) logger.info( f"Queued new version for document {pair.root_doc.id} after deleting pages {pages}", @@ -950,6 +985,7 @@ def edit_pdf( source_mode: SourceMode = SourceModeChoices.LATEST_VERSION, user: User | None = None, trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI, + parent_task_id: str | None = None, ) -> Literal["OK"]: """ Operations is a list of dictionaries describing the final PDF pages. @@ -994,16 +1030,15 @@ def edit_pdf( 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}, + queue_consumption( + input_doc=ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + root_document_id=pair.root_doc.id, + ), + overrides=overrides, + trigger_source=trigger_source, + parent_task_id=parent_task_id, ) else: version_filepaths = pdf_ops.build_pdfs( @@ -1043,7 +1078,17 @@ def edit_pdf( ).set(headers={"trigger_source": trigger_source}), ) - if delete_original: + if parent_task_id is not None: + backup = ( + release_archive_serial_numbers([doc.id]) if delete_original else {} + ) + queue_processing_tasks( + parent_task_id, + consume_tasks, + delete_ids=[doc.id] if delete_original else None, + asn_backup=backup, + ) + elif delete_original: backup = release_archive_serial_numbers([doc.id]) try: chord( @@ -1077,6 +1122,7 @@ def remove_password( source_mode: SourceMode = SourceModeChoices.LATEST_VERSION, user: User | None = None, trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI, + parent_task_id: str | None = None, source_paths_by_id: Mapping[int, Path] | None = None, ) -> Literal["OK"]: """ @@ -1119,16 +1165,15 @@ def remove_password( 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}, + queue_consumption( + input_doc=ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=filepath, + root_document_id=pair.root_doc.id, + ), + overrides=overrides, + trigger_source=trigger_source, + parent_task_id=parent_task_id, ) else: consume_tasks = [] @@ -1151,7 +1196,19 @@ def remove_password( ).set(headers={"trigger_source": trigger_source}), ) - if delete_original: + if parent_task_id is not None: + backup = ( + release_archive_serial_numbers([doc.id]) + if delete_original + else {} + ) + queue_processing_tasks( + parent_task_id, + consume_tasks, + delete_ids=[doc.id] if delete_original else None, + asn_backup=backup, + ) + elif delete_original: chord( header=consume_tasks, body=delete.si([doc.id]), diff --git a/src/documents/bulk_operations.py b/src/documents/bulk_operations.py index a1965c4cee..529a5c1ec0 100644 --- a/src/documents/bulk_operations.py +++ b/src/documents/bulk_operations.py @@ -115,7 +115,10 @@ def execute_bulk_operation( if operation not in MODIFIED_FIELDS: execution_parameters = parameters.copy() - if operation == "reprocess" and self.request.id is not None: + if ( + operation in TRIGGER_SOURCE_OPERATIONS | {"reprocess"} + and self.request.id is not None + ): execution_parameters["parent_task_id"] = self.request.id if operation in USER_OPERATIONS: execution_parameters["user"] = user @@ -128,7 +131,7 @@ def execute_bulk_operation( if settings.AUDIT_LOG_ENABLED and modified_field else {} ) - method(document_ids, **execution_parameters) + result = method(document_ids, **execution_parameters) if settings.AUDIT_LOG_ENABLED and modified_field: from auditlog.models import LogEntry @@ -143,7 +146,11 @@ def execute_bulk_operation( actor=user, additional_data={"reason": f"Bulk edit: {operation}"}, ) - return {"operation": operation, "document_count": len(document_ids)} + return { + "operation": operation, + "document_count": len(document_ids), + **(result if isinstance(result, dict) else {}), + } affected_ids: list[int] = [] modified_field = MODIFIED_FIELDS[operation] diff --git a/src/documents/signals/handlers.py b/src/documents/signals/handlers.py index 2f044afd97..36fd98f9cd 100644 --- a/src/documents/signals/handlers.py +++ b/src/documents/signals/handlers.py @@ -1079,6 +1079,7 @@ TRACKED_TASKS: dict[str, PaperlessTask.TaskType] = { "documents.tasks.update_document_content_maybe_archive_file": PaperlessTask.TaskType.REPROCESS_DOCUMENT, "documents.tasks.build_share_link_bundle": PaperlessTask.TaskType.BUILD_SHARE_LINK, "documents.bulk_edit.delete": PaperlessTask.TaskType.BULK_DELETE, + "documents.task_tracking.finish_file_operation": PaperlessTask.TaskType.BULK_DELETE, "documents.tasks.apply_ai_suggestions": PaperlessTask.TaskType.APPLY_AI_SUGGESTIONS, BULK_OPERATION_TASK_NAME: PaperlessTask.TaskType.BULK_UPDATE, } @@ -1127,6 +1128,8 @@ def _extract_input_data( data["overrides"] = override_dict return data + if task_type == PaperlessTask.TaskType.REPROCESS_DOCUMENT: + return {"document_id": task_kwargs["document_id"]} if task_type == PaperlessTask.TaskType.MAIL_FETCH: account_ids = task_kwargs.get("account_ids") if account_ids is not None: @@ -1214,6 +1217,19 @@ def before_task_publish_handler( input_data = _extract_input_data(task_type, task_kwargs) owner_id = _extract_owner_id(task_type, task_kwargs) trigger_source = _determine_trigger_source(headers) + # Consumption can fan out again (e.g. barcode splitting). Keep those + # descendants attached so their caller cannot finish before them. + parent = None + if task_type == PaperlessTask.TaskType.CONSUME_FILE and headers.get( + "parent_id", + ): + parent = PaperlessTask.objects.filter( + task_id=headers["parent_id"], + parent__isnull=False, + ).first() + if parent is not None: + owner_id = parent.owner_id + trigger_source = parent.trigger_source # A retried task is republished with the same task_id, so this fires # again for it; get_or_create keeps the original PENDING record @@ -1226,6 +1242,7 @@ def before_task_publish_handler( "status": PaperlessTask.Status.PENDING, "input_data": input_data, "owner_id": owner_id, + "parent": parent, }, ) except Exception: # pragma: no cover @@ -1337,6 +1354,19 @@ def task_postrun_handler( task_instance.status = PaperlessTask.Status.FAILURE changed_fields.append("status") + if ( + task_instance.parent_id is not None + and task_instance.task_type == PaperlessTask.TaskType.CONSUME_FILE + and (not isinstance(retval, dict) or "document_id" not in retval) + and not (isinstance(retval, dict) and "duplicate_of" in retval) + ): + task_instance.status = PaperlessTask.Status.FAILURE + task_instance.result_data = { + **(retval if isinstance(retval, dict) else {}), + "error_message": "Consumption stopped without creating a document.", + } + changed_fields.extend(["status", "result_data"]) + task_instance.save(update_fields=changed_fields) update_parent(task_instance) if task and task.name == BULK_OPERATION_TASK_NAME: diff --git a/src/documents/task_tracking.py b/src/documents/task_tracking.py index c74d5b74fb..8aca1d0236 100644 --- a/src/documents/task_tracking.py +++ b/src/documents/task_tracking.py @@ -1,12 +1,41 @@ """Completion tracking for operations which dispatch document processing tasks.""" -import uuid +from __future__ import annotations +import uuid +from typing import TYPE_CHECKING + +from celery import shared_task from django.db import transaction from django.utils import timezone from documents.models import PaperlessTask +if TYPE_CHECKING: + from celery.canvas import Signature + + from documents.data_models import ConsumableDocument + from documents.data_models import DocumentMetadataOverrides + + +def queue_consumption( + *, + input_doc: ConsumableDocument, + overrides: DocumentMetadataOverrides, + trigger_source: str, + parent_task_id: str | None, +) -> None: + from documents.tasks import consume_file + + kwargs = {"input_doc": input_doc, "overrides": overrides} + if parent_task_id is None: + consume_file.apply_async( + kwargs=kwargs, + headers={"trigger_source": trigger_source}, + ) + else: + queue_processing_tasks(parent_task_id, [consume_file.s(**kwargs)]) + def queue_reprocessing_tasks( parent_task_id: str, @@ -16,28 +45,63 @@ def queue_reprocessing_tasks( ) -> None: from documents.tasks import update_document_content_maybe_archive_file + queue_processing_tasks( + parent_task_id, + [ + update_document_content_maybe_archive_file.s( + document_id=document_id, + remote_ocr=remote_ocr, + ) + for document_id in document_ids + ], + ) + + +def queue_processing_tasks( + parent_task_id: str, + signatures: list[Signature], + *, + delete_ids: list[int] | None = None, + asn_backup: dict | None = None, +) -> None: + from documents.signals.handlers import TRACKED_TASKS + from documents.signals.handlers import _extract_input_data + parent = PaperlessTask.objects.get(task_id=parent_task_id) children = [ PaperlessTask( task_id=str(uuid.uuid4()), - task_type=PaperlessTask.TaskType.REPROCESS_DOCUMENT, + task_type=TRACKED_TASKS[signature.task], trigger_source=parent.trigger_source, owner_id=parent.owner_id, parent=parent, - input_data={"document_id": document_id}, + input_data=_extract_input_data( + TRACKED_TASKS[signature.task], + signature.kwargs, + ), ) - for document_id in document_ids + for signature in signatures ] - # Register the complete set before any can finish - PaperlessTask.objects.bulk_create(children) - for index, child in enumerate(children): - try: - update_document_content_maybe_archive_file.apply_async( - task_id=child.task_id, - kwargs={ - "document_id": child.input_data["document_id"], - "remote_ocr": remote_ocr, + # Dependencies and their final step must be registered together. + with transaction.atomic(): + PaperlessTask.objects.bulk_create(children) + if delete_ids: + PaperlessTask.objects.create( + task_id=str(uuid.uuid4()), + task_type=PaperlessTask.TaskType.BULK_DELETE, + owner_id=parent.owner_id, + trigger_source=parent.trigger_source, + parent=parent, + input_data={ + "dependencies": [child.pk for child in children], + "delete_ids": delete_ids, + "asn_backup": asn_backup or {}, }, + ) + for index, (signature, child) in enumerate(zip(signatures, children, strict=True)): + try: + signature.apply_async( + task_id=child.task_id, headers={"trigger_source": parent.trigger_source}, ) except Exception: @@ -46,11 +110,75 @@ def queue_reprocessing_tasks( ).update( status=PaperlessTask.Status.FAILURE, date_done=timezone.now(), - result_data={"error_message": "Could not publish reprocessing task."}, + result_data={"error_message": "Could not publish processing task."}, ) raise +@shared_task +def finish_file_operation( + dependencies: list[int], + delete_ids: list[int], + asn_backup: dict, +) -> dict[str, str]: + from documents.bulk_edit import delete + from documents.bulk_edit import restore_archive_serial_numbers + + statuses = list( + PaperlessTask.objects.filter(pk__in=dependencies).values_list( + "status", + flat=True, + ), + ) + if len(statuses) != len(dependencies): + raise ValueError( + "Processing task records are missing; originals were preserved.", + ) + if any(status != PaperlessTask.Status.SUCCESS for status in statuses): + restore_archive_serial_numbers(asn_backup) + return {"reason": "Originals preserved; archive serial numbers restored."} + delete(delete_ids) + return {"reason": "Originals moved to trash."} + + +def _publish_finalizer(task: PaperlessTask) -> None: + try: + finish_file_operation.apply_async( + task_id=task.task_id, + kwargs=task.input_data, + headers={"trigger_source": task.trigger_source}, + ) + except Exception as error: + from documents.signals.handlers import task_failure_handler + + task_failure_handler( + sender=finish_file_operation, + task_id=task.task_id, + exception=error, + ) + + +def _schedule_finalizers(parent: PaperlessTask) -> None: + for task in parent.children.filter( + task_type=PaperlessTask.TaskType.BULK_DELETE, + status=PaperlessTask.Status.PENDING, + ): + dependencies = task.input_data.get("dependencies") + if dependencies is None or task.result_data: + continue + if ( + PaperlessTask.objects.filter(pk__in=dependencies) + .exclude( + status__in=PaperlessTask.COMPLETE_STATUSES, + ) + .exists() + ): + continue + task.result_data = {"scheduled": True} + task.save(update_fields=["result_data"]) + transaction.on_commit(lambda task=task: _publish_finalizer(task)) + + def finish_dispatch(task: PaperlessTask, result_data: dict) -> None: """Record the scheduling outcome without completing an active parent.""" with transaction.atomic(): @@ -74,6 +202,7 @@ def update_parent(task: PaperlessTask) -> None: def _finalize_parent(parent: PaperlessTask) -> None: if parent.is_complete or not parent.dispatched: return + _schedule_finalizers(parent) statuses = list(parent.children.values_list("status", flat=True)) if any(status not in PaperlessTask.COMPLETE_STATUSES for status in statuses): return @@ -83,7 +212,7 @@ def _finalize_parent(parent: PaperlessTask) -> None: result_data = parent.result_data or {} if failed and not result_data.get("error_message"): result_data["error_message"] = ( - f"{succeeded} document(s) reprocessed; {failed} failed or were revoked. " + f"{succeeded} child task(s) succeeded; {failed} failed or were revoked. " "See child tasks for details." ) parent.status = ( @@ -108,4 +237,6 @@ def _finalize_parent(parent: PaperlessTask) -> None: parent.save() from documents.signals.handlers import _send_bulk_operation_status - transaction.on_commit(lambda: _send_bulk_operation_status(parent)) + if "operation" in parent.input_data: + transaction.on_commit(lambda: _send_bulk_operation_status(parent)) + update_parent(parent) diff --git a/src/documents/tests/test_task_tracking.py b/src/documents/tests/test_task_tracking.py index 5baf70baf4..5d1e5c33ab 100644 --- a/src/documents/tests/test_task_tracking.py +++ b/src/documents/tests/test_task_tracking.py @@ -1,3 +1,4 @@ +import shutil import uuid from unittest import mock @@ -33,6 +34,20 @@ def operation(paperless_dirs): return parent, kwargs +@pytest.fixture +def pdf_documents(paperless_dirs, tmp_path): + from documents.models import Document + from paperless_testing.samples import SIMPLE_DIGITAL_PDF + + documents = DocumentFactory.create_batch(2, mime_type="application/pdf") + for document in documents: + source = tmp_path / f"{document.pk}.pdf" + shutil.copy(SIMPLE_DIGITAL_PDF, source) + Document.objects.filter(pk=document.pk).update(filename=str(source)) + document.refresh_from_db() + return documents + + def complete(child, status="SUCCESS"): if status == "FAILURE": task_failure_handler( @@ -72,9 +87,7 @@ def test_parent_waits_for_all_children(operation, outcome): assert publish.call_count == 3 assert all(child.owner_id == parent.owner_id for child in children) assert all(child.trigger_source == parent.trigger_source for child in children) - assert all( - call.kwargs["kwargs"]["remote_ocr"] for call in publish.call_args_list - ) + assert all(call.args[1]["remote_ocr"] for call in publish.call_args_list) complete(children[0], outcome) complete(children[1]) parent.refresh_from_db() @@ -99,7 +112,7 @@ def test_parent_waits_for_all_children(operation, outcome): def test_fast_children_wait_for_dispatch_to_finish(operation, outcome): parent, kwargs = operation - def publish(*, task_id, **kwargs): + def publish(*args, task_id, **kwargs): # Every child is registered before the first publication. assert parent.children.count() == 3 complete(PaperlessTask.objects.get(task_id=task_id), outcome) @@ -174,3 +187,357 @@ def test_client_cannot_supply_parent_task_id(operation): ) assert response.status_code == 400 publish.assert_not_called() + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize( + "name,parameters", + [ + ("rotate", {"degrees": 90}), + ("merge", {}), + ("split", {"pages": [[1]]}), + ("delete_pages", {"pages": [1]}), + ("edit_pdf", {"operations": [{"page": 1}], "update_document": True}), + ("edit_pdf", {"operations": [{"page": 1}]}), + ("remove_password", {"password": "secret", "update_document": True}), + ("remove_password", {"password": "secret"}), + ], +) +def test_file_operations_wait_for_consumption( + operation, + pdf_documents, + name, + parameters, +): + from documents.tasks import consume_file + + parent, kwargs = operation + documents = pdf_documents + kwargs.update( + operation=name, + document_ids=[doc.pk for doc in documents] + if name == "merge" + else [documents[0].pk], + parameters=parameters, + ) + parent.input_data = { + "operation": name, + "document_count": len(kwargs["document_ids"]), + } + parent.save() + with ( + mock.patch("documents.tasks.consume_file.apply_async") as publish, + mock.patch("documents.bulk_edit.pdf_ops.needs_decrypt", return_value=True), + mock.patch( + "documents.bulk_edit.pdf_ops.decrypt_pdf", + return_value=documents[0].source_path, + ), + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + child = parent.children.get() + assert child.task_type == PaperlessTask.TaskType.CONSUME_FILE + publish.assert_called_once() + task_postrun_handler( + task=consume_file, + task_id=child.task_id, + state="SUCCESS", + retval={"document_id": 100}, + ) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.SUCCESS + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize( + "outcome", + ["success", "failure", "duplicate", "stopped", "revoked"], +) +def test_replacement_waits_for_deletion_or_recovery(operation, pdf_documents, outcome): + from documents.models import Document + from documents.task_tracking import finish_file_operation + from documents.tasks import consume_file + + parent, kwargs = operation + doc = pdf_documents[0] + doc.archive_serial_number = 123 + Document.objects.filter(pk=doc.pk).update(archive_serial_number=123) + kwargs.update( + operation="edit_pdf", + document_ids=[doc.pk], + parameters={"operations": [{"page": 1}], "delete_original": True}, + ) + parent.input_data = {"operation": "edit_pdf", "document_count": 1} + parent.save() + with ( + mock.patch("documents.tasks.consume_file.apply_async"), + mock.patch( + "documents.task_tracking.finish_file_operation.apply_async", + ) as finalize, + mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify, + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + consume = parent.children.get(task_type=PaperlessTask.TaskType.CONSUME_FILE) + finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE) + finalize.assert_not_called() + if outcome == "failure": + complete(consume, "FAILURE") + elif outcome == "revoked": + complete(consume, "REVOKED") + else: + retval = ( + {"document_id": 100} + if outcome == "success" + else ( + {"duplicate_of": 100} + if outcome == "duplicate" + else {"reason": "Stopped"} + ) + ) + task_postrun_handler( + task=consume_file, + task_id=consume.task_id, + state="SUCCESS", + retval=retval, + ) + finalize.assert_called_once() + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + assert Document.objects.filter(pk=doc.pk).exists() + notify.assert_not_called() + finish_file_operation.apply( + kwargs=finalizer.input_data, + task_id=finalizer.task_id, + throw=True, + ) + parent.refresh_from_db() + assert parent.status == ("success" if outcome == "success" else "failure") + notify.assert_called_once() + if outcome == "success": + assert not Document.objects.filter(pk=doc.pk).exists() + else: + doc.refresh_from_db() + assert not doc.is_deleted + assert doc.archive_serial_number == 123 + + +@pytest.mark.django_db(transaction=True) +def test_consumption_descendants_keep_operation_active(operation): + from documents.data_models import ConsumableDocument + from documents.data_models import DocumentSource + from documents.signals.handlers import before_task_publish_handler + from documents.task_tracking import finish_dispatch + from documents.task_tracking import queue_processing_tasks + from documents.tasks import consume_file + from paperless_testing.samples import SIMPLE_DIGITAL_PDF + + parent, _ = operation + parent.status = PaperlessTask.Status.STARTED + parent.save() + input_doc = ConsumableDocument( + source=DocumentSource.ConsumeFolder, + original_file=SIMPLE_DIGITAL_PDF, + ) + with mock.patch("documents.tasks.consume_file.apply_async"): + queue_processing_tasks(parent.task_id, [consume_file.s(input_doc=input_doc)]) + child = parent.children.get() + finish_dispatch(parent, {}) + descendant_id = str(uuid.uuid4()) + before_task_publish_handler( + headers={ + "task": consume_file.name, + "id": descendant_id, + "parent_id": child.task_id, + }, + body=((), {"input_doc": input_doc}, {}), + ) + descendant = PaperlessTask.objects.get(task_id=descendant_id) + assert descendant.parent_id == child.pk + assert descendant.owner_id == parent.owner_id + task_postrun_handler( + task=consume_file, + task_id=child.task_id, + state="SUCCESS", + retval={"reason": "Barcode splitting complete"}, + ) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + task_postrun_handler( + task=consume_file, + task_id=descendant_id, + state="SUCCESS", + retval={"document_id": 100}, + ) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.SUCCESS + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("failure", ["delete", "restore", "publish"]) +def test_final_step_failures_reach_parent(operation, pdf_documents, failure): + from documents.task_tracking import finish_file_operation + from documents.tasks import consume_file + + parent, kwargs = operation + doc = pdf_documents[0] + kwargs.update( + operation="edit_pdf", + document_ids=[doc.pk], + parameters={"operations": [{"page": 1}], "delete_original": True}, + ) + with ( + mock.patch("documents.tasks.consume_file.apply_async"), + mock.patch( + "documents.task_tracking.finish_file_operation.apply_async", + side_effect=RuntimeError("Publish failed") + if failure == "publish" + else None, + ), + mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify, + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + child = parent.children.get(task_type=PaperlessTask.TaskType.CONSUME_FILE) + if failure == "restore": + complete(child, "FAILURE") + else: + task_postrun_handler( + task=consume_file, + task_id=child.task_id, + state="SUCCESS", + retval={"document_id": 100}, + ) + finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE) + if failure != "publish": + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + notify.assert_not_called() + target = ( + "documents.bulk_edit.delete" + if failure == "delete" + else "documents.bulk_edit.restore_archive_serial_numbers" + ) + with mock.patch(target, side_effect=RuntimeError("Final step failed")): + finish_file_operation.apply( + kwargs=finalizer.input_data, + task_id=finalizer.task_id, + throw=False, + ) + parent.refresh_from_db() + finalizer.refresh_from_db() + assert finalizer.status == PaperlessTask.Status.FAILURE + assert parent.status == PaperlessTask.Status.FAILURE + notify.assert_called_once() + doc.refresh_from_db() + assert not doc.is_deleted + + +@pytest.mark.django_db(transaction=True) +def test_partial_merge_preserves_skipped_original_and_reports_it( + operation, + pdf_documents, +): + from documents.models import Document + from documents.task_tracking import finish_file_operation + from documents.tasks import consume_file + + parent, kwargs = operation + included, excluded = pdf_documents + excluded.source_path.unlink() + kwargs.update( + operation="merge", + document_ids=[included.pk, excluded.pk], + parameters={"delete_originals": True}, + ) + parent.input_data = {"operation": "merge", "document_count": 2} + parent.save() + with ( + mock.patch("documents.tasks.consume_file.apply_async") as consume, + mock.patch("documents.task_tracking.finish_file_operation.apply_async"), + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + consume.assert_called_once() + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + assert parent.result_data["failed_document_ids"] == [excluded.pk] + child = parent.children.get(task_type=PaperlessTask.TaskType.CONSUME_FILE) + task_postrun_handler( + task=consume_file, + task_id=child.task_id, + state="SUCCESS", + retval={"document_id": 100}, + ) + finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE) + assert finalizer.input_data["delete_ids"] == [included.pk] + finish_file_operation.apply( + kwargs=finalizer.input_data, + task_id=finalizer.task_id, + throw=True, + ) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.FAILURE + assert parent.result_data["failed_document_ids"] == [excluded.pk] + assert not Document.objects.filter(pk=included.pk).exists() + assert Document.objects.filter(pk=excluded.pk).exists() + + +@pytest.mark.django_db(transaction=True) +def test_merge_with_no_valid_inputs_fails(operation, pdf_documents): + parent, kwargs = operation + for doc in pdf_documents: + doc.source_path.unlink() + kwargs.update( + operation="merge", + document_ids=[doc.pk for doc in pdf_documents], + parameters={}, + ) + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=False) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.FAILURE + assert not parent.children.exists() + + +@pytest.mark.django_db(transaction=True) +def test_split_waits_for_every_output_before_recovery(operation, pdf_documents): + from documents.task_tracking import finish_file_operation + from documents.tasks import consume_file + + parent, kwargs = operation + doc = pdf_documents[0] + kwargs.update( + operation="split", + document_ids=[doc.pk], + parameters={"pages": [[1], [1]], "delete_originals": True}, + ) + with ( + mock.patch("documents.tasks.consume_file.apply_async"), + mock.patch( + "documents.task_tracking.finish_file_operation.apply_async", + ) as finalize, + ): + execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True) + children = list( + parent.children.filter(task_type=PaperlessTask.TaskType.CONSUME_FILE), + ) + assert len(children) == 2 + complete(children[0], "FAILURE") + finalize.assert_not_called() + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.STARTED + task_postrun_handler( + task=consume_file, + task_id=children[1].task_id, + state="SUCCESS", + retval={"document_id": 100}, + ) + finalize.assert_called_once() + finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE) + finish_file_operation.apply( + kwargs=finalizer.input_data, + task_id=finalizer.task_id, + throw=True, + ) + parent.refresh_from_db() + assert parent.status == PaperlessTask.Status.FAILURE + doc.refresh_from_db() + assert not doc.is_deleted