From a5c460a567fdb04363fe451018ebb97fc8616468 Mon Sep 17 00:00:00 2001 From: shamoon <4887959+shamoon@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:56:12 -0800 Subject: [PATCH] Add some backend tests for the manual wf --- src/documents/tests/test_api_bulk_edit.py | 83 +++++++++++++++++++++++ src/documents/tests/test_workflows.py | 50 ++++++++++++++ 2 files changed, 133 insertions(+) diff --git a/src/documents/tests/test_api_bulk_edit.py b/src/documents/tests/test_api_bulk_edit.py index 43dfc1064f..ef6f8267c0 100644 --- a/src/documents/tests/test_api_bulk_edit.py +++ b/src/documents/tests/test_api_bulk_edit.py @@ -14,7 +14,11 @@ from documents.models import Document from documents.models import DocumentType from documents.models import StoragePath from documents.models import Tag +from documents.models import Workflow +from documents.models import WorkflowAction +from documents.models import WorkflowTrigger from paperless_testing.dirs import DirectoriesMixin +from paperless_testing.factories import DocumentFactory from paperless_testing.factories import UserFactory from paperless_testing.permissions import grant_all_global from paperless_testing.permissions import grant_global @@ -583,6 +587,85 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase): self.assertEqual(args[0], [self.doc1.id]) self.assertEqual(kwargs, {"remote_ocr": True}) + def _make_manual_workflow(self) -> Workflow: + trigger = WorkflowTrigger.objects.create( + type=WorkflowTrigger.WorkflowTriggerType.MANUAL, + ) + action = WorkflowAction.objects.create() + workflow = Workflow.objects.create(name="Manual workflow", order=0) + workflow.triggers.add(trigger) + workflow.actions.add(action) + return workflow + + def _run_workflow(self, parameters, documents=None): + return self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": documents or [self.doc1.id], + "method": "run_workflow", + "parameters": parameters, + }, + ), + content_type="application/json", + ) + + @mock.patch("documents.views.bulk_edit.run_workflow") + def test_run_workflow(self, m) -> None: + """A valid run_workflow call passes the docs, workflow id and filter flag through.""" + self.setup_mock(m, "run_workflow") + workflow = self._make_manual_workflow() + + response = self._run_workflow({"workflow": workflow.id}) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + args, kwargs = m.call_args + self.assertEqual(args[0], [self.doc1.id]) + self.assertEqual(kwargs, {"workflow": workflow.id, "filter_documents": True}) + + self._run_workflow({"workflow": workflow.id, "filter_documents": False}) + self.assertFalse(m.call_args.kwargs["filter_documents"]) + + @mock.patch("documents.views.bulk_edit.run_workflow") + def test_run_workflow_invalid_parameters(self, m) -> None: + """Invalid run_workflow parameters are rejected before dispatch.""" + self.setup_mock(m, "run_workflow") + manual = self._make_manual_workflow() + non_manual = Workflow.objects.create(name="Non-manual", order=0) + non_manual.triggers.add( + WorkflowTrigger.objects.create( + type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED, + ), + ) + for label, parameters in ( + ("missing workflow", {}), + ("nonexistent workflow", {"workflow": 9999}), + ("no manual trigger", {"workflow": non_manual.id}), + ("non-bool filter", {"workflow": manual.id, "filter_documents": "nope"}), + ): + with self.subTest(label): + response = self._run_workflow(parameters) + self.assertEqual( + response.status_code, + status.HTTP_400_BAD_REQUEST, + ) + m.assert_not_called() + + @mock.patch("documents.views.bulk_edit.run_workflow") + def test_run_workflow_requires_ownership(self, m) -> None: + """A non-owner without document ownership is forbidden (actions mutate docs).""" + self.setup_mock(m, "run_workflow") + user = UserFactory(username="no_perms") + grant_all_global(user) + doc = DocumentFactory(owner=UserFactory(username="other")) + workflow = self._make_manual_workflow() + self.client.force_authenticate(user=user) + + response = self._run_workflow({"workflow": workflow.id}, documents=[doc.id]) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + m.assert_not_called() + @mock.patch("documents.serialisers.bulk_edit.set_storage_path") def test_api_set_storage_path(self, m) -> None: """ diff --git a/src/documents/tests/test_workflows.py b/src/documents/tests/test_workflows.py index 820882b490..74d358a91c 100644 --- a/src/documents/tests/test_workflows.py +++ b/src/documents/tests/test_workflows.py @@ -2288,6 +2288,56 @@ class TestWorkflows( doc.refresh_from_db() self.assertEqual(doc.owner, self.user2) + def _make_manual_workflow(self, *, require_tag: Tag | None = None) -> Workflow: + trigger = WorkflowTrigger.objects.create( + type=WorkflowTrigger.WorkflowTriggerType.MANUAL, + ) + if require_tag is not None: + trigger.filter_has_all_tags.add(require_tag) + action = WorkflowAction.objects.create(assign_owner=self.user2) + workflow = Workflow.objects.create(name="Manual workflow", order=0) + workflow.triggers.add(trigger) + workflow.actions.add(action) + return workflow + + @mock.patch("documents.tasks.send_websocket_document_updated") + def test_manual_workflow_run_applies_and_notifies(self, mock_ws) -> None: + """A manual run applies the workflow's actions and pushes a websocket update.""" + workflow = self._make_manual_workflow() + doc = DocumentFactory() + + tasks.run_workflows_for_documents([doc.pk], workflow.pk) + + doc.refresh_from_db() + self.assertEqual(doc.owner, self.user2) + self.assertEqual(mock_ws.call_args.kwargs["document"].pk, doc.pk) + + def test_manual_workflow_run_respects_filters(self) -> None: + """With filter_documents=True, a document failing the trigger filters is skipped.""" + workflow = self._make_manual_workflow(require_tag=self.t1) + doc = DocumentFactory() + + tasks.run_workflows_for_documents([doc.pk], workflow.pk, filter_documents=True) + + doc.refresh_from_db() + self.assertIsNone(doc.owner) + + def test_manual_workflow_run_force_ignores_filters(self) -> None: + """With filter_documents=False, actions run regardless of the trigger filters.""" + workflow = self._make_manual_workflow(require_tag=self.t1) + doc = DocumentFactory() + + tasks.run_workflows_for_documents([doc.pk], workflow.pk, filter_documents=False) + + doc.refresh_from_db() + self.assertEqual(doc.owner, self.user2) + + def test_manual_workflow_run_missing_workflow(self) -> None: + """A missing workflow id logs a warning rather than raising.""" + doc = DocumentFactory() + with self.assertLogs("paperless.tasks", level="WARNING"): + tasks.run_workflows_for_documents([doc.pk], 9999) + @mock.patch("documents.tasks.send_websocket_document_updated") def test_workflow_scheduled_trigger_sends_websocket_update( self,