import datetime import json import shutil import tempfile import uuid import zoneinfo from binascii import hexlify from datetime import date from datetime import timedelta from pathlib import Path from typing import TYPE_CHECKING from unittest import mock import celery from dateutil import parser from django.conf import settings from django.contrib.auth.models import Permission from django.contrib.auth.models import User from django.core import mail from django.core.cache import cache from django.core.files.uploadedfile import SimpleUploadedFile from django.db import DataError from django.test import override_settings from django.utils import timezone from guardian.shortcuts import assign_perm from rest_framework import status from rest_framework.test import APITestCase from documents.caching import CACHE_50_MINUTES from documents.caching import CLASSIFIER_HASH_KEY from documents.caching import CLASSIFIER_MODIFIED_KEY from documents.caching import CLASSIFIER_VERSION_KEY from documents.data_models import DocumentSource from documents.models import Correspondent from documents.models import CustomField from documents.models import CustomFieldInstance from documents.models import Document from documents.models import DocumentType from documents.models import MatchingModel from documents.models import Note from documents.models import SavedView from documents.models import ShareLink from documents.models import StoragePath from documents.models import Tag from documents.models import UiSettings from documents.models import Workflow from documents.models import WorkflowAction from documents.models import WorkflowTrigger from documents.signals.handlers import run_workflows from documents.tests.utils import ConsumeTaskMixin from documents.tests.utils import DirectoriesMixin from documents.tests.utils import read_streaming_response class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase): def setUp(self) -> None: super().setUp() self.user = User.objects.create_superuser(username="temp_admin") self.client.force_authenticate(user=self.user) cache.clear() def testDocuments(self) -> None: response = self.client.get("/api/documents/").data self.assertEqual(response["count"], 0) c = Correspondent.objects.create(name="c", pk=41) dt = DocumentType.objects.create(name="dt", pk=63) tag = Tag.objects.create(name="t", pk=85) doc = Document.objects.create( title="WOW", content="the content", correspondent=c, document_type=dt, checksum="123", mime_type="application/pdf", ) doc.tags.add(tag) response = self.client.get("/api/documents/", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["count"], 1) returned_doc = response.data["results"][0] self.assertEqual(returned_doc["id"], doc.id) self.assertEqual(returned_doc["title"], doc.title) self.assertEqual(returned_doc["correspondent"], c.id) self.assertEqual(returned_doc["document_type"], dt.id) self.assertListEqual(returned_doc["tags"], [tag.id]) c2 = Correspondent.objects.create(name="c2") returned_doc["correspondent"] = c2.pk returned_doc["title"] = "the new title" response = self.client.put( f"/api/documents/{doc.pk}/", returned_doc, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) doc_after_save = Document.objects.get(id=doc.id) self.assertEqual(doc_after_save.correspondent, c2) self.assertEqual(doc_after_save.title, "the new title") self.client.delete(f"/api/documents/{doc_after_save.pk}/") self.assertEqual(len(Document.objects.all()), 0) def test_document_fields(self) -> None: c = Correspondent.objects.create(name="c", pk=41) dt = DocumentType.objects.create(name="dt", pk=63) Tag.objects.create(name="t", pk=85) storage_path = StoragePath.objects.create(name="sp", pk=77, path="p") Document.objects.create( title="WOW", content="the content", correspondent=c, document_type=dt, checksum="123", mime_type="application/pdf", storage_path=storage_path, ) response = self.client.get("/api/documents/", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results_full = response.data["results"] self.assertIn("content", results_full[0]) self.assertIn("id", results_full[0]) # Content length is used internally for performance reasons. # No need to expose this field. self.assertNotIn("content_length", results_full[0]) response = self.client.get("/api/documents/?fields=id", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertFalse("content" in results[0]) self.assertIn("id", results[0]) self.assertEqual(len(results[0]), 1) response = self.client.get("/api/documents/?fields=content", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertIn("content", results[0]) self.assertFalse("id" in results[0]) self.assertEqual(len(results[0]), 1) response = self.client.get("/api/documents/?fields=id,content", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertIn("content", results[0]) self.assertIn("id", results[0]) self.assertEqual(len(results[0]), 2) response = self.client.get( "/api/documents/?fields=id,conteasdnt", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertFalse("content" in results[0]) self.assertIn("id", results[0]) self.assertEqual(len(results[0]), 1) response = self.client.get("/api/documents/?fields=", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results_full[0]), len(results[0])) response = self.client.get("/api/documents/?fields=dgfhs", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results[0]), 0) def test_document_fields_respects_created(self) -> None: Document.objects.create( title="legacy", checksum="123", mime_type="application/pdf", created=date(2024, 1, 15), ) response = self.client.get( "/api/documents/?fields=id", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertIn("id", results[0]) self.assertNotIn("created", results[0]) response = self.client.get( "/api/documents/?fields=id,created", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertIn("id", results[0]) self.assertIn("created", results[0]) self.assertEqual(results[0]["created"], "2024-01-15") def test_document_created_format(self) -> None: """ GIVEN: - Existing document WHEN: - Document is requested THEN: - Document created field is returned as date """ doc = Document.objects.create( title="none", checksum="123", mime_type="application/pdf", created=date(2023, 1, 1), ) response = self.client.get( f"/api/documents/{doc.pk}/", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["created"], "2023-01-01") # legacy datetime format response = self.client.patch( f"/api/documents/{doc.pk}/", {"created": "2023-02-01T23:00:00Z"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) doc.refresh_from_db() self.assertEqual(doc.created, date(2023, 2, 1)) # naive datetime response = self.client.patch( f"/api/documents/{doc.pk}/", {"created": "2023-06-28T23:00:00"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) doc.refresh_from_db() self.assertEqual(doc.created, date(2023, 6, 28)) def test_document_update_legacy_created_format(self) -> None: """ GIVEN: - Existing document WHEN: - Document is updated with created in datetime format THEN: - Document created field is updated as date """ doc = Document.objects.create( title="none", checksum="123", mime_type="application/pdf", created=date(2023, 1, 1), ) created_datetime = datetime.datetime(2023, 2, 1, 12, 0, 0) response = self.client.patch( f"/api/documents/{doc.pk}/", {"created": created_datetime}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) doc.refresh_from_db() self.assertEqual(doc.created, date(2023, 2, 1)) def test_document_update_with_created_date(self) -> None: """ GIVEN: - Existing document WHEN: - Document is updated with created_date and not created THEN: - Document created field is updated """ doc = Document.objects.create( title="none", checksum="123", mime_type="application/pdf", created=date(2023, 1, 1), ) created_date = date(2023, 2, 1) self.client.patch( f"/api/documents/{doc.pk}/", {"created_date": created_date}, format="json", ) doc.refresh_from_db() self.assertEqual(doc.created_date, created_date) def test_document_actions(self) -> None: _, filename = tempfile.mkstemp(dir=self.dirs.originals_dir) content = b"This is a test" content_thumbnail = b"thumbnail content" with Path(filename).open("wb") as f: f.write(content) doc = Document.objects.create( title="none", filename=Path(filename).name, mime_type="application/pdf", ) if TYPE_CHECKING: assert isinstance(self.dirs.thumbnail_dir, Path), self.dirs.thumbnail_dir with (self.dirs.thumbnail_dir / f"{doc.pk:07d}.webp").open("wb") as f: f.write(content_thumbnail) response = self.client.get(f"/api/documents/{doc.pk}/download/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content) response = self.client.get(f"/api/documents/{doc.pk}/preview/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content) response = self.client.get(f"/api/documents/{doc.pk}/thumb/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content_thumbnail) def test_document_actions_with_perms(self) -> None: """ GIVEN: - Document with owner and without granted permissions - User is then granted permissions WHEN: - User tries to load preview, thumbnail THEN: - Initially, HTTP 403 Forbidden - With permissions, HTTP 200 OK """ _, filename = tempfile.mkstemp(dir=self.dirs.originals_dir) content = b"This is a test" content_thumbnail = b"thumbnail content" with Path(filename).open("wb") as f: f.write(content) user1 = User.objects.create_user(username="test1") user2 = User.objects.create_user(username="test2") user1.user_permissions.add(*Permission.objects.filter(codename="view_document")) user2.user_permissions.add(*Permission.objects.filter(codename="view_document")) self.client.force_authenticate(user2) doc = Document.objects.create( title="none", filename=Path(filename).name, mime_type="application/pdf", owner=user1, ) with (Path(self.dirs.thumbnail_dir) / f"{doc.pk:07d}.webp").open("wb") as f: f.write(content_thumbnail) response = self.client.get(f"/api/documents/{doc.pk}/download/") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) response = self.client.get(f"/api/documents/{doc.pk}/preview/") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) response = self.client.get(f"/api/documents/{doc.pk}/thumb/") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) assign_perm("view_document", user2, doc) response = self.client.get(f"/api/documents/{doc.pk}/download/") self.assertEqual(response.status_code, status.HTTP_200_OK) response.close() response = self.client.get(f"/api/documents/{doc.pk}/preview/") self.assertEqual(response.status_code, status.HTTP_200_OK) response.close() response = self.client.get(f"/api/documents/{doc.pk}/thumb/") self.assertEqual(response.status_code, status.HTTP_200_OK) response.close() @override_settings(FILENAME_FORMAT="") def test_download_with_archive(self) -> None: content = b"This is a test" content_archive = b"This is the same test but archived" doc = Document.objects.create( title="none", filename="my_document.pdf", archive_filename="archived.pdf", mime_type="application/pdf", ) with Path(doc.source_path).open("wb") as f: f.write(content) with Path(doc.archive_path).open("wb") as f: f.write(content_archive) response = self.client.get(f"/api/documents/{doc.pk}/download/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content_archive) response = self.client.get( f"/api/documents/{doc.pk}/download/?original=true", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content) response = self.client.get(f"/api/documents/{doc.pk}/preview/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content_archive) response = self.client.get( f"/api/documents/{doc.pk}/preview/?original=true", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content) @override_settings(FILENAME_FORMAT="") def test_download_follow_formatting(self) -> None: content = b"This is a test" content_archive = b"This is the same test but archived" doc = Document.objects.create( title="none", filename="my_document.pdf", archive_filename="archived.pdf", mime_type="application/pdf", ) with Path(doc.source_path).open("wb") as f: f.write(content) with Path(doc.archive_path).open("wb") as f: f.write(content_archive) # Without follow_formatting, should use public filename response = self.client.get(f"/api/documents/{doc.pk}/download/") self.assertIn("none.pdf", response["Content-Disposition"]) response.close() # With follow_formatting, should use actual filename on disk response = self.client.get( f"/api/documents/{doc.pk}/download/?follow_formatting=true", ) self.assertIn("archived.pdf", response["Content-Disposition"]) response.close() # With follow_formatting and original, should use source filename response = self.client.get( f"/api/documents/{doc.pk}/download/?original=true&follow_formatting=true", ) self.assertIn("my_document.pdf", response["Content-Disposition"]) response.close() def test_document_actions_not_existing_file(self) -> None: doc = Document.objects.create( title="none", filename=Path("asd").name, mime_type="application/pdf", ) response = self.client.get(f"/api/documents/{doc.pk}/download/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) response = self.client.get(f"/api/documents/{doc.pk}/preview/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) response = self.client.get(f"/api/documents/{doc.pk}/thumb/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_document_actions_trashed_document(self) -> None: """ GIVEN: - Document with files exists WHEN: - Document is soft-deleted (moved to trash) - Preview and thumb endpoints are requested THEN: - HTTP 200 OK for both (trashed documents remain previewable) """ _, filename = tempfile.mkstemp(dir=self.dirs.originals_dir) content = b"This is a test" content_thumbnail = b"thumbnail content" with Path(filename).open("wb") as f: f.write(content) doc = Document.objects.create( title="none", filename=Path(filename).name, mime_type="application/pdf", ) with (self.dirs.thumbnail_dir / f"{doc.pk:07d}.webp").open("wb") as f: f.write(content_thumbnail) doc.delete() response = self.client.get(f"/api/documents/{doc.pk}/preview/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content) response = self.client.get(f"/api/documents/{doc.pk}/thumb/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(read_streaming_response(response), content_thumbnail) def test_document_history_action(self) -> None: """ GIVEN: - Document WHEN: - Document is updated THEN: - Audit log contains changes """ doc = Document.objects.create( title="First title", checksum="123", mime_type="application/pdf", ) self.client.force_login(user=self.user) self.client.patch( f"/api/documents/{doc.pk}/", {"title": "New title"}, format="json", ) response = self.client.get(f"/api/documents/{doc.pk}/history/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 2) self.assertEqual(response.data[0]["actor"]["id"], self.user.id) self.assertEqual(response.data[0]["action"], "update") self.assertEqual( response.data[0]["changes"], {"title": ["First title", "New title"]}, ) def test_document_history_action_w_custom_fields(self) -> None: """ GIVEN: - Document with custom fields WHEN: - Document is updated THEN: - Audit log contains custom field changes """ doc = Document.objects.create( title="First title", checksum="123", mime_type="application/pdf", ) custom_field = CustomField.objects.create( name="custom field str", data_type=CustomField.FieldDataType.STRING, ) self.client.force_login(user=self.user) # Initial response should include only document's creation response = self.client.get(f"/api/documents/{doc.pk}/history/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 1) self.assertIsNone(response.data[0]["actor"]) self.assertEqual(response.data[0]["action"], "create") self.client.patch( f"/api/documents/{doc.pk}/", data={ "custom_fields": [ { "field": custom_field.pk, "value": "custom value", }, ], }, format="json", ) # Second response should include custom field addition response = self.client.get(f"/api/documents/{doc.pk}/history/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 2) self.assertEqual(response.data[0]["actor"]["id"], self.user.id) self.assertEqual(response.data[0]["action"], "create") self.assertEqual( response.data[0]["changes"], { "custom_fields": { "type": "custom_field", "field": "custom field str", "value": "custom value", }, }, ) self.assertIsNone(response.data[1]["actor"]) self.assertEqual(response.data[1]["action"], "create") def test_document_history_logs_version_deletion(self) -> None: root_doc = Document.objects.create( title="Root", checksum="123", mime_type="application/pdf", owner=self.user, ) version_doc = Document.objects.create( title="Version", checksum="456", mime_type="application/pdf", root_document=root_doc, owner=self.user, ) response = self.client.delete( f"/api/documents/{root_doc.pk}/versions/{version_doc.pk}/", ) self.assertEqual(response.status_code, status.HTTP_200_OK) response = self.client.get(f"/api/documents/{root_doc.pk}/history/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 2) self.assertEqual(response.data[0]["actor"]["id"], self.user.id) self.assertEqual(response.data[0]["action"], "update") self.assertEqual( response.data[0]["changes"], {"Version Deleted": ["None", version_doc.pk]}, ) @override_settings(AUDIT_LOG_ENABLED=False) def test_document_history_action_disabled(self) -> None: """ GIVEN: - Audit log is disabled WHEN: - Document is updated - Audit log is requested THEN: - Audit log returns HTTP 400 Bad Request """ doc = Document.objects.create( title="First title", checksum="123", mime_type="application/pdf", ) self.client.force_login(user=self.user) self.client.patch( f"/api/documents/{doc.pk}/", {"title": "New title"}, format="json", ) response = self.client.get(f"/api/documents/{doc.pk}/history/") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_document_history_insufficient_perms(self) -> None: """ GIVEN: - Audit log is enabled WHEN: - History is requested without auditlog permissions - Or is requested as superuser on document with another owner THEN: - History endpoint returns HTTP 403 Forbidden - History is returned """ # No auditlog permissions user = User.objects.create_user(username="test") user.user_permissions.add(*Permission.objects.filter(codename="view_document")) self.client.force_authenticate(user=user) doc = Document.objects.create( title="First title", checksum="123", mime_type="application/pdf", owner=user, ) response = self.client.get(f"/api/documents/{doc.pk}/history/") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) # superuser user.is_superuser = True user.save() user2 = User.objects.create_user(username="test2") doc2 = Document.objects.create( title="Second title", checksum="456", mime_type="application/pdf", owner=user2, ) response = self.client.get(f"/api/documents/{doc2.pk}/history/") self.assertEqual(response.status_code, status.HTTP_200_OK) def test_document_filters(self) -> None: doc1 = Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", ) doc2 = Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", ) doc3 = Document.objects.create( title="none3", checksum="C", mime_type="application/pdf", ) tag_inbox = Tag.objects.create(name="t1", is_inbox_tag=True) tag_2 = Tag.objects.create(name="t2") tag_3 = Tag.objects.create(name="t3") cf1 = CustomField.objects.create( name="stringfield", data_type=CustomField.FieldDataType.STRING, ) cf2 = CustomField.objects.create( name="numberfield", data_type=CustomField.FieldDataType.INT, ) doc1.tags.add(tag_inbox) doc2.tags.add(tag_2) doc3.tags.add(tag_2) doc3.tags.add(tag_3) cf1_d1 = CustomFieldInstance.objects.create( document=doc1, field=cf1, value_text="foobard1", ) CustomFieldInstance.objects.create( document=doc1, field=cf2, value_int=999, ) cf1_d3 = CustomFieldInstance.objects.create( document=doc3, field=cf1, value_text="foobard3", ) response = self.client.get("/api/documents/?is_in_inbox=true") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc1.id) response = self.client.get("/api/documents/?is_in_inbox=false") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) self.assertCountEqual([results[0]["id"], results[1]["id"]], [doc2.id, doc3.id]) response = self.client.get( f"/api/documents/?tags__id__in={tag_inbox.id},{tag_3.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) self.assertCountEqual([results[0]["id"], results[1]["id"]], [doc1.id, doc3.id]) response = self.client.get( f"/api/documents/?tags__id__in={tag_2.id},{tag_3.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) self.assertCountEqual([results[0]["id"], results[1]["id"]], [doc2.id, doc3.id]) response = self.client.get( f"/api/documents/?tags__id__all={tag_2.id},{tag_3.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc3.id) response = self.client.get( f"/api/documents/?tags__id__all={tag_inbox.id},{tag_3.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 0) response = self.client.get( f"/api/documents/?tags__id__all={tag_inbox.id}a{tag_3.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) response = self.client.get(f"/api/documents/?tags__id__none={tag_3.id}") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) self.assertCountEqual([results[0]["id"], results[1]["id"]], [doc1.id, doc2.id]) response = self.client.get( f"/api/documents/?tags__id__none={tag_3.id},{tag_2.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc1.id) response = self.client.get( f"/api/documents/?tags__id__none={tag_2.id},{tag_inbox.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 0) response = self.client.get( f"/api/documents/?id__in={doc1.id},{doc2.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) response = self.client.get( f"/api/documents/?id__range={doc1.id},{doc3.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) response = self.client.get( f"/api/documents/?id={doc2.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) # custom field name response = self.client.get( f"/api/documents/?custom_fields__icontains={cf1.name}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) # custom field value response = self.client.get( f"/api/documents/?custom_fields__icontains={cf1_d1.value}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc1.id) response = self.client.get( f"/api/documents/?custom_fields__icontains={cf1_d3.value}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc3.id) response = self.client.get( "/api/documents/?mime_type=pdf", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) def test_custom_field_select_filter(self) -> None: """ GIVEN: - Documents with select custom field values WHEN: - API request with custom field filtering is made THEN: - Only docs with selected custom field values are returned """ doc1 = Document.objects.create(checksum="1", content="test 1") Document.objects.create(checksum="2", content="test 2") custom_field_select = CustomField.objects.create( name="Test Custom Field Select", data_type=CustomField.FieldDataType.SELECT, extra_data={ "select_options": [ {"label": "Option 1", "id": "abc123"}, {"label": "Choice 2", "id": "def456"}, ], }, ) CustomFieldInstance.objects.create( document=doc1, field=custom_field_select, value_select="def456", ) r = self.client.get("/api/documents/?custom_fields__icontains=choice") self.assertEqual(r.status_code, status.HTTP_200_OK) self.assertEqual(r.data["count"], 1) r = self.client.get("/api/documents/?custom_fields__icontains=option") self.assertEqual(r.status_code, status.HTTP_200_OK) self.assertEqual(r.data["count"], 0) def test_document_checksum_filter(self) -> None: Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", ) doc2 = Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", ) Document.objects.create( title="none3", checksum="C", mime_type="application/pdf", ) response = self.client.get("/api/documents/?checksum__iexact=B") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc2.id) response = self.client.get("/api/documents/?checksum__iexact=X") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 0) def test_document_original_filename_filter(self) -> None: doc1 = Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", original_filename="docA.pdf", ) doc2 = Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", original_filename="docB.pdf", ) doc3 = Document.objects.create( title="none3", checksum="C", mime_type="application/pdf", original_filename="docC.pdf", ) response = self.client.get("/api/documents/?original_filename__iexact=DOCa.pdf") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc1.id) response = self.client.get("/api/documents/?original_filename__iexact=docx.pdf") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 0) response = self.client.get("/api/documents/?original_filename__istartswith=dOc") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) self.assertCountEqual( [results[0]["id"], results[1]["id"], results[2]["id"]], [doc1.id, doc2.id, doc3.id], ) def test_documents_title_content_filter(self) -> None: doc1 = Document.objects.create( title="title A", content="content A", checksum="A", mime_type="application/pdf", ) doc2 = Document.objects.create( title="title B", content="content A", checksum="B", mime_type="application/pdf", ) doc3 = Document.objects.create( title="title A", content="content B", checksum="C", mime_type="application/pdf", ) doc4 = Document.objects.create( title="title B", content="content B", checksum="D", mime_type="application/pdf", ) response = self.client.get("/api/documents/?title_content=A") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) self.assertCountEqual( [results[0]["id"], results[1]["id"], results[2]["id"]], [doc1.id, doc2.id, doc3.id], ) response = self.client.get("/api/documents/?title_content=B") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) self.assertCountEqual( [results[0]["id"], results[1]["id"], results[2]["id"]], [doc2.id, doc3.id, doc4.id], ) response = self.client.get("/api/documents/?title_content=X") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 0) def test_documents_title_content_filter_strips_boundary_whitespace(self) -> None: doc = Document.objects.create( title="Testwort", content="", checksum="A", mime_type="application/pdf", ) response = self.client.get( "/api/documents/", {"title_content": " Testwort "}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], doc.id) def test_document_permissions_filters(self) -> None: """ GIVEN: - Documents with owners, with and without granted permissions WHEN: - User filters by owner THEN: - Owner filters work correctly but still respect permissions """ u1 = User.objects.create_user("user1") u2 = User.objects.create_user("user2") u1.user_permissions.add(*Permission.objects.filter(codename="view_document")) u2.user_permissions.add(*Permission.objects.filter(codename="view_document")) u1_doc1 = Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", owner=u1, ) Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", owner=u2, ) u0_doc1 = Document.objects.create( title="none3", checksum="C", mime_type="application/pdf", ) u1_doc2 = Document.objects.create( title="none4", checksum="D", mime_type="application/pdf", owner=u1, ) u2_doc2 = Document.objects.create( title="none5", checksum="E", mime_type="application/pdf", owner=u2, ) self.client.force_authenticate(user=u1) assign_perm("view_document", u1, u2_doc2) # Will not show any u1 docs or u2_doc1 which isn't shared response = self.client.get(f"/api/documents/?owner__id__none={u1.id}") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 2) self.assertCountEqual( [results[0]["id"], results[1]["id"]], [u0_doc1.id, u2_doc2.id], ) # Will not show any u1 docs, u0_doc1 which has no owner or u2_doc1 which isn't shared response = self.client.get( f"/api/documents/?owner__id__none={u1.id}&owner__isnull=false", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertCountEqual([results[0]["id"]], [u2_doc2.id]) # Will not show any u1 docs, u2_doc2 which is shared but has owner response = self.client.get( f"/api/documents/?owner__id__none={u1.id}&owner__isnull=true", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertCountEqual([results[0]["id"]], [u0_doc1.id]) # Will not show any u1 docs or u2_doc1 which is not shared response = self.client.get(f"/api/documents/?owner__id={u2.id}") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertCountEqual([results[0]["id"]], [u2_doc2.id]) # Will not show u2_doc1 which is not shared response = self.client.get(f"/api/documents/?owner__id__in={u1.id},{u2.id}") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) self.assertCountEqual( [results[0]["id"], results[1]["id"], results[2]["id"]], [u1_doc1.id, u1_doc2.id, u2_doc2.id], ) assign_perm("view_document", u2, u1_doc1) # Will show only documents shared by user response = self.client.get(f"/api/documents/?shared_by__id={u1.id}") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertCountEqual( [results[0]["id"]], [u1_doc1.id], ) def test_pagination_results(self) -> None: """ GIVEN: - A set of 50 documents WHEN: - API request for document filtering THEN: - Results are paginated (25 items) and count reflects all results (50 items) """ t = Tag.objects.create(name="tag") for i in range(50): d = Document.objects.create(checksum=i, content=f"test{i}") d.tags.add(t) response = self.client.get( f"/api/documents/?tags__id__in={t.id}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 25) self.assertEqual(response.data["count"], 50) self.assertNotIn("all", response.data) def test_pagination_all_for_api_version_9(self) -> None: """ GIVEN: - A set of documents matching a filter WHEN: - API request uses legacy version 9 THEN: - Response includes "all" for backward compatibility """ t = Tag.objects.create(name="tag") docs = [] for i in range(4): d = Document.objects.create(checksum=i, content=f"test{i}") d.tags.add(t) docs.append(d) response = self.client.get( f"/api/documents/?tags__id__in={t.id}", headers={"Accept": "application/json; version=9"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn("all", response.data) self.assertCountEqual(response.data["all"], [d.id for d in docs]) def test_default_ordering_uses_id_as_tiebreaker(self) -> None: """ GIVEN: - Documents sharing the same created date WHEN: - API request for documents without an explicit ordering THEN: - Results are correctly ordered by created > id """ older_doc = Document.objects.create( checksum="older", content="older", created=date(2024, 1, 1), ) first_same_date_doc = Document.objects.create( checksum="same-date-1", content="same-date-1", created=date(2024, 1, 2), ) second_same_date_doc = Document.objects.create( checksum="same-date-2", content="same-date-2", created=date(2024, 1, 2), ) response = self.client.get("/api/documents/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( [result["id"] for result in response.data["results"]], [ second_same_date_doc.id, first_same_date_doc.id, older_doc.id, ], ) def test_list_with_include_selection_data(self) -> None: correspondent = Correspondent.objects.create(name="c1") doc_type = DocumentType.objects.create(name="dt1") storage_path = StoragePath.objects.create(name="sp1") tag = Tag.objects.create(name="tag") matching_doc = Document.objects.create( checksum="A", correspondent=correspondent, document_type=doc_type, storage_path=storage_path, ) matching_doc.tags.add(tag) non_matching_doc = Document.objects.create(checksum="B") non_matching_doc.tags.add(Tag.objects.create(name="other")) response = self.client.get( f"/api/documents/?tags__id__in={tag.id}&include_selection_data=true", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn("selection_data", response.data) selected_correspondent = next( item for item in response.data["selection_data"]["selected_correspondents"] if item["id"] == correspondent.id ) selected_tag = next( item for item in response.data["selection_data"]["selected_tags"] if item["id"] == tag.id ) selected_type = next( item for item in response.data["selection_data"]["selected_document_types"] if item["id"] == doc_type.id ) selected_storage_path = next( item for item in response.data["selection_data"]["selected_storage_paths"] if item["id"] == storage_path.id ) self.assertEqual(selected_correspondent["document_count"], 1) self.assertEqual(selected_tag["document_count"], 1) self.assertEqual(selected_type["document_count"], 1) self.assertEqual(selected_storage_path["document_count"], 1) def test_statistics(self) -> None: doc1 = Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", content="abc", ) Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", content="123", ) Document.objects.create( title="none3", checksum="C", mime_type="text/plain", content="hello", ) tag_inbox = Tag.objects.create(name="t1", is_inbox_tag=True) Tag.objects.create(name="t2") Tag.objects.create(name="t3") Correspondent.objects.create(name="c1") Correspondent.objects.create(name="c2") DocumentType.objects.create(name="dt1") StoragePath.objects.create(name="sp1") StoragePath.objects.create(name="sp2") doc1.tags.add(tag_inbox) response = self.client.get("/api/statistics/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["documents_total"], 3) self.assertEqual(response.data["documents_inbox"], 1) self.assertEqual(response.data["inbox_tags"], [tag_inbox.pk]) self.assertEqual( response.data["document_file_type_counts"][0]["mime_type_count"], 2, ) self.assertEqual( response.data["document_file_type_counts"][1]["mime_type_count"], 1, ) self.assertEqual(response.data["character_count"], 11) self.assertEqual(response.data["tag_count"], 3) self.assertEqual(response.data["correspondent_count"], 2) self.assertEqual(response.data["document_type_count"], 1) self.assertEqual(response.data["storage_path_count"], 2) def test_statistics_excludes_document_versions(self) -> None: root = Document.objects.create( title="root", checksum="A", mime_type="application/pdf", content="root", ) version = Document.objects.create( title="version", checksum="B", mime_type="application/pdf", content="version", root_document=root, version_index=1, ) tag_inbox = Tag.objects.create(name="t1", is_inbox_tag=True) version.tags.add(tag_inbox) response = self.client.get("/api/statistics/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["documents_total"], 1) self.assertEqual(response.data["documents_inbox"], 0) self.assertEqual(response.data["character_count"], 4) self.assertEqual( response.data["document_file_type_counts"][0]["mime_type_count"], 1, ) def test_statistics_no_inbox_tag(self) -> None: Document.objects.create(title="none1", checksum="A") response = self.client.get("/api/statistics/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["documents_inbox"], None) self.assertEqual(response.data["inbox_tags"], None) def test_statistics_multiple_users(self) -> None: """ GIVEN: - Inbox tags with different owners and documents that are accessible to different users WHEN: - Statistics are requested THEN: - Statistics only include inbox counts for tags accessible by the user """ u1 = User.objects.create_user("user1") u2 = User.objects.create_user("user2") inbox_tag_u1 = Tag.objects.create(name="inbox_u1", is_inbox_tag=True, owner=u1) Tag.objects.create(name="inbox_u2", is_inbox_tag=True, owner=u2) doc_u1 = Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", owner=u1, ) doc2_u1 = Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", ) doc_u1.tags.add(inbox_tag_u1) doc2_u1.save() doc2_u1.tags.add(inbox_tag_u1) doc2_u1.save() self.client.force_authenticate(user=u1) response = self.client.get("/api/statistics/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["documents_inbox"], 2) self.client.force_authenticate(user=u2) response = self.client.get("/api/statistics/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["documents_inbox"], 0) def test_statistics_with_statistics_permission(self) -> None: owner = User.objects.create_user("owner") stats_user = User.objects.create_user("stats-user") stats_user.user_permissions.add( Permission.objects.get(codename="view_global_statistics"), ) inbox_tag = Tag.objects.create( name="stats_inbox", is_inbox_tag=True, owner=owner, ) Document.objects.create( title="owned-doc", checksum="stats-A", mime_type="application/pdf", content="abcdef", owner=owner, ).tags.add(inbox_tag) Correspondent.objects.create(name="stats-correspondent", owner=owner) DocumentType.objects.create(name="stats-type", owner=owner) StoragePath.objects.create(name="stats-path", path="archive", owner=owner) self.client.force_authenticate(user=stats_user) response = self.client.get("/api/statistics/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["documents_total"], 1) self.assertEqual(response.data["documents_inbox"], 1) self.assertEqual(response.data["inbox_tags"], [inbox_tag.pk]) self.assertEqual(response.data["character_count"], 6) self.assertEqual(response.data["correspondent_count"], 1) self.assertEqual(response.data["document_type_count"], 1) self.assertEqual(response.data["storage_path_count"], 1) def test_upload(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "simple.pdf") self.assertTrue( input_doc.original_file.resolve(strict=False).is_relative_to( Path(settings.SCRATCH_DIR).resolve(strict=False), ), ) self.assertIsNone(overrides.title) self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.document_type_id) self.assertIsNone(overrides.tag_ids) def test_upload_with_path_traversal_filename_is_reduced_to_basename(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) payload = SimpleUploadedFile( "../../outside.pdf", (Path(__file__).parent / "samples" / "simple.pdf").read_bytes(), content_type="application/pdf", ) response = self.client.post( "/api/documents/post_document/", {"document": payload}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "outside.pdf") self.assertEqual(overrides.filename, "outside.pdf") self.assertNotIn("..", input_doc.original_file.name) self.assertNotIn("..", overrides.filename) self.assertTrue( input_doc.original_file.resolve(strict=False).is_relative_to( Path(settings.SCRATCH_DIR).resolve(strict=False), ), ) def test_upload_with_path_traversal_content_disposition_filename_is_reduced_to_basename( self, ) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) pdf_bytes = (Path(__file__).parent / "samples" / "simple.pdf").read_bytes() boundary = "paperless-boundary" payload = ( ( f"--{boundary}\r\n" 'Content-Disposition: form-data; name="document"; ' 'filename="../../outside.pdf"\r\n' "Content-Type: application/pdf\r\n\r\n" ).encode() + pdf_bytes + f"\r\n--{boundary}--\r\n".encode() ) response = self.client.generic( "POST", "/api/documents/post_document/", payload, content_type=f"multipart/form-data; boundary={boundary}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "outside.pdf") self.assertEqual(overrides.filename, "outside.pdf") self.assertNotIn("..", input_doc.original_file.name) self.assertNotIn("..", overrides.filename) self.assertTrue( input_doc.original_file.resolve(strict=False).is_relative_to( Path(settings.SCRATCH_DIR).resolve(strict=False), ), ) def test_document_filters_use_latest_version_content(self) -> None: root = Document.objects.create( title="versioned root", checksum="root", mime_type="application/pdf", content="root-content", ) version = Document.objects.create( title="versioned root", checksum="v1", mime_type="application/pdf", root_document=root, content="latest-version-content", ) response = self.client.get( "/api/documents/?content__icontains=latest-version-content", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], root.id) self.assertEqual(results[0]["content"], version.content) response = self.client.get( "/api/documents/?title_content=latest-version-content", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], root.id) def test_create_wrong_endpoint(self) -> None: response = self.client.post( "/api/documents/", {}, ) self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) def test_upload_insufficient_permissions(self) -> None: self.client.force_authenticate(user=User.objects.create_user("testuser2")) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f}, ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_upload_empty_metadata(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", { "document": f, "title": "", "correspondent": "", "document_type": "", "storage_path": "", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "simple.pdf") self.assertTrue( input_doc.original_file.resolve(strict=False).is_relative_to( Path(settings.SCRATCH_DIR).resolve(strict=False), ), ) self.assertIsNone(overrides.title) self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.document_type_id) self.assertIsNone(overrides.storage_path_id) self.assertIsNone(overrides.tag_ids) def test_upload_invalid_form(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"documenst": f}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_upload_invalid_file(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.zip").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_upload_with_title(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "title": "my custom title"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) _, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(overrides.title, "my custom title") self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.document_type_id) self.assertIsNone(overrides.tag_ids) def test_upload_with_correspondent(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) c = Correspondent.objects.create(name="test-corres") with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "correspondent": c.id}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) _, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(overrides.correspondent_id, c.id) self.assertIsNone(overrides.title) self.assertIsNone(overrides.document_type_id) self.assertIsNone(overrides.tag_ids) def test_upload_with_invalid_correspondent(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "correspondent": 3456}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_upload_with_document_type(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) dt = DocumentType.objects.create(name="invoice") with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "document_type": dt.id}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) _, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(overrides.document_type_id, dt.id) self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.title) self.assertIsNone(overrides.tag_ids) def test_upload_with_invalid_document_type(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "document_type": 34578}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_upload_with_storage_path(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) sp = StoragePath.objects.create(name="invoices") with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "storage_path": sp.id}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) _, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(overrides.storage_path_id, sp.id) self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.title) self.assertIsNone(overrides.tag_ids) def test_upload_with_invalid_storage_path(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "storage_path": 34578}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_upload_with_tags(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) t1 = Tag.objects.create(name="tag1") t2 = Tag.objects.create(name="tag2") with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "tags": [t2.id, t1.id]}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) _, overrides = self.assert_queue_consumption_task_call_args() self.assertCountEqual(overrides.tag_ids, [t1.id, t2.id]) self.assertIsNone(overrides.document_type_id) self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.title) def test_upload_with_invalid_tags(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) t1 = Tag.objects.create(name="tag1") t2 = Tag.objects.create(name="tag2") with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "tags": [t2.id, t1.id, 734563]}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_upload_with_created(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) created = datetime.datetime( 2022, 5, 12, 0, 0, 0, 0, tzinfo=zoneinfo.ZoneInfo("America/Los_Angeles"), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "created": created}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) _, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(overrides.created, created.date()) def test_upload_with_asn(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "archive_serial_number": 500}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "simple.pdf") self.assertEqual(overrides.filename, "simple.pdf") self.assertIsNone(overrides.correspondent_id) self.assertIsNone(overrides.document_type_id) self.assertIsNone(overrides.tag_ids) self.assertEqual(500, overrides.asn) def test_upload_with_custom_fields(self) -> None: self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) custom_field = CustomField.objects.create( name="stringfield", data_type=CustomField.FieldDataType.STRING, ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", { "document": f, "custom_fields": [custom_field.id], }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "simple.pdf") self.assertEqual(overrides.filename, "simple.pdf") self.assertEqual(overrides.custom_fields, {custom_field.id: None}) def test_upload_with_custom_fields_and_workflow(self) -> None: """ GIVEN: A document with a source file WHEN: Upload the document with custom fields and a workflow THEN: Metadata is set correctly, mimicking what happens in the real consumer plugin """ self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) cf = CustomField.objects.create( name="stringfield", data_type=CustomField.FieldDataType.STRING, ) cf2 = CustomField.objects.create( name="intfield", data_type=CustomField.FieldDataType.INT, ) trigger1 = WorkflowTrigger.objects.create( type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION, sources=f"{DocumentSource.ApiUpload},{DocumentSource.ConsumeFolder},{DocumentSource.MailFetch}", ) action1 = WorkflowAction.objects.create( assign_title="Doc title", ) action1.assign_custom_fields.add(cf2) action1.assign_custom_fields_values = {cf2.id: 123} action1.save() w1 = Workflow.objects.create( name="Workflow 1", order=0, ) w1.triggers.add(trigger1) w1.actions.add(action1) w1.save() with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", { "document": f, "custom_fields": [cf.id], }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() new_overrides, _ = run_workflows( trigger_type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION, document=input_doc, logging_group=None, overrides=overrides, ) overrides.update(new_overrides) self.assertEqual(overrides.custom_fields, {cf.id: None, cf2.id: 123}) def test_upload_with_custom_field_values(self) -> None: """ GIVEN: A document with a source file WHEN: Upload the document with custom fields and values THEN: Metadata is set correctly """ self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) cf_string = CustomField.objects.create( name="stringfield", data_type=CustomField.FieldDataType.STRING, ) cf_int = CustomField.objects.create( name="intfield", data_type=CustomField.FieldDataType.INT, ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", { "document": f, "custom_fields": json.dumps( { str(cf_string.id): "a string", str(cf_int.id): 123, }, ), }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, overrides = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.original_file.name, "simple.pdf") self.assertEqual(overrides.filename, "simple.pdf") self.assertEqual( overrides.custom_fields, {cf_string.id: "a string", cf_int.id: 123}, ) def test_upload_with_custom_fields_errors(self) -> None: """ GIVEN: A document with a source file WHEN: Upload the document with invalid custom fields payloads THEN: The upload is rejected """ self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) error_payloads = [ # Non-integer key in mapping {"custom_fields": json.dumps({"abc": "a string"})}, # List with non-integer entry {"custom_fields": json.dumps(["abc"])}, # Nonexistent id in mapping {"custom_fields": json.dumps({99999999: "a string"})}, # Nonexistent id in list {"custom_fields": json.dumps([99999999])}, # Invalid type (JSON string, not list/dict/int) {"custom_fields": json.dumps("not-a-supported-structure")}, ] for payload in error_payloads: with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: data = {"document": f, **payload} response = self.client.post( "/api/documents/post_document/", data, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.consume_file_mock.assert_not_called() def test_patch_document_integer_custom_field_out_of_range(self) -> None: """ GIVEN: - An integer custom field - A document WHEN: - Patching the document with an integer value exceeding PostgreSQL's range THEN: - HTTP 400 is returned (validation catches the overflow) - No custom field instance is created """ cf_int = CustomField.objects.create( name="intfield", data_type=CustomField.FieldDataType.INT, ) doc = Document.objects.create( title="Doc", checksum="123", mime_type="application/pdf", ) response = self.client.patch( f"/api/documents/{doc.pk}/", { "custom_fields": [ { "field": cf_int.pk, "value": 2**31, # overflow for PostgreSQL integer fields }, ], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("custom_fields", response.data) self.assertEqual(CustomFieldInstance.objects.count(), 0) def test_upload_with_webui_source(self) -> None: """ GIVEN: A document with a source file WHEN: Upload the document with 'from_webui' flag THEN: Consume is called with the source set as WebUI """ self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f, "from_webui": True}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) input_doc, _ = self.assert_queue_consumption_task_call_args() self.assertEqual(input_doc.source, WorkflowTrigger.DocumentSourceChoices.WEB_UI) def test_upload_invalid_pdf(self) -> None: """ GIVEN: Invalid PDF named "*.pdf" that mime_type is in settings.CONSUMER_PDF_RECOVERABLE_MIME_TYPES WHEN: Upload the file THEN: The file is not rejected """ self.consume_file_mock.return_value = celery.result.AsyncResult( id=str(uuid.uuid4()), ) with (Path(__file__).parent / "samples" / "invalid_pdf.pdf").open("rb") as f: response = self.client.post( "/api/documents/post_document/", {"document": f}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) def test_get_metadata(self) -> None: doc = Document.objects.create( title="test", filename="file.pdf", mime_type="image/png", archive_checksum="A", archive_filename="archive.pdf", ) source_file: Path = ( Path(__file__).parent / "samples" / "documents" / "thumbnails" / "0000001.webp" ) archive_file: Path = Path(__file__).parent / "samples" / "simple.pdf" shutil.copy(source_file, doc.source_path) shutil.copy(archive_file, doc.archive_path) response = self.client.get(f"/api/documents/{doc.pk}/metadata/") self.assertEqual(response.status_code, status.HTTP_200_OK) meta = response.data self.assertEqual(meta["original_mime_type"], "image/png") self.assertTrue(meta["has_archive_version"]) self.assertEqual(len(meta["original_metadata"]), 0) self.assertGreater(len(meta["archive_metadata"]), 0) self.assertEqual(meta["media_filename"], "file.pdf") self.assertEqual(meta["archive_media_filename"], "archive.pdf") self.assertEqual(meta["original_size"], Path(source_file).stat().st_size) self.assertEqual(meta["archive_size"], Path(archive_file).stat().st_size) response = self.client.get(f"/api/documents/{doc.pk}/metadata/") self.assertEqual(response.status_code, status.HTTP_200_OK) def test_get_metadata_invalid_doc(self) -> None: response = self.client.get("/api/documents/34576/metadata/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_get_metadata_no_archive(self) -> None: doc = Document.objects.create( title="test", filename="file.pdf", mime_type="application/pdf", ) shutil.copy(Path(__file__).parent / "samples" / "simple.pdf", doc.source_path) response = self.client.get(f"/api/documents/{doc.pk}/metadata/") self.assertEqual(response.status_code, status.HTTP_200_OK) meta = response.data self.assertEqual(meta["original_mime_type"], "application/pdf") self.assertFalse(meta["has_archive_version"]) self.assertGreater(len(meta["original_metadata"]), 0) self.assertIsNone(meta["archive_metadata"]) self.assertIsNone(meta["archive_media_filename"]) def test_get_metadata_missing_files(self) -> None: doc = Document.objects.create( title="test", filename="file.pdf", mime_type="application/pdf", archive_filename="file.pdf", archive_checksum="B", checksum="A", ) response = self.client.get(f"/api/documents/{doc.pk}/metadata/") self.assertEqual(response.status_code, status.HTTP_200_OK) meta = response.data self.assertTrue(meta["has_archive_version"]) self.assertIsNone(meta["original_metadata"]) self.assertIsNone(meta["original_size"]) self.assertIsNone(meta["archive_metadata"]) self.assertIsNone(meta["archive_size"]) def test_get_empty_suggestions(self) -> None: doc = Document.objects.create(title="test", mime_type="application/pdf") response = self.client.get(f"/api/documents/{doc.pk}/suggestions/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( response.data, { "correspondents": [], "tags": [], "document_types": [], "storage_paths": [], "dates": [], }, ) def test_get_suggestions_invalid_doc(self) -> None: response = self.client.get("/api/documents/34676/suggestions/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) @mock.patch("documents.views.get_ai_document_classification") @override_settings(AI_ENABLED=True) def test_suggestions_still_uses_classifier_when_ai_enabled( self, mock_get_ai_classification, ) -> None: doc = Document.objects.create(title="test", mime_type="application/pdf") response = self.client.get(f"/api/documents/{doc.pk}/suggestions/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( response.data, { "correspondents": [], "tags": [], "document_types": [], "storage_paths": [], "dates": [], }, ) mock_get_ai_classification.assert_not_called() @mock.patch("documents.views.match_storage_paths") @mock.patch("documents.views.match_document_types") @mock.patch("documents.views.match_tags") @mock.patch("documents.views.match_correspondents") @override_settings(NUMBER_OF_SUGGESTED_DATES=10) def test_get_suggestions( self, match_correspondents, match_tags, match_document_types, match_storage_paths, ) -> None: doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is an invoice from 12.04.2022!", ) match_correspondents.return_value = [Correspondent(id=88), Correspondent(id=2)] match_tags.return_value = [Tag(id=56), Tag(id=123)] match_document_types.return_value = [DocumentType(id=23)] match_storage_paths.return_value = [StoragePath(id=99), StoragePath(id=77)] response = self.client.get(f"/api/documents/{doc.pk}/suggestions/") self.assertEqual( response.data, { "correspondents": [88, 2], "tags": [56, 123], "document_types": [23], "storage_paths": [99, 77], "dates": ["2022-04-12"], }, ) @mock.patch("documents.views.load_classifier") @mock.patch("documents.views.match_storage_paths") @mock.patch("documents.views.match_document_types") @mock.patch("documents.views.match_tags") @mock.patch("documents.views.match_correspondents") @override_settings(NUMBER_OF_SUGGESTED_DATES=10) def test_get_suggestions_cached( self, match_correspondents, match_tags, match_document_types, match_storage_paths, mocked_load, ) -> None: """ GIVEN: - Request for suggestions for a document WHEN: - Classifier has not been modified THEN: - Subsequent requests are returned alright - ETag and last modified headers are set """ # setup the cache how the classifier does it from documents.classifier import DocumentClassifier settings.MODEL_FILE.touch() classifier_checksum_bytes = b"thisisachecksum" classifier_checksum_hex = hexlify(classifier_checksum_bytes).decode() # Two loads, so two side effects mocked_load.side_effect = [ mock.Mock( last_auto_type_hash=classifier_checksum_bytes, FORMAT_VERSION=DocumentClassifier.FORMAT_VERSION, ), mock.Mock( last_auto_type_hash=classifier_checksum_bytes, FORMAT_VERSION=DocumentClassifier.FORMAT_VERSION, ), ] last_modified = timezone.now() cache.set(CLASSIFIER_MODIFIED_KEY, last_modified, CACHE_50_MINUTES) cache.set(CLASSIFIER_HASH_KEY, classifier_checksum_hex, CACHE_50_MINUTES) cache.set( CLASSIFIER_VERSION_KEY, DocumentClassifier.FORMAT_VERSION, CACHE_50_MINUTES, ) # Mock the matching match_correspondents.return_value = [Correspondent(id=88), Correspondent(id=2)] match_tags.return_value = [Tag(id=56), Tag(id=123)] match_document_types.return_value = [DocumentType(id=23)] match_storage_paths.return_value = [StoragePath(id=99), StoragePath(id=77)] doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is an invoice from 12.04.2022!", ) response = self.client.get(f"/api/documents/{doc.pk}/suggestions/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( response.data, { "correspondents": [88, 2], "tags": [56, 123], "document_types": [23], "storage_paths": [99, 77], "dates": ["2022-04-12"], }, ) self.assertIn("Last-Modified", response.headers) self.assertEqual( response.headers["Last-Modified"], last_modified.strftime("%a, %d %b %Y %H:%M:%S %Z").replace("UTC", "GMT"), ) self.assertIn("ETag", response.headers) self.assertEqual( response.headers["ETag"], f'"{classifier_checksum_hex}:{settings.NUMBER_OF_SUGGESTED_DATES}"', ) response = self.client.get(f"/api/documents/{doc.pk}/suggestions/") self.assertEqual(response.status_code, status.HTTP_200_OK) @mock.patch("documents.views.get_date_parser") @override_settings(NUMBER_OF_SUGGESTED_DATES=0) def test_get_suggestions_dates_disabled( self, mock_get_date_parser: mock.MagicMock, ) -> None: """ GIVEN: - NUMBER_OF_SUGGESTED_DATES = 0 (disables feature) WHEN: - API request for document suggestions THEN: - Dont check for suggested dates at all """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is an invoice from 12.04.2022!", ) self.client.get(f"/api/documents/{doc.pk}/suggestions/") mock_get_date_parser.assert_not_called() def test_saved_views(self) -> None: u1 = User.objects.create_user("user1") u2 = User.objects.create_user("user2") u3 = User.objects.create_user("user3") view_perm = Permission.objects.get(codename="view_savedview") change_perm = Permission.objects.get(codename="change_savedview") for user in [u1, u2, u3]: user.user_permissions.add(view_perm, change_perm) v1 = SavedView.objects.create( owner=u1, name="test1", sort_field="", ) v2 = SavedView.objects.create( owner=u2, name="test2", sort_field="", ) v3 = SavedView.objects.create( owner=u2, name="test3", sort_field="", ) assign_perm("view_savedview", u1, v2) assign_perm("change_savedview", u1, v2) assign_perm("view_savedview", u1, v3) self.client.force_authenticate(user=u1) response = self.client.get("/api/saved_views/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["count"], 3) for view_id in [v1.id, v2.id, v3.id]: self.assertEqual( self.client.get(f"/api/saved_views/{view_id}/").status_code, status.HTTP_200_OK, ) response = self.client.patch( f"/api/saved_views/{v2.id}/", {"sort_field": "added"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) response = self.client.patch( f"/api/saved_views/{v3.id}/", {"sort_field": "added"}, format="json", ) self.assertEqual( response.status_code, status.HTTP_403_FORBIDDEN, ) response = self.client.patch( f"/api/saved_views/{v2.id}/", { "set_permissions": { "view": {"users": [u3.id]}, }, }, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) response = self.client.patch( f"/api/saved_views/{v2.id}/", {"owner": u1.id}, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.client.force_authenticate(user=u3) response = self.client.get("/api/saved_views/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["count"], 0) def test_saved_view_api_version_backward_compatibility(self) -> None: """ GIVEN: - Saved views and UiSettings with visibility preferences WHEN: - API request with version=9 (legacy) - API request with version=10 (current) THEN: - Version 9 returns show_on_dashboard and show_in_sidebar from UiSettings - Version 10 omits these fields (moved to UiSettings) """ v1 = SavedView.objects.create( owner=self.user, name="dashboard_view", sort_field="created", ) v2 = SavedView.objects.create( owner=self.user, name="sidebar_view", sort_field="created", ) v3 = SavedView.objects.create( owner=self.user, name="hidden_view", sort_field="created", ) UiSettings.objects.update_or_create( user=self.user, defaults={ "settings": { "saved_views": { "dashboard_views_visible_ids": [v1.id], "sidebar_views_visible_ids": [v2.id], }, }, }, ) response_v9 = self.client.get( "/api/saved_views/", headers={"Accept": "application/json; version=9"}, format="json", ) self.assertEqual(response_v9.status_code, status.HTTP_200_OK) results_v9 = {r["id"]: r for r in response_v9.data["results"]} self.assertIn("show_on_dashboard", results_v9[v1.id]) self.assertIn("show_in_sidebar", results_v9[v1.id]) self.assertTrue(results_v9[v1.id]["show_on_dashboard"]) self.assertFalse(results_v9[v1.id]["show_in_sidebar"]) self.assertTrue(results_v9[v2.id]["show_in_sidebar"]) self.assertFalse(results_v9[v2.id]["show_on_dashboard"]) self.assertFalse(results_v9[v3.id]["show_on_dashboard"]) self.assertFalse(results_v9[v3.id]["show_in_sidebar"]) response_v10 = self.client.get( "/api/saved_views/", headers={"Accept": "application/json; version=10"}, format="json", ) self.assertEqual(response_v10.status_code, status.HTTP_200_OK) results_v10 = {r["id"]: r for r in response_v10.data["results"]} self.assertNotIn("show_on_dashboard", results_v10[v1.id]) self.assertNotIn("show_in_sidebar", results_v10[v1.id]) def test_saved_view_api_version_9_user_without_ui_settings(self) -> None: """ GIVEN: - User with no UiSettings and a saved view WHEN: - API request with version=9 THEN: - show_on_dashboard and show_in_sidebar are False (default) """ SavedView.objects.create( owner=self.user, name="test_view", sort_field="created", ) UiSettings.objects.filter(user=self.user).delete() response = self.client.get( "/api/saved_views/", headers={"Accept": "application/json; version=9"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) result = response.data["results"][0] self.assertFalse(result["show_on_dashboard"]) self.assertFalse(result["show_in_sidebar"]) def test_saved_view_api_version_9_create_writes_visibility_to_ui_settings( self, ) -> None: """ GIVEN: - No UiSettings for the current user WHEN: - A saved view is created through API version 9 with visibility flags THEN: - Visibility is persisted in UiSettings.saved_views """ UiSettings.objects.filter(user=self.user).delete() response = self.client.post( "/api/saved_views/", { "name": "legacy-v9-create", "sort_field": "created", "filter_rules": [], "show_on_dashboard": True, "show_in_sidebar": False, }, headers={"Accept": "application/json; version=9"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue(response.data["show_on_dashboard"]) self.assertFalse(response.data["show_in_sidebar"]) self.user.refresh_from_db() self.assertTrue(hasattr(self.user, "ui_settings")) saved_view_settings = self.user.ui_settings.settings["saved_views"] self.assertListEqual( saved_view_settings["dashboard_views_visible_ids"], [response.data["id"]], ) self.assertListEqual(saved_view_settings["sidebar_views_visible_ids"], []) def test_saved_view_api_version_9_patch_writes_visibility_to_ui_settings( self, ) -> None: """ GIVEN: - Existing saved views and UiSettings visibility ids WHEN: - A saved view is updated through API version 9 visibility flags THEN: - The per-user UiSettings visibility ids are updated """ v1 = SavedView.objects.create( owner=self.user, name="legacy-v9-patch-1", sort_field="created", ) v2 = SavedView.objects.create( owner=self.user, name="legacy-v9-patch-2", sort_field="created", ) UiSettings.objects.update_or_create( user=self.user, defaults={ "settings": { "saved_views": { "dashboard_views_visible_ids": [v1.id], "sidebar_views_visible_ids": [v1.id, v2.id], }, }, }, ) response = self.client.patch( f"/api/saved_views/{v1.id}/", { "show_on_dashboard": False, }, headers={"Accept": "application/json; version=9"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertFalse(response.data["show_on_dashboard"]) self.assertTrue(response.data["show_in_sidebar"]) self.user.refresh_from_db() saved_view_settings = self.user.ui_settings.settings["saved_views"] self.assertListEqual(saved_view_settings["dashboard_views_visible_ids"], []) self.assertListEqual( saved_view_settings["sidebar_views_visible_ids"], [v1.id, v2.id], ) response = self.client.patch( f"/api/saved_views/{v1.id}/", { "show_in_sidebar": False, }, headers={"Accept": "application/json; version=9"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertFalse(response.data["show_on_dashboard"]) self.assertFalse(response.data["show_in_sidebar"]) self.user.refresh_from_db() saved_view_settings = self.user.ui_settings.settings["saved_views"] self.assertListEqual(saved_view_settings["dashboard_views_visible_ids"], []) self.assertListEqual(saved_view_settings["sidebar_views_visible_ids"], [v2.id]) def test_saved_view_create_update_patch(self) -> None: User.objects.create_user("user1") view = { "name": "test", "sort_field": "created2", "filter_rules": [{"rule_type": 4, "value": "test"}], } response = self.client.post("/api/saved_views/", view, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) v1 = SavedView.objects.get(name="test") self.assertEqual(v1.sort_field, "created2") self.assertEqual(v1.filter_rules.count(), 1) self.assertEqual(v1.owner, self.user) response = self.client.patch( f"/api/saved_views/{v1.id}/", {"sort_reverse": True}, format="json", ) v1 = SavedView.objects.get(id=v1.id) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertTrue(v1.sort_reverse) self.assertEqual(v1.filter_rules.count(), 1) view["filter_rules"] = [{"rule_type": 12, "value": "secret"}] response = self.client.put(f"/api/saved_views/{v1.id}/", view, format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) v1 = SavedView.objects.get(id=v1.id) self.assertEqual(v1.filter_rules.count(), 1) self.assertEqual(v1.filter_rules.first().value, "secret") view["filter_rules"] = [] response = self.client.put(f"/api/saved_views/{v1.id}/", view, format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) v1 = SavedView.objects.get(id=v1.id) self.assertEqual(v1.filter_rules.count(), 0) def test_saved_view_display_options(self) -> None: """ GIVEN: - Saved view WHEN: - Updating display options THEN: - Display options are updated - Display fields are validated """ User.objects.create_user("user1") view = { "name": "test", "sort_field": "created2", "filter_rules": [{"rule_type": 4, "value": "test"}], "page_size": 20, "display_mode": SavedView.DisplayMode.SMALL_CARDS, "display_fields": [ SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, ], } response = self.client.post("/api/saved_views/", view, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) v1 = SavedView.objects.get(name="test") self.assertEqual(v1.page_size, 20) self.assertEqual( v1.display_mode, SavedView.DisplayMode.SMALL_CARDS, ) self.assertEqual( v1.display_fields, [ SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, ], ) response = self.client.patch( f"/api/saved_views/{v1.id}/", { "display_fields": [ SavedView.DisplayFields.TAGS, SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, ], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) v1.refresh_from_db() self.assertEqual( v1.display_fields, [ SavedView.DisplayFields.TAGS, SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, ], ) # Invalid display field response = self.client.patch( f"/api/saved_views/{v1.id}/", { "display_fields": [ "foobar", ], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # empty display fields treated as none response = self.client.patch( f"/api/saved_views/{v1.id}/", { "display_fields": [], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) v1.refresh_from_db() self.assertEqual(v1.display_fields, None) def test_saved_view_display_customfields(self) -> None: """ GIVEN: - Saved view WHEN: - Updating display options with custom fields THEN: - Display filds for custom fields are updated - Display fields for custom fields are validated """ view = { "name": "test", "sort_field": "created2", "filter_rules": [{"rule_type": 4, "value": "test"}], "page_size": 20, "display_mode": SavedView.DisplayMode.SMALL_CARDS, "display_fields": [ SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, ], } response = self.client.post("/api/saved_views/", view, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) v1 = SavedView.objects.get(name="test") custom_field = CustomField.objects.create( name="stringfield", data_type=CustomField.FieldDataType.STRING, ) response = self.client.patch( f"/api/saved_views/{v1.id}/", { "display_fields": [ SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, SavedView.DisplayFields.CUSTOM_FIELD % custom_field.id, ], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) v1.refresh_from_db() self.assertEqual( v1.display_fields, [ str(SavedView.DisplayFields.TITLE), str(SavedView.DisplayFields.CREATED), SavedView.DisplayFields.CUSTOM_FIELD % custom_field.id, ], ) # Custom field not found response = self.client.patch( f"/api/saved_views/{v1.id}/", { "display_fields": [ SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, SavedView.DisplayFields.CUSTOM_FIELD % 99, ], }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_saved_view_cleanup_after_custom_field_deletion(self) -> None: """ GIVEN: - Saved view with custom field in display fields and as sort field WHEN: - Custom field is deleted THEN: - Custom field is removed from display fields and sort field """ custom_field = CustomField.objects.create( name="stringfield", data_type=CustomField.FieldDataType.STRING, ) view = SavedView.objects.create( owner=self.user, name="test", sort_field=SavedView.DisplayFields.CUSTOM_FIELD % custom_field.id, display_fields=[ SavedView.DisplayFields.TITLE, SavedView.DisplayFields.CREATED, SavedView.DisplayFields.CUSTOM_FIELD % custom_field.id, ], ) custom_field.delete() view.refresh_from_db() self.assertEqual(view.sort_field, SavedView.DisplayFields.CREATED) self.assertEqual( view.display_fields, [str(SavedView.DisplayFields.TITLE), str(SavedView.DisplayFields.CREATED)], ) def test_get_logs(self) -> None: log_data = "test\ntest2\n" with (Path(settings.LOGGING_DIR) / "mail.log").open("w") as f: f.write(log_data) with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f: f.write(log_data) response = self.client.get("/api/logs/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertCountEqual(response.data, ["mail", "paperless"]) def test_get_logs_only_when_exist(self) -> None: log_data = "test\ntest2\n" with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f: f.write(log_data) response = self.client.get("/api/logs/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertCountEqual(response.data, ["paperless"]) def test_get_invalid_log(self) -> None: response = self.client.get("/api/logs/bogus_log/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) @override_settings(LOGGING_DIR="bogus_dir") def test_get_nonexistent_log(self) -> None: response = self.client.get("/api/logs/paperless/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_get_log(self) -> None: log_data = "test\ntest2\n" with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f: f.write(log_data) response = self.client.get("/api/logs/paperless/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertListEqual(response.data, ["test", "test2"]) def test_get_log_with_limit(self) -> None: log_data = "test1\ntest2\ntest3\n" with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f: f.write(log_data) response = self.client.get("/api/logs/paperless/", {"limit": 2}) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertListEqual(response.data, ["test2", "test3"]) def test_get_log_with_invalid_limit(self) -> None: log_data = "test1\ntest2\n" with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f: f.write(log_data) response = self.client.get("/api/logs/paperless/", {"limit": "abc"}) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.get("/api/logs/paperless/", {"limit": -5}) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_invalid_regex_other_algorithm(self) -> None: for endpoint in ["correspondents", "tags", "document_types"]: response = self.client.post( f"/api/{endpoint}/", { "name": "test", "matching_algorithm": MatchingModel.MATCH_ANY, "match": "[", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED, endpoint) def test_invalid_regex(self) -> None: for endpoint in ["correspondents", "tags", "document_types"]: response = self.client.post( f"/api/{endpoint}/", { "name": "test", "matching_algorithm": MatchingModel.MATCH_REGEX, "match": "[", }, format="json", ) self.assertEqual( response.status_code, status.HTTP_400_BAD_REQUEST, endpoint, ) def test_valid_regex(self) -> None: for endpoint in ["correspondents", "tags", "document_types"]: response = self.client.post( f"/api/{endpoint}/", { "name": "test", "matching_algorithm": MatchingModel.MATCH_REGEX, "match": "[0-9]", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED, endpoint) def test_regex_no_algorithm(self) -> None: for endpoint in ["correspondents", "tags", "document_types"]: response = self.client.post( f"/api/{endpoint}/", {"name": "test", "match": "[0-9]"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED, endpoint) def test_get_existing_notes(self) -> None: """ GIVEN: - A document with a single note WHEN: - API request for document - API request for document notes is made THEN: - Note is included in the document response - The associated note is returned """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have notes!", ) note = Note.objects.create( note="This is a note.", document=doc, user=self.user, ) response = self.client.get( f"/api/documents/{doc.pk}/", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) resp_data = response.json() self.assertEqual(len(resp_data["notes"]), 1) self.assertEqual(resp_data["notes"][0]["note"], note.note) self.assertEqual(resp_data["notes"][0]["user"]["username"], self.user.username) response = self.client.get( f"/api/documents/{doc.pk}/notes/", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) resp_data = response.json() self.assertEqual(len(resp_data), 1) resp_data = resp_data[0] del resp_data["created"] self.assertDictEqual( resp_data, { "id": note.id, "note": note.note, "user": { "id": note.user.id, "username": note.user.username, "first_name": note.user.first_name, "last_name": note.user.last_name, }, }, ) def test_create_note(self) -> None: """ GIVEN: - Existing document WHEN: - API request is made to add a note THEN: - note is created and associated with document, modified time is updated """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have notes added", created=timezone.now() - timedelta(days=1), ) # set to yesterday doc.modified = timezone.now() - timedelta(days=1) self.assertEqual(doc.modified.day, (timezone.now() - timedelta(days=1)).day) resp = self.client.post( f"/api/documents/{doc.pk}/notes/", data={"note": "this is a posted note"}, ) self.assertEqual(resp.status_code, status.HTTP_200_OK) response = self.client.get( f"/api/documents/{doc.pk}/notes/", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) resp_data = response.json() self.assertEqual(len(resp_data), 1) resp_data = resp_data[0] self.assertEqual(resp_data["note"], "this is a posted note") doc = Document.objects.get(pk=doc.pk) # modified was updated to today self.assertEqual(doc.modified.day, timezone.now().day) def test_create_note_only_saves_document_modified_field(self) -> None: """ GIVEN: - Existing document with a created date WHEN: - API request is made to add a note THEN: - Only the document modified field is persisted by the note endpoint - Other document fields are not rewritten by the note endpoint """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have notes added", created=datetime.date(2026, 3, 31), ) original_save = Document.save with mock.patch.object( Document, "save", autospec=True, side_effect=original_save, ) as save_mock: resp = self.client.post( f"/api/documents/{doc.pk}/notes/", data={"note": "this is a posted note"}, ) self.assertEqual(resp.status_code, status.HTTP_200_OK) doc.refresh_from_db() self.assertEqual(doc.created, datetime.date(2026, 3, 31)) self.assertTrue( any( call.kwargs.get("update_fields") == ["modified"] for call in save_mock.call_args_list if call.args and call.args[0].pk == doc.pk ), ) def test_notes_permissions_aware(self) -> None: """ GIVEN: - Existing document owned by user2 but with granted view perms for user1 WHEN: - API request is made by user1 to add a note or delete THEN: - Notes are neither created nor deleted """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.all()) user1.save() user2 = User.objects.create_user(username="test2") user2.save() doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have notes added", ) doc.owner = user2 doc.save() self.client.force_authenticate(user1) resp = self.client.get( f"/api/documents/{doc.pk}/notes/", format="json", ) self.assertEqual(resp.content, b"Insufficient permissions to view notes") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) assign_perm("view_document", user1, doc) resp = self.client.post( f"/api/documents/{doc.pk}/notes/", data={"note": "this is a posted note"}, ) self.assertEqual(resp.content, b"Insufficient permissions to create notes") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) note = Note.objects.create( note="This is a note.", document=doc, user=user2, ) response = self.client.delete( f"/api/documents/{doc.pk}/notes/?id={note.pk}", format="json", ) self.assertEqual(response.content, b"Insufficient permissions to delete notes") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_delete_note(self) -> None: """ GIVEN: - Existing document, existing note WHEN: - API request is made to delete a note THEN: - note is deleted, document modified is updated """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have notes!", created=timezone.now() - timedelta(days=1), ) # set to yesterday doc.modified = timezone.now() - timedelta(days=1) self.assertEqual(doc.modified.day, (timezone.now() - timedelta(days=1)).day) note = Note.objects.create( note="This is a note.", document=doc, user=self.user, ) response = self.client.delete( f"/api/documents/{doc.pk}/notes/?id={note.pk}", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(Note.objects.all()), 0) doc = Document.objects.get(pk=doc.pk) # modified was updated to today self.assertEqual(doc.modified.day, timezone.now().day) def test_delete_note_missing_id(self) -> None: """ GIVEN: - Existing document WHEN: - API DELETE request to notes endpoint without an id query param - API DELETE request to notes endpoint with an empty id query param THEN: - HTTP 400 is returned """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document", ) response = self.client.delete( f"/api/documents/{doc.pk}/notes/", format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.delete( f"/api/documents/{doc.pk}/notes/?id=", format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_delete_note_invalid_id(self) -> None: """ GIVEN: - Existing document WHEN: - API DELETE request to notes endpoint with a non-integer note id THEN: - HTTP 400 is returned """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document", ) response = self.client.delete( f"/api/documents/{doc.pk}/notes/?id=notaninteger", format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_delete_note_nonexistent_id(self) -> None: """ GIVEN: - Existing document, no notes WHEN: - API DELETE request to notes endpoint with a non-existent note id THEN: - HTTP 404 is returned """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document", ) response = self.client.delete( f"/api/documents/{doc.pk}/notes/?id=99999", format="json", ) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_get_notes_no_doc(self) -> None: """ GIVEN: - A request to get notes from a non-existent document WHEN: - API request for document notes is made THEN: - HTTP status.HTTP_404_NOT_FOUND is returned """ response = self.client.get( "/api/documents/500/notes/", format="json", ) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_tag_unique_name_and_owner(self) -> None: """ GIVEN: - Multiple users - Tags owned by particular users WHEN: - API request for creating items which are unique by name and owner THEN: - Unique items are created - Non-unique items are not allowed """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.filter(codename="add_tag")) user1.save() user2 = User.objects.create_user(username="test2") user2.user_permissions.add(*Permission.objects.filter(codename="add_tag")) user2.save() # User 1 creates tag 1 owned by user 1 by default # No issue self.client.force_authenticate(user1) response = self.client.post("/api/tags/", {"name": "tag 1"}, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) # User 2 creates tag 1 owned by user 2 by default # No issue self.client.force_authenticate(user2) response = self.client.post("/api/tags/", {"name": "tag 1"}, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) # User 2 creates tag 2 owned by user 1 # No issue self.client.force_authenticate(user2) response = self.client.post( "/api/tags/", {"name": "tag 2", "owner": user1.pk}, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) # User 1 creates tag 2 owned by user 1 by default # Not allowed, would create tag2/user1 which already exists self.client.force_authenticate(user1) response = self.client.post( "/api/tags/", {"name": "tag 2"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # User 1 creates tag 2 owned by user 1 # Not allowed, would create tag2/user1 which already exists response = self.client.post( "/api/tags/", {"name": "tag 2", "owner": user1.pk}, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_tag_unique_name_and_owner_enforced_on_update(self) -> None: """ GIVEN: - Multiple users - Tags owned by particular users WHEN: - API request for to update tag in such as way as makes it non-unqiue THEN: - Unique items are created - Non-unique items are not allowed on update """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.filter(codename="change_tag")) user1.save() user2 = User.objects.create_user(username="test2") user2.user_permissions.add(*Permission.objects.filter(codename="change_tag")) user2.save() # Create name tag 1 owned by user 1 # Create name tag 1 owned by user 2 Tag.objects.create(name="tag 1", owner=user1) tag2 = Tag.objects.create(name="tag 1", owner=user2) # User 2 attempts to change the owner of tag to user 1 # Not allowed, would change to tag1/user1 which already exists self.client.force_authenticate(user2) response = self.client.patch( f"/api/tags/{tag2.id}/", {"owner": user1.pk}, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_create_share_links(self) -> None: """ GIVEN: - Existing document WHEN: - API request is made to generate a share_link - API request is made to view share_links on incorrect doc pk - Invalid method request is made to view share_links doc THEN: - Link is created with a slug and associated with document - 404 - Error """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have notes added", ) # never expires resp = self.client.post( "/api/share_links/", data={ "document": doc.pk, }, ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) resp = self.client.post( "/api/share_links/", data={ "expiration": (timezone.now() + timedelta(days=7)).isoformat(), "document": doc.pk, "file_version": "original", }, ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) response = self.client.get( f"/api/documents/{doc.pk}/share_links/", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) resp_data = response.json() self.assertEqual(len(resp_data), 2) self.assertGreater(len(resp_data[1]["slug"]), 0) self.assertIsNone(resp_data[1]["expiration"]) self.assertEqual( (parser.isoparse(resp_data[0]["expiration"]) - timezone.now()).days, 6, ) sl1 = ShareLink.objects.get(slug=resp_data[1]["slug"]) self.assertEqual(str(sl1), f"Share Link for {doc.title}") response = self.client.post( f"/api/documents/{doc.pk}/share_links/", format="json", ) self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) response = self.client.get( "/api/documents/99/share_links/", format="json", ) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_share_links_permissions_aware(self) -> None: """ GIVEN: - Existing document owned by user2 but with granted view perms for user1 WHEN: - API request is made by user1 to view share links THEN: - Links only shown if user has permissions """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.all()) user1.save() user2 = User.objects.create_user(username="test2") user2.save() doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will have share links added", ) doc.owner = user2 doc.save() self.client.force_authenticate(user1) resp = self.client.get( f"/api/documents/{doc.pk}/share_links/", format="json", ) self.assertEqual(resp.content, b"Insufficient permissions to add share link") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) assign_perm("change_document", user1, doc) resp = self.client.get( f"/api/documents/{doc.pk}/share_links/", format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) def test_create_share_link_requires_view_permission_for_document(self) -> None: """ GIVEN: - A user with add_sharelink but without view permission on a document WHEN: - API request is made to create a share link for that document THEN: - Share link creation is denied until view permission is granted """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.filter(codename="add_sharelink")) user1.save() user2 = User.objects.create_user(username="test2") user2.save() doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document which will be protected", owner=user2, ) self.client.force_authenticate(user1) create_resp = self.client.post( "/api/share_links/", data={ "document": doc.pk, "file_version": "original", }, format="json", ) self.assertEqual(create_resp.status_code, status.HTTP_403_FORBIDDEN) assign_perm("view_document", user1, doc) create_resp = self.client.post( "/api/share_links/", data={ "document": doc.pk, "file_version": "original", }, format="json", ) self.assertEqual(create_resp.status_code, status.HTTP_201_CREATED) self.assertEqual(create_resp.data["document"], doc.pk) def test_next_asn(self) -> None: """ GIVEN: - Existing documents with ASNs, highest owned by user2 WHEN: - API request is made by user1 to get next ASN THEN: - ASN +1 from user2's doc is returned for user1 """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.all()) user1.save() user2 = User.objects.create_user(username="test2") user2.save() doc1 = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", archive_serial_number=998, ) doc1.owner = user1 doc1.save() doc2 = Document.objects.create( title="test2", mime_type="application/pdf", content="this is a document 2 with higher ASN", checksum="2", archive_serial_number=999, ) doc2.owner = user2 doc2.save() self.client.force_authenticate(user1) resp = self.client.get( "/api/documents/next_asn/", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.content, b"1000") def test_next_asn_no_documents_with_asn(self) -> None: """ GIVEN: - Existing document, but with no ASN assugned WHEN: - API request to get next ASN THEN: - ASN 1 is returned """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.all()) user1.save() doc1 = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", ) doc1.save() self.client.force_authenticate(user1) resp = self.client.get( "/api/documents/next_asn/", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.content, b"1") def test_asn_not_unique_with_trashed_doc(self) -> None: """ GIVEN: - Existing document with ASN that is trashed WHEN: - API request to update document with same ASN THEN: - Explicit error is returned """ user1 = User.objects.create_superuser(username="test1") self.client.force_authenticate(user1) doc1 = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", archive_serial_number=1, ) doc1.delete() doc2 = Document.objects.create( title="test2", mime_type="application/pdf", content="this is a document 2", checksum="2", ) result = self.client.patch( f"/api/documents/{doc2.pk}/", { "archive_serial_number": 1, }, ) self.assertEqual(result.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( result.json(), { "archive_serial_number": [ "Document with this Archive Serial Number already exists in the trash.", ], }, ) def test_remove_inbox_tags(self) -> None: """ GIVEN: - Existing document with or without inbox tags WHEN: - API request to update document, with or without `remove_inbox_tags` flag THEN: - Inbox tags are removed as long as they are not being added """ tag1 = Tag.objects.create(name="tag1", color="#abcdef") inbox_tag1 = Tag.objects.create( name="inbox1", color="#abcdef", is_inbox_tag=True, ) inbox_tag2 = Tag.objects.create( name="inbox2", color="#abcdef", is_inbox_tag=True, ) doc1 = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", ) doc1.tags.add(tag1) doc1.tags.add(inbox_tag1) doc1.tags.add(inbox_tag2) doc1.save() # Remove inbox tags defaults to false resp = self.client.patch( f"/api/documents/{doc1.pk}/", { "title": "New title", }, ) doc1.refresh_from_db() self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(doc1.tags.count(), 3) # Remove inbox tags set to true resp = self.client.patch( f"/api/documents/{doc1.pk}/", { "remove_inbox_tags": True, }, ) doc1.refresh_from_db() self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(doc1.tags.count(), 1) # Remove inbox tags set to true but adding a new inbox tag resp = self.client.patch( f"/api/documents/{doc1.pk}/", { "remove_inbox_tags": True, "tags": [inbox_tag1.pk, tag1.pk], }, ) doc1.refresh_from_db() self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(doc1.tags.count(), 2) @override_settings( EMAIL_ENABLED=True, EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend", ) def test_email_document(self) -> None: """ GIVEN: - Existing document WHEN: - API request is made to email document action THEN: - Email is sent, with document (original or archive) attached """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", filename="test.pdf", archive_checksum="A", archive_filename="archive.pdf", ) doc2 = Document.objects.create( title="test2", mime_type="application/pdf", content="this is a document 2", checksum="2", filename="test2.pdf", ) archive_file = Path(__file__).parent / "samples" / "simple.pdf" source_file = Path(__file__).parent / "samples" / "simple.pdf" shutil.copy(archive_file, doc.archive_path) shutil.copy(source_file, doc2.source_path) self.client.post( f"/api/documents/{doc.pk}/email/", { "addresses": "hello@paperless-ngx.com", "subject": "test", "message": "hello", }, ) self.assertEqual(len(mail.outbox), 1) expected_filename = f"{doc.created} test.pdf" self.assertEqual(mail.outbox[0].attachments[0][0], expected_filename) self.client.post( f"/api/documents/{doc2.pk}/email/", { "addresses": "hello@paperless-ngx.com", "subject": "test", "message": "hello", "use_archive_version": False, }, ) self.assertEqual(len(mail.outbox), 2) expected_filename2 = f"{doc2.created} test2.pdf" self.assertEqual(mail.outbox[1].attachments[0][0], expected_filename2) @mock.patch("django.core.mail.message.EmailMessage.send", side_effect=Exception) def test_email_document_errors(self, mocked_send) -> None: """ GIVEN: - Existing document WHEN: - API request is made to email document action with insufficient permissions - API request is made to email document action with invalid document id - API request is made to email document action with missing data - API request is made to email document action with invalid email address - API request is made to email document action and error occurs during email send THEN: - Error response is returned """ user1 = User.objects.create_user(username="test1") user1.user_permissions.add(*Permission.objects.all()) user1.save() doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", filename="test.pdf", archive_checksum="A", archive_filename="archive.pdf", ) doc2 = Document.objects.create( title="test2", mime_type="application/pdf", content="this is a document 2", checksum="2", owner=self.user, ) self.client.force_authenticate(user1) resp = self.client.post( f"/api/documents/{doc2.pk}/email/", { "addresses": "hello@paperless-ngx.com", "subject": "test", "message": "hello", }, ) self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) resp = self.client.post( "/api/documents/999/email/", { "addresses": "hello@paperless-ngx.com", "subject": "test", "message": "hello", }, ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) resp = self.client.post( f"/api/documents/{doc.pk}/email/", { "addresses": "hello@paperless-ngx.com", }, ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) resp = self.client.post( f"/api/documents/{doc.pk}/email/", { "addresses": "hello@paperless-ngx.com,hello", "subject": "test", "message": "hello", }, ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) resp = self.client.post( f"/api/documents/{doc.pk}/email/", { "addresses": "hello@paperless-ngx.com", "subject": "test", "message": "hello", }, ) self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) @mock.patch("django_softdelete.models.SoftDeleteModel.delete") def test_warn_on_delete_with_old_uuid_field(self, mocked_delete) -> None: """ GIVEN: - Existing document in a (mocked) MariaDB database with an old UUID field WHEN: - API request to delete document is made which raises "Data too long for column" error THEN: - Warning is logged alerting the user of the issue (and link to the fix) """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document 1", checksum="1", ) mocked_delete.side_effect = DataError( "Data too long for column 'transaction_id' at row 1", ) with self.assertLogs(level="WARNING") as cm: self.client.delete(f"/api/documents/{doc.pk}/") self.assertIn( "Detected a possible incompatible database column", cm.output[0], ) class TestDocumentApiTagColors(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() self.user = User.objects.create_superuser(username="temp_admin") self.client.force_authenticate(user=self.user) def test_tag_validate_color(self) -> None: self.assertEqual( self.client.post( "/api/tags/", {"name": "test", "color": "#12fFaA"}, format="json", ).status_code, status.HTTP_201_CREATED, ) self.assertEqual( self.client.post( "/api/tags/", {"name": "test1", "color": "abcdef"}, format="json", ).status_code, status.HTTP_400_BAD_REQUEST, ) self.assertEqual( self.client.post( "/api/tags/", {"name": "test2", "color": "#abcdfg"}, format="json", ).status_code, status.HTTP_400_BAD_REQUEST, ) self.assertEqual( self.client.post( "/api/tags/", {"name": "test3", "color": "#asd"}, format="json", ).status_code, status.HTTP_400_BAD_REQUEST, ) self.assertEqual( self.client.post( "/api/tags/", {"name": "test4", "color": "#12121212"}, format="json", ).status_code, status.HTTP_400_BAD_REQUEST, ) def test_tag_text_color(self) -> None: t = Tag.objects.create(name="tag1", color="#000000") self.assertEqual( self.client.get(f"/api/tags/{t.id}/", format="json").data["text_color"], "#ffffff", ) t.color = "#ffffff" t.save() self.assertEqual( self.client.get(f"/api/tags/{t.id}/", format="json").data["text_color"], "#000000", ) t.color = "asdf" t.save() self.assertEqual( self.client.get(f"/api/tags/{t.id}/", format="json").data["text_color"], "#000000", ) t.color = "123" t.save() self.assertEqual( self.client.get(f"/api/tags/{t.id}/", format="json").data["text_color"], "#000000", ) class TestDocumentApiCustomFieldsSorting(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() self.user = User.objects.create_superuser(username="temp_admin") self.client.force_authenticate(user=self.user) self.doc1 = Document.objects.create( title="none1", checksum="A", mime_type="application/pdf", ) self.doc2 = Document.objects.create( title="none2", checksum="B", mime_type="application/pdf", ) self.doc3 = Document.objects.create( title="none3", checksum="C", mime_type="application/pdf", ) cache.clear() def test_document_custom_fields_sorting(self) -> None: """ GIVEN: - Documents with custom fields WHEN: - API request for document filtering with custom field sorting THEN: - Documents are sorted by custom field values """ values = { CustomField.FieldDataType.STRING: { "values": ["foo", "bar", "baz"], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.STRING ], }, CustomField.FieldDataType.INT: { "values": [3, 1, 2], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.INT ], }, CustomField.FieldDataType.FLOAT: { "values": [3.3, 1.1, 2.2], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.FLOAT ], }, CustomField.FieldDataType.BOOL: { "values": [True, False, False], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.BOOL ], }, CustomField.FieldDataType.DATE: { "values": [date(2021, 1, 3), date(2021, 1, 1), date(2021, 1, 2)], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.DATE ], }, CustomField.FieldDataType.URL: { "values": [ "http://example.org", "http://example.com", "http://example.net", ], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.URL ], }, CustomField.FieldDataType.MONETARY: { "values": ["USD789.00", "USD123.00", "USD456.00"], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.MONETARY ], }, CustomField.FieldDataType.DOCUMENTLINK: { "values": [self.doc3.pk, self.doc1.pk, self.doc2.pk], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.DOCUMENTLINK ], }, CustomField.FieldDataType.SELECT: { "values": ["ghi-789", "abc-123", "def-456"], "field_name": CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[ CustomField.FieldDataType.SELECT ], "extra_data": { "select_options": [ {"label": "Option 1", "id": "abc-123"}, {"label": "Option 2", "id": "def-456"}, {"label": "Option 3", "id": "ghi-789"}, ], }, }, } for data_type, data in values.items(): CustomField.objects.all().delete() CustomFieldInstance.objects.all().delete() custom_field = CustomField.objects.create( name=f"custom field {data_type}", data_type=data_type, extra_data=data.get("extra_data", {}), ) for i, value in enumerate(data["values"]): CustomFieldInstance.objects.create( document=[self.doc1, self.doc2, self.doc3][i], field=custom_field, **{data["field_name"]: value}, ) response = self.client.get( f"/api/documents/?ordering=custom_field_{custom_field.pk}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) self.assertEqual( [results[0]["id"], results[1]["id"], results[2]["id"]], [self.doc2.id, self.doc3.id, self.doc1.id], ) response = self.client.get( f"/api/documents/?ordering=-custom_field_{custom_field.pk}", ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data["results"] self.assertEqual(len(results), 3) if data_type == CustomField.FieldDataType.BOOL: # just check the first one for bools, as the rest are the same self.assertEqual( [results[0]["id"]], [self.doc1.id], ) else: self.assertEqual( [results[0]["id"], results[1]["id"], results[2]["id"]], [self.doc1.id, self.doc3.id, self.doc2.id], ) def test_document_custom_fields_sorting_invalid(self) -> None: """ GIVEN: - Documents with custom fields WHEN: - API request for document filtering with invalid custom field sorting THEN: - 400 is returned """ response = self.client.get( "/api/documents/?ordering=custom_field_999", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_document_custom_fields_sorting_invalid_data_type(self) -> None: """ GIVEN: - Documents with custom fields WHEN: - API request for document filtering with a custom field sorting with a new (unhandled) data type THEN: - Error is raised """ custom_field = CustomField.objects.create( name="custom field", data_type="foo", ) with self.assertRaises(ValueError): self.client.get( f"/api/documents/?ordering=custom_field_{custom_field.pk}", )