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 guardian.shortcuts import assign_perm
from rest_framework import status
from rest_framework.test import APITestCase
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="text")
- self.cf2 = CustomField.objects.create(name="cf2", data_type="text")
+ 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"):
+ m.return_value = return_value
+ m.__name__ = method_name
@mock.patch("documents.bulk_edit.bulk_update_documents.delay")
def test_api_set_correspondent(self, bulk_update_task_mock):
@mock.patch("documents.serialisers.bulk_edit.modify_tags")
def test_api_modify_tags(self, m):
- m.return_value = "OK"
+ self.setup_mock(m, "modify_tags")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
- API returns HTTP 400
- modify_tags is not called
"""
- m.return_value = "OK"
+ self.setup_mock(m, "modify_tags")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
@mock.patch("documents.serialisers.bulk_edit.modify_custom_fields")
def test_api_modify_custom_fields(self, m):
- m.return_value = "OK"
+ self.setup_mock(m, "modify_custom_fields")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
- API returns HTTP 400
- modify_custom_fields is not called
"""
- m.return_value = "OK"
-
+ self.setup_mock(m, "modify_custom_fields")
# Missing add_custom_fields
response = self.client.post(
"/api/documents/bulk_edit/",
@mock.patch("documents.serialisers.bulk_edit.delete")
def test_api_delete(self, m):
- m.return_value = "OK"
+ self.setup_mock(m, "delete")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
THEN:
- set_storage_path is called with correct document IDs and storage_path ID
"""
- m.return_value = "OK"
-
+ self.setup_mock(m, "set_storage_path")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
THEN:
- set_storage_path is called with correct document IDs and None storage_path
"""
- m.return_value = "OK"
-
+ self.setup_mock(m, "set_storage_path")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
@mock.patch("documents.serialisers.bulk_edit.set_permissions")
def test_set_permissions(self, m):
- m.return_value = "OK"
+ self.setup_mock(m, "set_permissions")
user1 = User.objects.create(username="user1")
user2 = User.objects.create(username="user2")
permissions = {
@mock.patch("documents.serialisers.bulk_edit.set_permissions")
def test_set_permissions_merge(self, m):
- m.return_value = "OK"
+ self.setup_mock(m, "set_permissions")
user1 = User.objects.create(username="user1")
user2 = User.objects.create(username="user2")
permissions = {
THEN:
- User is not able to change permissions
"""
- m.return_value = "OK"
+ self.setup_mock(m, "set_permissions")
self.doc1.owner = User.objects.get(username="temp_admin")
self.doc1.save()
user1 = User.objects.create(username="user1")
THEN:
- set_storage_path only called if user can edit all docs
"""
- m.return_value = "OK"
+ 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")
@mock.patch("documents.serialisers.bulk_edit.rotate")
def test_rotate(self, m):
- m.return_value = "OK"
-
+ self.setup_mock(m, "rotate")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
@mock.patch("documents.serialisers.bulk_edit.merge")
def test_merge(self, m):
- m.return_value = "OK"
-
+ self.setup_mock(m, "merge")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
user1 = User.objects.create(username="user1")
self.client.force_authenticate(user=user1)
- m.return_value = "OK"
-
+ self.setup_mock(m, "merge")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
THEN:
- The API fails with a correct error code
"""
- m.return_value = "OK"
-
+ self.setup_mock(m, "merge")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
@mock.patch("documents.serialisers.bulk_edit.split")
def test_split(self, m):
- m.return_value = "OK"
-
+ self.setup_mock(m, "split")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
@mock.patch("documents.serialisers.bulk_edit.delete_pages")
def test_delete_pages(self, m):
- m.return_value = "OK"
-
+ self.setup_mock(m, "delete_pages")
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn(b"pages must be a list of integers", response.content)
+
+ @override_settings(AUDIT_LOG_ENABLED=True)
+ def test_bulk_edit_audit_log_enabled_simple_field(self):
+ """
+ GIVEN:
+ - Audit log is enabled
+ WHEN:
+ - API to bulk edit documents is called
+ THEN:
+ - Audit log is created
+ """
+ LogEntry.objects.all().delete()
+ response = self.client.post(
+ "/api/documents/bulk_edit/",
+ json.dumps(
+ {
+ "documents": [self.doc1.id],
+ "method": "set_correspondent",
+ "parameters": {"correspondent": self.c2.id},
+ },
+ ),
+ content_type="application/json",
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(LogEntry.objects.filter(object_pk=self.doc1.id).count(), 1)
+
+ @override_settings(AUDIT_LOG_ENABLED=True)
+ def test_bulk_edit_audit_log_enabled_tags(self):
+ """
+ GIVEN:
+ - Audit log is enabled
+ WHEN:
+ - API to bulk edit tags is called
+ THEN:
+ - Audit log is created
+ """
+ LogEntry.objects.all().delete()
+ response = self.client.post(
+ "/api/documents/bulk_edit/",
+ json.dumps(
+ {
+ "documents": [self.doc1.id],
+ "method": "modify_tags",
+ "parameters": {
+ "add_tags": [self.t1.id],
+ "remove_tags": [self.t2.id],
+ },
+ },
+ ),
+ content_type="application/json",
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(LogEntry.objects.filter(object_pk=self.doc1.id).count(), 1)
+
+ @override_settings(AUDIT_LOG_ENABLED=True)
+ def test_bulk_edit_audit_log_enabled_custom_fields(self):
+ """
+ GIVEN:
+ - Audit log is enabled
+ WHEN:
+ - API to bulk edit custom fields is called
+ THEN:
+ - Audit log is created
+ """
+ LogEntry.objects.all().delete()
+ response = self.client.post(
+ "/api/documents/bulk_edit/",
+ json.dumps(
+ {
+ "documents": [self.doc1.id],
+ "method": "modify_custom_fields",
+ "parameters": {
+ "add_custom_fields": [self.cf1.id],
+ "remove_custom_fields": [],
+ },
+ },
+ ),
+ content_type="application/json",
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(LogEntry.objects.filter(object_pk=self.doc1.id).count(), 2)
from django.db.models import Count
from django.db.models import IntegerField
from django.db.models import Max
+from django.db.models import Model
from django.db.models import Q
from django.db.models import Sum
from django.db.models import When
from django.db.models.functions import Length
from django.db.models.functions import Lower
+from django.db.models.manager import Manager
from django.http import Http404
from django.http import HttpResponse
from django.http import HttpResponseBadRequest
class BulkEditView(PassUserMixin):
+ MODIFIED_FIELD_BY_METHOD = {
+ "set_correspondent": "correspondent",
+ "set_document_type": "document_type",
+ "set_storage_path": "storage_path",
+ "add_tag": "tags",
+ "remove_tag": "tags",
+ "modify_tags": "tags",
+ "modify_custom_fields": "custom_fields",
+ "set_permissions": None,
+ "delete": "deleted_at",
+ "rotate": "checksum",
+ "delete_pages": "checksum",
+ "split": None,
+ "merge": None,
+ }
+
permission_classes = (IsAuthenticated,)
serializer_class = BulkEditSerializer
parser_classes = (parsers.JSONParser,)
return HttpResponseForbidden("Insufficient permissions")
try:
+ modified_field = self.MODIFIED_FIELD_BY_METHOD[method.__name__]
+ if settings.AUDIT_LOG_ENABLED and modified_field:
+ old_documents = {
+ obj["pk"]: obj
+ for obj in Document.objects.filter(pk__in=documents).values(
+ "pk",
+ "correspondent",
+ "document_type",
+ "storage_path",
+ "tags",
+ "custom_fields",
+ "deleted_at",
+ "checksum",
+ )
+ }
+
# TODO: parameter validation
result = method(documents, **parameters)
+
+ if settings.AUDIT_LOG_ENABLED and modified_field:
+ new_documents = Document.objects.filter(pk__in=documents)
+ for doc in new_documents:
+ old_value = old_documents[doc.pk][modified_field]
+ new_value = getattr(doc, modified_field)
+
+ if isinstance(new_value, Model):
+ # correspondent, document type, etc.
+ new_value = new_value.pk
+ elif isinstance(new_value, Manager):
+ # tags, custom fields
+ new_value = list(new_value.values_list("pk", flat=True))
+
+ LogEntry.objects.log_create(
+ instance=doc,
+ changes={
+ modified_field: [
+ old_value,
+ new_value,
+ ],
+ },
+ action=LogEntry.Action.UPDATE,
+ actor=user,
+ additional_data={
+ "reason": f"Bulk edit: {method.__name__}",
+ },
+ )
+
return Response({"result": result})
except Exception as e:
logger.warning(f"An error occurred performing bulk edit: {e!s}")
projects / thirdparty / paperless-ngx.git / commitdiff
