Compare commits

..
Author SHA1 Message Date
stumpylog c40849e9bb Chore: make the backend test suite pass on MariaDB
Make the reference predicates for the string custom field filters follow
the column collation, which is case insensitive on MariaDB, and quote
identifiers in the effective content query matchers with the backend's
own quoting.
2026-10-30 12:56:16 -07:00
stumpylogandClaude Sonnet 5.5 f4358cef53 Fix: scheduled workflow filename filter fails on PostgreSQL
The prefilter passed fnmatch.translate output to an iregex lookup. That
output uses (?s:...), atomic groups and named group backreferences, which
PostgreSQL rejects with "invalid embedded option". Translate the glob with
a helper that only emits syntax shared by Python, PostgreSQL and MariaDB.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-30 12:16:41 -07:00
stumpylogandClaude Sonnet 5.5 2cb04b0b8b 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.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-30 12:16:35 -07:00
4 changed files with 135 additions and 10 deletions

No files matched your search

+45 -3
View File
@@ -3,7 +3,6 @@ from __future__ import annotations
import logging import logging
import re import re
from fnmatch import fnmatch from fnmatch import fnmatch
from fnmatch import translate as fnmatch_translate
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from rest_framework import serializers from rest_framework import serializers
@@ -548,6 +547,48 @@ def existing_document_matches_workflow(
return (True, None) 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( def prefilter_documents_by_workflowtrigger(
documents: QuerySet[Document], documents: QuerySet[Document],
trigger: WorkflowTrigger, trigger: WorkflowTrigger,
@@ -628,8 +669,9 @@ def prefilter_documents_by_workflowtrigger(
documents = documents.annotate(**annotations).filter(custom_field_q) documents = documents.annotate(**annotations).filter(custom_field_q)
if trigger.filter_filename: if trigger.filter_filename:
regex = fnmatch_translate(trigger.filter_filename).lstrip("^").rstrip("$") documents = documents.filter(
documents = documents.filter(original_filename__iregex=regex) original_filename__iregex=glob_to_database_regex(trigger.filter_filename),
)
return documents return documents
@@ -6,6 +6,7 @@ from unittest.mock import patch
from urllib.parse import quote from urllib.parse import quote
import time_machine import time_machine
from django.db import connection
from django.test import override_settings from django.test import override_settings
from django.utils import timezone from django.utils import timezone
from rest_framework.test import APITestCase from rest_framework.test import APITestCase
@@ -20,6 +21,13 @@ from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import UserFactory 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: class DocumentWrapper:
""" """
Allows Pythonic access to the custom fields associated with the wrapped document. Allows Pythonic access to the custom fields associated with the wrapped document.
@@ -287,7 +295,8 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
self._assert_query_match_predicate( self._assert_query_match_predicate(
["string_field", "exact", "paperless"], ["string_field", "exact", "paperless"],
lambda document: ( lambda document: (
"string_field" in document and document["string_field"] == "paperless" "string_field" in document
and _exact_equal(document["string_field"], "paperless")
), ),
) )
@@ -296,7 +305,10 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
["string_field", "in", ["paperless", "Paperless"]], ["string_field", "in", ["paperless", "Paperless"]],
lambda document: ( lambda document: (
"string_field" in document "string_field" in document
and document["string_field"] in ("paperless", "Paperless") and any(
_exact_equal(document["string_field"], expected)
for expected in ("paperless", "Paperless")
)
), ),
) )
@@ -761,7 +773,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
lambda document: ( lambda document: (
not ( not (
"string_field" in document "string_field" in document
and document["string_field"] == "paperless" and _exact_equal(document["string_field"], "paperless")
) )
), ),
) )
@@ -242,9 +242,7 @@ class TestDocumentListEffectiveContentAnnotation:
# Only the list query itself reads a content column: no extra query # Only the list query itself reads a content column: no extra query
# for the skipped prefetch, none for a per-instance fallback # for the skipped prefetch, none for a per-instance fallback
content_queries = [ content_queries = [
query query for query in ctx.captured_queries if _content_column() in query["sql"]
for query in ctx.captured_queries
if '"documents_document"."content"' in query["sql"]
] ]
assert len(content_queries) == 1 assert len(content_queries) == 1
@@ -365,6 +363,11 @@ def _get_document_queries(
return [q for q in ctx.captured_queries if '"django_content_type"' not in q["sql"]] 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( def _get_effective_content_fallback_queries(
ctx: CaptureQueriesContext, ctx: CaptureQueriesContext,
) -> list[dict[str, str]]: ) -> list[dict[str, str]]:
@@ -379,7 +382,7 @@ def _get_effective_content_fallback_queries(
return [ return [
q q
for q in ctx.captured_queries for q in ctx.captured_queries
if q["sql"].startswith('SELECT "documents_document"."content" FROM') if q["sql"].startswith(f"SELECT {_content_column()} FROM")
] ]
@@ -0,0 +1,68 @@
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]