import json from unittest import mock from django.contrib.auth.models import Permission from django.contrib.auth.models import User 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 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_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.bulk_edit.remove_document_from_index.apply_async") def test_merges_documents_in_creation_order( self, remove_from_index_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() self.assertEqual(source2.root_document_id, root.id) self.assertEqual(source2.version_index, 5) self.assertEqual(source1.root_document_id, root.id) self.assertEqual(source1.version_index, 4) self.assertIsNone(source1.archive_serial_number) self.assertIsNone(source2.archive_serial_number) self.assertGreater(root.modified, original_modified) self.assertEqual(existing_version.root_document_id, root.id) self.assertEqual( [call.kwargs["args"] for call in remove_from_index_mock.call_args_list], [[source1.id], [source2.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( [source1.id, source2.id], ) @mock.patch("documents.bulk_edit.DocumentsStatusManager") @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") @mock.patch("documents.bulk_edit.remove_document_from_index.apply_async") def test_sets_version_label_for_one_source_document( self, _remove_from_index_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") @mock.patch("documents.bulk_edit.DocumentsStatusManager") @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") @mock.patch("documents.bulk_edit.remove_document_from_index.apply_async") def test_rejects_source_document_with_versions( self, remove_from_index_mock, bulk_update_mock, status_manager_mock, ) -> None: source = Document.objects.create(checksum="A", title="Source") Document.objects.create( checksum="B", title="Source version", root_document=source, version_index=1, ) root = Document.objects.create(checksum="C", title="Root") with self.assertRaisesRegex(ValueError, "existing versions"): merge_as_versions( [source.id, root.id], root_document_id=root.id, ) source.refresh_from_db() self.assertIsNone(source.root_document_id) remove_from_index_mock.assert_not_called() bulk_update_mock.assert_not_called() status_manager_mock.assert_not_called() class TestMergeDocumentsAsVersionsAPI(APITestCase): def setUp(self) -> None: self.user = User.objects.create_user(username="user") self.user.user_permissions.add( Permission.objects.get(codename="change_document"), Permission.objects.get(codename="view_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", ) @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 = User.objects.create_user(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_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.bulk_edit.DocumentsStatusManager") @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") @mock.patch("documents.bulk_edit.remove_document_from_index.apply_async") def test_merges_and_returns_documents_as_versions( self, remove_from_index_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], ) remove_from_index_mock.assert_called_once_with(args=[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], )