Fix: correct doc version filename handling (#12223)

src/documents/tests/test_consumer.py

@@ -726,6 +726,13 @@ class TestConsumer(
         self.assertIsNotNone(root_doc)
         assert root_doc is not None
 
+        root_storage_path = StoragePath.objects.create(
+            name="version-root-path",
+            path="root/{{title}}",
+        )
+        root_doc.storage_path = root_storage_path
+        root_doc.save()
+
         actor = User.objects.create_user(
             username="actor",
             email="actor@example.com",
@@ -762,7 +769,7 @@ class TestConsumer(
         )
         consumer.setup()
         try:
-            self.assertTrue(consumer.filename.endswith("_v0.pdf"))
+            self.assertEqual(consumer.filename, version_file.name)
             consumer.run()
         finally:
             consumer.cleanup()
@@ -772,8 +779,9 @@ class TestConsumer(
         version = versions.first()
         assert version is not None
         assert version.original_filename is not None
+        self.assertEqual(version.version_index, 1)
         self.assertEqual(version.version_label, "v2")
-        self.assertTrue(version.original_filename.endswith("_v0.pdf"))
+        self.assertEqual(version.original_filename, version_file.name)
         self.assertTrue(bool(version.content))
 
     @override_settings(AUDIT_LOG_ENABLED=True)
@@ -822,7 +830,7 @@ class TestConsumer(
         )
         consumer.setup()
         try:
-            self.assertEqual(consumer.filename, "valid_pdf_version-upload_v0")
+            self.assertEqual(consumer.filename, "valid_pdf_version-upload")
             consumer.run()
         finally:
             consumer.cleanup()
@@ -832,9 +840,67 @@ class TestConsumer(
         )
         self.assertIsNotNone(version)
         assert version is not None
-        self.assertEqual(version.original_filename, "valid_pdf_version-upload_v0")
+        self.assertEqual(version.version_index, 1)
+        self.assertEqual(version.original_filename, "valid_pdf_version-upload")
         self.assertTrue(bool(version.content))
 
+    @override_settings(AUDIT_LOG_ENABLED=True)
+    @mock.patch("documents.consumer.load_classifier")
+    def test_consume_version_index_monotonic_after_version_deletion(self, m) -> None:
+        m.return_value = MagicMock()
+
+        with self.get_consumer(self.get_test_file()) as consumer:
+            consumer.run()
+
+        root_doc = Document.objects.first()
+        self.assertIsNotNone(root_doc)
+        assert root_doc is not None
+
+        def consume_version(version_file: Path) -> Document:
+            status = DummyProgressManager(version_file.name, None)
+            overrides = DocumentMetadataOverrides()
+            doc = ConsumableDocument(
+                DocumentSource.ApiUpload,
+                original_file=version_file,
+                root_document_id=root_doc.pk,
+            )
+            preflight = ConsumerPreflightPlugin(
+                doc,
+                overrides,
+                status,  # type: ignore[arg-type]
+                self.dirs.scratch_dir,
+                "task-id",
+            )
+            preflight.setup()
+            preflight.run()
+
+            consumer = ConsumerPlugin(
+                doc,
+                overrides,
+                status,  # type: ignore[arg-type]
+                self.dirs.scratch_dir,
+                "task-id",
+            )
+            consumer.setup()
+            try:
+                consumer.run()
+            finally:
+                consumer.cleanup()
+
+            version = (
+                Document.objects.filter(root_document=root_doc).order_by("-id").first()
+            )
+            assert version is not None
+            return version
+
+        v1 = consume_version(self.get_test_file2())
+        self.assertEqual(v1.version_index, 1)
+        v1.delete()
+
+        # The next version should have version_index 2, even though version_index 1 was deleted
+        v2 = consume_version(self.get_test_file())
+        self.assertEqual(v2.version_index, 2)
+
     @mock.patch("documents.consumer.load_classifier")
     def testClassifyDocument(self, m) -> None:
         correspondent = Correspondent.objects.create(

src/documents/tests/test_file_handling.py

@@ -77,6 +77,58 @@ class TestFileHandling(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
             settings.ORIGINALS_DIR / "test" / "test.pdf",
         )
 
+    @override_settings(FILENAME_FORMAT=None)
+    def test_root_storage_path_change_updates_version_files(self) -> None:
+        old_storage_path = StoragePath.objects.create(
+            name="old-path",
+            path="old/{{title}}",
+        )
+        new_storage_path = StoragePath.objects.create(
+            name="new-path",
+            path="new/{{title}}",
+        )
+
+        root_doc = Document.objects.create(
+            title="rootdoc",
+            mime_type="application/pdf",
+            checksum="root-checksum",
+            storage_path=old_storage_path,
+        )
+        version_doc = Document.objects.create(
+            title="version-title",
+            mime_type="application/pdf",
+            checksum="version-checksum",
+            root_document=root_doc,
+            version_index=1,
+        )
+
+        Document.objects.filter(pk=root_doc.pk).update(
+            filename=generate_filename(root_doc),
+        )
+        Document.objects.filter(pk=version_doc.pk).update(
+            filename=generate_filename(version_doc),
+        )
+        root_doc.refresh_from_db()
+        version_doc.refresh_from_db()
+
+        create_source_path_directory(root_doc.source_path)
+        Path(root_doc.source_path).touch()
+        create_source_path_directory(version_doc.source_path)
+        Path(version_doc.source_path).touch()
+
+        root_doc.storage_path = new_storage_path
+        root_doc.save()
+
+        root_doc.refresh_from_db()
+        version_doc.refresh_from_db()
+
+        self.assertEqual(root_doc.filename, "new/rootdoc.pdf")
+        self.assertEqual(version_doc.filename, "new/rootdoc_v1.pdf")
+        self.assertIsFile(root_doc.source_path)
+        self.assertIsFile(version_doc.source_path)
+        self.assertIsNotFile(settings.ORIGINALS_DIR / "old" / "rootdoc.pdf")
+        self.assertIsNotFile(settings.ORIGINALS_DIR / "old" / "rootdoc_v1.pdf")
+
     @override_settings(FILENAME_FORMAT="{correspondent}/{correspondent}")
     def test_file_renaming_missing_permissions(self) -> None:
         document = Document()
@@ -1226,6 +1278,94 @@ class TestFilenameGeneration(DirectoriesMixin, TestCase):
             Path("logs.pdf"),
         )
 
+    @override_settings(FILENAME_FORMAT="{title}")
+    def test_version_index_suffix_for_template_filename(self) -> None:
+        root_doc = Document.objects.create(
+            title="the_doc",
+            mime_type="application/pdf",
+            checksum="root-checksum",
+        )
+        version_doc = Document.objects.create(
+            title="the_doc",
+            mime_type="application/pdf",
+            checksum="version-checksum",
+            root_document=root_doc,
+            version_index=1,
+        )
+
+        self.assertEqual(generate_filename(version_doc), Path("the_doc_v1.pdf"))
+        self.assertEqual(
+            generate_filename(version_doc, counter=1),
+            Path("the_doc_v1_01.pdf"),
+        )
+
+    @override_settings(FILENAME_FORMAT=None)
+    def test_version_index_suffix_for_default_filename(self) -> None:
+        root_doc = Document.objects.create(
+            title="root",
+            mime_type="text/plain",
+            checksum="root-checksum",
+        )
+        version_doc = Document.objects.create(
+            title="root",
+            mime_type="text/plain",
+            checksum="version-checksum",
+            root_document=root_doc,
+            version_index=2,
+        )
+
+        self.assertEqual(
+            generate_filename(version_doc),
+            Path(f"{root_doc.pk:07d}_v2.txt"),
+        )
+        self.assertEqual(
+            generate_filename(version_doc, archive_filename=True),
+            Path(f"{root_doc.pk:07d}_v2.pdf"),
+        )
+
+    @override_settings(FILENAME_FORMAT="{original_name}")
+    def test_version_index_suffix_with_original_name_placeholder(self) -> None:
+        root_doc = Document.objects.create(
+            title="root",
+            mime_type="application/pdf",
+            checksum="root-checksum",
+            original_filename="root-upload.pdf",
+        )
+        version_doc = Document.objects.create(
+            title="root",
+            mime_type="application/pdf",
+            checksum="version-checksum",
+            root_document=root_doc,
+            version_index=1,
+            original_filename="version-upload.pdf",
+        )
+
+        self.assertEqual(generate_filename(version_doc), Path("root-upload_v1.pdf"))
+
+    def test_version_index_suffix_with_storage_path(self) -> None:
+        storage_path = StoragePath.objects.create(
+            name="vtest",
+            path="folder/{{title}}",
+        )
+        root_doc = Document.objects.create(
+            title="storage_doc",
+            mime_type="application/pdf",
+            checksum="root-checksum",
+            storage_path=storage_path,
+        )
+        version_doc = Document.objects.create(
+            title="version_title_should_not_be_used",
+            mime_type="application/pdf",
+            checksum="version-checksum",
+            root_document=root_doc,
+            version_index=3,
+        )
+
+        self.assertEqual(
+            generate_filename(version_doc),
+            Path("folder/storage_doc_v3.pdf"),
+        )
+
     @override_settings(
         FILENAME_FORMAT="XX{correspondent}/{title}",
         FILENAME_FORMAT_REMOVE_NONE=True,