import shutil from datetime import date from pathlib import Path from unittest import mock import pikepdf from django.contrib.auth.models import Group from django.contrib.auth.models import Permission from django.contrib.auth.models import User from django.test import TestCase from guardian.shortcuts import assign_perm from guardian.shortcuts import get_groups_with_perms from guardian.shortcuts import get_users_with_perms from documents import bulk_edit from documents.models import Correspondent from documents.models import CustomField from documents.models import CustomFieldInstance from documents.models import Document from documents.models import DocumentType from documents.models import StoragePath from documents.models import Tag from documents.permissions import set_permissions_for_objects from documents.tests.utils import DirectoriesMixin class TestBulkEdit(DirectoriesMixin, TestCase): def setUp(self) -> None: super().setUp() self.owner = User.objects.create(username="test_owner") self.user1 = User.objects.create(username="user1") self.user2 = User.objects.create(username="user2") self.group1 = Group.objects.create(name="group1") self.group2 = Group.objects.create(name="group2") patcher = mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") self.async_task = patcher.start() self.addCleanup(patcher.stop) self.c1 = Correspondent.objects.create(name="c1") self.c2 = Correspondent.objects.create(name="c2") self.dt1 = DocumentType.objects.create(name="dt1") self.dt2 = DocumentType.objects.create(name="dt2") self.t1 = Tag.objects.create(name="t1") self.t2 = Tag.objects.create(name="t2") self.doc1 = Document.objects.create( checksum="A", title="A", created=date(2023, 1, 1), ) self.doc2 = Document.objects.create( checksum="B", title="B", correspondent=self.c1, document_type=self.dt1, created=date(2023, 1, 2), ) self.doc3 = Document.objects.create( checksum="C", title="C", correspondent=self.c2, document_type=self.dt2, created=date(2023, 1, 3), ) self.doc4 = Document.objects.create(checksum="D", title="D") self.doc5 = Document.objects.create(checksum="E", title="E") self.doc2.tags.add(self.t1) self.doc3.tags.add(self.t2) self.doc4.tags.add(self.t1, self.t2) self.sp1 = StoragePath.objects.create(name="sp1", path="Something/{checksum}") def test_set_correspondent(self) -> None: self.assertEqual(Document.objects.filter(correspondent=self.c2).count(), 1) bulk_edit.set_correspondent( [self.doc1.id, self.doc2.id, self.doc3.id], self.c2.id, ) self.assertEqual(Document.objects.filter(correspondent=self.c2).count(), 3) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id, self.doc2.id]) def test_unset_correspondent(self) -> None: self.assertEqual(Document.objects.filter(correspondent=self.c2).count(), 1) bulk_edit.set_correspondent([self.doc1.id, self.doc2.id, self.doc3.id], None) self.assertEqual(Document.objects.filter(correspondent=self.c2).count(), 0) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc2.id, self.doc3.id]) def test_set_document_type(self) -> None: self.assertEqual(Document.objects.filter(document_type=self.dt2).count(), 1) bulk_edit.set_document_type( [self.doc1.id, self.doc2.id, self.doc3.id], self.dt2.id, ) self.assertEqual(Document.objects.filter(document_type=self.dt2).count(), 3) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id, self.doc2.id]) def test_unset_document_type(self) -> None: self.assertEqual(Document.objects.filter(document_type=self.dt2).count(), 1) bulk_edit.set_document_type([self.doc1.id, self.doc2.id, self.doc3.id], None) self.assertEqual(Document.objects.filter(document_type=self.dt2).count(), 0) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc2.id, self.doc3.id]) def test_set_document_storage_path(self) -> None: """ GIVEN: - 5 documents without defined storage path WHEN: - Bulk edit called to add storage path to 1 document THEN: - Single document storage path update """ self.assertEqual(Document.objects.filter(storage_path=None).count(), 5) bulk_edit.set_storage_path( [self.doc1.id], self.sp1.id, ) self.assertEqual(Document.objects.filter(storage_path=None).count(), 4) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id]) def test_unset_document_storage_path(self) -> None: """ GIVEN: - 4 documents without defined storage path - 1 document with a defined storage WHEN: - Bulk edit called to remove storage path from 1 document THEN: - Single document storage path removed """ self.assertEqual(Document.objects.filter(storage_path=None).count(), 5) bulk_edit.set_storage_path( [self.doc1.id], self.sp1.id, ) self.assertEqual(Document.objects.filter(storage_path=None).count(), 4) bulk_edit.set_storage_path( [self.doc1.id], None, ) self.assertEqual(Document.objects.filter(storage_path=None).count(), 5) self.async_task.assert_called() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id]) def test_add_tag(self) -> None: self.assertEqual(Document.objects.filter(tags__id=self.t1.id).count(), 2) bulk_edit.add_tag( [self.doc1.id, self.doc2.id, self.doc3.id, self.doc4.id], self.t1.id, ) self.assertEqual(Document.objects.filter(tags__id=self.t1.id).count(), 4) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id, self.doc3.id]) def test_remove_tag(self) -> None: self.assertEqual(Document.objects.filter(tags__id=self.t1.id).count(), 2) bulk_edit.remove_tag([self.doc1.id, self.doc3.id, self.doc4.id], self.t1.id) self.assertEqual(Document.objects.filter(tags__id=self.t1.id).count(), 1) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc4.id]) def test_modify_tags(self) -> None: tag_unrelated = Tag.objects.create(name="unrelated") self.doc2.tags.add(tag_unrelated) self.doc3.tags.add(tag_unrelated) bulk_edit.modify_tags( [self.doc2.id, self.doc3.id], add_tags=[self.t2.id], remove_tags=[self.t1.id], ) self.assertCountEqual(list(self.doc2.tags.all()), [self.t2, tag_unrelated]) self.assertCountEqual(list(self.doc3.tags.all()), [self.t2, tag_unrelated]) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] # TODO: doc3 should not be affected, but the query for that is rather complicated self.assertCountEqual(kwargs["document_ids"], [self.doc2.id, self.doc3.id]) def test_modify_custom_fields(self) -> None: """ GIVEN: - 2 documents with custom fields - 3 custom fields WHEN: - Custom fields are modified using old format (list of ids) THEN: - Custom fields are modified for the documents """ cf = CustomField.objects.create( name="cf1", data_type=CustomField.FieldDataType.STRING, ) cf2 = CustomField.objects.create( name="cf2", data_type=CustomField.FieldDataType.INT, ) cf3 = CustomField.objects.create( name="cf3", data_type=CustomField.FieldDataType.STRING, ) CustomFieldInstance.objects.create( document=self.doc1, field=cf, ) CustomFieldInstance.objects.create( document=self.doc2, field=cf, ) CustomFieldInstance.objects.create( document=self.doc2, field=cf3, ) bulk_edit.modify_custom_fields( [self.doc1.id, self.doc2.id], add_custom_fields=[cf2.id], remove_custom_fields=[cf.id], ) self.doc1.refresh_from_db() self.doc2.refresh_from_db() self.assertEqual( self.doc1.custom_fields.count(), 1, ) self.assertEqual( self.doc2.custom_fields.count(), 2, ) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id, self.doc2.id]) def test_modify_custom_fields_with_values(self) -> None: """ GIVEN: - 2 documents with custom fields - 3 custom fields WHEN: - Custom fields are modified using new format (dict) THEN: - Custom fields are modified for the documents """ cf = CustomField.objects.create( name="cf", data_type=CustomField.FieldDataType.STRING, ) cf1 = CustomField.objects.create( name="cf1", data_type=CustomField.FieldDataType.STRING, ) cf2 = CustomField.objects.create( name="cf2", data_type=CustomField.FieldDataType.MONETARY, ) cf3 = CustomField.objects.create( name="cf3", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) CustomFieldInstance.objects.create( document=self.doc2, field=cf, ) CustomFieldInstance.objects.create( document=self.doc2, field=cf1, ) CustomFieldInstance.objects.create( document=self.doc2, field=cf3, ) bulk_edit.modify_custom_fields( [self.doc1.id, self.doc2.id], add_custom_fields={cf2.id: None, cf3.id: [self.doc3.id]}, remove_custom_fields=[cf.id], ) self.doc1.refresh_from_db() self.doc2.refresh_from_db() self.assertEqual( self.doc1.custom_fields.count(), 2, ) self.assertEqual( self.doc1.custom_fields.get(field=cf2).value, None, ) self.assertEqual( self.doc1.custom_fields.get(field=cf3).value, [self.doc3.id], ) self.assertEqual( self.doc2.custom_fields.count(), 3, ) self.assertEqual( self.doc2.custom_fields.get(field=cf3).value, [self.doc3.id], ) # assert reflect document link _cf_1 = self.doc3.custom_fields.first() assert _cf_1 is not None self.assertEqual( _cf_1.value, [self.doc2.id, self.doc1.id], ) self.async_task.assert_called_once() kwargs = self.async_task.call_args.kwargs["kwargs"] self.assertCountEqual(kwargs["document_ids"], [self.doc1.id, self.doc2.id]) # removal of document link cf, should also remove symmetric link bulk_edit.modify_custom_fields( [self.doc3.id], add_custom_fields={}, remove_custom_fields=[cf3.id], ) _cf_2 = self.doc1.custom_fields.filter(field=cf3).first() assert _cf_2 is not None self.assertNotIn(self.doc3.id, _cf_2.value) _cf_3 = self.doc2.custom_fields.filter(field=cf3).first() assert _cf_3 is not None self.assertNotIn(self.doc3.id, _cf_3.value) def test_modify_custom_fields_doclink_self_link(self) -> None: """ GIVEN: - 2 existing documents - Existing doc link custom field WHEN: - Doc link field is modified to include self link THEN: - Self link should not be created """ cf = CustomField.objects.create( name="cf", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) bulk_edit.modify_custom_fields( [self.doc1.id, self.doc2.id], add_custom_fields={cf.id: [self.doc1.id]}, remove_custom_fields=[], ) _cf_4 = self.doc1.custom_fields.first() assert _cf_4 is not None self.assertEqual(_cf_4.value, [self.doc2.id]) _cf_5 = self.doc2.custom_fields.first() assert _cf_5 is not None self.assertEqual(_cf_5.value, [self.doc1.id]) def test_delete(self) -> None: self.assertEqual(Document.objects.count(), 5) bulk_edit.delete([self.doc1.id, self.doc2.id]) self.assertEqual(Document.objects.count(), 3) self.assertCountEqual( [doc.id for doc in Document.objects.all()], [self.doc3.id, self.doc4.id, self.doc5.id], ) def test_delete_root_document_deletes_all_versions(self) -> None: version = Document.objects.create( checksum="A-v1", title="A version", root_document=self.doc1, ) bulk_edit.delete([self.doc1.id]) self.assertFalse(Document.objects.filter(id=self.doc1.id).exists()) self.assertFalse(Document.objects.filter(id=version.id).exists()) def test_delete_version_document_keeps_root(self) -> None: version = Document.objects.create( checksum="A-v1", title="A version", root_document=self.doc1, ) bulk_edit.delete([version.id]) self.assertTrue(Document.objects.filter(id=self.doc1.id).exists()) self.assertFalse(Document.objects.filter(id=version.id).exists()) def test_resolve_root_and_source_doc_latest_version_prefers_newest_version( self, ) -> None: version1 = Document.objects.create( checksum="B-v1", title="B version 1", root_document=self.doc2, ) version2 = Document.objects.create( checksum="B-v2", title="B version 2", root_document=self.doc2, ) root_doc, source_doc = bulk_edit._resolve_root_and_source_doc( self.doc2, source_mode="latest_version", ) self.assertEqual(root_doc.id, self.doc2.id) self.assertEqual(source_doc.id, version2.id) self.assertNotEqual(source_doc.id, version1.id) @mock.patch("documents.tasks.bulk_update_documents.apply_async") def test_set_permissions(self, m) -> None: doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id] assign_perm("view_document", self.group1, self.doc1) permissions = { "view": { "users": [self.user1.id, self.user2.id], "groups": [self.group2.id], }, "change": { "users": [self.user1.id], "groups": [self.group2.id], }, } bulk_edit.set_permissions( doc_ids, set_permissions=permissions, owner=self.owner, merge=False, ) m.assert_called_once() self.assertEqual(Document.objects.filter(owner=self.owner).count(), 3) self.assertEqual(Document.objects.filter(id__in=doc_ids).count(), 3) users_with_perms = get_users_with_perms( self.doc1, ) self.assertEqual(users_with_perms.count(), 2) # group1 should be replaced by group2 groups_with_perms = get_groups_with_perms( self.doc1, ) self.assertEqual(groups_with_perms.count(), 1) @mock.patch("documents.tasks.bulk_update_documents.apply_async") def test_set_permissions_merge(self, m) -> None: doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id] self.doc1.owner = self.user1 self.doc1.save() assign_perm("view_document", self.user1, self.doc1) assign_perm("view_document", self.group1, self.doc1) permissions = { "view": { "users": [self.user2.id], "groups": [self.group2.id], }, "change": { "users": [self.user2.id], "groups": [self.group2.id], }, } bulk_edit.set_permissions( doc_ids, set_permissions=permissions, owner=self.owner, merge=True, ) m.assert_called_once() # when merge is true owner doesn't get replaced if its not empty self.assertEqual(Document.objects.filter(owner=self.owner).count(), 2) self.assertEqual(Document.objects.filter(id__in=doc_ids).count(), 3) # merge of user1 which was pre-existing and user2 users_with_perms = get_users_with_perms( self.doc1, ) self.assertEqual(users_with_perms.count(), 2) # group1 should be merged by group2 groups_with_perms = get_groups_with_perms( self.doc1, ) self.assertEqual(groups_with_perms.count(), 2) @mock.patch("documents.tasks.bulk_update_documents.apply_async") def test_set_permissions_batched_across_document_count( self, m, ) -> None: """ GIVEN: - Many documents are being bulk-edited to set permissions at once WHEN: - set_permissions runs over a small batch vs. a much larger one THEN: - Permissions are applied correctly at both scales """ permissions = { "view": { "users": [self.user1.id, self.user2.id], "groups": [self.group2.id], }, "change": { "users": [self.user1.id], "groups": [self.group2.id], }, } def run_with_n_documents(n: int) -> None: docs = [ Document.objects.create(checksum=f"perm-{n}-{i}", title=f"perm-{n}-{i}") for i in range(n) ] bulk_edit.set_permissions( [doc.id for doc in docs], set_permissions=permissions, owner=self.owner, merge=False, ) for doc in docs: self.assertEqual(get_users_with_perms(doc).count(), 2) self.assertEqual(get_groups_with_perms(doc).count(), 1) run_with_n_documents(5) run_with_n_documents(50) @mock.patch("documents.tasks.bulk_update_documents.apply_async") def test_set_permissions_grants_direct_perm_even_if_already_granted_via_group( self, m, ) -> None: """ GIVEN: - A user already has view access to a document via group membership, with no direct grant of their own WHEN: - set_permissions explicitly grants that same user direct view access via bulk_edit THEN: - A direct permission grant is created for the user, not skipped because they already have equivalent access via the group Regression test: guardian's queryset-aware assign_perm() (routed to when the target is a list/queryset) skips creating a direct row for anyone whose ObjectPermissionChecker.has_perm() already returns True -- which includes group-derived access. The single-object assign_perm this bulk path replaces has no such check; it always ensures a direct row via get_or_create. Losing that guarantee would mean revoking the group's grant later silently strips access that was supposed to be explicit. """ self.doc1.owner = self.user1 self.doc1.save() self.user1.groups.add(self.group1) assign_perm("view_document", self.group1, self.doc1) bulk_edit.set_permissions( [self.doc1.id], set_permissions={ "view": {"users": [self.user1.id], "groups": []}, }, merge=True, ) direct_users = get_users_with_perms( self.doc1, only_with_perms_in=["view_document"], with_group_users=False, ) self.assertIn(self.user1, direct_users) def test_set_permissions_for_objects_raises_for_unknown_action(self) -> None: """ GIVEN: - An unrecognized permission action name with users to grant it to WHEN: - set_permissions_for_objects is called THEN: - Permission.DoesNotExist is raised, not a silent no-op Regression test: the endpoint that calls this (BulkEditObjectPermissionsView) never actually validates action names against the raw client-supplied permissions dict -- BulkEditObjectsSerializer._validate_permissions calls validate_set_permissions() only for its side-effecting user/group id checks and discards the filtered dict it returns -- so a bogus action key reaches this function as-is. Resolving the Permission via a bare `.filter()` (which returns empty instead of raising) would silently drop the grant and report success. """ with self.assertRaises(Permission.DoesNotExist): set_permissions_for_objects( {"not_a_real_action": {"users": [self.user1.id], "groups": []}}, Document, [self.doc1.pk], ) @mock.patch("documents.models.Document.delete") def test_delete_documents_old_uuid_field(self, m) -> None: m.side_effect = Exception("Data too long for column 'transaction_id' at row 1") doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id] bulk_edit.delete(doc_ids) with self.assertLogs(level="WARNING") as cm: bulk_edit.delete(doc_ids) self.assertIn("possible incompatible database column", cm.output[0]) class TestPDFActions(DirectoriesMixin, TestCase): def setUp(self) -> None: super().setUp() sample1 = self.dirs.scratch_dir / "sample.pdf" shutil.copy( Path(__file__).parent / "samples" / "documents" / "originals" / "0000001.pdf", sample1, ) sample1_archive = self.dirs.archive_dir / "sample_archive.pdf" shutil.copy( Path(__file__).parent / "samples" / "documents" / "originals" / "0000001.pdf", sample1_archive, ) sample2 = self.dirs.scratch_dir / "sample2.pdf" shutil.copy( Path(__file__).parent / "samples" / "documents" / "originals" / "0000002.pdf", sample2, ) sample2_archive = self.dirs.archive_dir / "sample2_archive.pdf" shutil.copy( Path(__file__).parent / "samples" / "documents" / "originals" / "0000002.pdf", sample2_archive, ) sample3 = self.dirs.scratch_dir / "sample3.pdf" shutil.copy( Path(__file__).parent / "samples" / "documents" / "originals" / "0000003.pdf", sample3, ) self.doc1 = Document.objects.create( checksum="A", title="A", filename=sample1, mime_type="application/pdf", ) self.doc1.archive_filename = sample1_archive self.doc1.save() self.doc2 = Document.objects.create( checksum="B", title="B", filename=sample2, mime_type="application/pdf", page_count=8, created=date(2023, 1, 2), ) self.doc2.archive_filename = sample2_archive self.doc2.save() self.doc3 = Document.objects.create( checksum="C", title="C", filename=sample3, mime_type="application/pdf", ) img_doc = self.dirs.scratch_dir / "sample_image.jpg" shutil.copy( Path(__file__).parent / "samples" / "simple.jpg", img_doc, ) img_doc_archive = self.dirs.archive_dir / "sample_image.pdf" shutil.copy( Path(__file__).parent / "samples" / "documents" / "originals" / "0000001.pdf", img_doc_archive, ) self.img_doc = Document.objects.create( checksum="D", title="D", filename=img_doc, mime_type="image/jpeg", created=date(2023, 1, 3), ) self.img_doc.archive_filename = img_doc_archive self.img_doc.save() @staticmethod def mock_password_required_pdf( mock_open: mock.Mock, fake_pdf: mock.Mock, ) -> None: password_context = mock.MagicMock() password_context.__enter__.return_value = fake_pdf mock_open.side_effect = [ pikepdf.PasswordError("password required"), password_context, ] @mock.patch("documents.tasks.consume_file.s") def test_merge(self, mock_consume_file) -> None: """ GIVEN: - Existing documents WHEN: - Merge action is called with 3 documents THEN: - Consume file should be called """ doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id] metadata_document_id = self.doc2.id user = User.objects.create(username="test_user") result = bulk_edit.merge( doc_ids, metadata_document_id=None, delete_originals=False, user=user, ) expected_filename = ( f"{'_'.join([str(doc_id) for doc_id in doc_ids])[:100]}_merged.pdf" ) mock_consume_file.assert_called() call_kwargs = mock_consume_file.call_args.kwargs self.assertEqual( Path(call_kwargs["input_doc"].original_file).name, expected_filename, ) self.assertEqual(call_kwargs["overrides"].title, None) # No metadata_document_id, delete_originals False, so ASN should be None self.assertIsNone(call_kwargs["overrides"].asn) # With metadata_document_id overrides result = bulk_edit.merge(doc_ids, metadata_document_id=metadata_document_id) call_kwargs = mock_consume_file.call_args.kwargs self.assertEqual(call_kwargs["overrides"].title, "B (merged)") self.assertEqual(call_kwargs["overrides"].created, self.doc2.created) self.assertEqual(result, "OK") @mock.patch("pikepdf.open") @mock.patch("documents.tasks.consume_file.s") def test_merge_uses_latest_version_source_for_root_selection( self, mock_consume_file, mock_open_pdf, ) -> None: version_file = self.dirs.scratch_dir / "sample2_version_merge.pdf" shutil.copy(self.doc2.source_path, version_file) version = Document.objects.create( checksum="B-v1", title="B version 1", root_document=self.doc2, filename=version_file, mime_type="application/pdf", ) fake_pdf = mock.MagicMock() fake_pdf.pdf_version = "1.7" fake_pdf.pages = [mock.Mock()] mock_open_pdf.return_value.__enter__.return_value = fake_pdf result = bulk_edit.merge([self.doc2.id]) self.assertEqual(result, "OK") mock_open_pdf.assert_called_once_with(str(version.source_path)) mock_consume_file.assert_not_called() @mock.patch("documents.bulk_edit.delete.si") @mock.patch("documents.tasks.consume_file.s") def test_merge_and_delete_originals( self, mock_consume_file, mock_delete_documents, ) -> None: """ GIVEN: - Existing documents WHEN: - Merge action with deleting documents is called with 3 documents THEN: - Consume file task should be called - Document deletion task should be called """ doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id] self.doc1.archive_serial_number = 101 self.doc2.archive_serial_number = 102 self.doc3.archive_serial_number = 103 self.doc1.save() self.doc2.save() self.doc3.save() result = bulk_edit.merge(doc_ids, delete_originals=True) self.assertEqual(result, "OK") expected_filename = ( f"{'_'.join([str(doc_id) for doc_id in doc_ids])[:100]}_merged.pdf" ) mock_consume_file.assert_called() mock_delete_documents.assert_called() mock_consume_file.return_value.set.return_value.apply_async.assert_called_once() call_kwargs = mock_consume_file.call_args.kwargs self.assertEqual( Path(call_kwargs["input_doc"].original_file).name, expected_filename, ) self.assertEqual(call_kwargs["overrides"].title, None) self.assertEqual(call_kwargs["overrides"].asn, 101) delete_documents_args, _ = mock_delete_documents.call_args self.assertEqual( delete_documents_args[0], doc_ids, ) self.doc1.refresh_from_db() self.doc2.refresh_from_db() self.doc3.refresh_from_db() self.assertIsNone(self.doc1.archive_serial_number) self.assertIsNone(self.doc2.archive_serial_number) self.assertIsNone(self.doc3.archive_serial_number) @mock.patch("documents.bulk_edit.delete.si") @mock.patch("documents.tasks.consume_file.s") def test_merge_and_delete_originals_restore_on_failure( self, mock_consume_file, mock_delete_documents, ) -> None: """ GIVEN: - Existing documents WHEN: - Merge action with deleting documents is called with 1 document - Error occurs when queuing consume file task THEN: - Archive serial numbers are restored """ doc_ids = [self.doc1.id] self.doc1.archive_serial_number = 111 self.doc1.save() sig = mock.Mock() sig.set.return_value.apply_async.side_effect = Exception("boom") mock_consume_file.return_value = sig with self.assertRaises(Exception): bulk_edit.merge(doc_ids, delete_originals=True) self.doc1.refresh_from_db() self.assertEqual(self.doc1.archive_serial_number, 111) @mock.patch("documents.bulk_edit.delete.si") @mock.patch("documents.tasks.consume_file.s") def test_merge_and_delete_originals_metadata_handoff( self, mock_consume_file, mock_delete_documents, ) -> None: """ GIVEN: - Existing documents with ASNs WHEN: - Merge with delete_originals=True and metadata_document_id set THEN: - Handoff ASN uses metadata document ASN """ doc_ids = [self.doc1.id, self.doc2.id] self.doc1.archive_serial_number = 101 self.doc2.archive_serial_number = 202 self.doc1.save() self.doc2.save() result = bulk_edit.merge( doc_ids, metadata_document_id=self.doc2.id, delete_originals=True, ) self.assertEqual(result, "OK") self.assertEqual(mock_consume_file.call_args.kwargs["overrides"].asn, 202) def test_restore_archive_serial_numbers_task(self) -> None: """ GIVEN: - Existing document with no archive serial number WHEN: - Restore archive serial number task is called with backup data THEN: - Document archive serial number is restored """ self.doc1.archive_serial_number = 444 self.doc1.save() Document.objects.filter(pk=self.doc1.id).update(archive_serial_number=None) backup: dict[int, int | None] = {self.doc1.id: 444} bulk_edit.restore_archive_serial_numbers_task(backup) self.doc1.refresh_from_db() self.assertEqual(self.doc1.archive_serial_number, 444) @mock.patch("documents.tasks.consume_file.s") def test_merge_with_archive_fallback(self, mock_consume_file) -> None: """ GIVEN: - Existing documents WHEN: - Merge action is called with 2 documents, one of which is an image and archive_fallback is set to True THEN: - Image document should be included """ doc_ids = [self.doc2.id, self.img_doc.id] result = bulk_edit.merge(doc_ids, archive_fallback=True) self.assertEqual(result, "OK") expected_filename = ( f"{'_'.join([str(doc_id) for doc_id in doc_ids])[:100]}_merged.pdf" ) mock_consume_file.assert_called() self.assertEqual( Path(mock_consume_file.call_args.kwargs["input_doc"].original_file).name, expected_filename, ) @mock.patch("documents.tasks.consume_file.delay") @mock.patch("pikepdf.open") def test_merge_with_errors(self, mock_open_pdf, mock_consume_file) -> None: """ GIVEN: - Existing documents WHEN: - Merge action is called with 2 documents - Error occurs when opening both files THEN: - Consume file should not be called """ mock_open_pdf.side_effect = Exception("Error opening PDF") doc_ids = [self.doc2.id, self.doc3.id] with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm: bulk_edit.merge(doc_ids) error_str = cm.output[0] expected_str = ( "Error merging document 2, it will not be included in the merge" ) self.assertIn(expected_str, error_str) mock_consume_file.assert_not_called() @mock.patch("documents.tasks.consume_file.s") def test_split(self, mock_consume_file) -> None: """ GIVEN: - Existing documents WHEN: - Split action is called with 1 document and 2 pages THEN: - Consume file should be called twice """ doc_ids = [self.doc2.id] pages = [[1, 2], [3]] user = User.objects.create(username="test_user") result = bulk_edit.split(doc_ids, pages, delete_originals=False, user=user) self.assertEqual(mock_consume_file.call_count, 2) self.assertEqual( mock_consume_file.call_args.kwargs["overrides"].title, "B (split 2)", ) self.assertIsNone(mock_consume_file.call_args.kwargs["overrides"].asn) self.assertEqual(result, "OK") @mock.patch("documents.bulk_edit.group") @mock.patch("pikepdf.open") @mock.patch("documents.tasks.consume_file.s") def test_split_uses_latest_version_source_for_root_selection( self, mock_consume_file, mock_open_pdf, mock_group, ) -> None: version_file = self.dirs.scratch_dir / "sample2_version_split.pdf" shutil.copy(self.doc2.source_path, version_file) version = Document.objects.create( checksum="B-v1", title="B version 1", root_document=self.doc2, filename=version_file, mime_type="application/pdf", ) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock(), mock.Mock()] mock_open_pdf.return_value.__enter__.return_value = fake_pdf mock_group.return_value.delay.return_value = None result = bulk_edit.split([self.doc2.id], [[1], [2]]) self.assertEqual(result, "OK") mock_open_pdf.assert_called_once_with(version.source_path) mock_consume_file.assert_not_called() mock_group.return_value.delay.assert_not_called() @mock.patch("documents.bulk_edit.delete.si") @mock.patch("documents.tasks.consume_file.s") @mock.patch("documents.bulk_edit.chord") def test_split_and_delete_originals( self, mock_chord, mock_consume_file, mock_delete_documents, ) -> None: """ GIVEN: - Existing documents WHEN: - Split action with deleting documents is called with 1 document and 2 page groups - delete_originals is set to True THEN: - Consume file should be called twice - Document deletion task should be called """ doc_ids = [self.doc2.id] pages = [[1, 2], [3]] self.doc2.archive_serial_number = 200 self.doc2.save() errback = bulk_edit.restore_archive_serial_numbers_task.s( {self.doc2.id: 200}, ) mock_chord.return_value.on_error.return_value = mock_chord.return_value result = bulk_edit.split(doc_ids, pages, delete_originals=True) self.assertEqual(result, "OK") self.assertEqual(mock_consume_file.call_count, 2) self.assertEqual( mock_consume_file.call_args.kwargs["overrides"].title, "B (split 2)", ) mock_delete_documents.assert_called() mock_chord.assert_called_once() mock_chord.return_value.on_error.assert_called_once_with(errback) mock_chord.return_value.apply_async.assert_called_once_with() delete_documents_args, _ = mock_delete_documents.call_args self.assertEqual( delete_documents_args[0], doc_ids, ) self.doc2.refresh_from_db() self.assertIsNone(self.doc2.archive_serial_number) @mock.patch("documents.bulk_edit.delete.si") @mock.patch("documents.tasks.consume_file.s") @mock.patch("documents.bulk_edit.chord") def test_split_restore_on_failure( self, mock_chord, mock_consume_file, mock_delete_documents, ) -> None: """ GIVEN: - Existing documents WHEN: - Split action with deleting documents is called with 1 document and 2 page groups - Error occurs when queuing chord task THEN: - Archive serial numbers are restored """ doc_ids = [self.doc2.id] pages = [[1, 2]] self.doc2.archive_serial_number = 222 self.doc2.save() sig = mock.Mock() sig.on_error.return_value = sig sig.apply_async.side_effect = Exception("boom") mock_chord.return_value = sig result = bulk_edit.split(doc_ids, pages, delete_originals=True) self.assertEqual(result, "OK") self.doc2.refresh_from_db() self.assertEqual(self.doc2.archive_serial_number, 222) @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.Pdf.save") def test_split_with_errors(self, mock_save_pdf, mock_consume_file) -> None: """ GIVEN: - Existing documents WHEN: - Split action is called with 1 document and 2 page groups - Error occurs when saving the files THEN: - Consume file should not be called """ mock_save_pdf.side_effect = Exception("Error saving PDF") doc_ids = [self.doc2.id] pages = [[1, 2], [3]] with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm: bulk_edit.split(doc_ids, pages) error_str = cm.output[0] expected_str = "Error splitting document 2" self.assertIn(expected_str, error_str) mock_consume_file.assert_not_called() @mock.patch("documents.tasks.consume_file.apply_async") def test_rotate(self, mock_consume_delay) -> None: """ GIVEN: - Existing documents WHEN: - Rotate action is called with 2 documents THEN: - Rotate action should be called twice """ doc_ids = [self.doc1.id, self.doc2.id] result = bulk_edit.rotate(doc_ids, 90) self.assertEqual(mock_consume_delay.call_count, 2) for call, expected_id in zip( mock_consume_delay.call_args_list, doc_ids, ): task_kwargs = call.kwargs["kwargs"] self.assertEqual(task_kwargs["input_doc"].root_document_id, expected_id) self.assertIsNotNone(task_kwargs["overrides"]) self.assertEqual(result, "OK") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.Pdf.save") def test_rotate_with_error( self, mock_pdf_save, mock_consume_delay, ) -> None: """ GIVEN: - Existing documents WHEN: - Rotate action is called with 2 documents - PikePDF raises an error THEN: - Rotate action should be called 0 times """ mock_pdf_save.side_effect = Exception("Error saving PDF") doc_ids = [self.doc2.id, self.doc3.id] with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm: bulk_edit.rotate(doc_ids, 90) error_str = cm.output[0] expected_str = "Error rotating document" self.assertIn(expected_str, error_str) mock_consume_delay.assert_not_called() @mock.patch("documents.tasks.consume_file.apply_async") def test_rotate_non_pdf( self, mock_consume_delay, ) -> None: """ GIVEN: - Existing documents WHEN: - Rotate action is called with 2 documents, one of which is not a PDF THEN: - Rotate action should be performed 1 time, with the non-PDF document skipped """ with self.assertLogs("paperless.bulk_edit", level="INFO") as cm: result = bulk_edit.rotate([self.doc2.id, self.img_doc.id], 90) expected_str = f"Document {self.img_doc.id} is not a PDF, skipping rotation" self.assertTrue(any(expected_str in line for line in cm.output)) self.assertEqual(mock_consume_delay.call_count, 1) task_kwargs = mock_consume_delay.call_args.kwargs["kwargs"] self.assertEqual(task_kwargs["input_doc"].root_document_id, self.doc2.id) self.assertIsNotNone(task_kwargs["overrides"]) self.assertEqual(result, "OK") @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.open") def test_rotate_explicit_selection_uses_root_source_when_root_selected( self, mock_open, mock_consume_delay, mock_magic, ) -> None: Document.objects.create( checksum="B-v1", title="B version 1", root_document=self.doc2, ) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock()] mock_open.return_value.__enter__.return_value = fake_pdf result = bulk_edit.rotate( [self.doc2.id], 90, source_mode="explicit_selection", ) self.assertEqual(result, "OK") mock_open.assert_called_once_with(self.doc2.source_path) mock_consume_delay.assert_called_once() @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.Pdf.save") @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf") def test_delete_pages(self, mock_magic, mock_pdf_save, mock_consume_delay) -> None: """ GIVEN: - Existing documents WHEN: - Delete pages action is called with 1 document and 2 pages THEN: - Save should be called once - A new version should be enqueued via consume_file """ doc_ids = [self.doc2.id] pages = [1, 3] result = bulk_edit.delete_pages(doc_ids, pages) mock_pdf_save.assert_called_once() mock_consume_delay.assert_called_once() task_kwargs = mock_consume_delay.call_args.kwargs["kwargs"] self.assertEqual(task_kwargs["input_doc"].root_document_id, self.doc2.id) self.assertTrue( str(task_kwargs["input_doc"].original_file).endswith("_pages_deleted.pdf"), ) self.assertIsNotNone(task_kwargs["overrides"]) self.assertEqual(result, "OK") @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.open") def test_delete_pages_explicit_selection_uses_root_source_when_root_selected( self, mock_open, mock_consume_delay, mock_magic, ) -> None: Document.objects.create( checksum="B-v1", title="B version 1", root_document=self.doc2, ) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock(), mock.Mock()] mock_open.return_value.__enter__.return_value = fake_pdf result = bulk_edit.delete_pages( [self.doc2.id], [1], source_mode="explicit_selection", ) self.assertEqual(result, "OK") mock_open.assert_called_once_with(self.doc2.source_path) mock_consume_delay.assert_called_once() @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.Pdf.save") def test_delete_pages_with_error(self, mock_pdf_save, mock_consume_delay) -> None: """ GIVEN: - Existing documents WHEN: - Delete pages action is called with 1 document and 2 pages - PikePDF raises an error THEN: - Save should be called once - No new version should be enqueued """ mock_pdf_save.side_effect = Exception("Error saving PDF") doc_ids = [self.doc2.id] pages = [1, 3] with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm: bulk_edit.delete_pages(doc_ids, pages) error_str = cm.output[0] expected_str = "Error deleting pages from document" self.assertIn(expected_str, error_str) mock_consume_delay.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) -> None: """ 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) -> None: """ 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) -> None: """ 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}] self.doc2.archive_serial_number = 250 self.doc2.save() errback = bulk_edit.restore_archive_serial_numbers_task.s( {self.doc2.id: 250}, ) mock_chord.return_value.on_error.return_value = mock_chord.return_value result = bulk_edit.edit_pdf(doc_ids, operations, delete_original=True) self.assertEqual(result, "OK") mock_chord.assert_called_once() mock_chord.return_value.on_error.assert_called_once_with(errback) mock_chord.return_value.apply_async.assert_called_once_with() self.assertEqual(mock_consume_file.call_args.kwargs["overrides"].asn, 250) self.doc2.refresh_from_db() self.assertIsNone(self.doc2.archive_serial_number) @mock.patch("documents.bulk_edit.delete.si") @mock.patch("documents.tasks.consume_file.s") @mock.patch("documents.bulk_edit.chord") def test_edit_pdf_restore_on_failure( self, mock_chord: mock.Mock, mock_consume_file: mock.Mock, mock_delete_documents: mock.Mock, ) -> None: """ GIVEN: - Existing document WHEN: - edit_pdf is called with delete_original=True - Error occurs when queuing chord task THEN: - Archive serial numbers are restored """ doc_ids = [self.doc2.id] operations = [{"page": 1}] self.doc2.archive_serial_number = 333 self.doc2.save() sig = mock.Mock() sig.on_error.return_value = sig sig.apply_async.side_effect = Exception("boom") mock_chord.return_value = sig with self.assertRaises(Exception): bulk_edit.edit_pdf(doc_ids, operations, delete_original=True) self.doc2.refresh_from_db() self.assertEqual(self.doc2.archive_serial_number, 333) @mock.patch("documents.tasks.consume_file.apply_async") def test_edit_pdf_with_update_document(self, mock_consume_delay) -> None: """ GIVEN: - A single existing PDF document WHEN: - edit_pdf is called with update_document=True and a single output THEN: - A version update is enqueued targeting the existing document """ doc_ids = [self.doc2.id] operations = [{"page": 1}, {"page": 2}] result = bulk_edit.edit_pdf( doc_ids, operations=operations, update_document=True, delete_original=False, ) self.assertEqual(result, "OK") mock_consume_delay.assert_called_once() task_kwargs = mock_consume_delay.call_args.kwargs["kwargs"] self.assertEqual(task_kwargs["input_doc"].root_document_id, self.doc2.id) self.assertTrue( str(task_kwargs["input_doc"].original_file).endswith("_edited.pdf"), ) self.assertIsNotNone(task_kwargs["overrides"]) @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.new") @mock.patch("pikepdf.open") def test_edit_pdf_explicit_selection_uses_root_source_when_root_selected( self, mock_open, mock_new, mock_consume_delay, mock_magic, ) -> None: Document.objects.create( checksum="B-v1", title="B version 1", root_document=self.doc2, ) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock()] mock_open.return_value.__enter__.return_value = fake_pdf output_pdf = mock.MagicMock() output_pdf.pages = [] mock_new.return_value = output_pdf result = bulk_edit.edit_pdf( [self.doc2.id], operations=[{"page": 1}], update_document=True, source_mode="explicit_selection", ) self.assertEqual(result, "OK") mock_open.assert_called_once_with(self.doc2.source_path) mock_consume_delay.assert_called_once() @mock.patch("documents.bulk_edit.group") @mock.patch("documents.tasks.consume_file.s") def test_edit_pdf_without_metadata( self, mock_consume_file: mock.Mock, mock_group: mock.Mock, ) -> None: """ 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.Mock, mock_group: mock.Mock, ) -> None: """ 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"): with self.assertRaises(Exception): bulk_edit.edit_pdf(doc_ids, operations) 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, ) -> None: """ 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"): with self.assertRaises(ValueError): bulk_edit.edit_pdf(doc_ids, operations, update_document=True) mock_group.assert_not_called() mock_consume_file.assert_not_called() @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file.delay") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("documents.bulk_edit.tempfile.mkdtemp") @mock.patch("pikepdf.open") def test_remove_password_update_document( self, mock_open, mock_mkdtemp, mock_consume_delay, mock_update_document, ) -> None: doc = self.doc1 temp_dir = self.dirs.scratch_dir / "remove-password-update" temp_dir.mkdir(parents=True, exist_ok=True) mock_mkdtemp.return_value = str(temp_dir) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock(), mock.Mock(), mock.Mock()] fake_pdf.is_encrypted = True def save_side_effect(target_path): Path(target_path).write_bytes(b"new pdf content") fake_pdf.save.side_effect = save_side_effect mock_open.return_value.__enter__.return_value = fake_pdf result = bulk_edit.remove_password( [doc.id], password="secret", update_document=True, ) self.assertEqual(result, "OK") self.assertEqual( mock_open.call_args_list, [ mock.call(doc.source_path), mock.call(doc.source_path, password="secret"), ], ) fake_pdf.remove_unreferenced_resources.assert_called_once() mock_update_document.assert_not_called() mock_consume_delay.assert_called_once() task_kwargs = mock_consume_delay.call_args.kwargs["kwargs"] expected_path = temp_dir / f"{doc.id}_unprotected.pdf" self.assertTrue(expected_path.exists()) self.assertEqual( Path(task_kwargs["input_doc"].original_file).resolve(), expected_path.resolve(), ) self.assertEqual(task_kwargs["input_doc"].root_document_id, doc.id) self.assertIsNotNone(task_kwargs["overrides"]) @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("documents.bulk_edit.tempfile.mkdtemp") @mock.patch("pikepdf.open") def test_remove_password_update_document_skips_unencrypted_pdf( self, mock_open, mock_mkdtemp, mock_consume_delay, ) -> None: doc = self.doc1 fake_pdf = mock.MagicMock() fake_pdf.is_encrypted = False mock_open.return_value.__enter__.return_value = fake_pdf result = bulk_edit.remove_password( [doc.id], password="secret", update_document=True, ) self.assertEqual(result, "OK") mock_open.assert_called_once_with(doc.source_path) fake_pdf.remove_unreferenced_resources.assert_not_called() fake_pdf.save.assert_not_called() mock_mkdtemp.assert_not_called() mock_consume_delay.assert_not_called() @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file.delay") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("documents.bulk_edit.tempfile.mkdtemp") @mock.patch("pikepdf.open") def test_remove_password_update_document_uses_source_paths( self, mock_open, mock_mkdtemp, mock_consume_delay, mock_update_document, ) -> None: doc = self.doc1 source_file = self.dirs.scratch_dir / "consumption-source.pdf" source_file.write_bytes(b"protected pdf content") temp_dir = self.dirs.scratch_dir / "remove-password-source-file" temp_dir.mkdir(parents=True, exist_ok=True) mock_mkdtemp.return_value = str(temp_dir) fake_pdf = mock.MagicMock() self.mock_password_required_pdf(mock_open, fake_pdf) def save_side_effect(target_path): Path(target_path).write_bytes(b"new pdf content") fake_pdf.save.side_effect = save_side_effect result = bulk_edit.remove_password( [doc.id], password="secret", update_document=True, source_paths_by_id={doc.id: source_file}, ) self.assertEqual(result, "OK") self.assertEqual( mock_open.call_args_list, [ mock.call(source_file), mock.call(source_file, password="secret"), ], ) mock_update_document.assert_not_called() mock_consume_delay.assert_called_once() @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf") @mock.patch("documents.tasks.consume_file.apply_async") @mock.patch("pikepdf.open") def test_remove_password_explicit_selection_uses_root_source_when_root_selected( self, mock_open, mock_consume_delay, mock_magic, ) -> None: Document.objects.create( checksum="A-v1", title="A version 1", root_document=self.doc1, ) fake_pdf = mock.MagicMock() self.mock_password_required_pdf(mock_open, fake_pdf) result = bulk_edit.remove_password( [self.doc1.id], password="secret", update_document=True, source_mode="explicit_selection", ) self.assertEqual(result, "OK") self.assertEqual( mock_open.call_args_list, [ mock.call(self.doc1.source_path), mock.call(self.doc1.source_path, password="secret"), ], ) mock_consume_delay.assert_called_once() @mock.patch("documents.bulk_edit.chord") @mock.patch("documents.bulk_edit.group") @mock.patch("documents.tasks.consume_file.s") @mock.patch("documents.bulk_edit.tempfile.mkdtemp") @mock.patch("pikepdf.open") def test_remove_password_creates_consumable_document( self, mock_open: mock.Mock, mock_mkdtemp: mock.Mock, mock_consume_file: mock.Mock, mock_group: mock.Mock, mock_chord: mock.Mock, ) -> None: doc = self.doc2 temp_dir = self.dirs.scratch_dir / "remove-password" temp_dir.mkdir(parents=True, exist_ok=True) mock_mkdtemp.return_value = str(temp_dir) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock(), mock.Mock()] self.mock_password_required_pdf(mock_open, fake_pdf) def save_side_effect(target_path: Path) -> None: target_path.write_bytes(b"password removed") fake_pdf.save.side_effect = save_side_effect mock_group.return_value.delay.return_value = None user = User.objects.create(username="owner") result = bulk_edit.remove_password( [doc.id], password="secret", include_metadata=False, update_document=False, delete_original=False, user=user, ) self.assertEqual(result, "OK") self.assertEqual( mock_open.call_args_list, [ mock.call(doc.source_path), mock.call(doc.source_path, password="secret"), ], ) mock_consume_file.assert_called_once() call_kwargs = mock_consume_file.call_args.kwargs consumable_document = call_kwargs["input_doc"] overrides = call_kwargs["overrides"] expected_path = temp_dir / f"{doc.id}_unprotected.pdf" self.assertTrue(expected_path.exists()) self.assertEqual( Path(consumable_document.original_file).resolve(), expected_path.resolve(), ) self.assertEqual(overrides.owner_id, user.id) mock_group.assert_called_once_with( [mock_consume_file.return_value.set.return_value], ) mock_group.return_value.delay.assert_called_once() mock_chord.assert_not_called() @mock.patch("documents.bulk_edit.delete") @mock.patch("documents.bulk_edit.chord") @mock.patch("documents.bulk_edit.group") @mock.patch("documents.tasks.consume_file.s") @mock.patch("documents.bulk_edit.tempfile.mkdtemp") @mock.patch("pikepdf.open") def test_remove_password_skips_unencrypted_pdf_without_queueing( self, mock_open: mock.Mock, mock_mkdtemp: mock.Mock, mock_consume_file: mock.Mock, mock_group: mock.Mock, mock_chord: mock.Mock, mock_delete: mock.Mock, ) -> None: doc = self.doc2 fake_pdf = mock.MagicMock() fake_pdf.is_encrypted = False mock_open.return_value.__enter__.return_value = fake_pdf result = bulk_edit.remove_password( [doc.id], password="secret", update_document=False, delete_original=True, ) self.assertEqual(result, "OK") mock_open.assert_called_once_with(doc.source_path) fake_pdf.remove_unreferenced_resources.assert_not_called() fake_pdf.save.assert_not_called() mock_mkdtemp.assert_not_called() mock_consume_file.assert_not_called() mock_group.assert_not_called() mock_chord.assert_not_called() mock_delete.si.assert_not_called() @mock.patch("documents.bulk_edit.delete") @mock.patch("documents.bulk_edit.chord") @mock.patch("documents.bulk_edit.group") @mock.patch("documents.tasks.consume_file.s") @mock.patch("documents.bulk_edit.tempfile.mkdtemp") @mock.patch("pikepdf.open") def test_remove_password_deletes_original( self, mock_open: mock.Mock, mock_mkdtemp: mock.Mock, mock_consume_file: mock.Mock, mock_group: mock.Mock, mock_chord: mock.Mock, mock_delete: mock.Mock, ) -> None: doc = self.doc2 temp_dir = self.dirs.scratch_dir / "remove-password-delete" temp_dir.mkdir(parents=True, exist_ok=True) mock_mkdtemp.return_value = str(temp_dir) fake_pdf = mock.MagicMock() fake_pdf.pages = [mock.Mock(), mock.Mock()] self.mock_password_required_pdf(mock_open, fake_pdf) def save_side_effect(target_path: Path) -> None: target_path.write_bytes(b"password removed") fake_pdf.save.side_effect = save_side_effect mock_chord.return_value.delay.return_value = None result = bulk_edit.remove_password( [doc.id], password="secret", include_metadata=False, update_document=False, delete_original=True, ) self.assertEqual(result, "OK") self.assertEqual( mock_open.call_args_list, [ mock.call(doc.source_path), mock.call(doc.source_path, password="secret"), ], ) mock_consume_file.assert_called_once() mock_group.assert_not_called() mock_chord.assert_called_once() mock_chord.return_value.delay.assert_called_once() mock_delete.si.assert_called_once_with([doc.id]) @mock.patch("pikepdf.open") def test_remove_password_open_failure(self, mock_open: mock.Mock) -> None: mock_open.side_effect = RuntimeError("wrong password") with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm: with self.assertRaises(ValueError) as exc: bulk_edit.remove_password([self.doc1.id], password="secret") self.assertIn("wrong password", str(exc.exception)) self.assertIn("Error removing password from document", cm.output[0]) class TestBulkEditReprocess(DirectoriesMixin, TestCase): def setUp(self) -> None: super().setUp() self.doc = Document.objects.create( title="test", checksum="A", mime_type="application/pdf", ) @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file") def test_reprocess_defaults_to_local(self, mock_task: mock.Mock) -> None: """ GIVEN: - A reprocess request that says nothing about remote OCR WHEN: - reprocess is called THEN: - The task is queued without asking for the remote engine """ result = bulk_edit.reprocess([self.doc.id]) self.assertEqual(result, "OK") mock_task.apply_async.assert_called_once() _, kwargs = mock_task.apply_async.call_args self.assertEqual( kwargs["kwargs"], {"document_id": self.doc.id, "remote_ocr": False}, ) @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file") def test_reprocess_passes_remote_ocr(self, mock_task: mock.Mock) -> None: """ GIVEN: - A reprocess request that explicitly asks for remote OCR WHEN: - reprocess is called THEN: - The request is forwarded to the task for every document """ other = Document.objects.create( title="test2", checksum="B", mime_type="application/pdf", ) bulk_edit.reprocess([self.doc.id, other.id], remote_ocr=True) self.assertEqual(mock_task.apply_async.call_count, 2) for call in mock_task.apply_async.call_args_list: self.assertTrue(call.kwargs["kwargs"]["remote_ocr"])