Compare commits

..
Author SHA1 Message Date
stumpylog 5be95f8fcc Chore: make the backend test suite pass on PostgreSQL
Remove SQLite-only assumptions from the tests: hard-coded primary keys,
NULL ordering, the database vendor, and the CPU-dependent default process
count. Keep the test connection open inside the test transaction, since
close_old_connections() and connections.close_all() really close it on
PostgreSQL.
2026-10-30 12:55:37 -07:00
4 changed files with 10 additions and 135 deletions

No files matched your search

+3 -45
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import logging
import re
from fnmatch import fnmatch
from fnmatch import translate as fnmatch_translate
from typing import TYPE_CHECKING
from rest_framework import serializers
@@ -547,48 +548,6 @@ def existing_document_matches_workflow(
return (True, None)
def glob_to_database_regex(pattern: str) -> str:
"""
Translates an fnmatch style glob into a regular expression which the
database can evaluate, matching the whole value.
fnmatch.translate is not usable here: it emits syntax such as (?s:...),
atomic groups and named group backreferences, which PostgreSQL rejects.
Only constructs common to Python, PostgreSQL and MariaDB are produced.
"""
parts: list[str] = []
i = 0
n = len(pattern)
while i < n:
char = pattern[i]
i += 1
if char == "*":
parts.append(".*")
elif char == "?":
parts.append(".")
elif char == "[":
j = i
if j < n and pattern[j] == "!":
j += 1
if j < n and pattern[j] == "]":
j += 1
while j < n and pattern[j] != "]":
j += 1
if j >= n:
parts.append(r"\[")
else:
stuff = pattern[i:j].replace("\\", r"\\").replace("[", r"\[")
i = j + 1
if stuff.startswith("!"):
stuff = "^" + stuff[1:]
elif stuff.startswith("^"):
stuff = "\\" + stuff
parts.append(f"[{stuff}]")
else:
parts.append(re.escape(char))
return "^" + "".join(parts) + "$"
def prefilter_documents_by_workflowtrigger(
documents: QuerySet[Document],
trigger: WorkflowTrigger,
@@ -669,9 +628,8 @@ def prefilter_documents_by_workflowtrigger(
documents = documents.annotate(**annotations).filter(custom_field_q)
if trigger.filter_filename:
documents = documents.filter(
original_filename__iregex=glob_to_database_regex(trigger.filter_filename),
)
regex = fnmatch_translate(trigger.filter_filename).lstrip("^").rstrip("$")
documents = documents.filter(original_filename__iregex=regex)
return documents
@@ -6,7 +6,6 @@ from unittest.mock import patch
from urllib.parse import quote
import time_machine
from django.db import connection
from django.test import override_settings
from django.utils import timezone
from rest_framework.test import APITestCase
@@ -21,13 +20,6 @@ from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import UserFactory
def _exact_equal(value: str | None, expected: str) -> bool:
"""`exact` follows the column collation, which is case insensitive on MariaDB."""
if connection.vendor == "mysql" and value is not None:
return value.casefold() == expected.casefold()
return value == expected
class DocumentWrapper:
"""
Allows Pythonic access to the custom fields associated with the wrapped document.
@@ -295,8 +287,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
self._assert_query_match_predicate(
["string_field", "exact", "paperless"],
lambda document: (
"string_field" in document
and _exact_equal(document["string_field"], "paperless")
"string_field" in document and document["string_field"] == "paperless"
),
)
@@ -305,10 +296,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
["string_field", "in", ["paperless", "Paperless"]],
lambda document: (
"string_field" in document
and any(
_exact_equal(document["string_field"], expected)
for expected in ("paperless", "Paperless")
)
and document["string_field"] in ("paperless", "Paperless")
),
)
@@ -773,7 +761,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
lambda document: (
not (
"string_field" in document
and _exact_equal(document["string_field"], "paperless")
and document["string_field"] == "paperless"
)
),
)
@@ -242,7 +242,9 @@ class TestDocumentListEffectiveContentAnnotation:
# Only the list query itself reads a content column: no extra query
# for the skipped prefetch, none for a per-instance fallback
content_queries = [
query for query in ctx.captured_queries if _content_column() in query["sql"]
query
for query in ctx.captured_queries
if '"documents_document"."content"' in query["sql"]
]
assert len(content_queries) == 1
@@ -363,11 +365,6 @@ def _get_document_queries(
return [q for q in ctx.captured_queries if '"django_content_type"' not in q["sql"]]
def _content_column() -> str:
quote = connection.ops.quote_name
return f"{quote('documents_document')}.{quote('content')}"
def _get_effective_content_fallback_queries(
ctx: CaptureQueriesContext,
) -> list[dict[str, str]]:
@@ -382,7 +379,7 @@ def _get_effective_content_fallback_queries(
return [
q
for q in ctx.captured_queries
if q["sql"].startswith(f"SELECT {_content_column()} FROM")
if q["sql"].startswith('SELECT "documents_document"."content" FROM')
]
@@ -1,68 +0,0 @@
import re
from fnmatch import fnmatch
import pytest
from documents.matching import glob_to_database_regex
from documents.models import Document
from paperless_testing.factories import DocumentFactory
NAMES = [
"invoice.pdf",
"Invoice.PDF",
"invoice_2024.pdf",
"a.b.c",
"a-b",
"sample [1].pdf",
"sample 1.pdf",
"x^y",
"back\\slash",
"dir/file.pdf",
"",
]
PATTERNS = [
"*",
"invoice.pdf",
"*.pdf",
"*invoice*pdf",
"*a*b*c*",
"invoice_????.pdf",
"sample [[]1].pdf",
"sample [1-2].pdf",
"sample [!2].pdf",
"sample [^2].pdf",
"sample [",
"a.b.c",
"x^y",
"back\\slash",
"[]]",
]
@pytest.mark.parametrize("pattern", PATTERNS)
def test_regex_agrees_with_fnmatch(pattern: str) -> None:
regex = glob_to_database_regex(pattern)
for name in NAMES:
expected = fnmatch(name.lower(), pattern.lower())
actual = re.search(regex, name, re.IGNORECASE) is not None
assert actual is expected, f"{pattern!r} vs {name!r}"
@pytest.mark.parametrize("pattern", ["*", "*a*b*c*", "a*b*c*d*e"])
def test_regex_avoids_syntax_postgresql_rejects(pattern: str) -> None:
regex = glob_to_database_regex(pattern)
assert not any(marker in regex for marker in ("(?", "\\Z"))
@pytest.mark.django_db
def test_database_filter_matches_glob() -> None:
matching = DocumentFactory(original_filename="Invoice_2024.PDF")
DocumentFactory(original_filename="invoice_2024.txt")
DocumentFactory(original_filename="receipt.pdf")
found = Document.objects.filter(
original_filename__iregex=glob_to_database_regex("*invoice*.pdf"),
)
assert list(found) == [matching]