import dataclasses import time import uuid from collections import namedtuple from datetime import timedelta from unittest import mock import pytest from django.contrib.auth.models import Permission from django.core.management import call_command from django.db import DatabaseError from django.test import TestCase from django.test import override_settings from django.utils import timezone from imap_tools import NOT from imap_tools import EmailAddress from imap_tools import FolderInfo from imap_tools import MailboxFolderSelectError from imap_tools import errors from rest_framework import status from rest_framework.test import APITestCase from documents.models import Correspondent from documents.models import MatchingModel from paperless_mail import tasks from paperless_mail.mail import MailAccountHandler from paperless_mail.mail import MailError from paperless_mail.mail import TagMailAction from paperless_mail.mail import apply_mail_action from paperless_mail.mail import error_callback from paperless_mail.mail import get_mailbox from paperless_mail.models import MailAccount 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 from paperless_testing.factories import UserFactory from paperless_testing.permissions import grant_global def reset_bogus_mailbox( bogus_mailbox: BogusMailBox, message_builder: MessageBuilder, ) -> None: bogus_mailbox.messages = [] bogus_mailbox.messages_spam = [] bogus_mailbox.messages.append( message_builder.create_message( subject="Invoice 1", from_="amazon@amazon.de", to=["me@myselfandi.com", "helpdesk@mydomain.com"], body="cables", seen=True, flagged=False, processed=False, ), ) bogus_mailbox.messages.append( message_builder.create_message( subject="Invoice 2", body="from my favorite electronic store", to=["invoices@mycompany.com"], seen=False, flagged=True, processed=True, ), ) bogus_mailbox.messages.append( message_builder.create_message( subject="Claim your $10M price now!", from_="amazon@amazon-some-indian-site.org", to=["special@me.me"], seen=False, ), ) bogus_mailbox.updateClient() class MailMocker(DirectoriesMixin, FileSystemAssertsMixin, TestCase): def setUp(self) -> None: self.bogus_mailbox = BogusMailBox() self.messageBuilder = MessageBuilder() reset_bogus_mailbox(self.bogus_mailbox, self.messageBuilder) patcher = mock.patch("paperless_mail.mail.PinnedMailBox") m = patcher.start() m.return_value = self.bogus_mailbox self.addCleanup(patcher.stop) patcher = mock.patch("paperless_mail.mail.queue_consumption_tasks") self._queue_consumption_tasks_mock = patcher.start() self.addCleanup(patcher.stop) super().setUp() def assert_queue_consumption_tasks_call_args( self, expected_call_args: list[list[dict[str, str]]], ) -> None: """ Verifies that queue_consumption_tasks has been called with the expected arguments. expected_call_args is the following format: * List of calls to queue_consumption_tasks, called once per mail, where each element is: * List of signatures for the consume_file task, where each element is: * dictionary containing arguments that need to be present in the consume_file signature. """ # assert number of calls to queue_consumption_tasks match self.assertEqual( len(self._queue_consumption_tasks_mock.call_args_list), len(expected_call_args), ) for (mock_args, mock_kwargs), expected_signatures in zip( self._queue_consumption_tasks_mock.call_args_list, expected_call_args, ): consume_tasks = mock_kwargs["consume_tasks"] # assert number of consume_file tasks match self.assertEqual(len(consume_tasks), len(expected_signatures)) for consume_task, expected_signature in zip( consume_tasks, expected_signatures, ): input_doc = consume_task.kwargs["input_doc"] overrides = consume_task.kwargs["overrides"] # assert the file exists self.assertIsFile(input_doc.original_file) # assert all expected arguments are present in the signature for key, value in expected_signature.items(): if key == "override_correspondent_id": self.assertEqual(overrides.correspondent_id, value) elif key == "override_filename": self.assertEqual(overrides.filename, value) elif key == "override_title": self.assertEqual(overrides.title, value) else: self.fail("No match for expected arg") def apply_mail_actions(self) -> None: """ Applies pending actions to mails by inspecting calls to the queue_consumption_tasks method. """ for args, kwargs in self._queue_consumption_tasks_mock.call_args_list: message = kwargs["message"] rule = kwargs["rule"] apply_mail_action([], rule.pk, message.uid, message.subject, message.date) def assert_eventually_equals( getter_fn, expected_value, timeout=1.0, interval=0.05, ) -> None: """ Repeatedly calls `getter_fn()` until the result equals `expected_value`, or times out after `timeout` seconds. """ deadline = time.time() + timeout while time.time() < deadline: if getter_fn() == expected_value: return time.sleep(interval) actual = getter_fn() raise AssertionError(f"Expected {expected_value}, but got {actual}") @mock.patch("paperless_mail.mail.magic.from_buffer", fake_magic_from_buffer) class TestMail( DirectoriesMixin, FileSystemAssertsMixin, TestCase, ): def setUp(self) -> None: self.mailMocker = MailMocker() self.mailMocker.setUp() self.addCleanup(self.mailMocker.doCleanups) self.mail_account_handler = MailAccountHandler() super().setUp() @mock.patch("paperless_mail.mail.MAIL_FETCH_BATCH_SIZE", 5) def test_handle_mail_account_batches_body_fetch_for_large_backlog(self) -> None: """ GIVEN: - More new/unprocessed mail than MAIL_FETCH_BATCH_SIZE WHEN: - The mail account is processed THEN: - The body fetch is issued once, with all UIDs and the configured batch size handed to imap_tools so it can bulk-fetch in batches server-side - Every message is still processed (none dropped at a batch boundary) """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) rule = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY, ) message_count = 12 # more than the patched batch size of 5 self.mailMocker.bogus_mailbox.messages = [ self.mailMocker.messageBuilder.create_message( subject=f"No attachment {i}", attachments=[], ) for i in range(message_count) ] self.mailMocker.bogus_mailbox.updateClient() with mock.patch.object( self.mailMocker.bogus_mailbox, "fetch", wraps=self.mailMocker.bogus_mailbox.fetch, ) as fetch_spy: self.mail_account_handler.handle_mail_account(account) # A single fetch() call hands the full UID list and batch size to imap_tools, # which does its own bulk-fetching in batches of MAIL_FETCH_BATCH_SIZE. fetch_spy.assert_called_once() self.assertEqual(fetch_spy.call_args.kwargs["bulk"], 5) self.assertEqual(len(fetch_spy.call_args.kwargs["uid_list"]), message_count) self.assertEqual( ProcessedMail.objects.filter(rule=rule).count(), message_count, ) def test_get_correspondent(self) -> None: message = namedtuple("MailMessage", []) message.from_ = "someone@somewhere.com" message.from_values = EmailAddress( "Someone!", "someone@somewhere.com", ) message2 = namedtuple("MailMessage", []) message2.from_ = "me@localhost.com" message2.from_values = EmailAddress( "", "fake@localhost.com", ) me_localhost = CorrespondentFactory(name=message2.from_) someone_else = CorrespondentFactory(name="someone else") handler = MailAccountHandler() rule = MailRule( name="a", assign_correspondent_from=MailRule.CorrespondentSource.FROM_NOTHING, ) self.assertIsNone(handler._get_correspondent(message, rule)) rule = MailRule( name="b", assign_correspondent_from=MailRule.CorrespondentSource.FROM_EMAIL, ) c = handler._get_correspondent(message, rule) self.assertIsNotNone(c) self.assertEqual(c.name, "someone@somewhere.com") self.assertEqual(c.matching_algorithm, MatchingModel.MATCH_LITERAL) self.assertEqual(c.match, "someone@somewhere.com") c = handler._get_correspondent(message2, rule) self.assertIsNotNone(c) self.assertEqual(c.name, "me@localhost.com") self.assertEqual(c.id, me_localhost.id) rule = MailRule( name="c", assign_correspondent_from=MailRule.CorrespondentSource.FROM_NAME, ) c = handler._get_correspondent(message, rule) self.assertIsNotNone(c) self.assertEqual(c.name, "Someone!") c = handler._get_correspondent(message2, rule) self.assertIsNotNone(c) self.assertEqual(c.id, me_localhost.id) rule = MailRule( name="d", assign_correspondent_from=MailRule.CorrespondentSource.FROM_CUSTOM, assign_correspondent=someone_else, ) c = handler._get_correspondent(message, rule) self.assertEqual(c, someone_else) def test_get_title(self) -> None: message = namedtuple("MailMessage", []) message.subject = "the message title" att = namedtuple("Attachment", []) att.filename = "this_is_the_file.pdf" handler = MailAccountHandler() rule = MailRule( name="a", assign_title_from=MailRule.TitleSource.FROM_FILENAME, ) self.assertEqual(handler._get_title(message, att, rule), "this_is_the_file") rule = MailRule( name="b", assign_title_from=MailRule.TitleSource.FROM_SUBJECT, ) self.assertEqual(handler._get_title(message, att, rule), "the message title") rule = MailRule( name="b", assign_title_from=MailRule.TitleSource.NONE, ) self.assertEqual(handler._get_title(message, att, rule), None) def test_handle_message(self) -> None: message = self.mailMocker.messageBuilder.create_message( subject="the message title", from_="Myself", attachments=2, ) account = MailAccount.objects.create() rule = MailRule( assign_title_from=MailRule.TitleSource.FROM_FILENAME, account=account, ) rule.save() result = self.mail_account_handler._handle_message(message, rule) self.assertEqual(result, 2) self.mailMocker._queue_consumption_tasks_mock.assert_called() self.mailMocker.assert_queue_consumption_tasks_call_args( [ [ {"override_title": "file_0", "override_filename": "file_0.pdf"}, {"override_title": "file_1", "override_filename": "file_1.pdf"}, ], ], ) def test_bogus_mailbox_uids_and_uid_criteria(self) -> None: mailbox = self.mailMocker.bogus_mailbox all_messages = list(mailbox.messages) # uids() returns the UIDs of unseen messages, no bodies needed to call it unseen_uids = mailbox.uids("(UNSEEN)") self.assertEqual( set(unseen_uids), {m.uid for m in all_messages if not m.seen}, ) # fetch() with an explicit UID criteria returns only the matching messages target_uid = all_messages[0].uid from imap_tools import AND fetched = mailbox.fetch(AND(uid=[target_uid]), mark_seen=False) self.assertEqual([m.uid for m in fetched], [target_uid]) def test_handle_empty_message(self) -> None: message = self.mailMocker.messageBuilder.create_message( subject="No attachments here", attachments=[], ) account = MailAccount.objects.create() rule = MailRule.objects.create( account=account, consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY, ) result = self.mail_account_handler._handle_message(message, rule) self.mailMocker._queue_consumption_tasks_mock.assert_not_called() self.assertEqual(result, 0) processed = ProcessedMail.objects.get( rule=rule, uid=message.uid, folder=rule.folder, ) self.assertEqual(processed.status, "PROCESSED_WO_CONSUMPTION") # Calling it again must not create a second row self.mail_account_handler._handle_message(message, rule) self.assertEqual( ProcessedMail.objects.filter( rule=rule, uid=message.uid, folder=rule.folder, ).count(), 1, ) def test_handle_empty_message_long_subject(self) -> None: """ GIVEN: - A mail with no attachments and a subject longer than the ProcessedMail.subject field WHEN: - The mail is handled by an attachments-only rule THEN: - The subject is truncated to the field length instead of raising """ message = self.mailMocker.messageBuilder.create_message( subject="A" * 300, attachments=[], ) account = MailAccount.objects.create() rule = MailRule.objects.create( account=account, consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY, ) self.mail_account_handler._handle_message(message, rule) processed = ProcessedMail.objects.get( rule=rule, uid=message.uid, folder=rule.folder, ) self.assertEqual(processed.status, "PROCESSED_WO_CONSUMPTION") self.assertEqual(processed.subject, "A" * 256) def test_handle_unknown_mime_type(self) -> None: message = self.mailMocker.messageBuilder.create_message( attachments=[ _AttachmentDef(filename="f1.pdf"), _AttachmentDef( filename="f2.json", content=b"{'much': 'payload.', 'so': 'json', 'wow': true}", ), ], ) account = MailAccount.objects.create() rule = MailRule( assign_title_from=MailRule.TitleSource.FROM_FILENAME, account=account, ) rule.save() result = self.mail_account_handler._handle_message(message, rule) self.assertEqual(result, 1) self.mailMocker.assert_queue_consumption_tasks_call_args( [ [ {"override_filename": "f1.pdf"}, ], ], ) def test_handle_disposition(self) -> None: message = self.mailMocker.messageBuilder.create_message( attachments=[ _AttachmentDef( filename="f1.pdf", disposition="inline", ), _AttachmentDef(filename="f2.pdf"), ], ) account = MailAccount.objects.create() rule = MailRule( assign_title_from=MailRule.TitleSource.FROM_FILENAME, account=account, ) rule.save() result = self.mail_account_handler._handle_message(message, rule) self.assertEqual(result, 1) self.mailMocker.assert_queue_consumption_tasks_call_args( [ [ {"override_filename": "f2.pdf"}, ], ], ) def test_handle_inline_files(self) -> None: message = self.mailMocker.messageBuilder.create_message( attachments=[ _AttachmentDef( filename="f1.pdf", disposition="inline", ), _AttachmentDef(filename="f2.pdf"), ], ) account = MailAccount.objects.create() rule = MailRule( assign_title_from=MailRule.TitleSource.FROM_FILENAME, account=account, attachment_type=MailRule.AttachmentProcessing.EVERYTHING, ) rule.save() result = self.mail_account_handler._handle_message(message, rule) self.assertEqual(result, 2) self.mailMocker.assert_queue_consumption_tasks_call_args( [ [ {"override_filename": "f1.pdf"}, {"override_filename": "f2.pdf"}, ], ], ) def test_filename_filter(self) -> None: """ GIVEN: - Email with multiple similar named attachments - Rule with inclusive and exclusive filters WHEN: - Mail action filtering is checked THEN: - Mail action should not be performed for files excluded - Mail action should be performed for files included """ message = self.mailMocker.messageBuilder.create_message( attachments=[ _AttachmentDef(filename="f1.pdf"), _AttachmentDef(filename="f2.pdf"), _AttachmentDef(filename="f3.pdf"), _AttachmentDef(filename="f2.png"), _AttachmentDef(filename="file.PDf"), _AttachmentDef(filename="f1.Pdf"), ], ) @dataclasses.dataclass(frozen=True) class FilterTestCase: name: str include_pattern: str | None exclude_pattern: str | None expected_matches: list[str] tests = [ FilterTestCase( "PDF Wildcard", include_pattern="*.pdf", exclude_pattern=None, expected_matches=["f1.pdf", "f2.pdf", "f3.pdf", "file.PDf", "f1.Pdf"], ), FilterTestCase( "F1 PDF Only", include_pattern="f1.pdf", exclude_pattern=None, expected_matches=["f1.pdf", "f1.Pdf"], ), FilterTestCase( "All Files", include_pattern="*", exclude_pattern=None, expected_matches=[ "f1.pdf", "f2.pdf", "f3.pdf", "f2.png", "file.PDf", "f1.Pdf", ], ), FilterTestCase( "PNG Only", include_pattern="*.png", exclude_pattern=None, expected_matches=["f2.png"], ), FilterTestCase( "PDF Files with f2 and f3", include_pattern="f2.pdf,f3*", exclude_pattern=None, expected_matches=["f2.pdf", "f3.pdf"], ), FilterTestCase( "PDF Files without f1", include_pattern="*.pdf", exclude_pattern="f1*", expected_matches=["f2.pdf", "f3.pdf", "file.PDf"], ), FilterTestCase( "PDF Files without f1 and f2", include_pattern="*.pdf", exclude_pattern="f1*,f2*", expected_matches=["f3.pdf", "file.PDf"], ), FilterTestCase( "PDF Files without f1 and f2 and f3", include_pattern="*.pdf", exclude_pattern="f1*,f2*,f3*", expected_matches=["file.PDf"], ), FilterTestCase( "All Files, no PNG", include_pattern="*", exclude_pattern="*.png", expected_matches=[ "f1.pdf", "f2.pdf", "f3.pdf", "file.PDf", "f1.Pdf", ], ), ] for test_case in tests: with self.subTest(msg=test_case.name): self.mailMocker._queue_consumption_tasks_mock.reset_mock() account = MailAccount(name=str(uuid.uuid4())) account.save() rule = MailRule( name=str(uuid.uuid4()), assign_title_from=MailRule.TitleSource.FROM_FILENAME, account=account, filter_attachment_filename_include=test_case.include_pattern, filter_attachment_filename_exclude=test_case.exclude_pattern, ) rule.save() self.mail_account_handler._handle_message(message, rule) self.mailMocker.assert_queue_consumption_tasks_call_args( [ [{"override_filename": m} for m in test_case.expected_matches], ], ) @pytest.mark.flaky(reruns=4) def test_filename_filter_inline_no_consumption(self) -> None: """ GIVEN: - Rule that processes all attachments but filters by filename WHEN: - Given email with inline attachment that does not meet filename filter THEN: - Mail action should not be performed """ message = self.mailMocker.messageBuilder.create_message( attachments=[ _AttachmentDef( filename="test.png", disposition="inline", ), ], ) self.mailMocker.bogus_mailbox.messages.append(message) account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) account.save() rule = MailRule( name=str(uuid.uuid4()), assign_title_from=MailRule.TitleSource.FROM_FILENAME, account=account, filter_attachment_filename_include="*.pdf", attachment_type=MailRule.AttachmentProcessing.EVERYTHING, action=MailRule.MailAction.DELETE, ) rule.save() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 4) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 1) def test_handle_mail_account_mark_read(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) @pytest.mark.flaky(reruns=4) def test_handle_mail_account_delete(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, filter_subject="Invoice", ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() assert_eventually_equals(lambda: len(self.mailMocker.bogus_mailbox.messages), 1) def test_handle_mail_account_delete_no_filters(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, maximum_age=0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 0) def test_handle_mail_account_overlapping_rules_only_first_consumes(self) -> None: """ GIVEN: - Multiple rules that match the same mail WHEN: - Mail account is processed THEN: - Only the first rule should be applied """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) first_rule = MailRule.objects.create( name="testrule-first", account=account, action=MailRule.MailAction.DELETE, filter_subject="Claim", order=1, ) _ = MailRule.objects.create( name="testrule-second", account=account, action=MailRule.MailAction.DELETE, filter_subject="Claim", order=2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1) queued_rule = self.mailMocker._queue_consumption_tasks_mock.call_args.kwargs[ "rule" ] self.assertEqual(queued_rule.id, first_rule.id) def test_handle_mail_account_skips_body_fetch_for_already_processed_mail( self, ) -> None: """ GIVEN: - An attachment-less mail under an attachments-only mark-read rule, already recorded as PROCESSED_WO_CONSUMPTION WHEN: - The mail account is processed again and the mail still matches the search criteria (it was never marked read, since no mail action is applied for the no-consumption case) THEN: - No IMAP body fetch happens for that mail; only the cheap UID search runs. """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) rule = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY, ) message = self.mailMocker.messageBuilder.create_message( subject="No attachment", attachments=[], ) self.mailMocker.bogus_mailbox.messages = [message] self.mailMocker.bogus_mailbox.updateClient() # First run: records ProcessedMail without consuming anything. self.mail_account_handler.handle_mail_account(account) self.assertTrue( ProcessedMail.objects.filter( rule=rule, uid=message.uid, folder=rule.folder, ).exists(), ) self.mailMocker._queue_consumption_tasks_mock.assert_not_called() # Second run: message still matches UNSEEN (mark-read action never ran), # but its body must not be downloaded again. with mock.patch.object( self.mailMocker.bogus_mailbox, "fetch", wraps=self.mailMocker.bogus_mailbox.fetch, ) as fetch_spy: self.mail_account_handler.handle_mail_account(account) fetch_spy.assert_not_called() def test_handle_mail_account_skip_duplicate_uids_from_fetch(self) -> None: """ GIVEN: - Multiple mails with the same UID returned from the mailbox fetch method WHEN: - Mail account is processed THEN: - Only one of the mails should be processed, to avoid duplicate processing due to fetch issues """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, filter_subject="Duplicated mail", ) duplicated_message = self.mailMocker.messageBuilder.create_message( subject="Duplicated mail", ) self.mailMocker.bogus_mailbox.messages = [ duplicated_message, duplicated_message, ] self.mailMocker.bogus_mailbox.updateClient() self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1) def test_handle_mail_account_skips_mail_already_processed_in_same_uidvalidity( self, ) -> None: """ GIVEN: - A ProcessedMail row recorded under the mailbox's current UIDVALIDITY WHEN: - A mail with the same UID is fetched from the same UIDVALIDITY epoch THEN: - The mail is skipped as a duplicate. """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) rule = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, ) message = self.mailMocker.messageBuilder.create_message() self.mailMocker.bogus_mailbox.messages = [message] self.mailMocker.bogus_mailbox.updateClient() self.mailMocker.bogus_mailbox.folder.uidvalidity = "SAME" ProcessedMail.objects.create( rule=rule, folder=rule.folder, uid=message.uid, uid_validity="SAME", subject="Previously processed mail", status="SUCCESS", received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)), ) self.mail_account_handler.handle_mail_account(account) self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0) def test_handle_mail_account_processes_mail_after_uidvalidity_change( self, ) -> None: """ GIVEN: - A ProcessedMail row recorded under a previous UIDVALIDITY epoch WHEN: - A mail with the same UID is fetched after UIDVALIDITY has changed THEN: - The mail is processed, not skipped as a duplicate. """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) rule = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, ) message = self.mailMocker.messageBuilder.create_message() self.mailMocker.bogus_mailbox.messages = [message] self.mailMocker.bogus_mailbox.updateClient() self.mailMocker.bogus_mailbox.folder.uidvalidity = "NEW" ProcessedMail.objects.create( rule=rule, folder=rule.folder, uid=message.uid, uid_validity="OLD", subject="Previously processed mail", status="SUCCESS", received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)), ) self.mail_account_handler.handle_mail_account(account) self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1) def test_handle_mail_account_skips_mail_processed_before_uidvalidity_tracking( self, ) -> None: """ GIVEN: - A ProcessedMail row recorded before UIDVALIDITY tracking existed (uid_validity is NULL) WHEN: - A mail with the same UID is fetched THEN: - The mail is skipped as a duplicate, to avoid re-ingesting all previously processed mail after upgrading. """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) rule = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, ) message = self.mailMocker.messageBuilder.create_message() self.mailMocker.bogus_mailbox.messages = [message] self.mailMocker.bogus_mailbox.updateClient() ProcessedMail.objects.create( rule=rule, folder=rule.folder, uid=message.uid, uid_validity=None, subject="Previously processed mail", status="SUCCESS", received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)), ) self.mail_account_handler.handle_mail_account(account) self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0) def test_handle_mail_account_processes_mail_when_uidvalidity_unavailable( self, ) -> None: """ GIVEN: - The mail server fails to report a UIDVALIDITY for the folder WHEN: - A mail account is processed THEN: - The failure is logged and the rule still processes the mail, instead of the whole rule being disabled. """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, ) message = self.mailMocker.messageBuilder.create_message() self.mailMocker.bogus_mailbox.messages = [message] self.mailMocker.bogus_mailbox.updateClient() self.mailMocker.bogus_mailbox.folder.status = mock.MagicMock( side_effect=errors.MailboxFolderStatusError(("NO", [b"unsupported"]), "OK"), ) with self.assertLogs("paperless_mail", level="WARNING") as cm: self.mail_account_handler.handle_mail_account(account) self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 1) self.assertEqual(len(cm.output), 1) self.assertIn( "Server does not support retrieving UIDVALIDITY", cm.output[0], ) def test_handle_mail_account_skips_mail_when_uidvalidity_unavailable_but_prior_record_exists( self, ) -> None: """ GIVEN: - A ProcessedMail row recorded with a real uid_validity value - The mail server fails to report UIDVALIDITY (MailboxFolderStatusError), so _get_uid_validity returns None WHEN: - A mail with the same UID is fetched THEN: - The mail is skipped as already-processed rather than re-ingested, falling back to (rule, uid, folder) matching. """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) rule = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.DELETE, ) message = self.mailMocker.messageBuilder.create_message() self.mailMocker.bogus_mailbox.messages = [message] self.mailMocker.bogus_mailbox.updateClient() ProcessedMail.objects.create( rule=rule, folder=rule.folder, uid=message.uid, uid_validity="REAL_VALIDITY", subject="Previously processed mail", status="SUCCESS", received=timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)), ) self.mailMocker.bogus_mailbox.folder.status = mock.MagicMock( side_effect=errors.MailboxFolderStatusError(("NO", [b"unsupported"]), "OK"), ) self.mail_account_handler.handle_mail_account(account) self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0) @pytest.mark.flaky(reruns=4) def test_handle_mail_account_flag(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.FLAG, filter_subject="Invoice", ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNFLAGGED", mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNFLAGGED", mark_seen=False)), 1, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) @pytest.mark.flaky(reruns=4) def test_handle_mail_account_move(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", filter_subject="Claim", ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages_spam), 0) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 2) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages_spam), 1) def test_handle_mail_account_move_no_filters(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", maximum_age=0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages_spam), 0) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 0) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages_spam), 3) def test_handle_mail_account_tag(self) -> None: account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.TAG, action_parameter="processed", ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual( len( self.mailMocker.bogus_mailbox.fetch( "UNKEYWORD processed", mark_seen=False, ), ), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual( len( self.mailMocker.bogus_mailbox.fetch( "UNKEYWORD processed", mark_seen=False, ), ), 0, ) def test_handle_mail_account_tag_gmail(self) -> None: self.mailMocker.bogus_mailbox._host = "imap.gmail.com" self.mailMocker.bogus_mailbox.client.capabilities = ["X-GM-EXT-1"] account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.TAG, action_parameter="processed", ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) criteria = NOT(gmail_label="processed") self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch(criteria, mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch(criteria, mark_seen=False)), 0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) def test_tag_mail_action_applemail_wrong_input(self) -> None: self.assertRaises( MailError, TagMailAction, "apple:black", supports_gmail_labels=False, ) def test_handle_mail_account_tag_applemail(self) -> None: # all mails will be FLAGGED afterwards account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.TAG, action_parameter="apple:green", ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNFLAGGED", mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNFLAGGED", mark_seen=False)), 0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) def test_error_login(self) -> None: """ GIVEN: - Account configured with incorrect password WHEN: - Account tried to login THEN: - MailError with correct message raised """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="wrong", ) with self.assertRaisesRegex( MailError, "Error while authenticating account", ): self.mail_account_handler.handle_mail_account(account) @pytest.mark.flaky(reruns=4) def test_error_skip_account(self) -> None: _ = MailAccount.objects.create( name="test", imap_server="", username="admin", password="wroasdng", ) account = MailAccount.objects.create( name="test2", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", filter_subject="Claim", ) tasks.process_mail_accounts() self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 2) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages_spam), 1) @pytest.mark.flaky(reruns=4) def test_error_skip_rule(self) -> None: account = MailAccount.objects.create( name="test2", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", filter_subject="Claim", order=1, folder="uuuhhhh", ) _ = MailRule.objects.create( name="testrule2", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", filter_subject="Claim", order=2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 2) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages_spam), 1) def test_error_folder_set(self) -> None: """ GIVEN: - Mail rule with non-existent folder THEN: - Should call list to output all folders in the account - Should not process any messages """ account = MailAccount.objects.create( name="test2", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", filter_subject="Claim", order=1, folder="uuuhhhh", # Invalid folder name ) self.mailMocker.bogus_mailbox.folder.list = mock.Mock( return_value=[FolderInfo("SomeFoldername", "|", ())], ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.bogus_mailbox.folder.list.assert_called_once() self.mailMocker._queue_consumption_tasks_mock.assert_not_called() def test_error_folder_set_error_listing(self) -> None: """ GIVEN: - Mail rule with non-existent folder - Mail account folder listing raises exception THEN: - Should not process any messages """ account = MailAccount.objects.create( name="test2", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", filter_subject="Claim", order=1, folder="uuuhhhh", # Invalid folder name ) self.mailMocker.bogus_mailbox.folder.list = mock.Mock( side_effect=MailboxFolderSelectError(None, "uhm"), ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.bogus_mailbox.folder.list.assert_called_once() self.mailMocker._queue_consumption_tasks_mock.assert_not_called() @pytest.mark.flaky(reruns=4) @mock.patch("paperless_mail.mail.MailAccountHandler._get_correspondent") def test_error_skip_mail(self, m) -> None: def get_correspondent_fake(message, rule) -> None: if message.from_ == "amazon@amazon.de": raise ValueError("Does not compute.") else: return m.side_effect = get_correspondent_fake account = MailAccount.objects.create( name="test2", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() # test that we still consume mail even if some mails throw errors. self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 2) # faulty mail still in inbox, untouched self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 1) self.assertEqual( self.mailMocker.bogus_mailbox.messages[0].from_, "amazon@amazon.de", ) def test_error_create_correspondent(self) -> None: account = MailAccount.objects.create( name="test2", imap_server="", username="admin", password="secret", ) _ = MailRule.objects.create( name="testrule", filter_from="amazon@amazon.de", account=account, action=MailRule.MailAction.MOVE, action_parameter="spam", assign_correspondent_from=MailRule.CorrespondentSource.FROM_EMAIL, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker._queue_consumption_tasks_mock.assert_called_once() c = Correspondent.objects.get(name="amazon@amazon.de") self.mailMocker.assert_queue_consumption_tasks_call_args( [ [ {"override_correspondent_id": c.id}, ], ], ) self.mailMocker._queue_consumption_tasks_mock.reset_mock() reset_bogus_mailbox( self.mailMocker.bogus_mailbox, self.mailMocker.messageBuilder, ) with mock.patch("paperless_mail.mail.Correspondent.objects.get_or_create") as m: m.side_effect = DatabaseError() self.mail_account_handler.handle_mail_account(account) self.mailMocker.assert_queue_consumption_tasks_call_args( [ [ {"override_correspondent_id": None}, ], ], ) @pytest.mark.flaky(reruns=4) def test_filters(self) -> None: account = MailAccount.objects.create( name="test3", imap_server="", username="admin", password="secret", ) for f_body, f_from, f_to, f_subject, expected_mail_count in [ (None, None, None, "Claim", 1), ("electronic", None, None, None, 1), (None, "amazon", None, None, 2), ("cables", "amazon", None, "Invoice", 1), (None, None, "test@email.com", None, 0), (None, None, "invoices@mycompany.com", None, 1), ("electronic", None, "invoices@mycompany.com", None, 1), (None, "amazon", "me@myselfandi.com", None, 1), ]: with self.subTest( f_body=f_body, f_from=f_from, f_to=f_to, f_subject=f_subject, ): MailRule.objects.all().delete() _ = MailRule.objects.create( name="testrule3", account=account, action=MailRule.MailAction.DELETE, filter_subject=f_subject, filter_body=f_body, filter_from=f_from, filter_to=f_to, ) reset_bogus_mailbox( self.mailMocker.bogus_mailbox, self.mailMocker.messageBuilder, ) self.mailMocker._queue_consumption_tasks_mock.reset_mock() self.mailMocker._queue_consumption_tasks_mock.assert_not_called() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual( len(self.mailMocker.bogus_mailbox.messages), 3 - expected_mail_count, ) self.assertEqual( self.mailMocker._queue_consumption_tasks_mock.call_count, expected_mail_count, ) def test_auth_plain_fallback(self) -> None: """ GIVEN: - Mail account with password containing non-ASCII characters WHEN: - Mail account is handled THEN: - Should still authenticate to the mail account """ account = MailAccount.objects.create( name="test", imap_server="", username=BogusMailBox.USERNAME, # Note the non-ascii characters here password=BogusMailBox.UTF_PASSWORD, ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.mailMocker._queue_consumption_tasks_mock.assert_not_called() self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 2) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) def test_auth_plain_fallback_fails_still(self) -> None: """ GIVEN: - Mail account with password containing non-ASCII characters - Incorrect password value WHEN: - Mail account is handled THEN: - Should raise a MailError for the account """ account = MailAccount.objects.create( name="test", imap_server="", username=BogusMailBox.USERNAME, # Note the non-ascii characters here # Passes the check in login, not in authenticate password="réception", ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, ) self.assertRaises( MailError, self.mail_account_handler.handle_mail_account, account, ) def test_auth_with_valid_token(self) -> None: """ GIVEN: - Mail account configured with access token WHEN: - Mail account is handled THEN: - Should still authenticate to the mail account """ account = MailAccount.objects.create( name="test", imap_server="", username=BogusMailBox.USERNAME, # Note the non-ascii characters here password=BogusMailBox.ACCESS_TOKEN, is_token=True, ) _ = MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 0) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(self.mailMocker._queue_consumption_tasks_mock.call_count, 2) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 0, ) self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) def test_disabled_rule(self) -> None: """ GIVEN: - Mail rule is disabled WHEN: - Mail account is handled THEN: - Should not process any messages """ account = MailAccount.objects.create( name="test", imap_server="", username="admin", password="secret", ) MailRule.objects.create( name="testrule", account=account, action=MailRule.MailAction.MARK_READ, enabled=False, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual(len(self.mailMocker.bogus_mailbox.messages), 3) self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 2, ) self.mail_account_handler.handle_mail_account(account) self.mailMocker.apply_mail_actions() self.assertEqual( len(self.mailMocker.bogus_mailbox.fetch("UNSEEN", mark_seen=False)), 2, ) # still 2 class TestPostConsumeAction(TestCase): def setUp(self) -> None: self.account = MailAccountFactory() self.rule = MailRuleFactory(account=self.account) self.message_uid = "12345" self.message_subject = "Test Subject" self.message_date = timezone.make_aware(timezone.datetime(2023, 1, 1, 12, 0, 0)) @mock.patch("paperless_mail.mail.get_mailbox") @mock.patch("paperless_mail.mail.mailbox_login") @mock.patch("paperless_mail.mail.get_rule_action") def test_post_consume_success( self, mock_get_rule_action, mock_mailbox_login, mock_get_mailbox, ) -> None: mock_mailbox = mock.MagicMock() mock_get_mailbox.return_value.__enter__.return_value = mock_mailbox mock_action = mock.MagicMock() mock_get_rule_action.return_value = mock_action apply_mail_action( result=[], rule_id=self.rule.pk, message_uid=self.message_uid, message_subject=self.message_subject, message_date=self.message_date, ) mock_mailbox_login.assert_called_once_with(mock_mailbox, self.account) mock_mailbox.folder.set.assert_called_once_with(self.rule.folder) mock_action.post_consume.assert_called_once_with( mock_mailbox, self.message_uid, self.rule.action_parameter, ) processed_mail = ProcessedMail.objects.get(uid=self.message_uid) self.assertEqual(processed_mail.status, "SUCCESS") @mock.patch("paperless_mail.mail.get_mailbox") @mock.patch("paperless_mail.mail.mailbox_login") @mock.patch("paperless_mail.mail.get_rule_action") def test_post_consume_failure( self, mock_get_rule_action, mock_mailbox_login, mock_get_mailbox, ) -> None: mock_mailbox = mock.MagicMock() mock_get_mailbox.return_value.__enter__.return_value = mock_mailbox mock_action = mock.MagicMock() mock_get_rule_action.return_value = mock_action mock_action.post_consume.side_effect = errors.ImapToolsError("Test Exception") with ( self.assertRaises(errors.ImapToolsError), self.assertLogs("paperless_mail", level="ERROR") as cm, ): apply_mail_action( result=[], rule_id=self.rule.pk, message_uid=self.message_uid, message_subject=self.message_subject, message_date=self.message_date, ) error_str = cm.output[0] expected_str = "Error while processing mail action during post_consume" self.assertIn(expected_str, error_str) processed_mail = ProcessedMail.objects.get(uid=self.message_uid) self.assertEqual(processed_mail.status, "FAILED") self.assertIn("Test Exception", processed_mail.error) @pytest.mark.django_db class TestErrorCallback: def test_error_callback_is_idempotent_for_same_mail(self) -> None: """ GIVEN: - A mail rule and a mail that failed to be consumed WHEN: - error_callback is invoked more than once for the same mail, as happens when task_allow_error_cb_on_chord_header fires the errback once per failed header task in a chord THEN: - Only one ProcessedMail row is created for that mail """ rule = MailRuleFactory() message_uid = "12345" for _ in range(2): error_callback( None, Exception("Test Exception"), None, rule_id=rule.pk, message_uid=message_uid, message_subject="Test Subject", message_date=timezone.make_aware( timezone.datetime(2023, 1, 1, 12, 0, 0), ), ) processed_mails = ProcessedMail.objects.filter( rule=rule, uid=message_uid, folder=rule.folder, ) assert processed_mails.count() == 1 assert processed_mails.get().status == "FAILED" class TestManagementCommand(TestCase): @mock.patch( "paperless_mail.management.commands.mail_fetcher.tasks.process_mail_accounts", ) def test_mail_fetcher(self, m) -> None: call_command("mail_fetcher", skip_checks=True) m.assert_called_once() class TestTasks(TestCase): @mock.patch("paperless_mail.tasks.MailAccountHandler.handle_mail_account") def test_all_accounts(self, m) -> None: m.side_effect = lambda account: 6 MailAccount.objects.create( name="A", imap_server="A", username="A", password="A", ) MailAccount.objects.create( name="B", imap_server="A", username="A", password="A", ) MailRule.objects.create( name="A", account=MailAccount.objects.get(name="A"), ) MailRule.objects.create( name="B", account=MailAccount.objects.get(name="B"), ) result = tasks.process_mail_accounts() self.assertEqual(m.call_count, 2) self.assertIn("Added 12", result) m.side_effect = lambda account: 0 result = tasks.process_mail_accounts() self.assertIn("No new", result) @mock.patch("paperless_mail.tasks.MailAccountHandler.handle_mail_account") def test_accounts_no_enabled_rules(self, m) -> None: m.side_effect = lambda account: 6 MailAccount.objects.create( name="A", imap_server="A", username="A", password="A", ) MailAccount.objects.create( name="B", imap_server="A", username="A", password="A", ) MailRule.objects.create( name="A", account=MailAccount.objects.get(name="A"), enabled=False, ) MailRule.objects.create( name="B", account=MailAccount.objects.get(name="B"), enabled=False, ) tasks.process_mail_accounts() self.assertEqual(m.call_count, 0) @mock.patch("paperless_mail.tasks.MailAccountHandler.handle_mail_account") def test_process_with_account_ids(self, m) -> None: m.side_effect = lambda account: 6 account_a = MailAccount.objects.create( name="A", imap_server="A", username="A", password="A", ) account_b = MailAccount.objects.create( name="B", imap_server="A", username="A", password="A", ) MailRule.objects.create( name="A", account=account_a, ) MailRule.objects.create( name="B", account=account_b, ) result = tasks.process_mail_accounts(account_ids=[account_a.id]) self.assertEqual(m.call_count, 1) self.assertIn("Added 6", result) m.side_effect = lambda account: 0 result = tasks.process_mail_accounts(account_ids=[account_b.id]) self.assertIn("No new", result) @mock.patch("paperless_mail.tasks.MailAccountHandler.handle_mail_account") def test_rule_with_stop_processing(self, m) -> None: """ GIVEN: - Mail account with a rule with stop_processing=True WHEN: - Mail account is processed THEN: - Should only process the first rule """ m.side_effect = lambda account: 6 account = MailAccount.objects.create( name="A", imap_server="A", username="A", password="A", ) MailRule.objects.create( name="A", account=account, stop_processing=True, ) MailRule.objects.create( name="B", account=account, ) result = tasks.process_mail_accounts() self.assertEqual(m.call_count, 1) self.assertIn("Added 6", result) class TestMailAccountTestView(APITestCase): def setUp(self) -> None: self.mailMocker = MailMocker() self.mailMocker.setUp() self.addCleanup(self.mailMocker.doCleanups) self.user = UserFactory(username="testuser", password="testpassword") grant_global(self.user, "add_mailaccount", "change_mailaccount") self.user.save() self.client.force_authenticate(user=self.user) self.url = "/api/mail_accounts/test/" def test_mail_account_test_view_success(self) -> None: data = { "imap_server": "imap.example.com", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "secret", "account_type": MailAccount.MailAccountType.IMAP, "is_token": False, } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data, {"success": True}) def test_mail_account_test_view_mail_error(self) -> None: data = { "imap_server": "imap.example.com", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "wrong", "account_type": MailAccount.MailAccountType.IMAP, "is_token": False, } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(response.content.decode(), "Unable to connect to server") @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch("paperless_mail.mail.resolve_hostname_ips", return_value=["127.0.0.1"]) def test_mail_account_test_view_blocks_internal_host_when_disabled( self, _mock_resolve_hostname_ips, ) -> None: data = { "imap_server": "internal.example", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "secret", "account_type": MailAccount.MailAccountType.IMAP, "is_token": False, } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(response.content.decode(), "Unable to connect to server") @mock.patch( "paperless_mail.oauth.PaperlessMailOAuth2Manager.refresh_account_oauth_token", ) def test_mail_account_test_view_refresh_token( self, mock_refresh_account_oauth_token, ) -> None: """ GIVEN: - Mail account with expired token WHEN: - Mail account is tested THEN: - Should refresh the token """ existing_account = MailAccount.objects.create( imap_server="imap.example.com", imap_port=993, imap_security=MailAccount.ImapSecurity.SSL, username="admin", password="secret", account_type=MailAccount.MailAccountType.GMAIL_OAUTH, refresh_token="oldtoken", expiration=timezone.now() - timedelta(days=1), is_token=True, ) mock_refresh_account_oauth_token.return_value = True data = { "id": existing_account.id, "imap_server": "imap.example.com", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "****", "is_token": True, } self.client.post(self.url, data, format="json") self.assertEqual(mock_refresh_account_oauth_token.call_count, 1) @mock.patch( "paperless_mail.oauth.PaperlessMailOAuth2Manager.refresh_account_oauth_token", ) def test_mail_account_test_view_refresh_token_fails( self, mock_mock_refresh_account_oauth_token: mock.MagicMock, ) -> None: """ GIVEN: - Mail account with expired token WHEN: - Mail account is tested - Token refresh fails THEN: - Should log an error """ existing_account = MailAccount.objects.create( imap_server="imap.example.com", imap_port=993, imap_security=MailAccount.ImapSecurity.SSL, username="admin", password="secret", account_type=MailAccount.MailAccountType.GMAIL_OAUTH, refresh_token="oldtoken", expiration=timezone.now() - timedelta(days=1), is_token=True, ) mock_mock_refresh_account_oauth_token.return_value = False data = { "id": existing_account.id, "imap_server": "imap.example.com", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "****", "is_token": True, } with self.assertLogs("paperless_mail", level="ERROR") as cm: response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) error_str = cm.output[0] expected_str = "Unable to refresh oauth token" self.assertIn(expected_str, error_str) def test_mail_account_test_view_existing_forbidden_for_other_owner(self) -> None: other_user = UserFactory(username="otheruser", password="testpassword") existing_account = MailAccount.objects.create( name="Owned account", imap_server="imap.example.com", imap_port=993, imap_security=MailAccount.ImapSecurity.SSL, username="admin", password="secret", owner=other_user, ) data = { "id": existing_account.id, "imap_server": "imap.example.com", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "****", "is_token": False, } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(response.content.decode(), "Insufficient permissions") def test_mail_account_test_view_requires_add_permission_without_account_id( self, ) -> None: self.user.user_permissions.remove( *Permission.objects.filter(codename__in=["add_mailaccount"]), ) self.user.save() data = { "imap_server": "imap.example.com", "imap_port": 993, "imap_security": MailAccount.ImapSecurity.SSL, "username": "admin", "password": "secret", "is_token": False, } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(response.content.decode(), "Insufficient permissions") class TestGetMailboxHostPinning(TestCase): """ get_mailbox() must connect to the address it validated, so that a DNS answer which changes between the check and the connection (DNS rebinding) cannot redirect the connection to an internal host. """ @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch( "paperless_mail.mail.resolve_hostname_ips", return_value=["93.184.216.34"], ) def test_connects_to_validated_ip(self, _mock_resolve) -> None: with mock.patch( "paperless_mail.mail.socket.create_connection", side_effect=OSError("no connection in tests"), ) as mock_connection: with self.assertRaises(OSError): get_mailbox( "mail.example.com", 143, MailAccount.ImapSecurity.NONE, ) # the hostname is never handed to the socket layer for a second lookup mock_connection.assert_called_once_with(("93.184.216.34", 143)) @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch( "paperless_mail.mail.resolve_hostname_ips", return_value=["93.184.216.34"], ) def test_ssl_pins_ip_but_keeps_hostname_for_sni(self, _mock_resolve) -> None: ssl_context = mock.MagicMock() ssl_context.wrap_socket.return_value.makefile.side_effect = OSError( "no connection in tests", ) with ( mock.patch( "paperless_mail.mail.ssl.create_default_context", return_value=ssl_context, ), mock.patch( "paperless_mail.mail.socket.create_connection", ) as mock_connection, ): with self.assertRaises(OSError): get_mailbox( "mail.example.com", 993, MailAccount.ImapSecurity.SSL, ) mock_connection.assert_called_once_with(("93.184.216.34", 993)) # certificate verification still happens against the hostname ssl_context.wrap_socket.assert_called_once_with( mock_connection.return_value, server_hostname="mail.example.com", ) @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch( "paperless_mail.mail.resolve_hostname_ips", return_value=["93.184.216.34", "127.0.0.1"], ) def test_blocks_when_any_resolved_address_is_internal(self, _mock_resolve) -> None: with self.assertRaises(MailError): get_mailbox("mail.example.com", 993, MailAccount.ImapSecurity.SSL) class TestMailAccountProcess(APITestCase): def setUp(self) -> None: self.mailMocker = MailMocker() self.mailMocker.setUp() self.addCleanup(self.mailMocker.doCleanups) self.user = UserFactory( username="testuser", password="testpassword", superuser=True, ) self.client.force_authenticate(user=self.user) self.account = MailAccount.objects.create( imap_server="imap.example.com", imap_port=993, imap_security=MailAccount.ImapSecurity.SSL, username="admin", password="secret", account_type=MailAccount.MailAccountType.IMAP, owner=self.user, ) self.url = f"/api/mail_accounts/{self.account.pk}/process/" @mock.patch("paperless_mail.tasks.process_mail_accounts.apply_async") def test_mail_account_process_view(self, m) -> None: response = self.client.post(self.url) self.assertEqual(response.status_code, status.HTTP_200_OK) m.assert_called_once() class TestMailRuleAPI(APITestCase): def setUp(self) -> None: self.user = UserFactory( username="testuser", password="testpassword", superuser=True, ) self.client.force_authenticate(user=self.user) self.account = MailAccountFactory(owner=self.user) self.url = "/api/mail_rules/" def test_create_mail_rule(self) -> None: """ GIVEN: - Valid data for creating a mail rule WHEN: - A POST request is made to the mail rules endpoint THEN: - The rule should be created successfully - The response should contain the created rule's details """ data = { "name": "Test Rule", "account": self.account.pk, "action": MailRule.MailAction.MOVE, "action_parameter": "inbox", } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(MailRule.objects.count(), 1) rule = MailRule.objects.first() assert rule is not None self.assertEqual(rule.name, "Test Rule") def test_mail_rule_action_parameter_required_for_tag_or_move(self) -> None: """ GIVEN: - Valid data for creating a mail rule without action_parameter WHEN: - A POST request is made to the mail rules endpoint THEN: - The request should fail with a 400 Bad Request status - The response should indicate that action_parameter is required """ data = { "name": "Test Rule", "account": self.account.pk, "action": MailRule.MailAction.MOVE, } response = self.client.post(self.url, data, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn( "action parameter is required", str(response.data["non_field_errors"]), )