mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-02 14:20:31 +00:00
Trenton H
7424e7ce0b
Snapshot only the edited field, before and after the operation, gathering tags and custom field instances into sorted id lists per document (empty when there are none).
2655 lines
95 KiB
Python
2655 lines
95 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
from auditlog.models import LogEntry
|
|
from django.contrib.auth.models import User
|
|
from django.test import override_settings
|
|
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 CustomFieldInstance
|
|
from documents.models import Document
|
|
from documents.models import DocumentType
|
|
from documents.models import StoragePath
|
|
from documents.models import Tag
|
|
from paperless_testing.dirs import DirectoriesMixin
|
|
from paperless_testing.factories import UserFactory
|
|
from paperless_testing.permissions import grant_all_global
|
|
from paperless_testing.permissions import grant_global
|
|
from paperless_testing.permissions import grant_object
|
|
|
|
|
|
class TestBulkEditAPI(DirectoriesMixin, APITestCase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
|
|
user = UserFactory(username="temp_admin", superuser=True)
|
|
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 = UserFactory(username="doc-owner")
|
|
grant_global(user, "change_document")
|
|
other_user = UserFactory(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 = UserFactory(username="duplicate-owner")
|
|
grant_global(user, "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)
|
|
|
|
grant_object(user1, self.doc2, "view_document")
|
|
|
|
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)
|
|
|
|
grant_global(user1, "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")
|
|
grant_all_global(user1)
|
|
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")
|
|
grant_object(user1, self.doc1, "view_document")
|
|
grant_all_global(user1)
|
|
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")
|
|
|
|
grant_object(user1, self.doc1, "change_document")
|
|
|
|
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_rejects_unknown_page_count(self, m) -> None:
|
|
"""
|
|
GIVEN:
|
|
- A legacy split bulk edit of a document without a page count
|
|
WHEN:
|
|
- API to bulk edit is called
|
|
THEN:
|
|
- API returns HTTP 400
|
|
- split is not called
|
|
"""
|
|
self.setup_mock(m, "split")
|
|
|
|
for pages in ("1", "1-5000000"):
|
|
with self.subTest(pages=pages):
|
|
response = self.client.post(
|
|
"/api/documents/bulk_edit/",
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.id],
|
|
"method": "split",
|
|
"parameters": {"pages": pages},
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn(b"document page count is unknown", 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")
|
|
grant_all_global(user1)
|
|
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")
|
|
grant_all_global(user1)
|
|
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 = UserFactory(username="no-change")
|
|
grant_global(user, "add_document", "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")
|
|
grant_all_global(user1)
|
|
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 = UserFactory(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 = UserFactory(username="no-delete")
|
|
grant_global(user, "add_document", "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")
|
|
grant_all_global(user1)
|
|
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 with the old and new correspondent
|
|
"""
|
|
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)
|
|
entry = LogEntry.objects.get_for_object(self.doc1).get()
|
|
self.assertEqual(entry.changes, {"correspondent": [None, self.c2.id]})
|
|
|
|
@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 on an untagged document and a
|
|
document with several tags
|
|
THEN:
|
|
- Audit log is created for each document with its full tag list
|
|
before and after the edit
|
|
"""
|
|
LogEntry.objects.all().delete()
|
|
response = self.client.post(
|
|
"/api/documents/bulk_edit/",
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.id, self.doc4.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)
|
|
entry = LogEntry.objects.get_for_object(self.doc1).get()
|
|
self.assertEqual(entry.changes, {"tags": [[], [self.t1.id]]})
|
|
entry = LogEntry.objects.get_for_object(self.doc4).get()
|
|
self.assertEqual(
|
|
entry.changes,
|
|
{"tags": [[self.t1.id, self.t2.id], [self.t1.id]]},
|
|
)
|
|
|
|
@override_settings(AUDIT_LOG_ENABLED=True)
|
|
def test_bulk_edit_audit_log_enabled_custom_fields(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Audit log is enabled
|
|
- A document with two custom fields
|
|
WHEN:
|
|
- API to bulk edit custom fields is called to add a third
|
|
THEN:
|
|
- Audit log is created with every custom field instance before and
|
|
after the edit
|
|
- Audit log is created for the new custom field instance
|
|
"""
|
|
cf3 = CustomField.objects.create(name="cf3", data_type="string")
|
|
existing = [
|
|
CustomFieldInstance.objects.create(document=self.doc1, field=field)
|
|
for field in (self.cf2, cf3)
|
|
]
|
|
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)
|
|
added = CustomFieldInstance.objects.get(document=self.doc1, field=self.cf1)
|
|
existing_ids = [instance.id for instance in existing]
|
|
entry = LogEntry.objects.get_for_object(self.doc1).get()
|
|
self.assertEqual(
|
|
entry.changes,
|
|
{"custom_fields": [existing_ids, [*existing_ids, added.id]]},
|
|
)
|
|
self.assertEqual(LogEntry.objects.get_for_object(added).count(), 1)
|
|
|
|
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)
|