expected_str = "Error deleting pages from document"
self.assertIn(expected_str, error_str)
mock_update_archive_file.assert_not_called()
+
+ @mock.patch("documents.bulk_edit.group")
+ @mock.patch("documents.tasks.consume_file.s")
+ def test_edit_pdf_basic_operations(self, mock_consume_file, mock_group):
+ """
+ GIVEN:
+ - Existing document
+ WHEN:
+ - edit_pdf is called with two operations to split the doc and rotate pages
+ THEN:
+ - A grouped task is generated and delay() is called
+ """
+ mock_group.return_value.delay.return_value = None
+ doc_ids = [self.doc2.id]
+ operations = [{"page": 1, "doc": 0}, {"page": 2, "doc": 1, "rotate": 90}]
+
+ result = bulk_edit.edit_pdf(doc_ids, operations)
+ self.assertEqual(result, "OK")
+ mock_group.return_value.delay.assert_called_once()
+
+ @mock.patch("documents.bulk_edit.group")
+ @mock.patch("documents.tasks.consume_file.s")
+ def test_edit_pdf_with_user_override(self, mock_consume_file, mock_group):
+ """
+ GIVEN:
+ - Existing document
+ WHEN:
+ - edit_pdf is called with user override
+ THEN:
+ - Task is created with user context
+ """
+ mock_group.return_value.delay.return_value = None
+ doc_ids = [self.doc2.id]
+ operations = [{"page": 1, "doc": 0}, {"page": 2, "doc": 1}]
+ user = User.objects.create(username="editor")
+
+ result = bulk_edit.edit_pdf(doc_ids, operations, user=user)
+ self.assertEqual(result, "OK")
+ mock_group.return_value.delay.assert_called_once()
+
+ @mock.patch("documents.bulk_edit.chord")
+ @mock.patch("documents.tasks.consume_file.s")
+ def test_edit_pdf_with_delete_original(self, mock_consume_file, mock_chord):
+ """
+ GIVEN:
+ - Existing document
+ WHEN:
+ - edit_pdf is called with delete_original=True
+ THEN:
+ - Task group is triggered
+ """
+ mock_chord.return_value.delay.return_value = None
+ doc_ids = [self.doc2.id]
+ operations = [{"page": 1}, {"page": 2}]
+
+ result = bulk_edit.edit_pdf(doc_ids, operations, delete_original=True)
+ self.assertEqual(result, "OK")
+ mock_chord.assert_called_once()
+
+ @mock.patch("documents.tasks.update_document_content_maybe_archive_file.delay")
+ def test_edit_pdf_with_update_document(self, mock_update_document):
+ """
+ GIVEN:
+ - A single existing PDF document
+ WHEN:
+ - edit_pdf is called with update_document=True and a single output
+ THEN:
+ - The original document is updated in-place
+ - The update_document_content_maybe_archive_file task is triggered
+ """
+ doc_ids = [self.doc2.id]
+ operations = [{"page": 1}, {"page": 2}]
+ original_checksum = self.doc2.checksum
+ original_page_count = self.doc2.page_count
+
+ result = bulk_edit.edit_pdf(
+ doc_ids,
+ operations=operations,
+ update_document=True,
+ delete_original=False,
+ )
+
+ self.assertEqual(result, "OK")
+ self.doc2.refresh_from_db()
+ self.assertNotEqual(self.doc2.checksum, original_checksum)
+ self.assertNotEqual(self.doc2.page_count, original_page_count)
+ mock_update_document.assert_called_once_with(document_id=self.doc2.id)
+
+ @mock.patch("documents.bulk_edit.group")
+ @mock.patch("documents.tasks.consume_file.s")
+ def test_edit_pdf_without_metadata(self, mock_consume_file, mock_group):
+ """
+ GIVEN:
+ - Existing document
+ WHEN:
+ - edit_pdf is called with include_metadata=False
+ THEN:
+ - Tasks are created with empty metadata
+ """
+ mock_group.return_value.delay.return_value = None
+ doc_ids = [self.doc2.id]
+ operations = [{"page": 1}]
+
+ result = bulk_edit.edit_pdf(doc_ids, operations, include_metadata=False)
+ self.assertEqual(result, "OK")
+ mock_group.return_value.delay.assert_called_once()
+
+ @mock.patch("documents.bulk_edit.group")
+ @mock.patch("documents.tasks.consume_file.s")
+ def test_edit_pdf_open_failure(self, mock_consume_file, mock_group):
+ """
+ GIVEN:
+ - Existing document
+ WHEN:
+ - edit_pdf fails to open PDF
+ THEN:
+ - Task group is not called
+ """
+ doc_ids = [self.doc2.id]
+ operations = [
+ {"page": 9999}, # invalid page, forces error during PDF load
+ ]
+ with self.assertLogs("paperless.bulk_edit", level="ERROR"):
+ result = bulk_edit.edit_pdf(doc_ids, operations)
+ self.assertEqual(result, "ERROR")
+ mock_group.assert_not_called()
+ mock_consume_file.assert_not_called()
+
+ @mock.patch("documents.bulk_edit.group")
+ @mock.patch("documents.tasks.consume_file.s")
+ def test_edit_pdf_multiple_outputs_with_update_flag_errors(
+ self,
+ mock_consume_file,
+ mock_group,
+ ):
+ """
+ GIVEN:
+ - Existing document
+ WHEN:
+ - edit_pdf is called with multiple outputs and update_document=True
+ THEN:
+ - An error is logged and task group is not called
+ """
+ doc_ids = [self.doc2.id]
+ operations = [
+ {"page": 1, "doc": 0},
+ {"page": 2, "doc": 1},
+ ]
+ with self.assertLogs("paperless.bulk_edit", level="ERROR"):
+ result = bulk_edit.edit_pdf(doc_ids, operations, update_document=True)
+ self.assertEqual(result, "ERROR")
+ mock_group.assert_not_called()
+ mock_consume_file.assert_not_called()