mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-03 06:40:31 +00:00
540 lines
19 KiB
Python
540 lines
19 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
from auditlog.models import LogEntry
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.test import TestCase
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
|
|
from documents.bulk_edit import merge_as_versions
|
|
from documents.models import Document
|
|
from documents.serialisers import MergeDocumentsAsVersionsSerializer
|
|
from paperless_testing.factories import UserFactory
|
|
from paperless_testing.permissions import grant_global
|
|
|
|
|
|
class TestMergeDocumentsAsVersionsSerializer(TestCase):
|
|
def setUp(self) -> None:
|
|
self.doc1 = Document.objects.create(checksum="A", title="A")
|
|
self.doc2 = Document.objects.create(checksum="B", title="B")
|
|
self.doc3 = Document.objects.create(checksum="C", title="C")
|
|
|
|
def test_accepts_selected_root_document(self) -> None:
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id, self.doc3.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
)
|
|
|
|
self.assertTrue(serializer.is_valid(), serializer.errors)
|
|
self.assertEqual(
|
|
serializer.validated_data,
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id, self.doc3.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
)
|
|
|
|
def test_requires_at_least_two_documents(self) -> None:
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id],
|
|
"root_document_id": self.doc1.id,
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertEqual(
|
|
serializer.errors["non_field_errors"][0],
|
|
"At least two documents are required.",
|
|
)
|
|
|
|
def test_accepts_version_label_for_one_source_document(self) -> None:
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc1.id,
|
|
"version_label": " Imported ",
|
|
},
|
|
)
|
|
|
|
self.assertTrue(serializer.is_valid(), serializer.errors)
|
|
self.assertEqual(serializer.validated_data["version_label"], "Imported")
|
|
|
|
def test_rejects_version_label_for_multiple_source_documents(self) -> None:
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id, self.doc3.id],
|
|
"root_document_id": self.doc1.id,
|
|
"version_label": "Imported",
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertEqual(
|
|
serializer.errors["non_field_errors"][0],
|
|
"version_label can only be used when merging one source document.",
|
|
)
|
|
|
|
def test_requires_root_document_to_be_selected(self) -> None:
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc3.id,
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertEqual(
|
|
serializer.errors["non_field_errors"][0],
|
|
"root_document_id must be one of the selected documents.",
|
|
)
|
|
|
|
def test_rejects_duplicate_documents(self) -> None:
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc1.id],
|
|
"root_document_id": self.doc1.id,
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("documents", serializer.errors)
|
|
|
|
def test_rejects_selected_version(self) -> None:
|
|
version = Document.objects.create(
|
|
checksum="D",
|
|
title="D",
|
|
root_document=self.doc1,
|
|
version_index=1,
|
|
)
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [version.id, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertEqual(
|
|
serializer.errors["non_field_errors"][0],
|
|
"Only top-level documents can be merged as versions.",
|
|
)
|
|
|
|
def test_rejects_source_document_with_versions(self) -> None:
|
|
Document.objects.create(
|
|
checksum="D",
|
|
title="D",
|
|
root_document=self.doc1,
|
|
version_index=1,
|
|
)
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertEqual(
|
|
serializer.errors["non_field_errors"][0],
|
|
"Documents with existing versions cannot be merged into another document.",
|
|
)
|
|
|
|
def test_rejects_source_document_with_trashed_versions(self) -> None:
|
|
version = Document.objects.create(
|
|
checksum="D",
|
|
title="D",
|
|
root_document=self.doc1,
|
|
version_index=1,
|
|
)
|
|
version.delete() # trashed, but still points at doc1
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertEqual(
|
|
serializer.errors["non_field_errors"][0],
|
|
"Documents with existing versions cannot be merged into another document.",
|
|
)
|
|
|
|
def test_allows_root_document_with_versions(self) -> None:
|
|
Document.objects.create(
|
|
checksum="D",
|
|
title="D",
|
|
root_document=self.doc1,
|
|
version_index=1,
|
|
)
|
|
serializer = MergeDocumentsAsVersionsSerializer(
|
|
data={
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc1.id,
|
|
},
|
|
)
|
|
|
|
self.assertTrue(serializer.is_valid(), serializer.errors)
|
|
|
|
|
|
class TestMergeDocumentsAsVersions(TestCase):
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_merges_documents_in_selection_order(
|
|
self,
|
|
get_backend_mock,
|
|
bulk_update_mock,
|
|
status_manager_mock,
|
|
) -> None:
|
|
root = Document.objects.create(checksum="A", title="Root")
|
|
existing_version = Document.objects.create(
|
|
checksum="B",
|
|
title="Existing version",
|
|
root_document=root,
|
|
version_index=3,
|
|
)
|
|
source1 = Document.objects.create(
|
|
checksum="C",
|
|
title="Source 1",
|
|
archive_serial_number=1,
|
|
)
|
|
source2 = Document.objects.create(
|
|
checksum="D",
|
|
title="Source 2",
|
|
archive_serial_number=2,
|
|
)
|
|
original_modified = root.modified
|
|
|
|
result = merge_as_versions(
|
|
[source2.id, root.id, source1.id],
|
|
root_document_id=root.id,
|
|
)
|
|
|
|
self.assertEqual(result, "OK")
|
|
source1.refresh_from_db()
|
|
source2.refresh_from_db()
|
|
root.refresh_from_db()
|
|
# source2 was selected first, so it becomes the older of the two versions
|
|
self.assertEqual(source2.root_document_id, root.id)
|
|
self.assertEqual(source2.version_index, 4)
|
|
self.assertEqual(source1.root_document_id, root.id)
|
|
self.assertEqual(source1.version_index, 5)
|
|
self.assertIsNone(source1.archive_serial_number)
|
|
self.assertIsNone(source2.archive_serial_number)
|
|
# The root had no ASN of its own, so it takes the first one
|
|
self.assertEqual(root.archive_serial_number, 2)
|
|
self.assertGreater(root.modified, original_modified)
|
|
self.assertEqual(existing_version.root_document_id, root.id)
|
|
|
|
batch = get_backend_mock.return_value.batch_update.return_value.__enter__.return_value
|
|
self.assertEqual(
|
|
[call.args[0] for call in batch.remove.call_args_list],
|
|
[source2.id, source1.id],
|
|
)
|
|
bulk_update_mock.assert_called_once_with(
|
|
kwargs={"document_ids": [root.id]},
|
|
headers={"trigger_source": "system"},
|
|
)
|
|
status_manager_mock.return_value.send_documents_deleted.assert_called_once_with(
|
|
[source2.id, source1.id],
|
|
)
|
|
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_root_keeps_its_own_archive_serial_number(self, *_mocks) -> None:
|
|
root = Document.objects.create(
|
|
checksum="A",
|
|
title="Root",
|
|
archive_serial_number=1,
|
|
)
|
|
source = Document.objects.create(
|
|
checksum="B",
|
|
title="Source",
|
|
archive_serial_number=2,
|
|
)
|
|
|
|
with self.assertLogs("paperless.bulk_edit", level="WARNING") as logs:
|
|
merge_as_versions([root.id, source.id], root_document_id=root.id)
|
|
|
|
root.refresh_from_db()
|
|
source.refresh_from_db()
|
|
self.assertEqual(root.archive_serial_number, 1)
|
|
self.assertIsNone(source.archive_serial_number)
|
|
# Dropping an ASN is not silent
|
|
self.assertIn("[2]", logs.output[0])
|
|
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_root_without_asn_takes_the_source_archive_serial_number(
|
|
self,
|
|
*_mocks,
|
|
) -> None:
|
|
root = Document.objects.create(checksum="A", title="Root")
|
|
source = Document.objects.create(
|
|
checksum="B",
|
|
title="Source",
|
|
archive_serial_number=7,
|
|
)
|
|
|
|
merge_as_versions([root.id, source.id], root_document_id=root.id)
|
|
|
|
root.refresh_from_db()
|
|
source.refresh_from_db()
|
|
self.assertEqual(root.archive_serial_number, 7)
|
|
self.assertIsNone(source.archive_serial_number)
|
|
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_writes_audit_log_entry(self, *_mocks) -> None:
|
|
user = UserFactory(username="merger")
|
|
root = Document.objects.create(checksum="A", title="Root")
|
|
source = Document.objects.create(checksum="B", title="Source")
|
|
LogEntry.objects.all().delete()
|
|
|
|
merge_as_versions([root.id, source.id], root_document_id=root.id, user=user)
|
|
|
|
entry = LogEntry.objects.filter(
|
|
content_type=ContentType.objects.get_for_model(Document),
|
|
object_id=root.id,
|
|
).first()
|
|
self.assertIsNotNone(entry)
|
|
self.assertEqual(entry.actor, user)
|
|
self.assertEqual(entry.action, LogEntry.Action.UPDATE)
|
|
self.assertEqual(entry.changes, {"Merged As Versions": ["None", [source.id]]})
|
|
self.assertEqual(entry.additional_data["version_ids"], [source.id])
|
|
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_sets_version_label_for_one_source_document(
|
|
self,
|
|
_get_backend_mock,
|
|
_bulk_update_mock,
|
|
_status_manager_mock,
|
|
) -> None:
|
|
root = Document.objects.create(checksum="A", title="Root")
|
|
source = Document.objects.create(checksum="B", title="Source")
|
|
|
|
merge_as_versions(
|
|
[root.id, source.id],
|
|
root_document_id=root.id,
|
|
version_label="Imported",
|
|
)
|
|
|
|
source.refresh_from_db()
|
|
self.assertEqual(source.version_label, "Imported")
|
|
|
|
|
|
class TestMergeDocumentsAsVersionsAPI(APITestCase):
|
|
def setUp(self) -> None:
|
|
self.user = UserFactory(username="user")
|
|
grant_global(self.user, "change_document", "view_document", "delete_document")
|
|
self.doc1 = Document.objects.create(
|
|
checksum="A",
|
|
title="A",
|
|
owner=self.user,
|
|
)
|
|
self.doc2 = Document.objects.create(
|
|
checksum="B",
|
|
title="B",
|
|
owner=self.user,
|
|
)
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
@mock.patch("documents.views.bulk_edit.merge_as_versions")
|
|
def test_merges_documents_as_versions(self, merge_mock) -> None:
|
|
merge_mock.return_value = "OK"
|
|
merge_mock.__name__ = "merge_as_versions"
|
|
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
"version_label": "Imported",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data, {"result": "OK"})
|
|
merge_mock.assert_called_once_with(
|
|
[self.doc1.id, self.doc2.id],
|
|
root_document_id=self.doc2.id,
|
|
version_label="Imported",
|
|
user=self.user,
|
|
)
|
|
|
|
@mock.patch("documents.views.bulk_edit.merge_as_versions")
|
|
def test_requires_change_permission(self, merge_mock) -> None:
|
|
merge_mock.__name__ = "merge_as_versions"
|
|
user = UserFactory(username="no-change")
|
|
self.doc1.owner = user
|
|
self.doc1.save()
|
|
self.doc2.owner = user
|
|
self.doc2.save()
|
|
self.client.force_authenticate(user=user)
|
|
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc1.id,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
merge_mock.assert_not_called()
|
|
|
|
@mock.patch("documents.views.bulk_edit.merge_as_versions")
|
|
def test_requires_delete_permission(self, merge_mock) -> None:
|
|
merge_mock.__name__ = "merge_as_versions"
|
|
# Owns them and may change them, but may not make them stop being documents
|
|
user = UserFactory(username="no-delete")
|
|
grant_global(user, "change_document", "view_document")
|
|
for doc in (self.doc1, self.doc2):
|
|
doc.owner = user
|
|
doc.save()
|
|
self.client.force_authenticate(user=user)
|
|
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc1.id,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
merge_mock.assert_not_called()
|
|
|
|
@mock.patch("documents.views.bulk_edit.merge_as_versions")
|
|
def test_rejects_unselected_root(self, merge_mock) -> None:
|
|
doc3 = Document.objects.create(
|
|
checksum="C",
|
|
title="C",
|
|
owner=self.user,
|
|
)
|
|
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": doc3.id,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
merge_mock.assert_not_called()
|
|
|
|
@mock.patch("documents.views.bulk_edit.merge_as_versions")
|
|
def test_rejects_source_document_with_versions(self, merge_mock) -> None:
|
|
Document.objects.create(
|
|
checksum="C",
|
|
title="C",
|
|
root_document=self.doc1,
|
|
version_index=1,
|
|
owner=self.user,
|
|
)
|
|
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
merge_mock.assert_not_called()
|
|
self.doc1.refresh_from_db()
|
|
self.assertIsNone(self.doc1.root_document_id)
|
|
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_merges_and_returns_documents_as_versions(
|
|
self,
|
|
get_backend_mock,
|
|
bulk_update_mock,
|
|
status_manager_mock,
|
|
) -> None:
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
{
|
|
"documents": [self.doc1.id, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
"version_label": "Imported",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.doc1.refresh_from_db()
|
|
self.assertEqual(self.doc1.root_document_id, self.doc2.id)
|
|
self.assertEqual(self.doc1.version_label, "Imported")
|
|
|
|
detail_response = self.client.get(
|
|
f"/api/documents/{self.doc2.id}/?fields=id,versions",
|
|
)
|
|
self.assertEqual(detail_response.status_code, status.HTTP_200_OK)
|
|
versions = detail_response.data["versions"]
|
|
self.assertEqual(
|
|
{version["id"] for version in versions},
|
|
{self.doc1.id, self.doc2.id},
|
|
)
|
|
self.assertEqual(
|
|
[version["id"] for version in versions if version["is_root"]],
|
|
[self.doc2.id],
|
|
)
|
|
batch = get_backend_mock.return_value.batch_update.return_value.__enter__.return_value
|
|
batch.remove.assert_called_once_with(self.doc1.id)
|
|
bulk_update_mock.assert_called_once_with(
|
|
kwargs={"document_ids": [self.doc2.id]},
|
|
headers={"trigger_source": "system"},
|
|
)
|
|
status_manager_mock.return_value.send_documents_deleted.assert_called_once_with(
|
|
[self.doc1.id],
|
|
)
|
|
|
|
@mock.patch("documents.bulk_edit.DocumentsStatusManager")
|
|
@mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
|
|
@mock.patch("documents.search.get_backend")
|
|
def test_chosen_order_survives_to_the_versions_list(self, *_mocks) -> None:
|
|
doc3 = Document.objects.create(checksum="C", title="C", owner=self.user)
|
|
# Deliberately not in id order, as dragging the dialog rows produces
|
|
ordered = [doc3.id, self.doc1.id]
|
|
|
|
response = self.client.post(
|
|
"/api/documents/merge_as_versions/",
|
|
{
|
|
"documents": [*ordered, self.doc2.id],
|
|
"root_document_id": self.doc2.id,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail_response = self.client.get(
|
|
f"/api/documents/{self.doc2.id}/?fields=id,versions",
|
|
)
|
|
# Newest first, so the reverse of the order they were merged in
|
|
self.assertEqual(
|
|
[version["id"] for version in detail_response.data["versions"]],
|
|
[self.doc1.id, doc3.id, self.doc2.id],
|
|
)
|