Compare commits

..
Author SHA1 Message Date
Trenton HandClaude Sonnet 5.5 a2ead137a5 Chore: Use shared samples in tests that arrived from dev
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:20:12 -07:00
Trenton HandClaude Sonnet 5.5 7a4ea36760 Chore: Tidy leftover parentheses and the samples module docstring
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:16 -07:00
Trenton HandClaude Sonnet 5.5 480fa42159 Chore: Make the sample duplicate guard fail on a missing tree
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:16 -07:00
Trenton HandClaude Sonnet 5.5 58e2a26215 Chore: Add a guard test so duplicate sample files cannot come back
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:16 -07:00
Trenton HandClaude Sonnet 5.5 a55869629d Chore: Remove duplicate and unreferenced test sample files
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:16 -07:00
Trenton HandClaude Sonnet 5.5 86750e2803 Chore: Materialise duplicated media-tree samples from the shared copies
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:16 -07:00
Trenton HandClaude Sonnet 5.5 c5580b361f Chore: Use the shared simple-digital.pdf in documents tests and drop the unreferenced simple.txt
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:09 -07:00
Trenton HandClaude Sonnet 5.5 d642f02225 Chore: Move the four duplicated parser sample PDFs into a shared paperless_testing location
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 10:16:09 -07:00
shamoon 973af89f61 Chore: update frontend TS libs to ES2022 2026-10-11 15:39:56 -07:00
GitHub Actions 937bf2fa7a Auto translate strings 2026-10-10 22:05:15 +00:00
Trenton HandClaude Sonnet 5.5 2f2381578c Fix: authorize document versions by their root and speed up permission id sets (#14396)
* Fix: authorize document versions by their root and speed up permission id sets

permitted_document_ids judged a version by its own owner and grants, so a
version whose owner had drifted from its root's was visible to the wrong
people and hidden from the right ones. Callers patched this individually by
mapping each document to its root first. The query itself was also slow on
MariaDB: the guardian grants were a UNION cast to integers and tested with
IN inside an OR with the owner checks, which MariaDB cannot materialize, so it
re-scans the user's grants for every document. At 20k documents that took
seconds for a user with a couple of hundred grants.

permitted_object_ids now casts the row key to a string, as guardian stores
object_pk, and tests it against a single uncorrelated UNION ALL of the user's
and groups' grants. No integer cast is needed and the user's groups are
matched with an IN subquery rather than a join through the membership table.
Postgres and SQLite build the grant set once, and MariaDB probes guardian's
unique indexes per row, which is cheap. A correlated EXISTS per grant also
fixed MariaDB but was up to 3x slower than before on Postgres and SQLite.

permitted_object_ids also takes an optional parent_field naming a
self-referencing foreign key whose target authorizes the row, and
permitted_document_ids passes root_document, so a version is visible exactly
when its root is. The helper that mapped documents to their roots at the call
sites is no longer needed, so the email, selection data, share link bundle,
trash, bulk download and bulk edit checks use the id set directly.

Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-10 15:03:52 -07:00
shamoon e606dbffcd Chore: update the db options warning (#14398) 2026-10-10 07:17:54 -07:00
56 changed files with 846 additions and 605 deletions

No files matched your search

-11
View File
@@ -794,17 +794,6 @@ system. See the corresponding
Defaults to None
#### [`PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR=<comma-separated-list>`](#PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR) {#PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR}
: A list of authentication method values which, if present in the `amr` claim of the ID token sent by an OpenID Connect provider, cause Paperless-ngx to skip its own two-factor authentication prompt for that login. This avoids users having to enter a second factor twice when the identity provider already performed multi-factor authentication. Logins with a username and password, or via a provider which does not send a matching `amr` value, still require the Paperless-ngx code.
Which values to trust depends entirely on your identity provider, e.g. Authelia sends `mfa` and Pocket ID sends `phr` for passkey logins (but `otp` for single-factor email codes, which should _not_ be trusted). Some providers, such as Keycloak, do not send an `amr` claim by default and need to be configured to do so. You can see what your provider sends from the Django admin as a superuser by checking `id_token` → `amr` in the "Extra data" of the social account.
!!! warning
Only list values that mean a second factor was actually used for the login.
Defaults to `mfa`, the standard value for multi-factor authentication ([RFC 8176](https://www.rfc-editor.org/rfc/rfc8176.html)).
#### [`PAPERLESS_SOCIAL_ACCOUNT_DEFAULT_GROUPS=<comma-separated-list>`](#PAPERLESS_SOCIAL_ACCOUNT_DEFAULT_GROUPS) {#PAPERLESS_SOCIAL_ACCOUNT_DEFAULT_GROUPS}
: A list of group names that users who signup via social accounts will be added to upon signup. Groups listed here must already exist.
-2
View File
@@ -473,8 +473,6 @@ Users can enable two-factor authentication (2FA) for their accounts from the 'My
Should a user lose access to their 2FA device and all recovery codes, a superuser can disable 2FA for the user from the 'Users & Groups' management screen.
If users log in via an OpenID Connect provider which already enforces multi-factor authentication, see [`PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR`](configuration.md#PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR) to skip the Paperless-ngx prompt for those logins.
## Workflows
!!! note
+2 -2
View File
@@ -256,8 +256,8 @@ isort.force-single-line = true
[tool.codespell]
ignore-words-list = "criterias,afterall,valeu,ureue,equest,ure,assertIn,Oktober,commitish,NIN,nin,reprot"
skip = """\
src-ui/src/locale/*,src-ui/pnpm-lock.yaml,src-ui/e2e/*,src/paperless_mail/tests/samples/*,src/paperless/tests/samples\
/mail/*,src/documents/tests/samples/*,*.po,*.json\
src-ui/src/locale/*,src-ui/pnpm-lock.yaml,src-ui/e2e/*,src/paperless/tests/samples/mail/*,src/documents/tests/samples\
/*,src/paperless_testing/sample_files/*,*.po,*.json\
"""
write-changes = true
+1 -1
View File
@@ -15,7 +15,7 @@
"target": "ES2022",
"module": "es2020",
"lib": [
"es2020",
"es2022",
"dom"
],
"useDefineForClassFields": false
+38
View File
@@ -11,6 +11,8 @@ from typing import TYPE_CHECKING
import pytest
from paperless_testing import samples
if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path
@@ -149,3 +151,39 @@ def fake_progress_manager(
monkeypatch.setattr("documents.tasks.ProgressManager", FakeProgressManager)
return FakeProgressManager
@pytest.fixture(scope="session")
def shared_samples_dir() -> Path:
"""Directory of sample files used by more than one app's tests."""
return samples.SHARED_SAMPLES_DIR
@pytest.fixture(scope="session")
def simple_digital_pdf_file() -> Path:
"""One-page PDF with a text layer."""
return samples.SIMPLE_DIGITAL_PDF
@pytest.fixture(scope="session")
def multi_page_digital_pdf_file() -> Path:
"""Three-page PDF with a text layer."""
return samples.MULTI_PAGE_DIGITAL_PDF
@pytest.fixture(scope="session")
def with_form_pdf_file() -> Path:
"""PDF containing a fillable form."""
return samples.WITH_FORM_PDF
@pytest.fixture(scope="session")
def multi_page_images_pdf_file() -> Path:
"""Multi-page PDF of scanned images, no text layer."""
return samples.MULTI_PAGE_IMAGES_PDF
@pytest.fixture(scope="session")
def thumbnail_webp_file() -> Path:
"""Small WebP thumbnail."""
return samples.THUMBNAIL_WEBP
+62 -37
View File
@@ -6,7 +6,9 @@ from django.contrib.auth.models import Permission
from django.contrib.auth.models import User
from django.contrib.contenttypes.models import ContentType
from django.db.models import Case
from django.db.models import CharField
from django.db.models import Count
from django.db.models import F
from django.db.models import IntegerField
from django.db.models import Model
from django.db.models import Q
@@ -349,6 +351,7 @@ def permitted_object_ids(
perm: str,
*,
include_deleted: bool = False,
parent_field: str | None = None,
) -> QuerySet[int]:
"""
Generic version of ``permitted_document_ids`` for any model with an
@@ -357,6 +360,24 @@ def permitted_object_ids(
soft-delete pattern (currently only ``Document``); for every other model
it is accepted but has no effect, since those models have no soft-delete
concept.
``parent_field`` names a self-referencing foreign key whose target
authorizes the row (``Document.root_document``). A row with a parent is
visible exactly when its parent is, judged by the parent's owner and
grants, so the row's own owner and grants are ignored.
Guardian stores ``object_pk`` as a string, so the row key is cast to a
string and tested against the user's and groups' grants with a single
uncorrelated ``IN``. Postgres and SQLite build that set once. MariaDB
evaluates it as an index probe per row, which is cheap because the
lookups use guardian's unique indexes. Casting every ``object_pk`` to an
integer instead cannot use an index, and MariaDB cannot materialize it
inside the owner ``OR``, so it re-scans the user's grants for every row.
A correlated ``EXISTS`` per grant fixes MariaDB too, but Postgres and
SQLite re-run it for every row and end up slower than the original. The
user's groups are matched with an ``IN`` subquery rather than a join
through the membership table, which SQLite plans badly once the grant
tables grow.
"""
has_soft_delete = hasattr(model, "global_objects")
manager = (
@@ -364,8 +385,21 @@ def permitted_object_ids(
)
base_qs = manager.all().only("id", "owner")
owner_field, key_field = "owner", "pk"
if parent_field is not None:
owner_field, key_field = "authorizing_owner", "authorizing_id"
base_qs = base_qs.annotate(
authorizing_id=Coalesce(f"{parent_field}_id", "id"),
authorizing_owner=Case(
When(**{f"{parent_field}_id__isnull": True}, then=F("owner_id")),
default=F(f"{parent_field}__owner_id"),
output_field=IntegerField(),
),
)
unowned = Q(**{f"{owner_field}__isnull": True})
if user is None or not getattr(user, "is_authenticated", False):
return base_qs.filter(owner__isnull=True).values_list("id", flat=True)
return base_qs.filter(unowned).values_list("id", flat=True)
# Deactivated users get nothing, deactivated superusers included, so this
# has to come before the superuser shortcut. guardian's
@@ -389,21 +423,26 @@ def permitted_object_ids(
"permission__content_type": content_type,
}
user_perm_ids = (
UserObjectPermission.objects.filter(user=user, **perm_filter)
.annotate(object_pk_int=Cast("object_pk", IntegerField()))
.values_list("object_pk_int", flat=True)
)
group_perm_ids = (
GroupObjectPermission.objects.filter(group__user=user, **perm_filter)
.annotate(object_pk_int=Cast("object_pk", IntegerField()))
.values_list("object_pk_int", flat=True)
)
permitted_ids = user_perm_ids.union(group_perm_ids)
# Both grant sets are compared to the row key as strings, exactly as
# guardian stores them, and are uncorrelated, so each engine can build the
# set once instead of probing per row.
user_keys = UserObjectPermission.objects.filter(
user=user,
**perm_filter,
).values_list("object_pk", flat=True)
group_keys = GroupObjectPermission.objects.filter(
group_id__in=user.groups.values("id"),
**perm_filter,
).values_list("object_pk", flat=True)
permitted_keys = user_keys.union(group_keys, all=True)
return base_qs.filter(
Q(owner=user) | Q(owner__isnull=True) | Q(id__in=permitted_ids),
).values_list("id", flat=True)
return (
base_qs.annotate(permitted_key=Cast(key_field, CharField(max_length=64)))
.filter(
Q(**{owner_field: user.pk}) | unowned | Q(permitted_key__in=permitted_keys),
)
.values_list("id", flat=True)
)
ModelT = TypeVar("ModelT", bound=Model)
@@ -471,30 +510,16 @@ def permitted_document_ids(
``include_deleted=True`` for callers that need to check permission on
soft-deleted documents (e.g. trash restore). This intentionally avoids
``get_objects_for_user`` to keep the subquery small and index-friendly.
"""
return permitted_object_ids(user, Document, perm, include_deleted=include_deleted)
def documents_without_permitted_root(
documents: QuerySet[Document],
user: User | None,
*,
perm: str = "view_document",
include_deleted: bool = False,
) -> QuerySet[Document]:
A version is authorized by its root document, so a version's own owner and
grants never matter.
"""
The documents the user lacks ``perm`` on. Versions are authorized by their
root document, so a version's own owner is ignored. A single query, without
loading the documents or joining the root.
"""
return documents.annotate(
root_id=Coalesce("root_document_id", "id"),
).exclude(
root_id__in=permitted_document_ids(
user,
perm=perm,
include_deleted=include_deleted,
),
return permitted_object_ids(
user,
Document,
perm,
include_deleted=include_deleted,
parent_field="root_document",
)
+6 -10
View File
@@ -12,29 +12,25 @@ if TYPE_CHECKING:
from paperless_testing.dirs import PaperlessDirs
@pytest.fixture(scope="session")
def document_samples_dir() -> Path:
"""Path to the shared test sample documents."""
return Path(__file__).parent / "samples" / "documents"
@pytest.fixture()
def sample_doc(
paperless_dirs: "PaperlessDirs",
document_samples_dir: Path,
simple_digital_pdf_file: Path,
multi_page_images_pdf_file: Path,
thumbnail_webp_file: Path,
) -> "Document":
"""Create a document with valid files and matching checksums."""
with filelock.FileLock(paperless_dirs.media_lock):
shutil.copy(
document_samples_dir / "originals" / "0000001.pdf",
simple_digital_pdf_file,
paperless_dirs.originals_dir / "0000001.pdf",
)
shutil.copy(
document_samples_dir / "archive" / "0000001.pdf",
multi_page_images_pdf_file,
paperless_dirs.archive_dir / "0000001.pdf",
)
shutil.copy(
document_samples_dir / "thumbnails" / "0000001.webp",
thumbnail_webp_file,
paperless_dirs.thumbnail_dir / "0000001.webp",
)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 32 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.6 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.6 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.6 KiB

Binary file not shown.
-1
View File
@@ -1 +0,0 @@
This is a test file.
@@ -53,7 +53,7 @@ class TestBulkDownload(DirectoriesMixin, SampleDirMixin, APITestCase):
archive_checksum="D",
)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", self.doc2.source_path)
shutil.copy(self.SIMPLE_PDF, self.doc2.source_path)
shutil.copy(self.SAMPLE_DIR / "simple.png", self.doc2b.source_path)
shutil.copy(self.SAMPLE_DIR / "simple.jpg", self.doc3.source_path)
shutil.copy(self.SAMPLE_DIR / "test_with_bom.pdf", self.doc3.archive_path)
+49 -33
View File
@@ -56,6 +56,8 @@ from paperless_testing.http import read_streaming_response
from paperless_testing.permissions import grant_all_global
from paperless_testing.permissions import grant_global
from paperless_testing.permissions import grant_object
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
from paperless_testing.samples import THUMBNAIL_WEBP
class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
@@ -1852,7 +1854,11 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SimpleUploadedFile(
"simple.pdf",
SIMPLE_DIGITAL_PDF.read_bytes(),
content_type="application/pdf",
) as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f},
@@ -1880,7 +1886,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
payload = SimpleUploadedFile(
"../../outside.pdf",
(Path(__file__).parent / "samples" / "simple.pdf").read_bytes(),
SIMPLE_DIGITAL_PDF.read_bytes(),
content_type="application/pdf",
)
@@ -1909,7 +1915,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
pdf_bytes = (Path(__file__).parent / "samples" / "simple.pdf").read_bytes()
pdf_bytes = SIMPLE_DIGITAL_PDF.read_bytes()
boundary = "paperless-boundary"
payload = (
(
@@ -1985,7 +1991,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
def test_upload_insufficient_permissions(self) -> None:
self.client.force_authenticate(user=UserFactory(username="testuser2"))
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f},
@@ -1998,7 +2004,11 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SimpleUploadedFile(
"simple.pdf",
SIMPLE_DIGITAL_PDF.read_bytes(),
content_type="application/pdf",
) as f:
response = self.client.post(
"/api/documents/post_document/",
{
@@ -2031,7 +2041,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"documenst": f},
@@ -2057,7 +2067,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "title": "my custom title"},
@@ -2077,7 +2087,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
)
c = Correspondent.objects.create(name="test-corres")
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "correspondent": c.id},
@@ -2096,7 +2106,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "correspondent": 3456},
@@ -2111,7 +2121,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
)
dt = DocumentType.objects.create(name="invoice")
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "document_type": dt.id},
@@ -2130,7 +2140,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "document_type": 34578},
@@ -2145,7 +2155,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
)
sp = StoragePath.objects.create(name="invoices")
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "storage_path": sp.id},
@@ -2164,7 +2174,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "storage_path": 34578},
@@ -2180,7 +2190,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
t1 = Tag.objects.create(name="tag1")
t2 = Tag.objects.create(name="tag2")
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "tags": [t2.id, t1.id]},
@@ -2201,7 +2211,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
t1 = Tag.objects.create(name="tag1")
t2 = Tag.objects.create(name="tag2")
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "tags": [t2.id, t1.id, 734563]},
@@ -2225,7 +2235,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
0,
tzinfo=zoneinfo.ZoneInfo("America/Los_Angeles"),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "created": created},
@@ -2241,7 +2251,11 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SimpleUploadedFile(
"simple.pdf",
SIMPLE_DIGITAL_PDF.read_bytes(),
content_type="application/pdf",
) as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "archive_serial_number": 500},
@@ -2268,7 +2282,11 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
data_type=CustomField.FieldDataType.STRING,
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SimpleUploadedFile(
"simple.pdf",
SIMPLE_DIGITAL_PDF.read_bytes(),
content_type="application/pdf",
) as f:
response = self.client.post(
"/api/documents/post_document/",
{
@@ -2323,7 +2341,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
w1.actions.add(action1)
w1.save()
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{
@@ -2364,7 +2382,11 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
data_type=CustomField.FieldDataType.INT,
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SimpleUploadedFile(
"simple.pdf",
SIMPLE_DIGITAL_PDF.read_bytes(),
content_type="application/pdf",
) as f:
response = self.client.post(
"/api/documents/post_document/",
{
@@ -2413,7 +2435,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
]
for payload in error_payloads:
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
data = {"document": f, **payload}
response = self.client.post(
"/api/documents/post_document/",
@@ -2471,7 +2493,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
id=str(uuid.uuid4()),
)
with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
with SIMPLE_DIGITAL_PDF.open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "from_webui": True},
@@ -2510,14 +2532,8 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
archive_filename="archive.pdf",
)
source_file: Path = (
Path(__file__).parent
/ "samples"
/ "documents"
/ "thumbnails"
/ "0000001.webp"
)
archive_file: Path = Path(__file__).parent / "samples" / "simple.pdf"
source_file: Path = THUMBNAIL_WEBP
archive_file: Path = SIMPLE_DIGITAL_PDF
shutil.copy(source_file, doc.source_path)
shutil.copy(archive_file, doc.archive_path)
@@ -2550,7 +2566,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
mime_type="application/pdf",
)
shutil.copy(Path(__file__).parent / "samples" / "simple.pdf", doc.source_path)
shutil.copy(SIMPLE_DIGITAL_PDF, doc.source_path)
response = self.client.get(f"/api/documents/{doc.pk}/metadata/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
@@ -4288,8 +4304,8 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
filename="test2.pdf",
)
archive_file = Path(__file__).parent / "samples" / "simple.pdf"
source_file = Path(__file__).parent / "samples" / "simple.pdf"
archive_file = SIMPLE_DIGITAL_PDF
source_file = SIMPLE_DIGITAL_PDF
shutil.copy(archive_file, doc.archive_path)
shutil.copy(source_file, doc2.source_path)
+8 -5
View File
@@ -12,6 +12,9 @@ from documents.tests.utils import SampleDirMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import UserFactory
from paperless_testing.permissions import grant_global
from paperless_testing.samples import MULTI_PAGE_DIGITAL_PDF
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
from paperless_testing.samples import WITH_FORM_PDF
class TestEmail(DirectoriesMixin, SampleDirMixin, APITestCase):
@@ -42,15 +45,15 @@ class TestEmail(DirectoriesMixin, SampleDirMixin, APITestCase):
# Copy sample files to document paths (using different files to distinguish versions)
shutil.copy(
self.SAMPLE_DIR / "documents" / "originals" / "0000001.pdf",
SIMPLE_DIGITAL_PDF,
self.doc1.archive_path,
)
shutil.copy(
self.SAMPLE_DIR / "documents" / "originals" / "0000002.pdf",
MULTI_PAGE_DIGITAL_PDF,
self.doc1.source_path,
)
shutil.copy(
self.SAMPLE_DIR / "documents" / "originals" / "0000003.pdf",
WITH_FORM_PDF,
self.doc2.source_path,
)
@@ -377,7 +380,7 @@ class TestEmail(DirectoriesMixin, SampleDirMixin, APITestCase):
checksum="3",
filename="test3.pdf",
)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", doc3.source_path)
shutil.copy(self.SIMPLE_PDF, doc3.source_path)
doc4 = Document.objects.create(
title="test1",
@@ -386,7 +389,7 @@ class TestEmail(DirectoriesMixin, SampleDirMixin, APITestCase):
checksum="4",
filename="test4.pdf",
)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", doc4.source_path)
shutil.copy(self.SIMPLE_PDF, doc4.source_path)
response = self.client.post(
self.ENDPOINT,
+2 -2
View File
@@ -167,7 +167,7 @@ class TestBarcode(
- No barcodes detected
- No pages to split on
"""
test_file = self.SAMPLE_DIR / "simple.pdf"
test_file = self.SIMPLE_PDF
with self.get_reader(test_file) as reader:
reader.detect()
separator_page_numbers = reader.get_separation_pages()
@@ -459,7 +459,7 @@ class TestBarcode(
THEN:
- Nothing happens
"""
test_file = self.SAMPLE_DIR / "simple.pdf"
test_file = self.SIMPLE_PDF
with self.get_reader(test_file) as reader:
try:
+9 -30
View File
@@ -27,6 +27,9 @@ from documents.permissions import set_permissions_for_objects
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import DocumentFactory
from paperless_testing.permissions import grant_object
from paperless_testing.samples import MULTI_PAGE_DIGITAL_PDF
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
from paperless_testing.samples import WITH_FORM_PDF
class TestBulkEdit(DirectoriesMixin, TestCase):
@@ -704,47 +707,27 @@ class TestPDFActions(DirectoriesMixin, TestCase):
super().setUp()
sample1 = self.dirs.scratch_dir / "sample.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
SIMPLE_DIGITAL_PDF,
sample1,
)
sample1_archive = self.dirs.archive_dir / "sample_archive.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
SIMPLE_DIGITAL_PDF,
sample1_archive,
)
sample2 = self.dirs.scratch_dir / "sample2.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000002.pdf",
MULTI_PAGE_DIGITAL_PDF,
sample2,
)
sample2_archive = self.dirs.archive_dir / "sample2_archive.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000002.pdf",
MULTI_PAGE_DIGITAL_PDF,
sample2_archive,
)
sample3 = self.dirs.scratch_dir / "sample3.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000003.pdf",
WITH_FORM_PDF,
sample3,
)
self.doc1 = Document.objects.create(
@@ -778,11 +761,7 @@ class TestPDFActions(DirectoriesMixin, TestCase):
)
img_doc_archive = self.dirs.archive_dir / "sample_image.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
SIMPLE_DIGITAL_PDF,
img_doc_archive,
)
self.img_doc = Document.objects.create(
+9 -28
View File
@@ -36,6 +36,9 @@ from paperless_testing.assertions import FileSystemAssertsMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import UserFactory
from paperless_testing.fakes.progress import FakeProgressManager
from paperless_testing.samples import MULTI_PAGE_DIGITAL_PDF
from paperless_testing.samples import MULTI_PAGE_IMAGES_PDF
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
class _BaseNewStyleParser:
@@ -112,9 +115,7 @@ class _BaseNewStyleParser:
class DummyParser(_BaseNewStyleParser):
_ARCHIVE_SRC = (
Path(__file__).parent / "samples" / "documents" / "archive" / "0000001.pdf"
)
_ARCHIVE_SRC = MULTI_PAGE_IMAGES_PDF
def parse(self, document_path, mime_type, *, produce_archive: bool = True) -> None:
self._text = "The Text"
@@ -200,33 +201,19 @@ class TestConsumer(
self.addCleanup(patcher.stop)
def get_test_file(self):
src = (
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf"
)
src = SIMPLE_DIGITAL_PDF
dst = self.dirs.scratch_dir / "sample.pdf"
shutil.copy(src, dst)
return dst
def get_test_file2(self):
src = (
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000002.pdf"
)
src = MULTI_PAGE_DIGITAL_PDF
dst = self.dirs.scratch_dir / "sample2.pdf"
shutil.copy(src, dst)
return dst
def get_test_archive_file(self):
src = (
Path(__file__).parent / "samples" / "documents" / "archive" / "0000001.pdf"
)
src = MULTI_PAGE_IMAGES_PDF
dst = self.dirs.scratch_dir / "sample_archive.pdf"
shutil.copy(src, dst)
return dst
@@ -1062,7 +1049,7 @@ class TestConsumer(
@mock.patch("documents.consumer.get_parser_registry")
def test_similar_filenames(self, m) -> None:
shutil.copy(
Path(__file__).parent / "samples" / "simple.pdf",
SIMPLE_DIGITAL_PDF,
settings.CONSUMPTION_DIR / "simple.pdf",
)
shutil.copy(
@@ -1603,13 +1590,7 @@ class TestConsumerRemoteOCR(
self.addCleanup(patcher.stop)
def _consume(self, *, overrides: DocumentMetadataOverrides | None = None) -> bool:
src = (
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf"
)
src = SIMPLE_DIGITAL_PDF
dst = self.dirs.scratch_dir / "sample.pdf"
shutil.copy(src, dst)
+5 -2
View File
@@ -37,12 +37,15 @@ class TestDoubleSided(
self.double_sided_dir.mkdir()
self.staging_file = self.dirs.scratch_dir / STAGING_FILE_NAME
def _sample(self, name: str) -> Path:
return self.SIMPLE_PDF if name == "simple.pdf" else self.SAMPLE_DIR / name
def consume_file(self, srcname, dstname: str | Path = "foo.pdf"):
"""
Starts the consume process and also ensures the
destination file does not exist afterwards
"""
src = self.SAMPLE_DIR / srcname
src = self._sample(srcname)
dst = self.double_sided_dir / dstname
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy(src, dst)
@@ -57,7 +60,7 @@ class TestDoubleSided(
return msg
def create_staging_file(self, src="double-sided-odd.pdf", datetime=None) -> None:
shutil.copy(self.SAMPLE_DIR / src, self.staging_file)
shutil.copy(self._sample(src), self.staging_file)
if datetime is None:
datetime = dt.datetime.now()
os.utime(str(self.staging_file), (datetime.timestamp(),) * 2)
+2 -1
View File
@@ -22,8 +22,9 @@ from documents.models import Document
from documents.tasks import update_document_content_maybe_archive_file
from paperless_testing.assertions import FileSystemAssertsMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
sample_file: Path = Path(__file__).parent / "samples" / "simple.pdf"
sample_file: Path = SIMPLE_DIGITAL_PDF
@pytest.mark.management
+21 -80
View File
@@ -53,6 +53,7 @@ from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.dirs import paperless_environment
from paperless_testing.factories import DocumentBarcodeFactory
from paperless_testing.permissions import grant_object
from paperless_testing.samples import install_document_samples
@pytest.mark.management
@@ -198,10 +199,7 @@ class TestExportImport(
def test_exporter(self, *, use_filename_format=False) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
num_permission_objects = Permission.objects.count()
@@ -305,10 +303,7 @@ class TestExportImport(
def test_exporter_with_filename_format(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
with override_settings(
FILENAME_FORMAT="{created_year}/{correspondent}/{title}",
@@ -317,10 +312,7 @@ class TestExportImport(
def test_exporter_includes_share_links_and_bundles(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
share_link = ShareLink.objects.create(
slug="share-link-slug",
@@ -419,10 +411,7 @@ class TestExportImport(
def test_update_export_changed_time(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
self._do_export()
self.assertIsFile(self.target / "manifest.json")
@@ -458,10 +447,7 @@ class TestExportImport(
def test_update_export_changed_checksum(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
self._do_export()
@@ -488,10 +474,7 @@ class TestExportImport(
def test_update_export_deleted_document(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
manifest = self._do_export()
@@ -523,10 +506,7 @@ class TestExportImport(
@override_settings(FILENAME_FORMAT="{title}/{correspondent}")
def test_update_export_changed_location(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
self._do_export(use_filename_format=True)
self.assertIsFile(self.target / "wow1" / "c.pdf")
@@ -568,10 +548,7 @@ class TestExportImport(
- Zipfile contains exported files
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
args = ["document_exporter", self.target, "--zip"]
@@ -600,10 +577,7 @@ class TestExportImport(
- Zipfile contains exported files
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
args = ["document_exporter", self.target, "--zip", "--use-filename-format"]
@@ -639,10 +613,7 @@ class TestExportImport(
- The existing file and directory in target are removed
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
# Create stuff in target directory
existing_file = self.target / "test.txt"
@@ -738,10 +709,7 @@ class TestExportImport(
- Documents can be imported again
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
manifest = self._do_export()
has_archive = False
@@ -784,10 +752,7 @@ class TestExportImport(
- Documents can be imported again
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
manifest = self._do_export()
has_thumbnail = False
@@ -832,10 +797,7 @@ class TestExportImport(
- Documents can be imported again
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
manifest = self._do_export(split_manifest=True)
has_document = False
@@ -860,10 +822,7 @@ class TestExportImport(
def _export_import_barcodes(self, *, split_manifest: bool) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
DocumentBarcodeFactory(document=self.d1, value="https://example.com")
DocumentBarcodeFactory(document=self.d2, page=2, value="DE8937")
@@ -899,10 +858,7 @@ class TestExportImport(
- Documents can be imported again
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
self._do_export(use_folder_prefix=True)
@@ -929,10 +885,7 @@ class TestExportImport(
- Documents can be imported again
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
self._do_export(use_folder_prefix=True, split_manifest=True)
@@ -959,10 +912,7 @@ class TestExportImport(
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
num_content_type_objects = ContentType.objects.count()
num_permission_objects = Permission.objects.count()
@@ -999,10 +949,7 @@ class TestExportImport(
def test_exporter_with_auditlog_disabled(self) -> None:
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
with override_settings(
AUDIT_LOG_ENABLED=False,
@@ -1022,10 +969,7 @@ class TestExportImport(
survive the round-trip with deleted_at preserved
"""
shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
Path(__file__).parent / "samples" / "documents",
Path(self.dirs.media_dir) / "documents",
)
install_document_samples(Path(self.dirs.media_dir) / "documents")
# d1 has self.note and self.cfi1 attached via setUp
self.d1.delete()
@@ -1069,10 +1013,7 @@ class TestExportImport(
"""
shutil.rmtree(self.dirs.media_dir / "documents")
shutil.copytree(
self.SAMPLE_DIR / "documents",
self.dirs.media_dir / "documents",
)
install_document_samples(self.dirs.media_dir / "documents")
_ = self._do_export(data_only=True)
@@ -11,6 +11,7 @@ from documents.models import Document
from documents.parsers import get_default_thumbnail
from paperless_testing.assertions import FileSystemAssertsMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
@pytest.mark.management
@@ -24,7 +25,7 @@ class TestMakeThumbnails(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
filename="test.pdf",
)
shutil.copy(
Path(__file__).parent / "samples" / "simple.pdf",
SIMPLE_DIGITAL_PDF,
self.d1.source_path,
)
@@ -36,7 +37,7 @@ class TestMakeThumbnails(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
filename="test2.pdf",
)
shutil.copy(
Path(__file__).parent / "samples" / "simple.pdf",
SIMPLE_DIGITAL_PDF,
self.d2.source_path,
)
@@ -32,6 +32,8 @@ from paperless_testing.permissions import grant_global
from paperless_testing.permissions import grant_object
if TYPE_CHECKING:
from django.contrib.auth.models import User
from paperless_testing.dirs import PaperlessDirs
@@ -178,6 +180,296 @@ class TestPermittedDocumentIdsIncludeDeleted:
)
@pytest.mark.django_db
class TestPermittedDocumentIdsVersions:
"""
A version is authorized by its root document: the version's own owner and
grants never matter.
"""
@pytest.mark.parametrize(
("root_owner", "version_owner", "expected_visible"),
[
pytest.param(
"other",
"nobody",
False,
id="unowned-version-of-private-root",
),
pytest.param("other", "user", False, id="own-version-of-private-root"),
pytest.param("user", "other", True, id="foreign-version-of-own-root"),
pytest.param("user", "nobody", True, id="unowned-version-of-own-root"),
pytest.param("nobody", "other", True, id="private-version-of-unowned-root"),
],
)
def test_version_follows_root_owner(
self,
root_owner: str,
version_owner: str,
*,
expected_visible: bool,
) -> None:
"""
GIVEN:
- A root document and a version with differing owners
WHEN:
- The permitted document ids are resolved for the user
THEN:
- The version is visible exactly when its root is
"""
user = UserFactory()
owners = {"user": user, "other": UserFactory(), "nobody": None}
root = DocumentFactory(owner=owners[root_owner])
version = DocumentFactory(root_document=root, owner=owners[version_owner])
visible = set(permitted_document_ids(user))
assert (version.pk in visible) is expected_visible
assert (root.pk in visible) is expected_visible
@staticmethod
def grantee(user: User, kind: str) -> User | Group:
"""The user itself, or a new group the user belongs to."""
if kind == "user":
return user
group = Group.objects.create(name="shared")
user.groups.add(group)
return group
@pytest.mark.parametrize(
"grantee_kind",
[pytest.param("user", id="user"), pytest.param("group", id="group")],
)
def test_grant_on_root_applies_to_version(self, grantee_kind: str) -> None:
"""
GIVEN:
- A private root document shared with a user or one of their groups
- A version of it owned by someone else
WHEN:
- The permitted document ids are resolved for the user
THEN:
- Both the root and the version are visible
- A user without the grant sees neither
"""
user = UserFactory()
stranger = UserFactory()
root = DocumentFactory(owner=UserFactory())
version = DocumentFactory(root_document=root, owner=UserFactory())
grant_object(self.grantee(user, grantee_kind), root, "view_document")
assert_visible_document_ids(
permitted_document_ids(user),
expected_visible=[root.pk, version.pk],
expected_hidden=[],
)
assert_visible_document_ids(
permitted_document_ids(stranger),
expected_visible=[],
expected_hidden=[root.pk, version.pk],
)
@pytest.mark.parametrize(
"grantee_kind",
[pytest.param("user", id="user"), pytest.param("group", id="group")],
)
def test_grant_on_version_is_ignored(self, grantee_kind: str) -> None:
"""
GIVEN:
- A private root document
- A version with an explicit grant for the user or one of their groups
WHEN:
- The permitted document ids are resolved for the user
THEN:
- Neither the root nor the version is visible
"""
user = UserFactory()
root = DocumentFactory(owner=UserFactory())
version = DocumentFactory(root_document=root, owner=UserFactory())
grant_object(self.grantee(user, grantee_kind), version, "view_document")
assert_visible_document_ids(
permitted_document_ids(user),
expected_visible=[],
expected_hidden=[root.pk, version.pk],
)
def test_grant_on_one_root_does_not_reach_another_roots_version(self) -> None:
"""
GIVEN:
- Two private roots, each with a version
- The user may view only the first root
WHEN:
- The permitted document ids are resolved for the user
THEN:
- Only the first root and its version are visible
"""
user = UserFactory()
first = DocumentFactory(owner=UserFactory())
first_version = DocumentFactory(root_document=first, owner=UserFactory())
second = DocumentFactory(owner=UserFactory())
second_version = DocumentFactory(root_document=second, owner=user)
grant_object(user, first, "view_document")
assert_visible_document_ids(
permitted_document_ids(user),
expected_visible=[first.pk, first_version.pk],
expected_hidden=[second.pk, second_version.pk],
)
def test_user_in_several_groups(self) -> None:
"""
GIVEN:
- A user in two groups
- Two private roots shared with one group each, and a third shared with nobody
- A version of each root
WHEN:
- The permitted document ids are resolved for the user
THEN:
- The two shared roots and their versions are visible
- The third root and its version are not
"""
user = UserFactory()
groups = [Group.objects.create(name=f"group{i}") for i in range(2)]
user.groups.add(*groups)
shared = [DocumentFactory(owner=UserFactory()) for _ in groups]
for root, group in zip(shared, groups, strict=True):
grant_object(group, root, "view_document")
unshared = DocumentFactory(owner=UserFactory())
shared_versions = [
DocumentFactory(root_document=root, owner=UserFactory()) for root in shared
]
unshared_version = DocumentFactory(root_document=unshared, owner=None)
assert_visible_document_ids(
permitted_document_ids(user),
expected_visible=[
*(root.pk for root in shared),
*(version.pk for version in shared_versions),
],
expected_hidden=[unshared.pk, unshared_version.pk],
)
def test_permission_is_resolved_through_the_root(self) -> None:
"""
GIVEN:
- A private root document where the user may view and change
WHEN:
- The permitted ids are resolved for view, change and delete
THEN:
- The version is visible for view and change only
"""
user = UserFactory()
root = DocumentFactory(owner=UserFactory())
version = DocumentFactory(root_document=root, owner=UserFactory())
grant_object(user, root, "view_document", "change_document")
assert version.pk in set(permitted_document_ids(user))
assert version.pk in set(permitted_document_ids(user, perm="change_document"))
assert version.pk in set(
permitted_document_ids(user, perm="documents.change_document"),
)
assert version.pk not in set(
permitted_document_ids(user, perm="delete_document"),
)
def test_anonymous_sees_versions_of_unowned_roots_only(self) -> None:
"""
GIVEN:
- A version owned by nobody under a private root
- A version owned by someone under an unowned root
WHEN:
- The permitted document ids are resolved for an anonymous user
THEN:
- Only the version of the unowned root is visible
"""
private_root = DocumentFactory(owner=UserFactory())
private_version = DocumentFactory(root_document=private_root, owner=None)
open_root = DocumentFactory(owner=None)
open_version = DocumentFactory(root_document=open_root, owner=UserFactory())
assert_visible_document_ids(
permitted_document_ids(AnonymousUser()),
expected_visible=[open_root.pk, open_version.pk],
expected_hidden=[private_root.pk, private_version.pk],
)
def test_deleted_versions_follow_their_deleted_root(self) -> None:
"""
GIVEN:
- A soft-deleted root document and its version, which deleting the
root soft-deletes too; the version is owned by someone else
WHEN:
- The permitted document ids are resolved with and without deleted
documents
THEN:
- Nothing is visible by default
- With deleted documents included, the version is visible to the
root's owner and not to the version's own owner
"""
owner = UserFactory()
version_owner = UserFactory()
root = DocumentFactory(owner=owner)
version = DocumentFactory(root_document=root, owner=version_owner)
root.delete()
assert not {root.pk, version.pk} & set(permitted_document_ids(owner))
assert_visible_document_ids(
permitted_document_ids(owner, include_deleted=True),
expected_visible=[root.pk, version.pk],
expected_hidden=[],
)
assert_visible_document_ids(
permitted_document_ids(version_owner, include_deleted=True),
expected_visible=[],
expected_hidden=[root.pk, version.pk],
)
@pytest.mark.parametrize(
"is_superuser",
[
pytest.param(False, id="regular-user"),
pytest.param(True, id="superuser"),
],
)
def test_inactive_user_sees_no_versions(self, *, is_superuser: bool) -> None:
"""
GIVEN:
- An inactive user, possibly a superuser, who owns a root and its version
WHEN:
- The permitted document ids are resolved for them
THEN:
- Nothing is visible
"""
user = UserFactory(is_active=False, is_superuser=is_superuser)
root = DocumentFactory(owner=user)
version = DocumentFactory(root_document=root, owner=user)
assert_visible_document_ids(
permitted_document_ids(user),
expected_visible=[],
expected_hidden=[root.pk, version.pk],
)
def test_superuser_sees_all_versions(self) -> None:
"""
GIVEN:
- A private root owned by someone else, with a version
WHEN:
- The permitted document ids are resolved for a superuser
THEN:
- Both the root and the version are visible
"""
superuser = UserFactory(superuser=True)
root = DocumentFactory(owner=UserFactory())
version = DocumentFactory(root_document=root, owner=UserFactory())
assert_visible_document_ids(
permitted_document_ids(superuser),
expected_visible=[root.pk, version.pk],
expected_hidden=[],
)
@pytest.mark.django_db
class TestAiChatAllDocumentsPermissionBoundary:
"""
+61
View File
@@ -0,0 +1,61 @@
import hashlib
from collections import defaultdict
from pathlib import Path
from paperless_testing.samples import SHARED_SAMPLES_DIR
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
from paperless_testing.samples import install_document_samples
EXPECTED_MEDIA_FILES = [
"originals/0000001.pdf",
"originals/0000002.pdf",
"originals/0000003.pdf",
"originals/0000004.pdf",
"originals/0000005.pdf",
"originals/0000006.pdf",
"archive/0000001.pdf",
"thumbnails/0000001.webp",
"thumbnails/0000002.webp",
"thumbnails/0000003.webp",
"thumbnails/0000004.webp",
]
def _sha(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def test_install_recreates_every_historical_opaque_name(tmp_path: Path) -> None:
install_document_samples(tmp_path)
for relpath in EXPECTED_MEDIA_FILES:
assert (tmp_path / relpath).is_file(), relpath
def test_install_uses_the_shared_bytes(tmp_path: Path) -> None:
install_document_samples(tmp_path)
assert _sha(tmp_path / "originals" / "0000001.pdf") == _sha(SIMPLE_DIGITAL_PDF)
APP_SAMPLE_TREES = [
Path(__file__).parent / "samples",
Path(__file__).parents[2] / "paperless" / "tests" / "samples",
SHARED_SAMPLES_DIR,
]
def test_no_two_sample_files_have_identical_bytes() -> None:
by_hash: dict[str, list[Path]] = defaultdict(list)
for tree in APP_SAMPLE_TREES:
assert tree.is_dir(), f"Sample tree does not exist: {tree}"
for path in tree.rglob("*"):
# Skip empty placeholders, which would all hash identically
if path.is_file() and path.stat().st_size > 0:
by_hash[_sha(path)].append(path)
assert by_hash, "No sample files were found to compare"
duplicates = {h: ps for h, ps in by_hash.items() if len(ps) > 1}
lines = [" " + ", ".join(str(p) for p in ps) for ps in duplicates.values()]
message = (
"Duplicate sample files, move one to paperless_testing/sample_files:\n"
+ "\n".join(lines)
)
assert not duplicates, message
+5 -23
View File
@@ -21,6 +21,7 @@ from documents.tests.helpers import dummy_preprocess
from paperless_testing.assertions import FileSystemAssertsMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import DocumentFactory
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
@pytest.mark.django_db
@@ -229,20 +230,12 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
"""
sample1 = self.dirs.scratch_dir / "sample.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
SIMPLE_DIGITAL_PDF,
sample1,
)
sample1_archive = self.dirs.archive_dir / "sample_archive.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
SIMPLE_DIGITAL_PDF,
sample1_archive,
)
doc = Document.objects.create(
@@ -270,11 +263,7 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
"""
sample1 = self.dirs.scratch_dir / "sample.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
SIMPLE_DIGITAL_PDF,
sample1,
)
doc = Document.objects.create(
@@ -290,14 +279,7 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
def _create_root_with_version(self) -> tuple[Document, Document]:
sample1 = self.dirs.scratch_dir / "sample.pdf"
shutil.copy(
Path(__file__).parent
/ "samples"
/ "documents"
/ "originals"
/ "0000001.pdf",
sample1,
)
shutil.copy(SIMPLE_DIGITAL_PDF, sample1)
root = DocumentFactory(content="root content", mime_type="application/pdf")
version = DocumentFactory(
content="my document",
+27 -27
View File
@@ -178,7 +178,7 @@ class TestWorkflows(
self.assertEqual(action.__str__(), "WorkflowAction 1")
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -290,7 +290,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -409,7 +409,7 @@ class TestWorkflows(
w2.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -478,7 +478,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -531,7 +531,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -608,7 +608,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -687,7 +687,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -765,7 +765,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -876,7 +876,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -928,7 +928,7 @@ class TestWorkflows(
generated = generate_unique_filename(doc)
destination = (settings.ORIGINALS_DIR / generated).resolve()
create_source_path_directory(destination)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", destination)
shutil.copy(self.SIMPLE_PDF, destination)
Document.objects.filter(pk=doc.pk).update(filename=generated.as_posix())
doc.refresh_from_db()
@@ -2023,7 +2023,7 @@ class TestWorkflows(
superuser = UserFactory(username="superuser", superuser=True)
self.client.force_authenticate(user=superuser)
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
tasks.consume_file(
@@ -3009,7 +3009,7 @@ class TestWorkflows(
generated = generate_unique_filename(doc)
destination = (settings.ORIGINALS_DIR / generated).resolve()
create_source_path_directory(destination)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", destination)
shutil.copy(self.SIMPLE_PDF, destination)
Document.objects.filter(pk=doc.pk).update(filename=generated.as_posix())
doc.refresh_from_db()
doc.tags.set([self.t1, self.t2])
@@ -3073,7 +3073,7 @@ class TestWorkflows(
generated = generate_unique_filename(doc)
destination = (settings.ORIGINALS_DIR / generated).resolve()
create_source_path_directory(destination)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", destination)
shutil.copy(self.SIMPLE_PDF, destination)
Document.objects.filter(pk=doc.pk).update(filename=generated.as_posix())
doc.refresh_from_db()
doc.tags.set([self.t1])
@@ -3222,7 +3222,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -3346,7 +3346,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -3496,7 +3496,7 @@ class TestWorkflows(
workflow.actions.set([assignment_action, email_action])
temp_working_copy = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "working-copy.pdf",
)
@@ -3596,7 +3596,7 @@ class TestWorkflows(
# move the file
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -3704,7 +3704,7 @@ class TestWorkflows(
generated = generate_unique_filename(doc)
destination = (settings.ORIGINALS_DIR / generated).resolve()
create_source_path_directory(destination)
shutil.copy(self.SAMPLE_DIR / "simple.pdf", destination)
shutil.copy(self.SIMPLE_PDF, destination)
Document.objects.filter(pk=doc.pk).update(filename=generated.as_posix())
run_workflows(WorkflowTrigger.WorkflowTriggerType.DOCUMENT_UPDATED, doc)
@@ -3915,7 +3915,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -4036,7 +4036,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -4387,7 +4387,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -4835,7 +4835,7 @@ class TestWorkflows(
w.save()
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
@@ -5027,7 +5027,7 @@ class TestWorkflows(
# Create a test file to be consumed
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
test_file_path = Path(test_file)
@@ -5091,7 +5091,7 @@ class TestWorkflows(
# Create a test file to be consumed
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple2.pdf",
)
test_file_path = Path(test_file)
@@ -5529,7 +5529,7 @@ class TestDateWorkflowLocalization(
)
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
tmp_path / "simple.pdf",
)
@@ -5596,7 +5596,7 @@ class TestRemoteOCRWorkflowAction(DirectoriesMixin, SampleDirMixin, APITestCase)
self._make_workflow(WorkflowTrigger.WorkflowTriggerType.CONSUMPTION)
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
overrides = DocumentMetadataOverrides()
@@ -5814,7 +5814,7 @@ class TestApplyAISuggestionsWorkflowAction(
)
test_file = shutil.copy(
self.SAMPLE_DIR / "simple.pdf",
self.SIMPLE_PDF,
self.dirs.scratch_dir / "simple.pdf",
)
+3
View File
@@ -11,6 +11,7 @@ from documents.data_models import ConsumableDocument
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
from paperless_testing.fakes.progress import FakeProgressManager
from paperless_testing.samples import SIMPLE_DIGITAL_PDF
class ConsumeTaskMixin:
@@ -53,6 +54,8 @@ class SampleDirMixin:
BARCODE_SAMPLE_DIR = SAMPLE_DIR / "barcodes"
SIMPLE_PDF = SIMPLE_DIGITAL_PDF
class GetConsumerMixin:
@contextmanager
+17 -26
View File
@@ -174,7 +174,6 @@ from documents.permissions import TrashPermissions
from documents.permissions import ViewDocumentsPermissions
from documents.permissions import annotate_document_count_by_ids
from documents.permissions import annotate_document_count_for_related_queryset
from documents.permissions import documents_without_permitted_root
from documents.permissions import get_document_count_filter_for_user
from documents.permissions import get_objects_for_user_owner_aware
from documents.permissions import has_global_statistics_permission
@@ -2117,7 +2116,7 @@ class DocumentViewSet(
documents = Document.objects.filter(pk__in=document_ids)
if (
request.user is not None
and documents_without_permitted_root(documents, request.user).exists()
and documents.exclude(id__in=permitted_document_ids(request.user)).exists()
):
return HttpResponseForbidden("Insufficient permissions")
@@ -3009,12 +3008,8 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
user.has_perm(
"documents.change_document",
)
and not Document.global_objects.filter(
pk__in=[doc.pk for doc in root_docs],
)
.exclude(
pk__in=permitted_document_ids(user, perm="change_document"),
)
and not Document.global_objects.filter(pk__in=documents)
.exclude(pk__in=permitted_document_ids(user, perm="change_document"))
.exists()
)
@@ -3627,7 +3622,7 @@ class SelectionDataView(DocumentSelectionMixin, GenericAPIView[Any]):
documents = Document.objects.filter(pk__in=ids)
if (
documents.count() != len(ids)
or documents_without_permitted_root(documents, request.user).exists()
or documents.exclude(id__in=permitted_document_ids(request.user)).exists()
):
return HttpResponseForbidden("Insufficient permissions")
@@ -4124,21 +4119,16 @@ class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]):
validated_data=serializer.validated_data,
)
documents = Document.objects.filter(pk__in=ids)
versioned_documents = []
compression = serializer.validated_data.get("compression")
content = serializer.validated_data.get("content")
follow_filename_format = serializer.validated_data.get("follow_formatting")
permitted_ids = set(permitted_document_ids(request.user))
for document in documents:
root_doc = get_root_document(document)
if root_doc.pk not in permitted_ids:
return HttpResponseForbidden("Insufficient permissions")
versioned_documents.append(
get_latest_version_for_root(
root_doc,
),
)
if documents.exclude(id__in=permitted_document_ids(request.user)).exists():
return HttpResponseForbidden("Insufficient permissions")
versioned_documents = [
get_latest_version_for_root(get_root_document(document))
for document in documents
]
if content == "both":
strategy_class = OriginalAndArchiveStrategy
@@ -4811,7 +4801,7 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]):
)
denied_id = (
documents_without_permitted_root(documents_qs, request.user)
documents_qs.exclude(id__in=permitted_document_ids(request.user))
.order_by("pk")
.values_list("pk", flat=True)
.first()
@@ -5657,11 +5647,12 @@ class TrashView(ListModelMixin, PassUserMixin):
if doc_ids is not None
else self.filter_queryset(self.get_queryset()).all()
)
if documents_without_permitted_root(
docs,
request.user,
perm="delete_document",
include_deleted=True,
if docs.exclude(
id__in=permitted_document_ids(
request.user,
perm="delete_document",
include_deleted=True,
),
).exists():
return HttpResponseForbidden("Insufficient permissions")
action = serializer.validated_data.get("action")
+12 -12
View File
@@ -2,7 +2,7 @@ msgid ""
msgstr ""
"Project-Id-Version: paperless-ngx\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-10-08 18:39+0000\n"
"POT-Creation-Date: 2026-10-10 22:04+0000\n"
"PO-Revision-Date: 2022-02-17 04:17\n"
"Last-Translator: \n"
"Language-Team: English\n"
@@ -1653,7 +1653,7 @@ msgid "workflow runs"
msgstr ""
#: documents/serialisers.py:516 documents/serialisers.py:873
#: documents/serialisers.py:2903 documents/views.py:344 documents/views.py:2751
#: documents/serialisers.py:2903 documents/views.py:343 documents/views.py:2750
#: paperless_mail/serialisers.py:156
msgid "Insufficient permissions."
msgstr ""
@@ -1694,7 +1694,7 @@ msgstr ""
msgid "Duplicate document identifiers are not allowed."
msgstr ""
#: documents/serialisers.py:2989 documents/views.py:4807
#: documents/serialisers.py:2989 documents/views.py:4797
#, python-format
msgid "Documents not found: %(ids)s"
msgstr ""
@@ -1945,40 +1945,40 @@ msgstr ""
msgid ", "
msgstr ""
#: documents/views.py:337 documents/views.py:2748
#: documents/views.py:336 documents/views.py:2747
msgid "Invalid more_like_id"
msgstr ""
#: documents/views.py:1683
#: documents/views.py:1682
msgid "Invalid AI configuration."
msgstr ""
#: documents/views.py:1694
#: documents/views.py:1693
msgid "AI backend request timed out."
msgstr ""
#: documents/views.py:1706
#: documents/views.py:1705
msgid "AI backend rejected the request. Check logs for details."
msgstr ""
#: documents/views.py:2573 documents/views.py:2889
#: documents/views.py:2572 documents/views.py:2888
msgid "Specify only one of text, title_search, query, or more_like_id."
msgstr ""
#: documents/views.py:4823
#: documents/views.py:4813
#, python-format
msgid "Insufficient permissions to share document %(id)s."
msgstr ""
#: documents/views.py:4870
#: documents/views.py:4860
msgid "Bundle is already being processed."
msgstr ""
#: documents/views.py:4934
#: documents/views.py:4924
msgid "The share link bundle is still being prepared. Please try again later."
msgstr ""
#: documents/views.py:4948
#: documents/views.py:4938
msgid "The share link bundle is unavailable."
msgstr ""
-28
View File
@@ -4,7 +4,6 @@ from urllib.parse import quote
from allauth.account.adapter import DefaultAccountAdapter
from allauth.core import context
from allauth.headless.tokens.strategies.sessions import SessionTokenStrategy
from allauth.mfa.stages import AuthenticateStage
from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
from django.conf import settings
from django.contrib.auth.models import Group
@@ -20,34 +19,7 @@ from paperless.signals import handle_social_account_updated
logger = logging.getLogger("paperless.auth")
class SocialAccountAuthenticateStage(AuthenticateStage):
"""
Skips the 2FA prompt for a social login if the ID token's `amr` claim shows the
identity provider already used one of the SOCIAL_ACCOUNT_MFA_TRUSTED_AMR methods.
"""
def _should_handle(self, request: HttpRequest) -> bool:
sociallogin = (self.login.signal_kwargs or {}).get("sociallogin")
if sociallogin is not None and settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR:
id_token = (sociallogin.account.extra_data or {}).get("id_token") or {}
amr = id_token.get("amr") or []
if set(amr) & set(settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR):
logger.debug(
f"Skipping 2FA for `{self.login.user}`, provider `{sociallogin.account.provider}` reported amr {amr}",
)
return False
return super()._should_handle(request)
class CustomAccountAdapter(DefaultAccountAdapter):
def get_login_stages(self) -> list[str]:
return [
"paperless.adapter.SocialAccountAuthenticateStage"
if stage == "allauth.mfa.stages.AuthenticateStage"
else stage
for stage in super().get_login_stages()
]
def is_open_for_signup(self, request):
"""
Check whether the site is open for signups, which can be
+10 -11
View File
@@ -272,27 +272,26 @@ def check_deprecated_db_settings(
Detects legacy advanced options that should be migrated to
PAPERLESS_DB_OPTIONS. Returns one Warning per deprecated variable found.
"""
deprecated_vars: dict[str, str] = {
"PAPERLESS_DB_TIMEOUT": "timeout",
"PAPERLESS_DB_POOLSIZE": "pool.min_size / pool.max_size",
"PAPERLESS_DBSSLMODE": "sslmode",
"PAPERLESS_DBSSLROOTCERT": "sslrootcert",
"PAPERLESS_DBSSLCERT": "sslcert",
"PAPERLESS_DBSSLKEY": "sslkey",
}
deprecated_vars = (
"PAPERLESS_DB_TIMEOUT",
"PAPERLESS_DB_POOLSIZE",
"PAPERLESS_DBSSLMODE",
"PAPERLESS_DBSSLROOTCERT",
"PAPERLESS_DBSSLCERT",
"PAPERLESS_DBSSLKEY",
)
warnings: list[Warning] = []
for var_name, db_option_key in deprecated_vars.items():
for var_name in deprecated_vars:
if not os.getenv(var_name):
continue
warnings.append(
Warning(
f"Deprecated environment variable: {var_name}",
hint=(
f"{var_name} is no longer supported and will be removed in v3.2. "
f"{var_name} is deprecated. "
f"Set the equivalent option via PAPERLESS_DB_OPTIONS instead. "
f'Example: PAPERLESS_DB_OPTIONS=\'{{"{db_option_key}": "<value>"}}\'. '
"See https://docs.paperless-ngx.com/migration-v3/ for the full reference."
),
id="paperless.W001",
-5
View File
@@ -348,11 +348,6 @@ SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP: Final[str | None] = os.getenv(
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP: Final[str | None] = os.getenv(
"PAPERLESS_SOCIAL_ACCOUNT_SYNC_STAFF_GROUP",
)
SOCIAL_ACCOUNT_MFA_TRUSTED_AMR = [
value.strip()
for value in os.getenv("PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR", "mfa").split(",")
if value.strip()
]
HEADLESS_TOKEN_STRATEGY = "paperless.adapter.DrfTokenStrategy"
-24
View File
@@ -432,30 +432,6 @@ def tesseract_samples_dir(parser_samples_dir: Path) -> Path:
return parser_samples_dir / "tesseract"
@pytest.fixture(scope="session")
def multi_page_images_pdf_file(tesseract_samples_dir: Path) -> Path:
"""Path to a multi-page PDF with images.
Returns
-------
Path
Absolute path to ``tesseract/multi-page-images.pdf``.
"""
return tesseract_samples_dir / "multi-page-images.pdf"
@pytest.fixture(scope="session")
def simple_digital_pdf_file(tesseract_samples_dir: Path) -> Path:
"""Path to a simple digital PDF sample file.
Returns
-------
Path
Absolute path to ``tesseract/simple-digital.pdf``.
"""
return tesseract_samples_dir / "simple-digital.pdf"
@pytest.fixture(scope="session")
def simple_no_dpi_png_file(tesseract_samples_dir: Path) -> Path:
"""Path to a simple PNG without DPI information.
@@ -160,11 +160,11 @@ class TestGetPageCount:
def test_single_page_pdf(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
assert (
tesseract_parser.get_page_count(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
)
== 1
@@ -187,7 +187,7 @@ class TestGetPageCount:
self,
mocker: MockerFixture,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
caplog,
) -> None:
"""
@@ -203,7 +203,7 @@ class TestGetPageCount:
with caplog.at_level(logging.WARNING):
page_count = tesseract_parser.get_page_count(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
)
assert page_count is None
@@ -272,10 +272,10 @@ class TestGetThumbnail:
def test_thumbnail_is_file(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
thumb = tesseract_parser.get_thumbnail(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
)
assert thumb.is_file()
@@ -284,7 +284,7 @@ class TestGetThumbnail:
self,
mocker: MockerFixture,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
def _raise_on_pdf(input_file, output_file, **kwargs) -> None:
if ".pdf" in str(input_file):
@@ -294,7 +294,7 @@ class TestGetThumbnail:
mocker.patch("documents.parsers.run_convert", side_effect=_raise_on_pdf)
thumb = tesseract_parser.get_thumbnail(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
)
assert thumb.is_file()
@@ -320,11 +320,11 @@ class TestExtractText:
def test_extract_text_from_digital_pdf(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
text = tesseract_parser.extract_text(
None,
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
)
assert text is not None
assert "This is a test document." in text.strip()
@@ -339,7 +339,7 @@ class TestParsePdf:
def test_simple_digital_creates_archive(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_digital_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -353,7 +353,7 @@ class TestParsePdf:
- Text is extracted
"""
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-digital.pdf",
multi_page_digital_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -368,10 +368,10 @@ class TestParsePdf:
def test_with_form_default(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
with_form_pdf_file: Path,
) -> None:
tesseract_parser.parse(
tesseract_samples_dir / "with-form.pdf",
with_form_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -384,11 +384,11 @@ class TestParsePdf:
def test_with_form_redo_no_archive_when_not_requested(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
with_form_pdf_file: Path,
) -> None:
tesseract_parser.settings.mode = ModeChoices.REDO
tesseract_parser.parse(
tesseract_samples_dir / "with-form.pdf",
with_form_pdf_file,
"application/pdf",
produce_archive=False,
)
@@ -401,11 +401,11 @@ class TestParsePdf:
def test_with_form_force(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
with_form_pdf_file: Path,
) -> None:
tesseract_parser.settings.mode = ModeChoices.FORCE
tesseract_parser.parse(
tesseract_samples_dir / "with-form.pdf",
with_form_pdf_file,
"application/pdf",
)
assert_ordered_substrings(
@@ -446,7 +446,7 @@ class TestParsePdf:
self,
mocker: MockerFixture,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
mocker.patch(
"ocrmypdf.ocr",
@@ -454,7 +454,7 @@ class TestParsePdf:
)
with pytest.raises(ParseError):
tesseract_parser.parse(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
)
@@ -530,10 +530,10 @@ class TestParseMultiPage:
def test_multi_page_digital(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_digital_pdf_file: Path,
) -> None:
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-digital.pdf",
multi_page_digital_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -557,12 +557,12 @@ class TestParseMultiPage:
self,
mode: str,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_digital_pdf_file: Path,
) -> None:
tesseract_parser.settings.pages = 2
tesseract_parser.settings.mode = mode
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-digital.pdf",
multi_page_digital_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -576,11 +576,11 @@ class TestParseMultiPage:
def test_multi_page_images_skip(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_images_pdf_file: Path,
) -> None:
tesseract_parser.settings.mode = ModeChoices.AUTO
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-images.pdf",
multi_page_images_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -594,7 +594,7 @@ class TestParseMultiPage:
def test_multi_page_images_redo_pages_2(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_images_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -609,7 +609,7 @@ class TestParseMultiPage:
tesseract_parser.settings.pages = 2
tesseract_parser.settings.mode = ModeChoices.REDO
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-images.pdf",
multi_page_images_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -622,7 +622,7 @@ class TestParseMultiPage:
def test_multi_page_images_force_page_1(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_images_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -637,7 +637,7 @@ class TestParseMultiPage:
tesseract_parser.settings.pages = 1
tesseract_parser.settings.mode = ModeChoices.FORCE
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-images.pdf",
multi_page_images_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -733,7 +733,7 @@ class TestSkipArchive:
def test_skip_noarchive_with_text_layer(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_digital_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -747,7 +747,7 @@ class TestSkipArchive:
"""
tesseract_parser.settings.mode = ModeChoices.AUTO
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-digital.pdf",
multi_page_digital_pdf_file,
"application/pdf",
produce_archive=False,
)
@@ -762,7 +762,7 @@ class TestSkipArchive:
def test_skip_noarchive_image_only_creates_archive(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
multi_page_images_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -775,7 +775,7 @@ class TestSkipArchive:
"""
tesseract_parser.settings.mode = ModeChoices.AUTO
tesseract_parser.parse(
tesseract_samples_dir / "multi-page-images.pdf",
multi_page_images_pdf_file,
"application/pdf",
)
assert tesseract_parser.archive_path is not None
@@ -787,29 +787,29 @@ class TestSkipArchive:
)
@pytest.mark.parametrize(
("produce_archive", "filename", "expect_archive"),
("produce_archive", "sample_fixture", "expect_archive"),
[
pytest.param(
True,
"multi-page-digital.pdf",
"multi_page_digital_pdf_file",
True,
id="produce-archive-with-text",
),
pytest.param(
True,
"multi-page-images.pdf",
"multi_page_images_pdf_file",
True,
id="produce-archive-no-text",
),
pytest.param(
False,
"multi-page-digital.pdf",
"multi_page_digital_pdf_file",
False,
id="no-archive-with-text-layer",
),
pytest.param(
False,
"multi-page-images.pdf",
"multi_page_images_pdf_file",
False,
id="no-archive-no-text-layer",
),
@@ -818,10 +818,10 @@ class TestSkipArchive:
def test_produce_archive_flag(
self,
produce_archive: bool, # noqa: FBT001
filename: str,
sample_fixture: str,
expect_archive: bool, # noqa: FBT001
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
request: pytest.FixtureRequest,
) -> None:
"""
GIVEN:
@@ -834,8 +834,9 @@ class TestSkipArchive:
- Text is always extracted
"""
tesseract_parser.settings.mode = ModeChoices.AUTO
sample = request.getfixturevalue(sample_fixture)
tesseract_parser.parse(
tesseract_samples_dir / filename,
sample,
"application/pdf",
produce_archive=produce_archive,
)
@@ -852,7 +853,7 @@ class TestSkipArchive:
self,
mocker: MockerFixture,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -868,7 +869,7 @@ class TestSkipArchive:
tesseract_parser.settings.mode = ModeChoices.AUTO
mock_ocr = mocker.patch("ocrmypdf.ocr")
tesseract_parser.parse(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
produce_archive=False,
)
@@ -907,7 +908,7 @@ class TestSkipArchive:
def test_tagged_pdf_produces_pdfa_archive_without_ocr(
self,
tesseract_parser: RasterisedDocumentParser,
tesseract_samples_dir: Path,
simple_digital_pdf_file: Path,
) -> None:
"""
GIVEN:
@@ -922,7 +923,7 @@ class TestSkipArchive:
"""
tesseract_parser.settings.mode = ModeChoices.AUTO
tesseract_parser.parse(
tesseract_samples_dir / "simple-digital.pdf",
simple_digital_pdf_file,
"application/pdf",
produce_archive=True,
)
@@ -1,19 +0,0 @@
<html>
<head>
<meta http-equiv="content-type" content="text/html; charset=UTF-8">
</head>
<body>
<p>Some Text</p>
<p>
<img src="cid:part1.pNdUSz0s.D3NqVtPg@example.de" alt="Has to be rewritten to work..">
<img src="http://localhost:8080/assets/logo_full_white.svg" alt="This image should not be shown.">
</p>
<p>and an embedded image.<br>
</p>
<p id="changeme">Paragraph unchanged.</p>
<scRipt>
document.getElementById("changeme").innerHTML = "Paragraph changed via Java Script.";
</script>
</body>
</html>
Binary file not shown.
Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.8 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 6.9 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 32 KiB

-90
View File
@@ -2,26 +2,19 @@ import logging
import pytest
from allauth.account.adapter import get_adapter
from allauth.account.models import Login
from allauth.core import context
from allauth.mfa.totp.internal.auth import TOTP
from allauth.mfa.totp.internal.auth import generate_totp_secret
from allauth.socialaccount.adapter import get_adapter as get_social_adapter
from allauth.socialaccount.models import SocialAccount
from allauth.socialaccount.models import SocialLogin
from django.contrib.auth.models import AnonymousUser
from django.contrib.auth.models import Group
from django.contrib.auth.models import User
from django.forms import ValidationError
from django.http import HttpRequest
from django.test import RequestFactory
from django.urls import reverse
from pytest_django.fixtures import Settings
from pytest_mock import MockerFixture
from rest_framework.authtoken.models import Token
from paperless.adapter import DrfTokenStrategy
from paperless.adapter import SocialAccountAuthenticateStage
from paperless_testing.factories import UserFactory
@@ -136,89 +129,6 @@ class TestCustomAccountAdapter:
assert not user2.is_superuser
@pytest.mark.django_db
class TestSocialAccountAuthenticateStage:
@staticmethod
def _should_handle(user: User, extra_data: dict | None) -> bool:
signal_kwargs = None
if extra_data is not None:
account = SocialAccount(
provider="openid_connect",
uid="uid",
extra_data=extra_data,
)
signal_kwargs = {"sociallogin": SocialLogin(user=user, account=account)}
request = RequestFactory().get("/")
request.session = {}
stage = SocialAccountAuthenticateStage(
None,
request,
Login(user=user, signal_kwargs=signal_kwargs),
)
return stage._should_handle(request)
def test_stage_replaces_allauth_stage(self) -> None:
stages = get_adapter().get_login_stages()
assert "paperless.adapter.SocialAccountAuthenticateStage" in stages
assert "allauth.mfa.stages.AuthenticateStage" not in stages
@pytest.mark.parametrize(
("trusted_amr", "extra_data", "expected"),
[
pytest.param(
["mfa"],
{"id_token": {"amr": ["pwd", "mfa"]}},
False,
id="trusted-amr",
),
pytest.param(
["phr"],
{"id_token": {"amr": ["otp"]}},
True,
id="untrusted-amr",
),
pytest.param([], {"id_token": {"amr": ["mfa"]}}, True, id="setting-unset"),
pytest.param(["mfa"], {"id_token": {}}, True, id="no-amr-claim"),
pytest.param(
["mfa"],
{"userinfo": {"amr": ["mfa"]}},
True,
id="userinfo-only",
),
pytest.param(["mfa"], None, True, id="regular-login"),
],
)
def test_mfa_skipped_only_for_trusted_amr(
self,
settings: Settings,
trusted_amr: list[str],
extra_data: dict | None,
*,
expected: bool,
) -> None:
"""
GIVEN:
- A user with TOTP enabled
WHEN:
- The user logs in, via a social account or not
THEN:
- The 2FA prompt is only skipped if the ID token's amr claim contains
one of SOCIAL_ACCOUNT_MFA_TRUSTED_AMR
"""
settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR = trusted_amr
user = UserFactory(username="testuser")
TOTP.activate(user, generate_totp_secret())
assert self._should_handle(user, extra_data) is expected
def test_no_mfa_for_user_without_authenticator(self, settings: Settings) -> None:
settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR = ["mfa"]
user = UserFactory(username="testuser")
assert not self._should_handle(user, {"id_token": {"amr": ["pwd"]}})
class TestCustomSocialAccountAdapter:
@pytest.mark.django_db
def test_is_open_for_signup(self, settings: Settings) -> None:
+11 -25
View File
@@ -211,14 +211,14 @@ class TestAuditLogChecks:
assert "auditlog table was found but audit log is disabled." in msgs[0].msg
DEPRECATED_VARS: dict[str, str] = {
"PAPERLESS_DB_TIMEOUT": "timeout",
"PAPERLESS_DB_POOLSIZE": "pool.min_size / pool.max_size",
"PAPERLESS_DBSSLMODE": "sslmode",
"PAPERLESS_DBSSLROOTCERT": "sslrootcert",
"PAPERLESS_DBSSLCERT": "sslcert",
"PAPERLESS_DBSSLKEY": "sslkey",
}
DEPRECATED_VARS = (
"PAPERLESS_DB_TIMEOUT",
"PAPERLESS_DB_POOLSIZE",
"PAPERLESS_DBSSLMODE",
"PAPERLESS_DBSSLROOTCERT",
"PAPERLESS_DBSSLCERT",
"PAPERLESS_DBSSLKEY",
)
class TestDeprecatedDbSettings:
@@ -234,26 +234,11 @@ class TestDeprecatedDbSettings:
result = check_deprecated_db_settings(None)
assert result == []
@pytest.mark.parametrize(
("env_var", "db_option_key"),
[
pytest.param("PAPERLESS_DB_TIMEOUT", "timeout", id="db-timeout"),
pytest.param(
"PAPERLESS_DB_POOLSIZE",
"pool.min_size / pool.max_size",
id="db-poolsize",
),
pytest.param("PAPERLESS_DBSSLMODE", "sslmode", id="ssl-mode"),
pytest.param("PAPERLESS_DBSSLROOTCERT", "sslrootcert", id="ssl-rootcert"),
pytest.param("PAPERLESS_DBSSLCERT", "sslcert", id="ssl-cert"),
pytest.param("PAPERLESS_DBSSLKEY", "sslkey", id="ssl-key"),
],
)
@pytest.mark.parametrize("env_var", DEPRECATED_VARS)
def test_single_deprecated_var_produces_one_warning(
self,
mocker: MockerFixture,
env_var: str,
db_option_key: str,
) -> None:
"""Each deprecated var in isolation produces exactly one warning."""
mocker.patch.dict(os.environ, {env_var: "some_value"}, clear=True)
@@ -264,7 +249,8 @@ class TestDeprecatedDbSettings:
assert isinstance(warning, Warning)
assert warning.id == "paperless.W001"
assert env_var in warning.hint
assert db_option_key in warning.hint
assert "PAPERLESS_DB_OPTIONS" in warning.hint
assert "https://docs.paperless-ngx.com/migration-v3/" in warning.hint
def test_multiple_deprecated_vars_produce_one_warning_each(
self,
+4 -6
View File
@@ -16,8 +16,6 @@ from paperless.parsers.utils import read_file_handle_unicode_errors
if TYPE_CHECKING:
from pytest_mock import MockerFixture
SAMPLES = Path(__file__).parent / "samples" / "tesseract"
class TestReadFileHandleUnicodeErrors:
def test_plain_utf8(self, tmp_path: Path) -> None:
@@ -55,11 +53,11 @@ class TestReadFileHandleUnicodeErrors:
class TestIsTaggedPdf:
def test_tagged_pdf_returns_true(self) -> None:
assert is_tagged_pdf(SAMPLES / "simple-digital.pdf") is True
def test_tagged_pdf_returns_true(self, simple_digital_pdf_file: Path) -> None:
assert is_tagged_pdf(simple_digital_pdf_file) is True
def test_untagged_pdf_returns_false(self) -> None:
assert is_tagged_pdf(SAMPLES / "multi-page-images.pdf") is False
def test_untagged_pdf_returns_false(self, multi_page_images_pdf_file: Path) -> None:
assert is_tagged_pdf(multi_page_images_pdf_file) is False
def test_nonexistent_path_returns_false(self) -> None:
assert is_tagged_pdf(Path("/nonexistent/file.pdf")) is False
+8 -10
View File
@@ -4,8 +4,7 @@ from django.conf import settings
from django.contrib.auth.models import User
from documents.models import Document
from documents.permissions import permitted_object_ids
from documents.permissions import restrict_queryset_to_visible
from documents.permissions import permitted_document_ids
from documents.permissions import user_is_unrestricted
from paperless.config import AIConfig
from paperless_ai.base_model import ClassificationSuggestions
@@ -54,8 +53,8 @@ def _fulltext_similar_documents(
active superuser - see user_is_unrestricted) is normalized to ``None``
before calling, since the backend's permission filter has no superuser
short-circuit of its own. Results are re-checked with
restrict_queryset_to_visible() since Tantivy's indexed permission fields
lag the DB via async reindexing.
permitted_document_ids() since Tantivy's indexed permission fields lag the
DB via async reindexing and judge a version by its own owner.
"""
from documents.search import get_backend
@@ -69,10 +68,9 @@ def _fulltext_similar_documents(
)
if not unrestricted:
allowed_ids = set(
restrict_queryset_to_visible(
Document.objects.filter(pk__in=similar_ids),
user,
"view_document",
Document.objects.filter(
pk__in=similar_ids,
id__in=permitted_document_ids(user),
).values_list("pk", flat=True),
)
similar_ids = [doc_id for doc_id in similar_ids if doc_id in allowed_ids]
@@ -200,13 +198,13 @@ def get_taxonomy_context(
# quadratic scan in the vector store at best, and past ~32,763
# documents a hard sqlite3.OperationalError (SQLite's
# bound-parameter limit) at worst.
# permitted_object_ids() has its own superuser shortcut that would
# permitted_document_ids() has its own superuser shortcut that would
# return every Document's id anyway, so this changes nothing about
# which documents are considered -- only how we get there.
visible_document_ids = (
None
if user_is_unrestricted(user)
else list(permitted_object_ids(user, Document, "view_document"))
else list(permitted_document_ids(user))
)
nodes = retrieve_similar_nodes(
document,
+83 -5
View File
@@ -552,7 +552,7 @@ class TestGetTaxonomyContextVisibility:
return_value=[],
)
mock_permitted = mocker.patch(
"paperless_ai.ai_classifier.permitted_object_ids",
"paperless_ai.ai_classifier.permitted_document_ids",
)
user = UserFactory.create(is_superuser=True)
@@ -582,7 +582,7 @@ class TestGetTaxonomyContextVisibility:
return_value=[],
)
mock_permitted = mocker.patch(
"paperless_ai.ai_classifier.permitted_object_ids",
"paperless_ai.ai_classifier.permitted_document_ids",
)
get_taxonomy_context(document, None)
@@ -611,16 +611,55 @@ class TestGetTaxonomyContextVisibility:
return_value=[],
)
mock_permitted = mocker.patch(
"paperless_ai.ai_classifier.permitted_object_ids",
"paperless_ai.ai_classifier.permitted_document_ids",
return_value=[1, 2, 3],
)
user = UserFactory.create(is_superuser=False)
get_taxonomy_context(document, user)
mock_permitted.assert_called_once_with(user, Document, "view_document")
mock_permitted.assert_called_once_with(user)
assert mock_retrieve.call_args.kwargs["document_ids"] == [1, 2, 3]
@pytest.mark.django_db
@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
def test_version_of_private_root_is_not_visible(
self,
mocker: pytest_mock.MockerFixture,
) -> None:
"""
GIVEN:
- A private root document owned by someone else
- A version of it whose own owner is unset, as when the root
changed hands after the version was created
WHEN:
- get_taxonomy_context() is called for a non-superuser
THEN:
- Neither the root nor the version is in the visible ids passed to
retrieve_similar_nodes(), since a version follows its root
"""
owner = UserFactory.create()
viewer = UserFactory.create(is_superuser=False)
root = DocumentFactory.create(content="private", owner=owner)
version = DocumentFactory.create(
content="private",
owner=None,
root_document=root,
version_index=1,
)
source = DocumentFactory.create(content="Some content", owner=viewer)
mock_retrieve = mocker.patch(
"paperless_ai.ai_classifier.retrieve_similar_nodes",
return_value=[],
)
get_taxonomy_context(source, viewer)
visible = mock_retrieve.call_args.kwargs["document_ids"]
assert source.pk in visible
assert root.pk not in visible
assert version.pk not in visible
@pytest.mark.django_db
class TestFulltextSimilarDocuments:
@@ -803,7 +842,7 @@ class TestFulltextSimilarDocuments:
- _fulltext_similar_documents() is called with that user
THEN:
- Only the still-permitted document is returned - the DB
re-check via restrict_queryset_to_visible() must catch the
re-check via permitted_document_ids() must catch the
document Tantivy's stale index still thinks is visible
"""
owner = UserFactory.create()
@@ -834,6 +873,45 @@ class TestFulltextSimilarDocuments:
assert [s["document_id"] for s in result] == [permitted.pk]
def test_excludes_version_of_private_root_for_regular_user(
self,
fulltext_backend: TantivyBackend,
) -> None:
"""
GIVEN:
- A regular user and a private root owned by someone else
- A version of that root with no owner of its own, which the
Tantivy index therefore treats as visible to everyone
WHEN:
- _fulltext_similar_documents() is called with that user
THEN:
- The version is not returned, since the DB re-check judges it by
its root
"""
owner = UserFactory.create()
viewer = UserFactory.create(is_superuser=False)
source = DocumentFactory.create(
content="shared content phrase",
owner=viewer,
)
root = DocumentFactory.create(
content="shared content phrase",
owner=owner,
)
version = DocumentFactory.create(
content="shared content phrase",
owner=None,
root_document=root,
version_index=1,
)
fulltext_backend.add_or_update(source)
fulltext_backend.add_or_update(root)
fulltext_backend.add_or_update(version)
result = _fulltext_similar_documents(source, user=viewer, top_k=5)
assert version.pk not in [s["document_id"] for s in result]
@pytest.mark.django_db
@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
+48
View File
@@ -0,0 +1,48 @@
"""Test sample files shared across tests.
A sample used by a single app stays in that app's ``tests/samples`` tree.
Samples needed by more than one app, or stored under more than one name, live
here exactly once.
"""
import shutil
from pathlib import Path
SHARED_SAMPLES_DIR = (Path(__file__).parent / "sample_files").resolve()
SIMPLE_DIGITAL_PDF = SHARED_SAMPLES_DIR / "simple-digital.pdf"
MULTI_PAGE_DIGITAL_PDF = SHARED_SAMPLES_DIR / "multi-page-digital.pdf"
WITH_FORM_PDF = SHARED_SAMPLES_DIR / "with-form.pdf"
MULTI_PAGE_IMAGES_PDF = SHARED_SAMPLES_DIR / "multi-page-images.pdf"
THUMBNAIL_WEBP = SHARED_SAMPLES_DIR / "thumbnail.webp"
# The fake media tree under documents/tests/samples/documents keeps only the
# files unique to it. The files that duplicate a shared sample are laid back
# down under their historical opaque names (DB rows and exporter manifests
# refer to them) by install_document_samples().
_DOCUMENT_SAMPLES_DIR = (
Path(__file__).parent.parent / "documents" / "tests" / "samples" / "documents"
).resolve()
_SHARED_AS_MEDIA = (
("originals/0000001.pdf", SIMPLE_DIGITAL_PDF),
("originals/0000002.pdf", MULTI_PAGE_DIGITAL_PDF),
("originals/0000003.pdf", WITH_FORM_PDF),
("archive/0000001.pdf", MULTI_PAGE_IMAGES_PDF),
("thumbnails/0000001.webp", THUMBNAIL_WEBP),
("thumbnails/0000002.webp", THUMBNAIL_WEBP),
("thumbnails/0000003.webp", THUMBNAIL_WEBP),
("thumbnails/0000004.webp", THUMBNAIL_WEBP),
)
def install_document_samples(dest: Path) -> None:
"""Populate ``dest`` with the full fake media tree.
Replaces ``shutil.copytree(<samples>/documents, dest)`` in tests.
"""
shutil.copytree(_DOCUMENT_SAMPLES_DIR, dest, dirs_exist_ok=True)
for relpath, source in _SHARED_AS_MEDIA:
target = dest / relpath
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)