diff --git a/src/documents/tests/test_bulk_edit.py b/src/documents/tests/test_bulk_edit.py index 245b56ad3..8af641ed4 100644 --- a/src/documents/tests/test_bulk_edit.py +++ b/src/documents/tests/test_bulk_edit.py @@ -909,3 +909,156 @@ class TestPDFActions(DirectoriesMixin, TestCase): 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()