Backend tests for the bulk operatoins stuff, task tracking

src/documents/tests/test_api_bulk_edit.py

@@ -7,6 +7,7 @@ from django.test import override_settings
 from rest_framework import status
 from rest_framework.test import APITestCase
 
+from documents.bulk_operations import execute_bulk_operation
 from documents.models import Correspondent
 from documents.models import CustomField
 from documents.models import CustomFieldInstance
@@ -29,9 +30,18 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.user = user
         self.client.force_authenticate(user=user)
 
-        patcher = mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+        patcher = mock.patch("documents.bulk_operations.bulk_update_documents")
         self.async_task = patcher.start()
         self.addCleanup(patcher.stop)
+        # Exercise accepted requests and worker execution together in these
+        # operation/validation tests. Queue-only behavior is tested separately.
+        enqueue = mock.patch(
+            "documents.views.execute_bulk_operation.apply_async",
+            side_effect=lambda *, kwargs, headers: self.run_operation(kwargs),
+        )
+        enqueue.start()
+        self.addCleanup(enqueue.stop)
+
         self.c1 = Correspondent.objects.create(name="c1")
         self.c2 = Correspondent.objects.create(name="c2")
         self.dt1 = DocumentType.objects.create(name="dt1")
@@ -61,11 +71,16 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.cf1 = CustomField.objects.create(name="cf1", data_type="string")
         self.cf2 = CustomField.objects.create(name="cf2", data_type="string")
 
+    @staticmethod
+    def run_operation(kwargs):
+        execute_bulk_operation(**kwargs)
+        return mock.Mock(id="test-bulk-operation")
+
     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")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_set_correspondent(self, bulk_update_task_mock) -> None:
         self.assertNotEqual(self.doc1.correspondent, self.c1)
         response = self.client.post(
@@ -84,11 +99,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         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"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_unset_correspondent(self, bulk_update_task_mock) -> None:
         self.doc1.correspondent = self.c1
         self.doc1.save()
@@ -110,7 +125,7 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.doc1.refresh_from_db()
         self.assertIsNone(self.doc1.correspondent)
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_set_type(self, bulk_update_task_mock) -> None:
         self.assertNotEqual(self.doc1.document_type, self.dt1)
         response = self.client.post(
@@ -129,11 +144,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         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"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_unset_type(self, bulk_update_task_mock) -> None:
         self.doc1.document_type = self.dt1
         self.doc1.save()
@@ -154,11 +169,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         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"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_add_tag(self, bulk_update_task_mock) -> None:
         self.assertFalse(self.doc1.tags.filter(pk=self.t1.pk).exists())
 
@@ -180,11 +195,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
 
         bulk_update_task_mock.assert_called_once()
         self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_remove_tag(self, bulk_update_task_mock) -> None:
         self.doc1.tags.add(self.t1)
 
@@ -223,6 +238,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         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])
@@ -277,6 +295,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         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])
@@ -345,6 +366,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         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])
@@ -506,6 +530,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(len(kwargs), 0)
 
@@ -520,6 +547,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(len(kwargs), 0)
 
@@ -544,6 +574,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(len(kwargs), 0)
 
@@ -558,6 +591,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs, {"remote_ocr": False})
 
@@ -580,6 +616,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs, {"remote_ocr": True})
 
@@ -609,6 +648,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
 
         self.assertListEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
@@ -639,6 +681,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
 
         self.assertListEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs["storage_path"], None)
@@ -761,6 +806,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -784,6 +832,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc1.id, self.doc3.id, self.doc5.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -820,6 +871,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [first_duplicate.id, second_duplicate.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -856,6 +910,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
                 self.assertEqual(response.status_code, status.HTTP_200_OK)
                 get_backend.return_value.search_ids.assert_called_once()
                 args, kwargs = m.call_args
+                kwargs = {
+                    key: value
+                    for key, value in kwargs.items()
+                    if key != "update_callback"
+                }
                 self.assertEqual(args[0], [self.doc2.id])
                 self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -1232,6 +1291,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
 
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
         self.assertEqual(len(kwargs["set_permissions"]["view"]["users"]), 2)
 
@@ -1644,6 +1706,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
         self.assertEqual(kwargs["degrees"], 90)
         self.assertEqual(kwargs["source_mode"], "latest_version")
@@ -1930,6 +1995,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         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")
@@ -2137,6 +2205,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["operations"], [{"page": 1}])
         self.assertEqual(kwargs["source_mode"], "explicit_selection")
@@ -2482,6 +2553,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["password"], "secret")
         self.assertTrue(kwargs["update_document"])

src/documents/tests/test_bulk_edit.py

@@ -698,10 +698,10 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
     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])
+            with self.assertRaisesRegex(Exception, "Data too long for column"):
+                bulk_edit.delete(doc_ids)
+        self.assertIn("possible incompatible database column", cm.output[0])
 
 
 class TestPDFActions(DirectoriesMixin, TestCase):

src/documents/tests/test_bulk_operations.py

@@ -0,0 +1,184 @@
+"""Request acceptance, worker execution, and task tracking."""
+
+import uuid
+from unittest import mock
+
+import pytest
+from auditlog.models import LogEntry
+from django.contrib.auth.models import User
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.test import APIClient
+
+from documents.bulk_operations import execute_bulk_operation
+from documents.models import Correspondent
+from documents.models import Document
+from documents.models import PaperlessTask
+from documents.signals.handlers import before_task_publish_handler
+from documents.signals.handlers import task_failure_handler
+from documents.signals.handlers import task_postrun_handler
+from paperless_testing.factories import UserFactory
+
+
+@pytest.fixture
+def bulk_setup(paperless_dirs):
+    user = UserFactory(superuser=True)
+    client = APIClient()
+    client.force_authenticate(user)
+    doc = Document.objects.create(title="Example", checksum="bulk-op-example")
+    correspondent = Correspondent.objects.create(name="Example correspondent")
+    return user, client, doc, correspondent
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "endpoint,operation,parameters",
+    [
+        ("bulk_edit", "set_correspondent", {"correspondent": None}),
+        ("delete", "delete", {}),
+    ],
+)
+def test_request_only_queues(bulk_setup, endpoint, operation, parameters):
+    user, client, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    data = {"documents": [doc.pk], "from_webui": True}
+    if endpoint == "bulk_edit":
+        data.update(method=operation, parameters=parameters)
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = task_id
+        response = client.post(f"/api/documents/{endpoint}/", data, format="json")
+    assert response.status_code == 200
+    assert response.data == {"result": "OK", "task_id": task_id}
+    assert Document.objects.filter(pk=doc.pk).exists()
+    assert enqueue.call_args.kwargs["kwargs"] == {
+        "operation": operation,
+        "document_ids": [doc.pk],
+        "parameters": parameters,
+        "user_id": user.pk,
+    }
+    assert enqueue.call_args.kwargs["headers"]["trigger_source"] == "web_ui"
+
+
+@pytest.mark.django_db
+def test_metadata_task_includes_audit_and_followup(bulk_setup, settings):
+    user, _, doc, correspondent = bulk_setup
+    settings.AUDIT_LOG_ENABLED = True
+    with (
+        mock.patch("documents.bulk_operations.bulk_update_documents") as followup,
+        mock.patch(
+            "documents.bulk_edit.bulk_update_documents.apply_async",
+        ) as extra_task,
+    ):
+
+        def check_followup(ids):
+            doc.refresh_from_db()
+            assert doc.correspondent_id == correspondent.pk
+            assert LogEntry.objects.filter(object_pk=str(doc.pk), actor=user).exists()
+            assert ids == [doc.pk]
+
+        followup.side_effect = check_followup
+        result = execute_bulk_operation(
+            "set_correspondent",
+            [doc.pk],
+            {"correspondent": correspondent.pk},
+            user.pk,
+        )
+    followup.assert_called_once()
+    extra_task.assert_not_called()
+    assert result["document_count"] == 1
+
+
+@pytest.mark.django_db
+def test_permissions_rechecked_on_worker(bulk_setup):
+    user, _, doc, correspondent = bulk_setup
+    User.objects.filter(pk=user.pk).update(is_superuser=False)
+    with pytest.raises(PermissionDenied):
+        execute_bulk_operation(
+            "set_correspondent",
+            [doc.pk],
+            {"correspondent": correspondent.pk},
+            user.pk,
+        )
+    doc.refresh_from_db()
+    assert doc.correspondent_id is None
+
+
+@pytest.mark.django_db
+def test_queue_failure_does_not_change_documents(bulk_setup):
+    _, client, doc, correspondent = bulk_setup
+    with mock.patch(
+        "documents.views.execute_bulk_operation.apply_async",
+        side_effect=RuntimeError("broker unavailable"),
+    ):
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "documents": [doc.pk],
+                "method": "set_correspondent",
+                "parameters": {"correspondent": correspondent.pk},
+            },
+            format="json",
+        )
+    assert response.status_code == 400
+    doc.refresh_from_db()
+    assert doc.correspondent_id is None
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("failed", [False, True])
+def test_tracked_outcome_records_initiator_and_result(bulk_setup, failed):
+    user, _, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    before_task_publish_handler(
+        headers={
+            "task": "documents.bulk_operations.execute_bulk_operation",
+            "id": task_id,
+        },
+        body=(
+            (),
+            {"operation": "delete", "document_ids": [doc.pk], "user_id": user.pk},
+            {},
+        ),
+    )
+    task = PaperlessTask.objects.get(task_id=task_id)
+    assert task.owner_id == user.pk
+    assert task.task_type == PaperlessTask.TaskType.BULK_DELETE
+    if failed:
+        task_failure_handler(task_id=task_id, exception=ValueError("Could not update"))
+    else:
+        task_postrun_handler(
+            task_id=task_id,
+            state="SUCCESS",
+            retval={"document_count": 1},
+        )
+    task.refresh_from_db()
+    assert task.status == ("failure" if failed else "success")
+    if failed:
+        assert task.result_data["error_message"] == "Could not update"
+    else:
+        assert task.result_data == {"document_count": 1}
+
+
+@pytest.mark.django_db
+def test_large_filtered_selection_is_one_task(bulk_setup):
+    _, client, _, _ = bulk_setup
+    Document.objects.bulk_create(
+        [
+            Document(title="bulk-large-selection", checksum=f"large-{i}")
+            for i in range(10_000)
+        ],
+    )
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = str(uuid.uuid4())
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "all": True,
+                "filters": {"title__icontains": "bulk-large-selection"},
+                "method": "set_correspondent",
+                "parameters": {"correspondent": None},
+            },
+            format="json",
+        )
+    assert response.status_code == 200
+    enqueue.assert_called_once()
+    assert len(enqueue.call_args.kwargs["kwargs"]["document_ids"]) == 10_000