Merge branch 'main' into dev

# Conflicts:
#	docs/setup.md
#	src-ui/src/main.ts
#	src/documents/tests/test_api_bulk_edit.py
#	src/documents/tests/test_api_custom_fields.py
#	src/documents/tests/test_api_search.py
#	src/documents/tests/test_api_status.py
#	src/documents/tests/test_workflows.py
#	src/paperless_mail/tests/test_api.py
This commit is contained in:
shamoon
2026-03-21 02:12:19 -07:00
23 changed files with 896 additions and 33 deletions
+44
View File
@@ -262,6 +262,50 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
self.assertEqual(kwargs["add_custom_fields"], [self.cf1.id])
self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id])
@mock.patch("documents.serialisers.bulk_edit.modify_custom_fields")
def test_api_modify_custom_fields_documentlink_forbidden_for_unpermitted_target(
self,
m,
) -> None:
self.setup_mock(m, "modify_custom_fields")
user = User.objects.create_user(username="doc-owner")
user.user_permissions.add(Permission.objects.get(codename="change_document"))
other_user = User.objects.create_user(username="other-user")
source_doc = Document.objects.create(
checksum="source",
title="Source",
owner=user,
)
target_doc = Document.objects.create(
checksum="target",
title="Target",
owner=other_user,
)
doclink_field = CustomField.objects.create(
name="doclink",
data_type=CustomField.FieldDataType.DOCUMENTLINK,
)
self.client.force_authenticate(user=user)
response = self.client.post(
"/api/documents/bulk_edit/",
json.dumps(
{
"documents": [source_doc.id],
"method": "modify_custom_fields",
"parameters": {
"add_custom_fields": {doclink_field.id: [target_doc.id]},
"remove_custom_fields": [],
},
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
m.assert_not_called()
@mock.patch("documents.serialisers.bulk_edit.modify_custom_fields")
def test_api_modify_custom_fields_with_values(self, m) -> None:
self.setup_mock(m, "modify_custom_fields")
@@ -6,6 +6,7 @@ from unittest.mock import ANY
from django.contrib.auth.models import Permission
from django.contrib.auth.models import User
from django.test import override_settings
from guardian.shortcuts import assign_perm
from rest_framework import status
from rest_framework.test import APITestCase
@@ -1140,6 +1141,102 @@ class TestCustomFieldsAPI(DirectoriesMixin, APITestCase):
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(doc5.custom_fields.first().value, [1])
def test_documentlink_patch_requires_change_permission_on_target_documents(
self,
) -> None:
source_owner = User.objects.create_user(username="source-owner")
source_owner.user_permissions.add(
Permission.objects.get(codename="change_document"),
)
other_user = User.objects.create_user(username="other-user")
source_doc = Document.objects.create(
title="Source",
checksum="source",
mime_type="application/pdf",
owner=source_owner,
)
target_doc = Document.objects.create(
title="Target",
checksum="target",
mime_type="application/pdf",
owner=other_user,
)
custom_field_doclink = CustomField.objects.create(
name="Test Custom Field Doc Link",
data_type=CustomField.FieldDataType.DOCUMENTLINK,
)
self.client.force_authenticate(user=source_owner)
resp = self.client.patch(
f"/api/documents/{source_doc.id}/",
data={
"custom_fields": [
{
"field": custom_field_doclink.id,
"value": [target_doc.id],
},
],
},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
self.assertEqual(
CustomFieldInstance.objects.filter(field=custom_field_doclink).count(),
0,
)
def test_documentlink_patch_allowed_with_change_permission_on_target_documents(
self,
) -> None:
source_owner = User.objects.create_user(username="source-owner")
source_owner.user_permissions.add(
Permission.objects.get(codename="change_document"),
)
other_user = User.objects.create_user(username="other-user")
source_doc = Document.objects.create(
title="Source",
checksum="source",
mime_type="application/pdf",
owner=source_owner,
)
target_doc = Document.objects.create(
title="Target",
checksum="target",
mime_type="application/pdf",
owner=other_user,
)
custom_field_doclink = CustomField.objects.create(
name="Test Custom Field Doc Link",
data_type=CustomField.FieldDataType.DOCUMENTLINK,
)
assign_perm("change_document", source_owner, target_doc)
self.client.force_authenticate(user=source_owner)
resp = self.client.patch(
f"/api/documents/{source_doc.id}/",
data={
"custom_fields": [
{
"field": custom_field_doclink.id,
"value": [target_doc.id],
},
],
},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
target_doc.refresh_from_db()
self.assertEqual(
target_doc.custom_fields.get(field=custom_field_doclink).value,
[source_doc.id],
)
def test_custom_field_filters(self) -> None:
custom_field_string = CustomField.objects.create(
name="Test Custom Field String",
+165
View File
@@ -702,6 +702,40 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
self.assertEqual(correction, None)
def test_search_spelling_suggestion_suppressed_for_private_terms(self):
owner = User.objects.create_user("owner")
attacker = User.objects.create_user("attacker")
attacker.user_permissions.add(
Permission.objects.get(codename="view_document"),
)
with AsyncWriter(index.open_index()) as writer:
for i in range(55):
private_doc = Document.objects.create(
checksum=f"p{i}",
pk=100 + i,
title=f"Private Document {i + 1}",
content=f"treasury document {i + 1}",
owner=owner,
)
visible_doc = Document.objects.create(
checksum=f"v{i}",
pk=200 + i,
title=f"Visible Document {i + 1}",
content=f"public ledger {i + 1}",
owner=attacker,
)
index.update_document(writer, private_doc)
index.update_document(writer, visible_doc)
self.client.force_authenticate(user=attacker)
response = self.client.get("/api/documents/?query=treasurx")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["count"], 0)
self.assertIsNone(response.data["corrected_query"])
@mock.patch(
"whoosh.searching.Searcher.correct_query",
side_effect=Exception("Test error"),
@@ -772,6 +806,60 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
self.assertEqual(results[0]["id"], d3.id)
self.assertEqual(results[1]["id"], d1.id)
def test_search_more_like_requires_view_permission_on_seed_document(
self,
) -> None:
"""
GIVEN:
- A user can search documents they own
- Another user's private document exists with similar content
WHEN:
- The user requests more-like-this for the private seed document
THEN:
- The request is rejected
"""
owner = User.objects.create_user("owner")
attacker = User.objects.create_user("attacker")
attacker.user_permissions.add(
Permission.objects.get(codename="view_document"),
)
private_seed = Document.objects.create(
title="private bank statement",
content="quarterly treasury bank statement wire transfer",
checksum="seed",
owner=owner,
pk=10,
)
visible_doc = Document.objects.create(
title="attacker-visible match",
content="quarterly treasury bank statement wire transfer summary",
checksum="visible",
owner=attacker,
pk=11,
)
other_doc = Document.objects.create(
title="unrelated",
content="completely different topic",
checksum="other",
owner=attacker,
pk=12,
)
with AsyncWriter(index.open_index()) as writer:
index.update_document(writer, private_seed)
index.update_document(writer, visible_doc)
index.update_document(writer, other_doc)
self.client.force_authenticate(user=attacker)
response = self.client.get(
f"/api/documents/?more_like_id={private_seed.id}",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertEqual(response.content, b"Insufficient permissions.")
def test_search_filtering(self) -> None:
t = Tag.objects.create(name="tag")
t2 = Tag.objects.create(name="tag2")
@@ -1356,6 +1444,83 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
self.assertEqual(results["custom_fields"][0]["id"], custom_field1.id)
self.assertEqual(results["workflows"][0]["id"], workflow1.id)
def test_global_search_filters_owned_mail_objects(self) -> None:
user1 = User.objects.create_user("mail-search-user")
user2 = User.objects.create_user("other-mail-search-user")
user1.user_permissions.add(
Permission.objects.get(codename="view_mailaccount"),
Permission.objects.get(codename="view_mailrule"),
)
own_account = MailAccount.objects.create(
name="bank owned account",
username="owner@example.com",
password="secret",
imap_server="imap.owner.example.com",
imap_port=993,
imap_security=MailAccount.ImapSecurity.SSL,
character_set="UTF-8",
owner=user1,
)
other_account = MailAccount.objects.create(
name="bank other account",
username="other@example.com",
password="secret",
imap_server="imap.other.example.com",
imap_port=993,
imap_security=MailAccount.ImapSecurity.SSL,
character_set="UTF-8",
owner=user2,
)
unowned_account = MailAccount.objects.create(
name="bank shared account",
username="shared@example.com",
password="secret",
imap_server="imap.shared.example.com",
imap_port=993,
imap_security=MailAccount.ImapSecurity.SSL,
character_set="UTF-8",
)
own_rule = MailRule.objects.create(
name="bank owned rule",
account=own_account,
action=MailRule.MailAction.MOVE,
owner=user1,
)
other_rule = MailRule.objects.create(
name="bank other rule",
account=other_account,
action=MailRule.MailAction.MOVE,
owner=user2,
)
unowned_rule = MailRule.objects.create(
name="bank shared rule",
account=unowned_account,
action=MailRule.MailAction.MOVE,
)
self.client.force_authenticate(user1)
response = self.client.get("/api/search/?query=bank")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertCountEqual(
[account["id"] for account in response.data["mail_accounts"]],
[own_account.id, unowned_account.id],
)
self.assertCountEqual(
[rule["id"] for rule in response.data["mail_rules"]],
[own_rule.id, unowned_rule.id],
)
self.assertNotIn(
other_account.id,
[account["id"] for account in response.data["mail_accounts"]],
)
self.assertNotIn(
other_rule.id,
[rule["id"] for rule in response.data["mail_rules"]],
)
def test_global_search_bad_request(self) -> None:
"""
WHEN:
+7
View File
@@ -86,11 +86,18 @@ class TestSystemStatus(APITestCase):
"""
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
self.assertEqual(response["WWW-Authenticate"], "Token")
normal_user = User.objects.create_user(username="normal_user")
self.client.force_login(normal_user)
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_system_status_with_bad_basic_auth_challenges(self) -> None:
self.client.credentials(HTTP_AUTHORIZATION="Basic invalid")
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
self.assertEqual(response["WWW-Authenticate"], 'Basic realm="api"')
def test_system_status_container_detection(self) -> None:
"""
GIVEN:
+47
View File
@@ -1,4 +1,5 @@
import datetime
import hashlib
import logging
import tempfile
from pathlib import Path
@@ -204,6 +205,52 @@ class TestFileHandling(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
)
self.assertEqual(document.filename, "none/none.pdf")
@override_settings(FILENAME_FORMAT=None)
def test_stale_save_recovers_already_moved_files(self) -> None:
old_storage_path = StoragePath.objects.create(
name="old-path",
path="old/{{title}}",
)
new_storage_path = StoragePath.objects.create(
name="new-path",
path="new/{{title}}",
)
original_bytes = b"original"
archive_bytes = b"archive"
doc = Document.objects.create(
title="document",
mime_type="application/pdf",
checksum=hashlib.md5(original_bytes).hexdigest(),
archive_checksum=hashlib.md5(archive_bytes).hexdigest(),
filename="old/document.pdf",
archive_filename="old/document.pdf",
storage_path=old_storage_path,
)
create_source_path_directory(doc.source_path)
doc.source_path.write_bytes(original_bytes)
create_source_path_directory(doc.archive_path)
doc.archive_path.write_bytes(archive_bytes)
stale_doc = Document.objects.get(pk=doc.pk)
fresh_doc = Document.objects.get(pk=doc.pk)
fresh_doc.storage_path = new_storage_path
fresh_doc.save()
doc.refresh_from_db()
self.assertEqual(doc.filename, "new/document.pdf")
self.assertEqual(doc.archive_filename, "new/document.pdf")
stale_doc.storage_path = new_storage_path
stale_doc.save()
doc.refresh_from_db()
self.assertEqual(doc.filename, "new/document.pdf")
self.assertEqual(doc.archive_filename, "new/document.pdf")
self.assertIsFile(doc.source_path)
self.assertIsFile(doc.archive_path)
self.assertIsNotFile(settings.ORIGINALS_DIR / "old" / "document.pdf")
self.assertIsNotFile(settings.ARCHIVE_DIR / "old" / "document.pdf")
@override_settings(FILENAME_FORMAT="{correspondent}/{correspondent}")
def test_document_delete(self) -> None:
document = Document()
+58
View File
@@ -963,6 +963,64 @@ class TestWorkflows(
self.assertEqual(Path(doc.filename), expected_filename)
self.assertTrue(doc.source_path.is_file())
def test_workflow_document_updated_does_not_overwrite_filename(self) -> None:
"""
GIVEN:
- A document whose filename has been updated in the DB by a concurrent
bulk_update_documents task (simulating update_filename_and_move_files
completing and writing the new filename to the DB)
- A stale in-memory document instance still holding the old filename
- An active DOCUMENT_UPDATED workflow
WHEN:
- run_workflows is called with the stale in-memory instance
(as would happen in the second concurrent bulk_update_documents task)
THEN:
- The DB filename is NOT overwritten with the stale in-memory value
(regression test for GH #12386 — the race window between
refresh_from_db and document.save in run_workflows)
"""
trigger = WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_UPDATED,
)
action = WorkflowAction.objects.create(
type=WorkflowAction.WorkflowActionType.ASSIGNMENT,
assign_title="Updated by workflow",
)
workflow = Workflow.objects.create(name="Race condition test workflow", order=0)
workflow.triggers.add(trigger)
workflow.actions.add(action)
workflow.save()
doc = Document.objects.create(
title="race condition test",
mime_type="application/pdf",
checksum="racecondition123",
original_filename="old.pdf",
filename="old/path/old.pdf",
)
# Simulate BUD-1 completing update_filename_and_move_files:
# the DB now holds the new filename while BUD-2's in-memory instance is stale.
new_filename = "new/path/new.pdf"
Document.global_objects.filter(pk=doc.pk).update(filename=new_filename)
# The stale instance still has filename="old/path/old.pdf" in memory.
# Mock refresh_from_db so the stale value persists through run_workflows,
# replicating the race window between refresh and save.
# Mock update_filename_and_move_files to prevent file-not-found errors
# since we are only testing DB state here.
with (
mock.patch(
"documents.signals.handlers.update_filename_and_move_files",
),
mock.patch.object(Document, "refresh_from_db"),
):
run_workflows(WorkflowTrigger.WorkflowTriggerType.DOCUMENT_UPDATED, doc)
# The DB filename must not have been reverted to the stale old value.
doc.refresh_from_db()
self.assertEqual(doc.filename, new_filename)
def test_document_added_workflow(self) -> None:
trigger = WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED,