From 74ce218b787825ae3575ef3d5cfe46f9d635edf7 Mon Sep 17 00:00:00 2001 From: shamoon <4887959+shamoon@users.noreply.github.com> Date: Wed, 11 Feb 2026 22:41:59 -0800 Subject: [PATCH] more backend coverage --- src/documents/tasks.py | 5 +- .../tests/test_api_document_versions.py | 214 ++++++++++++++++++ src/documents/tests/test_consumer.py | 80 +++++++ 3 files changed, 298 insertions(+), 1 deletion(-) diff --git a/src/documents/tasks.py b/src/documents/tasks.py index 019042c05..cee038072 100644 --- a/src/documents/tasks.py +++ b/src/documents/tasks.py @@ -157,7 +157,10 @@ def consume_file( overrides = DocumentMetadataOverrides() plugins: list[type[ConsumeTaskPlugin]] = ( - [ConsumerPreflightPlugin, ConsumerPlugin] + [ + ConsumerPreflightPlugin, + ConsumerPlugin, + ] if input_doc.root_document_id is not None else [ ConsumerPreflightPlugin, diff --git a/src/documents/tests/test_api_document_versions.py b/src/documents/tests/test_api_document_versions.py index dabbe97f8..48ed37fa8 100644 --- a/src/documents/tests/test_api_document_versions.py +++ b/src/documents/tests/test_api_document_versions.py @@ -1,8 +1,10 @@ from __future__ import annotations +from typing import TYPE_CHECKING from unittest import mock from auditlog.models import LogEntry # type: ignore[import-untyped] +from django.contrib.auth.models import Permission from django.contrib.auth.models import User from django.contrib.contenttypes.models import ContentType from django.core.files.uploadedfile import SimpleUploadedFile @@ -13,6 +15,9 @@ from documents.data_models import DocumentSource from documents.models import Document from documents.tests.utils import DirectoriesMixin +if TYPE_CHECKING: + from pathlib import Path + class TestDocumentVersioningApi(DirectoriesMixin, APITestCase): def setUp(self) -> None: @@ -28,6 +33,27 @@ class TestDocumentVersioningApi(DirectoriesMixin, APITestCase): content_type="application/pdf", ) + def _write_file(self, path: Path, content: bytes = b"data") -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + + def _create_pdf( + self, + *, + title: str, + checksum: str, + root_document: Document | None = None, + ) -> Document: + doc = Document.objects.create( + title=title, + checksum=checksum, + mime_type="application/pdf", + root_document=root_document, + ) + self._write_file(doc.source_path, b"pdf") + self._write_file(doc.thumbnail_path, b"thumb") + return doc + def test_root_endpoint_returns_root_for_version_and_root(self) -> None: root = Document.objects.create( title="root", @@ -49,6 +75,44 @@ class TestDocumentVersioningApi(DirectoriesMixin, APITestCase): self.assertEqual(resp_version.status_code, status.HTTP_200_OK) self.assertEqual(resp_version.data["root_id"], root.id) + def test_root_endpoint_returns_404_for_missing_document(self) -> None: + resp = self.client.get("/api/documents/9999/root/") + + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + def test_root_endpoint_returns_404_when_root_document_missing(self) -> None: + doc = Document( + title="orphan", + checksum="orphan", + mime_type="application/pdf", + ) + doc.root_document_id = 123 + doc.root_document = None + + with mock.patch("documents.views.Document.global_objects") as manager: + manager.select_related.return_value.get.return_value = doc + resp = self.client.get("/api/documents/123/root/") + + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + def test_root_endpoint_returns_403_when_user_lacks_permission(self) -> None: + owner = User.objects.create_user(username="owner") + viewer = User.objects.create_user(username="viewer") + viewer.user_permissions.add( + Permission.objects.get(codename="view_document"), + ) + root = Document.objects.create( + title="root", + checksum="root", + mime_type="application/pdf", + owner=owner, + ) + self.client.force_authenticate(user=viewer) + + resp = self.client.get(f"/api/documents/{root.id}/root/") + + self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) + def test_delete_version_disallows_deleting_root(self) -> None: root = Document.objects.create( title="root", @@ -185,6 +249,138 @@ class TestDocumentVersioningApi(DirectoriesMixin, APITestCase): self.assertFalse(Document.objects.filter(id=version.id).exists()) self.assertEqual(resp.data["current_version_id"], root.id) + def test_delete_version_returns_404_when_root_missing(self) -> None: + resp = self.client.delete("/api/documents/9999/versions/123/") + + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + def test_delete_version_returns_403_without_permission(self) -> None: + owner = User.objects.create_user(username="owner") + other = User.objects.create_user(username="other") + other.user_permissions.add( + Permission.objects.get(codename="delete_document"), + ) + root = Document.objects.create( + title="root", + checksum="root", + mime_type="application/pdf", + owner=owner, + ) + version = Document.objects.create( + title="v1", + checksum="v1", + mime_type="application/pdf", + root_document=root, + ) + self.client.force_authenticate(user=other) + + resp = self.client.delete( + f"/api/documents/{root.id}/versions/{version.id}/", + ) + + self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) + + def test_delete_version_returns_404_when_version_missing(self) -> None: + root = Document.objects.create( + title="root", + checksum="root", + mime_type="application/pdf", + ) + + resp = self.client.delete(f"/api/documents/{root.id}/versions/9999/") + + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + def test_download_version_param_errors(self) -> None: + root = self._create_pdf(title="root", checksum="root") + + resp = self.client.get( + f"/api/documents/{root.id}/download/?version=not-a-number", + ) + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + resp = self.client.get(f"/api/documents/{root.id}/download/?version=9999") + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + other_root = self._create_pdf(title="other", checksum="other") + other_version = self._create_pdf( + title="other-v1", + checksum="other-v1", + root_document=other_root, + ) + resp = self.client.get( + f"/api/documents/{root.id}/download/?version={other_version.id}", + ) + self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) + + def test_download_preview_thumb_with_version_param(self) -> None: + root = self._create_pdf(title="root", checksum="root") + version = self._create_pdf( + title="v1", + checksum="v1", + root_document=root, + ) + self._write_file(version.source_path, b"version") + self._write_file(version.thumbnail_path, b"thumb") + + resp = self.client.get( + f"/api/documents/{root.id}/download/?version={version.id}", + ) + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.content, b"version") + + resp = self.client.get( + f"/api/documents/{root.id}/preview/?version={version.id}", + ) + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.content, b"version") + + resp = self.client.get( + f"/api/documents/{root.id}/thumb/?version={version.id}", + ) + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual(resp.content, b"thumb") + + def test_metadata_version_param_uses_version(self) -> None: + root = Document.objects.create( + title="root", + checksum="root", + mime_type="application/pdf", + ) + version = Document.objects.create( + title="v1", + checksum="v1", + mime_type="application/pdf", + root_document=root, + ) + + with mock.patch("documents.views.DocumentViewSet.get_metadata") as metadata: + metadata.return_value = [] + resp = self.client.get( + f"/api/documents/{root.id}/metadata/?version={version.id}", + ) + + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertTrue(metadata.called) + + def test_metadata_returns_403_when_user_lacks_permission(self) -> None: + owner = User.objects.create_user(username="owner") + other = User.objects.create_user(username="other") + other.user_permissions.add( + Permission.objects.get(codename="view_document"), + ) + doc = Document.objects.create( + title="root", + checksum="root", + mime_type="application/pdf", + owner=owner, + ) + self.client.force_authenticate(user=other) + + resp = self.client.get(f"/api/documents/{doc.id}/metadata/") + + self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) + def test_update_version_enqueues_consume_with_overrides(self) -> None: root = Document.objects.create( title="root", @@ -213,6 +409,24 @@ class TestDocumentVersioningApi(DirectoriesMixin, APITestCase): self.assertEqual(overrides.version_label, "New Version") self.assertEqual(overrides.actor_id, self.user.id) + def test_update_version_returns_500_on_consume_failure(self) -> None: + root = Document.objects.create( + title="root", + checksum="root", + mime_type="application/pdf", + ) + upload = self._make_pdf_upload() + + with mock.patch("documents.views.consume_file") as consume_mock: + consume_mock.delay.side_effect = Exception("boom") + resp = self.client.post( + f"/api/documents/{root.id}/update_version/", + {"document": upload}, + format="multipart", + ) + + self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) + def test_update_version_returns_403_without_permission(self) -> None: owner = User.objects.create_user(username="owner") other = User.objects.create_user(username="other") diff --git a/src/documents/tests/test_consumer.py b/src/documents/tests/test_consumer.py index 717cffd6e..ea5beb08a 100644 --- a/src/documents/tests/test_consumer.py +++ b/src/documents/tests/test_consumer.py @@ -16,6 +16,9 @@ from guardian.core import ObjectPermissionChecker from documents.barcodes import BarcodePlugin from documents.consumer import ConsumerError +from documents.consumer import ConsumerPlugin +from documents.consumer import ConsumerPreflightPlugin +from documents.data_models import ConsumableDocument from documents.data_models import DocumentMetadataOverrides from documents.data_models import DocumentSource from documents.models import Correspondent @@ -29,6 +32,7 @@ from documents.parsers import ParseError from documents.plugins.helpers import ProgressStatusOptions from documents.tasks import sanity_check from documents.tests.utils import DirectoriesMixin +from documents.tests.utils import DummyProgressManager from documents.tests.utils import FileSystemAssertsMixin from documents.tests.utils import GetConsumerMixin from paperless_mail.models import MailRule @@ -664,6 +668,82 @@ class TestConsumer( self._assert_first_last_send_progress() + @mock.patch("documents.consumer.load_classifier") + def test_version_label_override_applies(self, m) -> None: + m.return_value = MagicMock() + + with self.get_consumer( + self.get_test_file(), + DocumentMetadataOverrides(version_label="v1"), + ) as consumer: + consumer.run() + + document = Document.objects.first() + + self.assertEqual(document.version_label, "v1") + + self._assert_first_last_send_progress() + + @override_settings(AUDIT_LOG_ENABLED=True) + @mock.patch("documents.consumer.load_classifier") + def test_consume_version_creates_new_version(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 + + actor = User.objects.create_user( + username="actor", + email="actor@example.com", + password="password", + ) + + version_file = self.get_test_file2() + status = DummyProgressManager(version_file.name, None) + overrides = DocumentMetadataOverrides( + version_label="v2", + actor_id=actor.pk, + ) + 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: + self.assertTrue(consumer.filename.endswith("_v0")) + consumer.run() + finally: + consumer.cleanup() + + versions = Document.objects.filter(root_document=root_doc) + self.assertEqual(versions.count(), 1) + version = versions.first() + assert version is not None + self.assertEqual(version.version_label, "v2") + self.assertTrue(version.original_filename.endswith("_v0")) + @mock.patch("documents.consumer.load_classifier") def testClassifyDocument(self, m) -> None: correspondent = Correspondent.objects.create(