From 02d355061f8ad36cbe042b07c191c81adee0c137 Mon Sep 17 00:00:00 2001 From: Trenton H <797416+stumpylog@users.noreply.github.com> Date: Tue, 22 Sep 2026 08:02:18 -0700 Subject: [PATCH] Chore: Move helpers that other test modules imported out of test files (#14223) The mail message and mailbox builders, the fake libmagic and the classifier preprocessor stub lived inside test_mail.py and test_classifier.py, so other test modules imported them by importing a test module. They now live in helpers modules beside the tests that use them. --- src/documents/tests/helpers.py | 10 + src/documents/tests/test_classifier.py | 11 +- src/documents/tests/test_tasks.py | 2 +- src/paperless_mail/tests/helpers.py | 262 ++++++++++++++++++ src/paperless_mail/tests/test_api.py | 2 +- src/paperless_mail/tests/test_mail.py | 262 +----------------- src/paperless_mail/tests/test_mail_nfc.py | 6 +- src/paperless_mail/tests/test_preprocessor.py | 2 +- 8 files changed, 283 insertions(+), 274 deletions(-) create mode 100644 src/documents/tests/helpers.py create mode 100644 src/paperless_mail/tests/helpers.py diff --git a/src/documents/tests/helpers.py b/src/documents/tests/helpers.py new file mode 100644 index 0000000000..e90b5fcf8a --- /dev/null +++ b/src/documents/tests/helpers.py @@ -0,0 +1,10 @@ +import re + + +def dummy_preprocess(content: str) -> str: + """ + Simpler, faster pre-processing for testing purposes + """ + content = content.lower().strip() + content = re.sub(r"\s+", " ", content) + return content diff --git a/src/documents/tests/test_classifier.py b/src/documents/tests/test_classifier.py index cb8049ce84..ae8921ad69 100644 --- a/src/documents/tests/test_classifier.py +++ b/src/documents/tests/test_classifier.py @@ -1,5 +1,4 @@ import pickle -import re import warnings from datetime import UTC from datetime import datetime @@ -28,6 +27,7 @@ from documents.models import DocumentType from documents.models import MatchingModel from documents.models import StoragePath from documents.models import Tag +from documents.tests.helpers import dummy_preprocess from paperless.settings import CLASSIFIER_LANGUAGES from paperless.signed_pickle import HMAC_SIZE from paperless.signed_pickle import signed_pickle_dumps @@ -36,15 +36,6 @@ from paperless_testing.factories import DocumentFactory from paperless_testing.factories import TagFactory -def dummy_preprocess(content: str) -> str: - """ - Simpler, faster pre-processing for testing purposes - """ - content = content.lower().strip() - content = re.sub(r"\s+", " ", content) - return content - - class TestClassifier(DirectoriesMixin, TestCase): def setUp(self) -> None: super().setUp() diff --git a/src/documents/tests/test_tasks.py b/src/documents/tests/test_tasks.py index 6829beb6b3..247b6a529b 100644 --- a/src/documents/tests/test_tasks.py +++ b/src/documents/tests/test_tasks.py @@ -17,7 +17,7 @@ from documents.models import Tag from documents.models import WorkflowAction from documents.sanity_checker import SanityCheckFailedException from documents.sanity_checker import SanityCheckMessages -from documents.tests.test_classifier import dummy_preprocess +from documents.tests.helpers import dummy_preprocess from paperless_testing.assertions import FileSystemAssertsMixin from paperless_testing.dirs import DirectoriesMixin diff --git a/src/paperless_mail/tests/helpers.py b/src/paperless_mail/tests/helpers.py new file mode 100644 index 0000000000..76d308a6b1 --- /dev/null +++ b/src/paperless_mail/tests/helpers.py @@ -0,0 +1,262 @@ +import dataclasses +import email.message +import uuid +from contextlib import AbstractContextManager + +from imap_tools import MailboxFolderSelectError +from imap_tools import MailboxLoginError +from imap_tools import MailMessage +from imap_tools import MailMessageFlags + + +@dataclasses.dataclass +class _AttachmentDef: + filename: str = "a_file.pdf" + maintype: str = "application/pdf" + subtype: str = "pdf" + disposition: str = "attachment" + content: bytes = b"a PDF document" + + +class BogusFolderManager: + current_folder = "INBOX" + uidvalidity = "1" + + def set(self, new_folder) -> None: + if new_folder not in ["INBOX", "spam"]: + raise MailboxFolderSelectError(None, "uhm") + self.current_folder = new_folder + + def status(self, folder, options): + return {"UIDVALIDITY": self.uidvalidity} + + +class BogusClient: + def __init__(self, messages) -> None: + self.messages: list[MailMessage] = messages + self.capabilities: list[str] = [] + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + pass + + def authenticate(self, mechanism, authobject) -> None: + # authobject must be a callable object + auth_bytes = authobject(None) + if auth_bytes != b"\x00admin\x00w57\xc3\xa4\xc3\xb6\xc3\xbcw4b6huwb6nhu": + raise MailboxLoginError("BAD", "OK") + + def uid(self, command, *args) -> None: + if command == "STORE": + for message in self.messages: + if message.uid == args[0]: + flag = args[2] + if flag == "processed": + message._raw_flag_data.append(b"+FLAGS (processed)") + if hasattr(message, "flags"): + del message.flags + + +class BogusMailBox(AbstractContextManager): + # Common values so tests don't need to remember an accepted login + USERNAME: str = "admin" + ASCII_PASSWORD: str = "secret" + # Note the non-ascii characters here + UTF_PASSWORD: str = "w57äöüw4b6huwb6nhu" + # A dummy access token + ACCESS_TOKEN = "ea7e075cd3acf2c54c48e600398d5d5a" + + def __init__(self) -> None: + self.messages: list[MailMessage] = [] + self.messages_spam: list[MailMessage] = [] + self.folder = BogusFolderManager() + self.client = BogusClient(self.messages) + self._host = "" + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + pass + + def updateClient(self) -> None: + self.client = BogusClient(self.messages) + + def login(self, username, password) -> None: + # This will raise a UnicodeEncodeError if the password is not ASCII only + password.encode("ascii") + # Otherwise, check for correct values + if username != self.USERNAME or password != self.ASCII_PASSWORD: + raise MailboxLoginError("BAD", "OK") + + def login_utf8(self, username, password) -> None: + # Expected to only be called with the UTF-8 password + if username != self.USERNAME or password != self.UTF_PASSWORD: + raise MailboxLoginError("BAD", "OK") + + def xoauth2(self, username: str, access_token: str) -> None: + if username != self.USERNAME or access_token != self.ACCESS_TOKEN: + raise MailboxLoginError("BAD", "OK") + + def fetch( + self, + criteria="ALL", + charset="", + *, + mark_seen=True, + bulk=True, + uid_list=None, + ): + if uid_list is not None: + return [m for m in self.messages if m.uid in uid_list] + return self._filter_messages(criteria) + + def uids(self, criteria, charset="") -> list[str]: + return [m.uid for m in self._filter_messages(criteria)] + + def _filter_messages(self, criteria): + msg = self.messages + + criteria = str(criteria).strip("()").split(" ") + + if "UNSEEN" in criteria: + msg = filter(lambda m: not m.seen, msg) + + if "SUBJECT" in criteria: + subject = criteria[criteria.index("SUBJECT") + 1].strip('"') + msg = filter(lambda m: subject in m.subject, msg) + + if "BODY" in criteria: + body = criteria[criteria.index("BODY") + 1].strip('"') + msg = filter(lambda m: body in m.text, msg) + + if "FROM" in criteria: + from_ = criteria[criteria.index("FROM") + 1].strip('"') + msg = filter(lambda m: from_ in m.from_, msg) + + if "TO" in criteria: + to_ = criteria[criteria.index("TO") + 1].strip('"') + msg = filter(lambda m: any(to_ in to_addr for to_addr in m.to), msg) + + if "UNFLAGGED" in criteria: + msg = filter(lambda m: not m.flagged, msg) + + if "UNKEYWORD" in criteria: + tag = criteria[criteria.index("UNKEYWORD") + 1].strip("'") + msg = filter(lambda m: tag not in m.flags, msg) + + if "(X-GM-LABELS" in criteria: # ['NOT', '(X-GM-LABELS', '"processed"'] + msg = filter(lambda m: "processed" not in m.flags, msg) + + if "UID" in criteria: + uid_list = criteria[criteria.index("UID") + 1].split(",") + msg = filter(lambda m: m.uid in uid_list, msg) + + return list(msg) + + def delete(self, uid_list) -> None: + self.messages = list(filter(lambda m: m.uid not in uid_list, self.messages)) + + def flag(self, uid_list, flag_set, value) -> None: + for message in self.messages: + if message.uid in uid_list: + for flag in flag_set: + if flag == MailMessageFlags.FLAGGED: + message.flagged = value + if flag == MailMessageFlags.SEEN: + message.seen = value + if flag == "processed": + message._raw_flag_data.append(b"+FLAGS (processed)") + if hasattr(message, "flags"): + del message.flags + + def move(self, uid_list, folder) -> None: + if folder == "spam": + self.messages_spam += list( + filter(lambda m: m.uid in uid_list, self.messages), + ) + self.messages = list(filter(lambda m: m.uid not in uid_list, self.messages)) + else: + raise Exception + + +def fake_magic_from_buffer(buffer, *, mime=False): + if mime: + if "PDF" in str(buffer): + return "application/pdf" + else: + return "unknown/type" + else: + return "Some verbose file description" + + +class MessageBuilder: + def __init__(self) -> None: + self._next_uid = 1 + + def create_message( + self, + *, + attachments: int | list[_AttachmentDef] = 1, + body: str = "", + subject: str = "the subject", + from_: str = "no_one@mail.com", + to: list[str] | None = None, + seen: bool = False, + flagged: bool = False, + processed: bool = False, + ) -> MailMessage: + if to is None: + to = ["tosomeone@somewhere.com"] + + email_msg = email.message.EmailMessage() + # TODO: This does NOT set the UID + email_msg["Message-ID"] = str(uuid.uuid4()) + email_msg["Subject"] = subject + email_msg["From"] = from_ + email_msg["To"] = str(" ,".join(to)) + email_msg.set_content(body) + + # Either add some default number of attachments + # or the provided attachments + if isinstance(attachments, int): + for i in range(attachments): + attachment = _AttachmentDef(filename=f"file_{i}.pdf") + email_msg.add_attachment( + attachment.content, + maintype=attachment.maintype, + subtype=attachment.subtype, + disposition=attachment.disposition, + filename=attachment.filename, + ) + else: + for attachment in attachments: + email_msg.add_attachment( + attachment.content, + maintype=attachment.maintype, + subtype=attachment.subtype, + disposition=attachment.disposition, + filename=attachment.filename, + ) + + # Convert the EmailMessage to an imap_tools MailMessage + imap_msg = MailMessage.from_bytes(email_msg.as_bytes()) + + # TODO: Unsure how to add a uid to the actual EmailMessage. This hacks it in, + # based on how imap_tools uses regex to extract it. + # This should be a large enough pool + uid = self._next_uid + self._next_uid += 1 + + imap_msg._raw_uid_data = f"UID {uid}".encode() + + imap_msg.seen = seen + imap_msg.flagged = flagged + if processed: + imap_msg._raw_flag_data.append(b"+FLAGS (processed)") + if hasattr(imap_msg, "flags"): + del imap_msg.flags + + return imap_msg diff --git a/src/paperless_mail/tests/test_api.py b/src/paperless_mail/tests/test_api.py index bb75bc6c6d..fa23a16ccc 100644 --- a/src/paperless_mail/tests/test_api.py +++ b/src/paperless_mail/tests/test_api.py @@ -11,7 +11,7 @@ from paperless_mail.models import ProcessedMail from paperless_mail.tests.factories import MailAccountFactory from paperless_mail.tests.factories import MailRuleFactory from paperless_mail.tests.factories import ProcessedMailFactory -from paperless_mail.tests.test_mail import BogusMailBox +from paperless_mail.tests.helpers import BogusMailBox from paperless_testing.dirs import DirectoriesMixin from paperless_testing.factories import CorrespondentFactory from paperless_testing.factories import DocumentTypeFactory diff --git a/src/paperless_mail/tests/test_mail.py b/src/paperless_mail/tests/test_mail.py index 7634f1522f..c249a9d628 100644 --- a/src/paperless_mail/tests/test_mail.py +++ b/src/paperless_mail/tests/test_mail.py @@ -1,9 +1,7 @@ import dataclasses -import email.contentmanager import time import uuid from collections import namedtuple -from contextlib import AbstractContextManager from datetime import timedelta from unittest import mock @@ -18,9 +16,6 @@ from imap_tools import NOT from imap_tools import EmailAddress from imap_tools import FolderInfo from imap_tools import MailboxFolderSelectError -from imap_tools import MailboxLoginError -from imap_tools import MailMessage -from imap_tools import MailMessageFlags from imap_tools import errors from rest_framework import status from rest_framework.test import APITestCase @@ -39,6 +34,10 @@ from paperless_mail.models import MailRule from paperless_mail.models import ProcessedMail from paperless_mail.tests.factories import MailAccountFactory from paperless_mail.tests.factories import MailRuleFactory +from paperless_mail.tests.helpers import BogusMailBox +from paperless_mail.tests.helpers import MessageBuilder +from paperless_mail.tests.helpers import _AttachmentDef +from paperless_mail.tests.helpers import fake_magic_from_buffer from paperless_testing.assertions import FileSystemAssertsMixin from paperless_testing.dirs import DirectoriesMixin from paperless_testing.factories import CorrespondentFactory @@ -46,259 +45,6 @@ from paperless_testing.factories import UserFactory from paperless_testing.permissions import grant_global -@dataclasses.dataclass -class _AttachmentDef: - filename: str = "a_file.pdf" - maintype: str = "application/pdf" - subtype: str = "pdf" - disposition: str = "attachment" - content: bytes = b"a PDF document" - - -class BogusFolderManager: - current_folder = "INBOX" - uidvalidity = "1" - - def set(self, new_folder) -> None: - if new_folder not in ["INBOX", "spam"]: - raise MailboxFolderSelectError(None, "uhm") - self.current_folder = new_folder - - def status(self, folder, options): - return {"UIDVALIDITY": self.uidvalidity} - - -class BogusClient: - def __init__(self, messages) -> None: - self.messages: list[MailMessage] = messages - self.capabilities: list[str] = [] - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb) -> None: - pass - - def authenticate(self, mechanism, authobject) -> None: - # authobject must be a callable object - auth_bytes = authobject(None) - if auth_bytes != b"\x00admin\x00w57\xc3\xa4\xc3\xb6\xc3\xbcw4b6huwb6nhu": - raise MailboxLoginError("BAD", "OK") - - def uid(self, command, *args) -> None: - if command == "STORE": - for message in self.messages: - if message.uid == args[0]: - flag = args[2] - if flag == "processed": - message._raw_flag_data.append(b"+FLAGS (processed)") - if hasattr(message, "flags"): - del message.flags - - -class BogusMailBox(AbstractContextManager): - # Common values so tests don't need to remember an accepted login - USERNAME: str = "admin" - ASCII_PASSWORD: str = "secret" - # Note the non-ascii characters here - UTF_PASSWORD: str = "w57äöüw4b6huwb6nhu" - # A dummy access token - ACCESS_TOKEN = "ea7e075cd3acf2c54c48e600398d5d5a" - - def __init__(self) -> None: - self.messages: list[MailMessage] = [] - self.messages_spam: list[MailMessage] = [] - self.folder = BogusFolderManager() - self.client = BogusClient(self.messages) - self._host = "" - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb) -> None: - pass - - def updateClient(self) -> None: - self.client = BogusClient(self.messages) - - def login(self, username, password) -> None: - # This will raise a UnicodeEncodeError if the password is not ASCII only - password.encode("ascii") - # Otherwise, check for correct values - if username != self.USERNAME or password != self.ASCII_PASSWORD: - raise MailboxLoginError("BAD", "OK") - - def login_utf8(self, username, password) -> None: - # Expected to only be called with the UTF-8 password - if username != self.USERNAME or password != self.UTF_PASSWORD: - raise MailboxLoginError("BAD", "OK") - - def xoauth2(self, username: str, access_token: str) -> None: - if username != self.USERNAME or access_token != self.ACCESS_TOKEN: - raise MailboxLoginError("BAD", "OK") - - def fetch( - self, - criteria="ALL", - charset="", - *, - mark_seen=True, - bulk=True, - uid_list=None, - ): - if uid_list is not None: - return [m for m in self.messages if m.uid in uid_list] - return self._filter_messages(criteria) - - def uids(self, criteria, charset="") -> list[str]: - return [m.uid for m in self._filter_messages(criteria)] - - def _filter_messages(self, criteria): - msg = self.messages - - criteria = str(criteria).strip("()").split(" ") - - if "UNSEEN" in criteria: - msg = filter(lambda m: not m.seen, msg) - - if "SUBJECT" in criteria: - subject = criteria[criteria.index("SUBJECT") + 1].strip('"') - msg = filter(lambda m: subject in m.subject, msg) - - if "BODY" in criteria: - body = criteria[criteria.index("BODY") + 1].strip('"') - msg = filter(lambda m: body in m.text, msg) - - if "FROM" in criteria: - from_ = criteria[criteria.index("FROM") + 1].strip('"') - msg = filter(lambda m: from_ in m.from_, msg) - - if "TO" in criteria: - to_ = criteria[criteria.index("TO") + 1].strip('"') - msg = filter(lambda m: any(to_ in to_addr for to_addr in m.to), msg) - - if "UNFLAGGED" in criteria: - msg = filter(lambda m: not m.flagged, msg) - - if "UNKEYWORD" in criteria: - tag = criteria[criteria.index("UNKEYWORD") + 1].strip("'") - msg = filter(lambda m: tag not in m.flags, msg) - - if "(X-GM-LABELS" in criteria: # ['NOT', '(X-GM-LABELS', '"processed"'] - msg = filter(lambda m: "processed" not in m.flags, msg) - - if "UID" in criteria: - uid_list = criteria[criteria.index("UID") + 1].split(",") - msg = filter(lambda m: m.uid in uid_list, msg) - - return list(msg) - - def delete(self, uid_list) -> None: - self.messages = list(filter(lambda m: m.uid not in uid_list, self.messages)) - - def flag(self, uid_list, flag_set, value) -> None: - for message in self.messages: - if message.uid in uid_list: - for flag in flag_set: - if flag == MailMessageFlags.FLAGGED: - message.flagged = value - if flag == MailMessageFlags.SEEN: - message.seen = value - if flag == "processed": - message._raw_flag_data.append(b"+FLAGS (processed)") - if hasattr(message, "flags"): - del message.flags - - def move(self, uid_list, folder) -> None: - if folder == "spam": - self.messages_spam += list( - filter(lambda m: m.uid in uid_list, self.messages), - ) - self.messages = list(filter(lambda m: m.uid not in uid_list, self.messages)) - else: - raise Exception - - -def fake_magic_from_buffer(buffer, *, mime=False): - if mime: - if "PDF" in str(buffer): - return "application/pdf" - else: - return "unknown/type" - else: - return "Some verbose file description" - - -class MessageBuilder: - def __init__(self) -> None: - self._next_uid = 1 - - def create_message( - self, - *, - attachments: int | list[_AttachmentDef] = 1, - body: str = "", - subject: str = "the subject", - from_: str = "no_one@mail.com", - to: list[str] | None = None, - seen: bool = False, - flagged: bool = False, - processed: bool = False, - ) -> MailMessage: - if to is None: - to = ["tosomeone@somewhere.com"] - - email_msg = email.message.EmailMessage() - # TODO: This does NOT set the UID - email_msg["Message-ID"] = str(uuid.uuid4()) - email_msg["Subject"] = subject - email_msg["From"] = from_ - email_msg["To"] = str(" ,".join(to)) - email_msg.set_content(body) - - # Either add some default number of attachments - # or the provided attachments - if isinstance(attachments, int): - for i in range(attachments): - attachment = _AttachmentDef(filename=f"file_{i}.pdf") - email_msg.add_attachment( - attachment.content, - maintype=attachment.maintype, - subtype=attachment.subtype, - disposition=attachment.disposition, - filename=attachment.filename, - ) - else: - for attachment in attachments: - email_msg.add_attachment( - attachment.content, - maintype=attachment.maintype, - subtype=attachment.subtype, - disposition=attachment.disposition, - filename=attachment.filename, - ) - - # Convert the EmailMessage to an imap_tools MailMessage - imap_msg = MailMessage.from_bytes(email_msg.as_bytes()) - - # TODO: Unsure how to add a uid to the actual EmailMessage. This hacks it in, - # based on how imap_tools uses regex to extract it. - # This should be a large enough pool - uid = self._next_uid - self._next_uid += 1 - - imap_msg._raw_uid_data = f"UID {uid}".encode() - - imap_msg.seen = seen - imap_msg.flagged = flagged - if processed: - imap_msg._raw_flag_data.append(b"+FLAGS (processed)") - if hasattr(imap_msg, "flags"): - del imap_msg.flags - - return imap_msg - - def reset_bogus_mailbox( bogus_mailbox: BogusMailBox, message_builder: MessageBuilder, diff --git a/src/paperless_mail/tests/test_mail_nfc.py b/src/paperless_mail/tests/test_mail_nfc.py index 930c861bcc..68f5cfa55d 100644 --- a/src/paperless_mail/tests/test_mail_nfc.py +++ b/src/paperless_mail/tests/test_mail_nfc.py @@ -15,9 +15,9 @@ import pytest from paperless_mail.models import MailRule from paperless_mail.tests.factories import MailAccountFactory -from paperless_mail.tests.test_mail import MessageBuilder -from paperless_mail.tests.test_mail import _AttachmentDef -from paperless_mail.tests.test_mail import fake_magic_from_buffer +from paperless_mail.tests.helpers import MessageBuilder +from paperless_mail.tests.helpers import _AttachmentDef +from paperless_mail.tests.helpers import fake_magic_from_buffer @pytest.fixture() diff --git a/src/paperless_mail/tests/test_preprocessor.py b/src/paperless_mail/tests/test_preprocessor.py index 33c9b3839a..e439b52096 100644 --- a/src/paperless_mail/tests/test_preprocessor.py +++ b/src/paperless_mail/tests/test_preprocessor.py @@ -16,8 +16,8 @@ from paperless_mail.mail import MailAccountHandler from paperless_mail.models import MailRule from paperless_mail.preprocessor import MailMessageDecryptor from paperless_mail.tests.factories import MailAccountFactory +from paperless_mail.tests.helpers import _AttachmentDef from paperless_mail.tests.test_mail import TestMail -from paperless_mail.tests.test_mail import _AttachmentDef class MessageEncryptor: