import json from unittest import mock from auditlog.models import LogEntry from django.contrib.auth.models import Permission from django.contrib.auth.models import User from django.test import override_settings from guardian.shortcuts import assign_perm from rest_framework import status from rest_framework.test import APITestCase from documents.models import Correspondent from documents.models import CustomField from documents.models import Document from documents.models import DocumentType from documents.models import StoragePath from documents.models import Tag from documents.tests.utils import DirectoriesMixin class TestBulkEditAPI(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() user = User.objects.create_superuser(username="temp_admin") self.user = user self.client.force_authenticate(user=user) 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") self.doc2 = Document.objects.create( checksum="B", title="B", correspondent=self.c1, document_type=self.dt1, page_count=5, ) self.doc3 = Document.objects.create( checksum="C", title="C", correspondent=self.c2, document_type=self.dt2, ) 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}") self.cf1 = CustomField.objects.create(name="cf1", data_type="string") self.cf2 = CustomField.objects.create(name="cf2", data_type="string") def setup_mock(self, m, method_name, return_value="OK") -> None: m.return_value = return_value m.__name__ = method_name @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") def test_api_set_correspondent(self, bulk_update_task_mock) -> None: self.assertNotEqual(self.doc1.correspondent, self.c1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_correspondent", "parameters": {"correspondent": self.c1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.doc1.refresh_from_db() self.assertEqual(self.doc1.correspondent, self.c1) bulk_update_task_mock.assert_called_once() self.assertCountEqual( bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"], [self.doc1.pk], ) @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") def test_api_unset_correspondent(self, bulk_update_task_mock) -> None: self.doc1.correspondent = self.c1 self.doc1.save() self.assertIsNotNone(self.doc1.correspondent) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_correspondent", "parameters": {"correspondent": None}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) bulk_update_task_mock.assert_called_once() self.doc1.refresh_from_db() self.assertIsNone(self.doc1.correspondent) @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") def test_api_set_type(self, bulk_update_task_mock) -> None: self.assertNotEqual(self.doc1.document_type, self.dt1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_document_type", "parameters": {"document_type": self.dt1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.doc1.refresh_from_db() self.assertEqual(self.doc1.document_type, self.dt1) bulk_update_task_mock.assert_called_once() self.assertCountEqual( bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"], [self.doc1.pk], ) @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") def test_api_unset_type(self, bulk_update_task_mock) -> None: self.doc1.document_type = self.dt1 self.doc1.save() response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_document_type", "parameters": {"document_type": None}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.doc1.refresh_from_db() self.assertIsNone(self.doc1.document_type) bulk_update_task_mock.assert_called_once() self.assertCountEqual( bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"], [self.doc1.pk], ) @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") def test_api_add_tag(self, bulk_update_task_mock) -> None: self.assertFalse(self.doc1.tags.filter(pk=self.t1.pk).exists()) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "add_tag", "parameters": {"tag": self.t1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.doc1.refresh_from_db() self.assertTrue(self.doc1.tags.filter(pk=self.t1.pk).exists()) bulk_update_task_mock.assert_called_once() self.assertCountEqual( bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"], [self.doc1.pk], ) @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") def test_api_remove_tag(self, bulk_update_task_mock) -> None: self.doc1.tags.add(self.t1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "remove_tag", "parameters": {"tag": self.t1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.doc1.refresh_from_db() self.assertFalse(self.doc1.tags.filter(pk=self.t1.pk).exists()) @mock.patch("documents.serialisers.bulk_edit.modify_tags") def test_api_modify_tags(self, m) -> None: self.setup_mock(m, "modify_tags") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_tags", "parameters": { "add_tags": [self.t1.id], "remove_tags": [self.t2.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertListEqual(args[0], [self.doc1.id, self.doc3.id]) self.assertEqual(kwargs["add_tags"], [self.t1.id]) self.assertEqual(kwargs["remove_tags"], [self.t2.id]) @mock.patch("documents.serialisers.bulk_edit.modify_tags") def test_api_modify_tags_not_provided(self, m) -> None: """ GIVEN: - API data to modify tags is missing remove_tags field WHEN: - API to edit tags is called THEN: - API returns HTTP 400 - modify_tags is not called """ self.setup_mock(m, "modify_tags") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_tags", "parameters": { "add_tags": [self.t1.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.modify_custom_fields") def test_api_modify_custom_fields(self, m) -> None: self.setup_mock(m, "modify_custom_fields") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": [ self.cf1.id, ], # old format accepts list of IDs "remove_custom_fields": [self.cf2.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertListEqual(args[0], [self.doc1.id, self.doc3.id]) self.assertEqual(kwargs["add_custom_fields"], [self.cf1.id]) self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id]) @mock.patch("documents.serialisers.bulk_edit.modify_custom_fields") def test_api_modify_custom_fields_documentlink_forbidden_for_unpermitted_target( self, m, ) -> None: self.setup_mock(m, "modify_custom_fields") user = User.objects.create_user(username="doc-owner") user.user_permissions.add(Permission.objects.get(codename="change_document")) other_user = User.objects.create_user(username="other-user") source_doc = Document.objects.create( checksum="source", title="Source", owner=user, ) target_doc = Document.objects.create( checksum="target", title="Target", owner=other_user, ) doclink_field = CustomField.objects.create( name="doclink", data_type=CustomField.FieldDataType.DOCUMENTLINK, ) self.client.force_authenticate(user=user) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [source_doc.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": {doclink_field.id: [target_doc.id]}, "remove_custom_fields": [], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.modify_custom_fields") def test_api_modify_custom_fields_with_values(self, m) -> None: self.setup_mock(m, "modify_custom_fields") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": {self.cf1.id: "foo"}, "remove_custom_fields": [self.cf2.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertListEqual(args[0], [self.doc1.id, self.doc3.id]) self.assertEqual(kwargs["add_custom_fields"], {self.cf1.id: "foo"}) self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id]) @mock.patch("documents.serialisers.bulk_edit.modify_custom_fields") def test_api_modify_custom_fields_rejects_invalid_value(self, m) -> None: self.setup_mock(m, "modify_custom_fields") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": {self.cf1.id: "x" * 129}, "remove_custom_fields": [], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("add_custom_fields", response.data) self.assertIn(str(self.cf1.id), response.data["add_custom_fields"]) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.modify_custom_fields") def test_api_modify_custom_fields_invalid_params(self, m) -> None: """ GIVEN: - API data to modify custom fields is malformed WHEN: - API to edit custom fields is called THEN: - API returns HTTP 400 - modify_custom_fields is not called """ self.setup_mock(m, "modify_custom_fields") # Missing add_custom_fields response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": [self.cf1.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() # Missing remove_custom_fields response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "remove_custom_fields": [self.cf1.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() # Not a list response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": self.cf1.id, "remove_custom_fields": self.cf2.id, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() # Invalid dict response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": {"foo": 99}, "remove_custom_fields": [self.cf2.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() # Missing remove_custom_fields response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": ["foo"], "remove_custom_fields": ["bar"], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() # Custom field ID not found # Missing remove_custom_fields response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc3.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": [self.cf1.id], "remove_custom_fields": [99], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.delete") def test_api_delete(self, m) -> None: self.setup_mock(m, "delete") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( {"documents": [self.doc1.id], "method": "delete"}, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc1.id]) self.assertEqual(len(kwargs), 0) @mock.patch("documents.views.bulk_edit.delete") def test_delete_documents_endpoint(self, m) -> None: self.setup_mock(m, "delete") response = self.client.post( "/api/documents/delete/", json.dumps({"documents": [self.doc1.id]}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc1.id]) self.assertEqual(len(kwargs), 0) @mock.patch("documents.views.bulk_edit.delete") def test_delete_documents_endpoint_with_excluded_documents(self, m) -> None: self.setup_mock(m, "delete") response = self.client.post( "/api/documents/delete/", json.dumps( { "all": True, "excluded_documents": [ self.doc2.id, self.doc3.id, self.doc4.id, self.doc5.id, ], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc1.id]) self.assertEqual(len(kwargs), 0) @mock.patch("documents.views.bulk_edit.reprocess") def test_reprocess_documents_endpoint(self, m) -> None: self.setup_mock(m, "reprocess") response = self.client.post( "/api/documents/reprocess/", json.dumps({"documents": [self.doc1.id]}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc1.id]) self.assertEqual(kwargs, {"remote_ocr": False}) @mock.patch("documents.views.bulk_edit.reprocess") def test_reprocess_documents_endpoint_remote_ocr(self, m) -> None: """ GIVEN: - API data to reprocess a document with remote OCR requested WHEN: - API is called THEN: - reprocess is called with remote_ocr=True """ self.setup_mock(m, "reprocess") response = self.client.post( "/api/documents/reprocess/", json.dumps({"documents": [self.doc1.id], "remote_ocr": True}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc1.id]) self.assertEqual(kwargs, {"remote_ocr": True}) @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_set_storage_path(self, m) -> None: """ GIVEN: - API data to set the storage path of a document WHEN: - API is called THEN: - set_storage_path is called with correct document IDs and storage_path ID """ self.setup_mock(m, "set_storage_path") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertListEqual(args[0], [self.doc1.id]) self.assertEqual(kwargs["storage_path"], self.sp1.id) @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_unset_storage_path(self, m) -> None: """ GIVEN: - API data to clear/unset the storage path of a document WHEN: - API is called THEN: - set_storage_path is called with correct document IDs and None storage_path """ self.setup_mock(m, "set_storage_path") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_storage_path", "parameters": {"storage_path": None}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertListEqual(args[0], [self.doc1.id]) self.assertEqual(kwargs["storage_path"], None) def test_api_invalid_storage_path(self) -> None: """ GIVEN: - API data to set the storage path of a document - Given storage_path ID isn't valid WHEN: - API is called THEN: - set_storage_path is called with correct document IDs and storage_path ID """ response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id + 10}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.async_task.assert_not_called() def test_api_set_storage_path_not_provided(self) -> None: """ GIVEN: - API data to set the storage path of a document - API data is missing storage path ID WHEN: - API is called THEN: - set_storage_path is called with correct document IDs and storage_path ID """ response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_storage_path", "parameters": {}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.async_task.assert_not_called() def test_api_invalid_doc(self) -> None: self.assertEqual(Document.objects.count(), 5) response = self.client.post( "/api/documents/bulk_edit/", json.dumps({"documents": [-235], "method": "delete"}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(Document.objects.count(), 5) def test_api_requires_documents_unless_all_is_true(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"documents is required unless all is true", response.content) def test_api_rejects_excluded_documents_unless_all_is_true(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "excluded_documents": [self.doc2.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn( b"excluded_documents is only supported when all is true", response.content, ) @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_bulk_edit_with_all_true_resolves_documents_from_filters( self, m, ) -> None: self.setup_mock(m, "set_storage_path") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "filters": {"title__icontains": "B"}, "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc2.id]) self.assertEqual(kwargs["storage_path"], self.sp1.id) @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_bulk_edit_with_all_true_excludes_documents(self, m) -> None: self.setup_mock(m, "set_storage_path") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "excluded_documents": [self.doc2.id, self.doc4.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [self.doc1.id, self.doc3.id, self.doc5.id]) self.assertEqual(kwargs["storage_path"], self.sp1.id) @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_bulk_edit_with_all_true_resolves_owned_duplicates(self, m) -> None: self.setup_mock(m, "set_storage_path") user = User.objects.create_user(username="duplicate-owner") user.user_permissions.add( Permission.objects.get(codename="change_document"), ) first_duplicate = Document.objects.create( checksum="owned-duplicate", title="First duplicate", owner=user, ) second_duplicate = Document.objects.create( checksum="owned-duplicate", title="Second duplicate", owner=user, ) self.client.force_authenticate(user=user) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "filters": {"has_duplicates": True}, "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [first_duplicate.id, second_duplicate.id]) self.assertEqual(kwargs["storage_path"], self.sp1.id) @mock.patch("documents.search.get_backend") @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_bulk_edit_with_all_true_resolves_documents_from_search_filters( self, m, get_backend, ) -> None: self.setup_mock(m, "set_storage_path") for filters in ( {"text": "new doc 2017-03-16"}, {"title_search": "apple"}, ): with self.subTest(filters=filters): get_backend.return_value.search_ids.return_value = [self.doc2.id] response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "filters": filters, "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) get_backend.return_value.search_ids.assert_called_once() args, kwargs = m.call_args self.assertEqual(args[0], [self.doc2.id]) self.assertEqual(kwargs["storage_path"], self.sp1.id) m.reset_mock() get_backend.return_value.search_ids.reset_mock() # more_like_id is a different path get_backend.return_value.more_like_this_ids.return_value = [self.doc2.id] response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "filters": {"more_like_id": self.doc1.id}, "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) get_backend.return_value.more_like_this_ids.assert_called_once() def test_api_bulk_edit_with_all_true_rejects_multiple_filters(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "filters": { "text": "B", "query": "c1", }, "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"Specify only one of", response.content) def test_api_bulk_edit_with_all_true_rejects_unsupported_methods(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "method": "merge", "parameters": {"metadata_document_id": self.doc2.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"This method does not support all=true", response.content) def test_api_invalid_method(self) -> None: self.assertEqual(Document.objects.count(), 5) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "exterminate", "parameters": {}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(Document.objects.count(), 5) def test_api_invalid_correspondent(self) -> None: self.assertEqual(self.doc2.correspondent, self.c1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_correspondent", "parameters": {"correspondent": 345657}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) doc2 = Document.objects.get(id=self.doc2.id) self.assertEqual(doc2.correspondent, self.c1) def test_api_no_correspondent(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_correspondent", "parameters": {}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_api_invalid_document_type(self) -> None: self.assertEqual(self.doc2.document_type, self.dt1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_document_type", "parameters": {"document_type": 345657}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) doc2 = Document.objects.get(id=self.doc2.id) self.assertEqual(doc2.document_type, self.dt1) def test_api_no_document_type(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_document_type", "parameters": {}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_api_add_invalid_tag(self) -> None: self.assertEqual(list(self.doc2.tags.all()), [self.t1]) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "add_tag", "parameters": {"tag": 345657}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(list(self.doc2.tags.all()), [self.t1]) def test_api_add_tag_no_tag(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( {"documents": [self.doc2.id], "method": "add_tag"}, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_api_delete_invalid_tag(self) -> None: self.assertEqual(list(self.doc2.tags.all()), [self.t1]) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "remove_tag", "parameters": {"tag": 345657}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(list(self.doc2.tags.all()), [self.t1]) def test_api_delete_tag_no_tag(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( {"documents": [self.doc2.id], "method": "remove_tag"}, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_api_modify_invalid_tags(self) -> None: self.assertEqual(list(self.doc2.tags.all()), [self.t1]) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "modify_tags", "parameters": { "add_tags": [self.t2.id, 1657], "remove_tags": [1123123], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_api_modify_tags_no_tags(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "modify_tags", "parameters": {"remove_tags": [1123123]}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "modify_tags", "parameters": {"add_tags": [self.t2.id, 1657]}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_api_selection_data_empty(self) -> None: response = self.client.post( "/api/documents/selection_data/", json.dumps({"documents": []}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) for field, Entity in [ ("selected_correspondents", Correspondent), ("selected_tags", Tag), ("selected_document_types", DocumentType), ]: self.assertEqual(len(response.data[field]), Entity.objects.count()) for correspondent in response.data[field]: self.assertEqual(correspondent["document_count"], 0) self.assertCountEqual( map(lambda c: c["id"], response.data[field]), map(lambda c: c["id"], Entity.objects.values("id")), ) def test_api_selection_data(self) -> None: response = self.client.post( "/api/documents/selection_data/", json.dumps( {"documents": [self.doc1.id, self.doc2.id, self.doc4.id, self.doc5.id]}, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertCountEqual( response.data["selected_correspondents"], [ {"id": self.c1.id, "document_count": 1}, {"id": self.c2.id, "document_count": 0}, ], ) self.assertCountEqual( response.data["selected_tags"], [ {"id": self.t1.id, "document_count": 2}, {"id": self.t2.id, "document_count": 1}, ], ) self.assertCountEqual( response.data["selected_document_types"], [ {"id": self.c1.id, "document_count": 1}, {"id": self.c2.id, "document_count": 0}, ], ) def test_api_selection_data_with_excluded_documents(self) -> None: response = self.client.post( "/api/documents/selection_data/", json.dumps( { "all": True, "excluded_documents": [self.doc2.id], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertCountEqual( response.data["selected_correspondents"], [ {"id": self.c1.id, "document_count": 0}, {"id": self.c2.id, "document_count": 1}, ], ) self.assertCountEqual( response.data["selected_tags"], [ {"id": self.t1.id, "document_count": 1}, {"id": self.t2.id, "document_count": 2}, ], ) def test_api_selection_data_requires_view_permission(self) -> None: self.doc2.owner = self.user self.doc2.save() user1 = User.objects.create(username="user1") self.client.force_authenticate(user=user1) assign_perm("view_document", user1, self.doc2) response = self.client.post( "/api/documents/selection_data/", json.dumps({"documents": [self.doc2.id]}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) user1.user_permissions.add( Permission.objects.get(codename="view_document"), ) user1 = User.objects.get(pk=user1.pk) self.client.force_authenticate(user=user1) response = self.client.post( "/api/documents/selection_data/", json.dumps({"documents": [self.doc2.id]}), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions(self, m) -> None: self.setup_mock(m, "set_permissions") user1 = User.objects.create(username="user1") user2 = User.objects.create(username="user2") permissions = { "view": { "users": [user1.id, user2.id], "groups": None, }, "change": { "users": [user1.id], "groups": None, }, } response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "method": "set_permissions", "parameters": {"set_permissions": permissions}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id]) self.assertEqual(len(kwargs["set_permissions"]["view"]["users"]), 2) @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_requires_set_permissions_parameter(self, m) -> None: self.setup_mock(m, "set_permissions") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_permissions", "parameters": { "owner": self.user.id, "merge": True, "permissions": {"view": {"users": [self.user.id]}}, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"set_permissions not specified", response.content) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_rejects_malformed_set_permissions(self, m) -> None: """ GIVEN: - A set_permissions bulk edit where set_permissions is not an object, has a non-list users value, or has an unknown action WHEN: - API to bulk edit is called THEN: - API returns HTTP 400 describing the problem - set_permissions is not called """ self.setup_mock(m, "set_permissions") for set_permissions, expected_message in ( (False, b"Expected a dictionary"), ({"view": {"users": False}}, b"Expected a list"), ({"not_a_real_action": {"users": [1]}}, b"Unknown permission action"), ): with self.subTest(set_permissions=set_permissions): response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_permissions", "parameters": {"set_permissions": set_permissions}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(expected_message, response.content) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_rejects_invalid_owner(self, m) -> None: """ GIVEN: - A set_permissions bulk edit with an owner that is a nonexistent id, a boolean, a list, a dict, or a non-numeric string WHEN: - API to bulk edit is called THEN: - API returns HTTP 400 - set_permissions is not called """ self.setup_mock(m, "set_permissions") for bad_owner in ( 999999, True, ["not", "an", "id"], {"nested": "dict"}, "not-a-number", ): with self.subTest(owner=bad_owner): response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_permissions", "parameters": { "set_permissions": { "view": {"users": [self.user.id]}, }, "owner": bad_owner, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"Specified owner cannot be found", response.content) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_null_is_a_noop(self, m) -> None: """ GIVEN: - A set_permissions bulk edit with set_permissions null and an owner, i.e. a request that only changes the owner WHEN: - API to bulk edit is called THEN: - Request succeeds - set_permissions receives no users or groups for any action """ self.setup_mock(m, "set_permissions") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_permissions", "parameters": { "set_permissions": None, "owner": self.user.id, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() self.assertEqual( m.call_args.kwargs["set_permissions"], {"view": {}, "change": {}}, ) @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_passes_validated_owner_id(self, m) -> None: """ GIVEN: - A set_permissions bulk edit with the owner id given as a string WHEN: - API to bulk edit is called THEN: - set_permissions receives the owner as an integer id """ self.setup_mock(m, "set_permissions") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "set_permissions", "parameters": { "set_permissions": {"view": {"users": [self.user.id]}}, "owner": str(self.user.id), }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() self.assertEqual(m.call_args.kwargs["owner"], self.user.id) @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_merge(self, m) -> None: self.setup_mock(m, "set_permissions") user1 = User.objects.create(username="user1") user2 = User.objects.create(username="user2") permissions = { "view": { "users": [user1.id, user2.id], "groups": None, }, "change": { "users": [user1.id], "groups": None, }, } response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "method": "set_permissions", "parameters": {"set_permissions": permissions}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called() _, kwargs = m.call_args self.assertEqual(kwargs["merge"], False) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "method": "set_permissions", "parameters": {"set_permissions": permissions, "merge": True}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called() _, kwargs = m.call_args self.assertEqual(kwargs["merge"], True) @mock.patch("documents.serialisers.bulk_edit.set_storage_path") @mock.patch("documents.views.bulk_edit.merge") def test_insufficient_global_perms(self, mock_merge, mock_set_storage) -> None: """ GIVEN: - User has no global permissions to change a document - User has no global permissions to add a document - User has no global permissions to delete a document WHEN: - API is called to set storage path - API is called to merge documents - API is called to merge with delete THEN: - API returns HTTP 403 for all calls unless global permissions are granted """ user1 = User.objects.create(username="user1") user1.save() self.client.force_authenticate(user=user1) self.setup_mock(mock_set_storage, "set_storage_path") self.setup_mock(mock_merge, "merge") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) mock_set_storage.assert_not_called() response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc1.id], "metadata_document_id": self.doc1.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) mock_merge.assert_not_called() response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc1.id], "metadata_document_id": self.doc1.id, "delete_originals": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) mock_merge.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_insufficient_permissions_ownership(self, m) -> None: """ GIVEN: - Documents owned by user other than logged in user WHEN: - set_permissions bulk edit API endpoint is called THEN: - User is not able to change permissions """ self.setup_mock(m, "set_permissions") self.doc1.owner = User.objects.get(username="temp_admin") self.doc1.save() user1 = User.objects.create(username="user1") user1.user_permissions.add(*Permission.objects.all()) user1.save() self.client.force_authenticate(user=user1) permissions = { "view": {"users": [user1.id]}, } response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc2.id, self.doc3.id], "method": "set_permissions", "parameters": {"set_permissions": permissions}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() self.assertEqual(response.content, b"Insufficient permissions") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "method": "set_permissions", "parameters": {"set_permissions": permissions}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_insufficient_permissions_edit(self, m) -> None: """ GIVEN: - Documents for which current user only has view permissions WHEN: - API is called THEN: - set_storage_path only called if user can edit all docs """ self.setup_mock(m, "set_storage_path") self.doc1.owner = User.objects.get(username="temp_admin") self.doc1.save() user1 = User.objects.create(username="user1") assign_perm("view_document", user1, self.doc1) user1.user_permissions.add(*Permission.objects.all()) user1.save() self.client.force_authenticate(user=user1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc2.id, self.doc3.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() self.assertEqual(response.content, b"Insufficient permissions") assign_perm("change_document", user1, self.doc1) response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id, self.doc2.id, self.doc3.id], "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() @mock.patch("documents.views.bulk_edit.rotate") def test_rotate(self, m) -> None: self.setup_mock(m, "rotate") response = self.client.post( "/api/documents/rotate/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "degrees": 90, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id]) self.assertEqual(kwargs["degrees"], 90) self.assertEqual(kwargs["source_mode"], "latest_version") self.assertEqual(kwargs["user"], self.user) @mock.patch("documents.views.bulk_edit.rotate") def test_rotate_invalid_params(self, m) -> None: response = self.client.post( "/api/documents/rotate/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "degrees": "foo", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.post( "/api/documents/rotate/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "degrees": 90.5, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.post( "/api/documents/rotate/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "degrees": 45, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"degrees must be a multiple of 90", response.content) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.rotate") def test_bulk_edit_rotate_rejects_invalid_degrees(self, m) -> None: """ GIVEN: - A legacy rotate bulk edit with degrees that are null, not an integer, or not a multiple of 90 WHEN: - API to bulk edit is called THEN: - API returns HTTP 400 - rotate is not called """ self.setup_mock(m, "rotate") for degrees, expected_message in ( (None, b"invalid rotation degrees"), (True, b"invalid rotation degrees"), ("foo", b"invalid rotation degrees"), (90.5, b"invalid rotation degrees"), (45, b"degrees must be a multiple of 90"), ): with self.subTest(degrees=degrees): response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "rotate", "parameters": {"degrees": degrees}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(expected_message, response.content) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.rotate") def test_bulk_edit_rotate_passes_integer_degrees(self, m) -> None: """ GIVEN: - A legacy rotate bulk edit with degrees given as a string WHEN: - API to bulk edit is called THEN: - rotate receives the degrees as an integer """ self.setup_mock(m, "rotate") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "rotate", "parameters": {"degrees": "-90"}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() self.assertEqual(m.call_args.kwargs["degrees"], -90) @mock.patch("documents.serialisers.bulk_edit.split") def test_bulk_edit_split_rejects_invalid_pages(self, m) -> None: """ GIVEN: - A legacy split bulk edit of a 5 page document with pages that are not a string, not numeric, zero, a reversed range, or past the last page WHEN: - API to bulk edit is called THEN: - API returns HTTP 400 - split is not called """ self.setup_mock(m, "split") for pages in (None, "", "a", "1-2-3", "0", "3-1", "1-6", "1-5000000"): with self.subTest(pages=pages): response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "split", "parameters": {"pages": pages}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"invalid pages specified", response.content) m.assert_not_called() @mock.patch("documents.serialisers.bulk_edit.split") def test_bulk_edit_split_parses_pages(self, m) -> None: """ GIVEN: - A legacy split bulk edit with single pages and ranges WHEN: - API to bulk edit is called THEN: - split receives one page list per comma separated group """ self.setup_mock(m, "split") response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "split", "parameters": {"pages": "1,2-4,5"}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() self.assertEqual(m.call_args.kwargs["pages"], [[1], [2, 3, 4], [5]]) @mock.patch("documents.views.bulk_edit.rotate") def test_rotate_insufficient_permissions(self, m) -> None: self.doc1.owner = User.objects.get(username="temp_admin") self.doc1.save() user1 = User.objects.create(username="user1") user1.user_permissions.add(*Permission.objects.all()) user1.save() self.client.force_authenticate(user=user1) self.setup_mock(m, "rotate") response = self.client.post( "/api/documents/rotate/", json.dumps( { "documents": [self.doc1.id, self.doc2.id], "degrees": 90, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() self.assertEqual(response.content, b"Insufficient permissions") response = self.client.post( "/api/documents/rotate/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "degrees": 90, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() @mock.patch("documents.views.bulk_edit.merge") def test_merge(self, m) -> None: self.setup_mock(m, "merge") response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "metadata_document_id": self.doc3.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id]) self.assertEqual(kwargs["metadata_document_id"], self.doc3.id) self.assertEqual(kwargs["source_mode"], "latest_version") self.assertEqual(kwargs["user"], self.user) @mock.patch("documents.views.bulk_edit.merge") def test_merge_and_delete_insufficient_permissions(self, m) -> None: self.doc1.owner = User.objects.get(username="temp_admin") self.doc1.save() user1 = User.objects.create(username="user1") user1.user_permissions.add(*Permission.objects.all()) user1.save() self.client.force_authenticate(user=user1) self.setup_mock(m, "merge") response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc1.id, self.doc2.id], "metadata_document_id": self.doc2.id, "delete_originals": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() self.assertEqual(response.content, b"Insufficient permissions") response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "metadata_document_id": self.doc2.id, "delete_originals": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() @mock.patch("documents.views.bulk_edit.merge") def test_merge_and_delete_requires_change_permission(self, m) -> None: self.setup_mock(m, "merge") user = User.objects.create_user(username="no-change") user.user_permissions.add( Permission.objects.get(codename="add_document"), Permission.objects.get(codename="delete_document"), ) self.client.force_authenticate(user=user) response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "delete_originals": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() @mock.patch("documents.views.bulk_edit.merge") def test_merge_invalid_parameters(self, m) -> None: self.setup_mock(m, "merge") response = self.client.post( "/api/documents/merge/", json.dumps( { "documents": [self.doc1.id, self.doc2.id], "delete_originals": "not_boolean", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) m.assert_not_called() def test_bulk_edit_allows_legacy_file_methods_with_warning(self) -> None: method_payloads = { "delete": {}, "reprocess": {}, "rotate": {"degrees": 90}, "merge": {"metadata_document_id": self.doc2.id}, "edit_pdf": {"operations": [{"page": 1}]}, "remove_password": {"password": "secret"}, "split": {"pages": "1,2-4"}, "delete_pages": {"pages": [1, 2]}, } for version in (9, 10): for method, parameters in method_payloads.items(): with self.subTest(method=method, version=version): with mock.patch( f"documents.views.bulk_edit.{method}", ) as mocked_method: self.setup_mock(mocked_method, method) with self.assertLogs("paperless.api", level="WARNING") as logs: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": method, "parameters": parameters, }, ), content_type="application/json", headers={ "Accept": f"application/json; version={version}", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) mocked_method.assert_called_once() self.assertTrue( any( "Deprecated bulk_edit method" in entry and f"'{method}'" in entry for entry in logs.output ), ) def test_legacy_bulk_edit_reprocess_invalid_remote_ocr(self) -> None: """ GIVEN: - The deprecated bulk_edit endpoint with a non-boolean remote_ocr WHEN: - API is called THEN: - The request is rejected rather than passed through to the task """ response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "reprocess", "parameters": {"remote_ocr": "yes please"}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_legacy_bulk_edit_rejects_out_of_bounds_pdf_doc_index(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "edit_pdf", "parameters": { "operations": [{"page": 1, "doc": 2**32}], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"doc index is out of bounds", response.content) def test_legacy_bulk_edit_rejects_empty_pdf_operations(self) -> None: response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "edit_pdf", "parameters": {"operations": []}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( response.json(), {"operations": ["This list may not be empty."]}, ) @mock.patch("documents.views.bulk_edit.edit_pdf") def test_edit_pdf(self, m) -> None: self.setup_mock(m, "edit_pdf") response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1}], "source_mode": "explicit_selection", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [self.doc2.id]) self.assertEqual(kwargs["operations"], [{"page": 1}]) self.assertEqual(kwargs["source_mode"], "explicit_selection") self.assertEqual(kwargs["user"], self.user) def test_edit_pdf_invalid_params(self) -> None: response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id, self.doc3.id], "operations": [{"page": 1}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"Edit PDF method only supports one document", response.content) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": "not_a_list", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"Expected a list of items", response.content) response = self.client.post( "/api/documents/edit_pdf/", {"documents": [self.doc2.id], "operations": []}, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": ["invalid_operation"], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"Expected a dictionary", response.content) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": "not_an_int"}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"valid integer is required", response.content) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1, "rotate": "not_an_int"}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"valid integer is required", response.content) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1, "doc": "not_an_int"}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"valid integer is required", response.content) # A negative index fails the field's min_value before the # object-level bound against len(operations) is checked for doc_index, expected_message in ( (-1, b"greater than or equal to 0"), (2**32, b"doc index is out of bounds"), ): with self.subTest(doc_index=doc_index): response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1, "doc": doc_index}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(expected_message, response.content) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "update_document": True, "operations": [{"page": 1, "doc": 0}, {"page": 2, "doc": 1}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn( b"update_document only allowed with a single output document", response.content, ) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1}], "source_mode": "not_a_mode", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"Invalid source_mode", response.content) @mock.patch("documents.views.bulk_edit.edit_pdf") def test_edit_pdf_rejects_invalid_operation_values(self, m) -> None: """ GIVEN: - An edit_pdf operation with a non-positive page, a boolean in place of an integer, or a rotation that is not a multiple of 90 WHEN: - API to edit the PDF is called THEN: - API returns HTTP 400 with the field error - edit_pdf is not called """ self.setup_mock(m, "edit_pdf") for operation, expected_message in ( ({"page": 0}, b"greater than or equal to 1"), ({"page": True}, b"valid integer is required"), ({"page": 1, "rotate": True}, b"valid integer is required"), ({"page": 1, "doc": True}, b"valid integer is required"), ({"page": 1, "rotate": 45}, b"rotate must be a multiple of 90"), ): with self.subTest(operation=operation): response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [operation], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(expected_message, response.content) m.assert_not_called() def test_legacy_bulk_edit_keys_pdf_operation_errors_under_operations( self, ) -> None: """ GIVEN: - A legacy bulk_edit edit_pdf request with an invalid operation WHEN: - API to bulk edit is called THEN: - API returns HTTP 400 - The error is keyed under operations and the operation index, matching the edit_pdf endpoint """ response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc2.id], "method": "edit_pdf", "parameters": {"operations": [{"page": 1, "rotate": 45}]}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( response.json(), {"operations": {"0": {"rotate": ["rotate must be a multiple of 90"]}}}, ) @mock.patch("documents.views.bulk_edit.edit_pdf") def test_edit_pdf_page_out_of_bounds(self, m) -> None: self.setup_mock(m, "edit_pdf") response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 99}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"out of bounds", response.content) m.assert_not_called() @mock.patch("documents.views.bulk_edit.edit_pdf") def test_edit_pdf_insufficient_permissions(self, m) -> None: self.doc1.owner = User.objects.get(username="temp_admin") self.doc1.save() user1 = User.objects.create(username="user1") user1.user_permissions.add(*Permission.objects.all()) user1.save() self.client.force_authenticate(user=user1) self.setup_mock(m, "edit_pdf") response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc1.id], "operations": [{"page": 1}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() self.assertEqual(response.content, b"Insufficient permissions") response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1}], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() @mock.patch("documents.views.bulk_edit.edit_pdf") def test_edit_pdf_update_requires_change_permission(self, m) -> None: self.setup_mock(m, "edit_pdf") user = User.objects.create_user(username="no-change") self.client.force_authenticate(user=user) response = self.client.post( "/api/documents/edit_pdf/", json.dumps( { "documents": [self.doc2.id], "operations": [{"page": 1}], "update_document": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() @mock.patch("documents.views.bulk_edit.remove_password") @mock.patch("documents.views.bulk_edit.edit_pdf") def test_delete_original_requires_delete_permission( self, edit_pdf_mock, remove_password_mock, ) -> None: self.setup_mock(edit_pdf_mock, "edit_pdf") self.setup_mock(remove_password_mock, "remove_password") user = User.objects.create_user(username="no-delete") user.user_permissions.add( Permission.objects.get(codename="add_document"), Permission.objects.get(codename="change_document"), ) self.client.force_authenticate(user=user) cases = [ ( "/api/documents/edit_pdf/", { "documents": [self.doc2.id], "operations": [{"page": 1}], "delete_original": True, }, edit_pdf_mock, ), ( "/api/documents/remove_password/", { "documents": [self.doc2.id], "password": "secret", "delete_original": True, }, remove_password_mock, ), ] for endpoint, payload, operation_mock in cases: with self.subTest(endpoint=endpoint): response = self.client.post( endpoint, json.dumps(payload), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) operation_mock.assert_not_called() @mock.patch("documents.views.bulk_edit.remove_password") def test_remove_password(self, m) -> None: self.setup_mock(m, "remove_password") response = self.client.post( "/api/documents/remove_password/", json.dumps( { "documents": [self.doc2.id], "password": "secret", "update_document": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() args, kwargs = m.call_args self.assertCountEqual(args[0], [self.doc2.id]) self.assertEqual(kwargs["password"], "secret") self.assertTrue(kwargs["update_document"]) self.assertEqual(kwargs["source_mode"], "latest_version") self.assertEqual(kwargs["user"], self.user) def test_remove_password_invalid_params(self) -> None: response = self.client.post( "/api/documents/remove_password/", json.dumps( { "documents": [self.doc2.id], }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.post( "/api/documents/remove_password/", json.dumps( { "documents": [self.doc2.id], "password": 123, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) @mock.patch("documents.views.bulk_edit.remove_password") def test_remove_password_insufficient_permissions(self, m) -> None: self.doc1.owner = User.objects.get(username="temp_admin") self.doc1.save() user1 = User.objects.create(username="user1") user1.user_permissions.add(*Permission.objects.all()) user1.save() self.client.force_authenticate(user=user1) self.setup_mock(m, "remove_password") response = self.client.post( "/api/documents/remove_password/", json.dumps( { "documents": [self.doc1.id], "password": "secret", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) m.assert_not_called() self.assertEqual(response.content, b"Insufficient permissions") response = self.client.post( "/api/documents/remove_password/", json.dumps( { "documents": [self.doc2.id], "password": "secret", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() @override_settings(AUDIT_LOG_ENABLED=True) def test_bulk_edit_audit_log_enabled_simple_field(self) -> None: """ GIVEN: - Audit log is enabled WHEN: - API to bulk edit documents is called THEN: - Audit log is created """ LogEntry.objects.all().delete() response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "set_correspondent", "parameters": {"correspondent": self.c2.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(LogEntry.objects.filter(object_pk=self.doc1.id).count(), 1) @override_settings(AUDIT_LOG_ENABLED=True) def test_bulk_edit_audit_log_enabled_tags(self) -> None: """ GIVEN: - Audit log is enabled WHEN: - API to bulk edit tags is called THEN: - Audit log is created """ LogEntry.objects.all().delete() response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "modify_tags", "parameters": { "add_tags": [self.t1.id], "remove_tags": [self.t2.id], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(LogEntry.objects.filter(object_pk=self.doc1.id).count(), 1) @override_settings(AUDIT_LOG_ENABLED=True) def test_bulk_edit_audit_log_enabled_custom_fields(self) -> None: """ GIVEN: - Audit log is enabled WHEN: - API to bulk edit custom fields is called THEN: - Audit log is created """ LogEntry.objects.all().delete() response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "documents": [self.doc1.id], "method": "modify_custom_fields", "parameters": { "add_custom_fields": [self.cf1.id], "remove_custom_fields": [], }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(LogEntry.objects.filter(object_pk=self.doc1.id).count(), 2) def test_api_bulk_edit_with_bad_search_query_returns_400(self) -> None: """ GIVEN: - Bulk edit request selects documents via a saved-search query filter WHEN: - The query contains a malformed field value (an invalid date) THEN: - The response is a 400 naming the bad value, exactly like the search list endpoint, never a 500 """ response = self.client.post( "/api/documents/bulk_edit/", json.dumps( { "all": True, "filters": {"query": "added:notadate"}, "method": "set_storage_path", "parameters": {"storage_path": self.sp1.id}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(b"notadate", response.content)