From a204f7986c2851f3a4009242f00d07958a1694b3 Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Thu, 30 Apr 2026 14:53:54 -0700 Subject: [PATCH] Simplifications and improvements --- src/paperless_mail/tests/conftest.py | 33 ++-- src/paperless_mail/tests/test_mail_oauth.py | 98 +++++------ src/paperless_mail/tests/test_preprocessor.py | 159 +++++++++--------- 3 files changed, 136 insertions(+), 154 deletions(-) diff --git a/src/paperless_mail/tests/conftest.py b/src/paperless_mail/tests/conftest.py index ae4d78c9c..173fe5a81 100644 --- a/src/paperless_mail/tests/conftest.py +++ b/src/paperless_mail/tests/conftest.py @@ -1,14 +1,18 @@ from collections.abc import Generator import pytest +from django.contrib.auth.models import Permission +from django.contrib.auth.models import User from django.test import Client +from pytest_django.fixtures import SettingsWrapper from paperless_mail.mail import MailAccountHandler from paperless_mail.models import MailAccount from paperless_mail.tests.factories import MailAccountFactory +from paperless_mail.tests.test_mail import MailMocker -@pytest.fixture() +@pytest.fixture def greenmail_mail_account(db: None) -> Generator[MailAccount, None, None]: """ Create a mail account configured for local Greenmail server. @@ -25,34 +29,25 @@ def greenmail_mail_account(db: None) -> Generator[MailAccount, None, None]: account.delete() -@pytest.fixture() +@pytest.fixture def mail_account_handler() -> MailAccountHandler: return MailAccountHandler() -@pytest.fixture() -def mail_user( - db: None, - django_user_model, - client: Client, -): +@pytest.fixture +def mail_user(db: None, django_user_model, client: Client) -> User: """ Create a user with the `add_mailaccount` permission and log them in via the test client. Returned so tests can mutate permissions if needed. """ - from django.contrib.auth.models import Permission - user = django_user_model.objects.create_user("testuser") - user.user_permissions.add( - *Permission.objects.filter(codename__in=["add_mailaccount"]), - ) - user.save() + user.user_permissions.add(*Permission.objects.filter(codename="add_mailaccount")) client.force_login(user) return user -@pytest.fixture() -def oauth_settings(settings): +@pytest.fixture +def oauth_settings(settings: SettingsWrapper) -> SettingsWrapper: """ Apply the OAuth callback / client-id settings the OAuth flow needs. Uses pytest-django's `settings` fixture so values are reverted automatically. @@ -65,15 +60,13 @@ def oauth_settings(settings): return settings -@pytest.fixture() -def mail_mocker(db: None): +@pytest.fixture +def mail_mocker(db: None) -> Generator[MailMocker, None, None]: """ Provides a MailMocker instance with its `MailBox` and `queue_consumption_tasks` patches active. Cleanups registered via TestCase.addCleanup are run on teardown by calling doCleanups(). """ - from paperless_mail.tests.test_mail import MailMocker - mocker = MailMocker() mocker.setUp() try: diff --git a/src/paperless_mail/tests/test_mail_oauth.py b/src/paperless_mail/tests/test_mail_oauth.py index 4d82b42a8..ecaaf1efa 100644 --- a/src/paperless_mail/tests/test_mail_oauth.py +++ b/src/paperless_mail/tests/test_mail_oauth.py @@ -17,12 +17,12 @@ from paperless_mail.oauth import PaperlessMailOAuth2Manager from paperless_mail.tests.factories import MailAccountFactory -@pytest.fixture() +@pytest.fixture def oauth_manager() -> PaperlessMailOAuth2Manager: return PaperlessMailOAuth2Manager() -@pytest.fixture() +@pytest.fixture def oauth_session(client: Client) -> Client: """Seed the test client session with a known oauth_state.""" session = client.session @@ -130,10 +130,7 @@ class TestOAuthCallbackView: """ response = client.get("/api/oauth/callback/") assert response.status_code == status.HTTP_400_BAD_REQUEST - assert not MailAccount.objects.filter(imap_server="imap.gmail.com").exists() - assert not MailAccount.objects.filter( - imap_server="outlook.office365.com", - ).exists() + assert not MailAccount.objects.exists() def test_invalid_state( self, @@ -153,10 +150,7 @@ class TestOAuthCallbackView: "/api/oauth/callback/?code=test_code&state=invalid_state", ) assert response.status_code == status.HTTP_400_BAD_REQUEST - assert not MailAccount.objects.filter(imap_server="imap.gmail.com").exists() - assert not MailAccount.objects.filter( - imap_server="outlook.office365.com", - ).exists() + assert not MailAccount.objects.exists() def test_insufficient_permissions( self, @@ -173,20 +167,15 @@ class TestOAuthCallbackView: THEN: - 400 Bad Request is returned and no mail account is created """ - mail_user.user_permissions.remove( - *Permission.objects.filter(codename__in=["add_mailaccount"]), + *Permission.objects.filter(codename="add_mailaccount"), ) - mail_user.save() response = client.get( "/api/oauth/callback/?code=test_code&scope=https://mail.google.com/", ) assert response.status_code == status.HTTP_400_BAD_REQUEST - assert not MailAccount.objects.filter(imap_server="imap.gmail.com").exists() - assert not MailAccount.objects.filter( - imap_server="outlook.office365.com", - ).exists() + assert not MailAccount.objects.exists() @pytest.mark.parametrize( ("provider", "callback_query", "expected_imap"), @@ -232,12 +221,10 @@ class TestOAuthCallbackView: "refresh_token": "test_refresh_token", "expires_in": 3600, } - target = ( - "paperless_mail.oauth.PaperlessMailOAuth2Manager.get_gmail_access_token" - if provider == "gmail" - else "paperless_mail.oauth.PaperlessMailOAuth2Manager.get_outlook_access_token" + mocked = mocker.patch( + f"paperless_mail.oauth.PaperlessMailOAuth2Manager.get_{provider}_access_token", + return_value=token_payload, ) - mocked = mocker.patch(target, return_value=token_payload) response = client.get(f"/api/oauth/callback/?{callback_query}") @@ -300,31 +287,45 @@ class TestOAuthCallbackView: ) +@pytest.fixture +def expired_oauth_account_factory(db: None, mocker: pytest_mock.MockerFixture): + """ + Build an OAuth-backed MailAccount whose access token has already expired, + while patching `get_mailbox` so no real IMAP connection is attempted. + """ + mocker.patch( + "paperless_mail.mail.get_mailbox", + ).return_value.__enter__.return_value = mocker.MagicMock() + + def _make(account_type: MailAccount.MailAccountType) -> MailAccount: + return MailAccountFactory( + username="test_username", + account_type=account_type, + is_token=True, + refresh_token="test_refresh_token", + expiration=timezone.now() - timedelta(days=1), + ) + + return _make + + @pytest.mark.django_db class TestRefreshTokenOnHandleMailAccount: """OAuth refresh-token flow exercised through MailAccountHandler.handle_mail_account.""" @pytest.mark.parametrize( - ("account_type", "name"), + "account_type", [ - pytest.param( - MailAccount.MailAccountType.GMAIL_OAUTH, - "Test Gmail", - id="gmail", - ), - pytest.param( - MailAccount.MailAccountType.OUTLOOK_OAUTH, - "Test Outlook", - id="outlook", - ), + pytest.param(MailAccount.MailAccountType.GMAIL_OAUTH, id="gmail"), + pytest.param(MailAccount.MailAccountType.OUTLOOK_OAUTH, id="outlook"), ], ) def test_refresh_token_called( self, mocker: pytest_mock.MockerFixture, mail_account_handler: MailAccountHandler, + expired_oauth_account_factory, account_type: MailAccount.MailAccountType, - name: str, ) -> None: """ GIVEN: @@ -334,10 +335,6 @@ class TestRefreshTokenOnHandleMailAccount: THEN: - The OAuth refresh_token endpoint is invoked exactly once """ - mock_mailbox = mocker.MagicMock() - mocker.patch( - "paperless_mail.mail.get_mailbox", - ).return_value.__enter__.return_value = mock_mailbox mock_refresh = mocker.patch( "httpx_oauth.oauth2.BaseOAuth2.refresh_token", return_value={ @@ -347,16 +344,9 @@ class TestRefreshTokenOnHandleMailAccount: }, ) - account = MailAccountFactory( - name=name, - username="test_username", - account_type=account_type, - is_token=True, - refresh_token="test_refresh_token", - expiration=timezone.now() - timedelta(days=1), + mail_account_handler.handle_mail_account( + expired_oauth_account_factory(account_type), ) - - mail_account_handler.handle_mail_account(account) mock_refresh.assert_called_once() def test_refresh_token_failure( @@ -364,6 +354,7 @@ class TestRefreshTokenOnHandleMailAccount: mocker: pytest_mock.MockerFixture, caplog: pytest.LogCaptureFixture, mail_account_handler: MailAccountHandler, + expired_oauth_account_factory, ) -> None: """ GIVEN: @@ -375,22 +366,13 @@ class TestRefreshTokenOnHandleMailAccount: - 0 processed mails is returned - The failure is logged at ERROR level with the account context """ - mock_mailbox = mocker.MagicMock() - mocker.patch( - "paperless_mail.mail.get_mailbox", - ).return_value.__enter__.return_value = mock_mailbox mock_refresh = mocker.patch( "httpx_oauth.oauth2.BaseOAuth2.refresh_token", side_effect=RefreshTokenError("test_error"), ) - account = MailAccountFactory( - name="Test Gmail Mail Account", - username="test_username", - account_type=MailAccount.MailAccountType.GMAIL_OAUTH, - is_token=True, - refresh_token="test_refresh_token", - expiration=timezone.now() - timedelta(days=1), + account = expired_oauth_account_factory( + MailAccount.MailAccountType.GMAIL_OAUTH, ) with caplog.at_level("ERROR", logger="paperless_mail"): diff --git a/src/paperless_mail/tests/test_preprocessor.py b/src/paperless_mail/tests/test_preprocessor.py index d5f946386..fc9bdc7d2 100644 --- a/src/paperless_mail/tests/test_preprocessor.py +++ b/src/paperless_mail/tests/test_preprocessor.py @@ -10,55 +10,59 @@ from pathlib import Path import gnupg import pytest from imap_tools import MailMessage +from pytest_django.fixtures import SettingsWrapper +from pytest_mock import MockerFixture 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.test_mail import MailMocker from paperless_mail.tests.test_mail import _AttachmentDef +def _kill_gpg_agent(gpg_home: str) -> None: + """ + Terminate any gpg-agent attached to `gpg_home` so the directory is + removable. Uses `gpgconf --kill`, the GnuPG project's recommended cleanup + path; python-gnupg has no built-in cleanup since it only wraps the CLI. + """ + try: + subprocess.run( + ["gpgconf", "--kill", "gpg-agent"], + env={"GNUPGHOME": gpg_home}, + check=False, + capture_output=True, + timeout=5, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + pass + + class MessageEncryptor: """ Test helper: generates a throwaway GPG keypair in a tempdir and exposes `encrypt(MailMessage) -> MailMessage`. """ + TEST_USER = "testuser@example.com" + def __init__(self, gpg_home: Path) -> None: self.gpg_home = str(gpg_home) self.gpg = gnupg.GPG(gnupghome=self.gpg_home) - self._testUser = "testuser@example.com" - # Generate a new key - input_data = self.gpg.gen_key_input( - name_email=self._testUser, - passphrase=None, - key_type="RSA", - key_length=2048, - expire_date=0, - no_protection=True, + self.gpg.gen_key( + self.gpg.gen_key_input( + name_email=self.TEST_USER, + passphrase=None, + key_type="RSA", + key_length=2048, + expire_date=0, + no_protection=True, + ), ) - self.gpg.gen_key(input_data) def kill_agent(self) -> None: - """ - Kill the gpg-agent so pytest can remove the GPG home. - - This uses gpgconf to properly terminate the agent, which is the officially - recommended cleanup method from the GnuPG project. python-gnupg does not - provide built-in cleanup methods as it's only a wrapper around the gpg CLI. - """ - # Kill the gpg-agent using the official GnuPG cleanup tool - try: - subprocess.run( - ["gpgconf", "--kill", "gpg-agent"], - env={"GNUPGHOME": self.gpg_home}, - check=False, - capture_output=True, - timeout=5, - ) - except (FileNotFoundError, subprocess.TimeoutExpired): - # gpgconf not found or hung - agent will timeout eventually - pass + _kill_gpg_agent(self.gpg_home) @staticmethod def get_email_body_without_headers(email_message: Message) -> bytes: @@ -66,7 +70,6 @@ class MessageEncryptor: Filters some relevant headers from an EmailMessage and returns just the body. """ message_copy = email.message_from_bytes(email_message.as_bytes()) - message_copy._headers = [ header for header in message_copy._headers @@ -74,30 +77,27 @@ class MessageEncryptor: ] return message_copy.as_bytes() - def encrypt(self, message) -> MailMessage: - original_email: email.message.Message = message.obj + def encrypt(self, message: MailMessage) -> MailMessage: + original_email: Message = message.obj encrypted_data = self.gpg.encrypt( self.get_email_body_without_headers(original_email), - self._testUser, + self.TEST_USER, armor=True, ) if not encrypted_data.ok: raise Exception(f"Encryption failed: {encrypted_data.stderr}") - encrypted_email_content = encrypted_data.data new_email = MIMEMultipart("encrypted", protocol="application/pgp-encrypted") new_email["From"] = original_email["From"] new_email["To"] = original_email["To"] new_email["Subject"] = original_email["Subject"] - # Add the control part control_part = MIMEApplication(_data=b"", _subtype="pgp-encrypted") control_part.set_payload("Version: 1") new_email.attach(control_part) - # Add the encrypted data part encrypted_part = MIMEApplication(_data=b"", _subtype="octet-stream") - encrypted_part.set_payload(encrypted_email_content.decode("ascii")) + encrypted_part.set_payload(encrypted_data.data.decode("ascii")) encrypted_part.add_header( "Content-Disposition", 'attachment; filename="encrypted.asc"', @@ -119,21 +119,26 @@ def message_encryptor( comes from `tmp_path_factory` so pytest cleans it up at session end; we still kill the gpg-agent ourselves so the dir is removable. """ - gpg_home = tmp_path_factory.mktemp("gpg-home") - encryptor = MessageEncryptor(gpg_home) + encryptor = MessageEncryptor(tmp_path_factory.mktemp("gpg-home")) yield encryptor encryptor.kill_agent() -@pytest.fixture() -def gpg_settings(settings, message_encryptor: MessageEncryptor): +@pytest.fixture +def gpg_settings( + settings: SettingsWrapper, + message_encryptor: MessageEncryptor, +) -> SettingsWrapper: settings.EMAIL_GNUPG_HOME = message_encryptor.gpg_home settings.EMAIL_ENABLE_GPG_DECRYPTOR = True return settings -@pytest.fixture() -def encrypted_pair(mail_mocker, message_encryptor: MessageEncryptor): +@pytest.fixture +def encrypted_pair( + mail_mocker: MailMocker, + message_encryptor: MessageEncryptor, +) -> tuple[MailMessage, MailMessage]: """ Build a (encrypted, plaintext) MailMessage pair sharing the same UID and headers, with two PDF attachments on the plaintext side. @@ -145,19 +150,23 @@ def encrypted_pair(mail_mocker, message_encryptor: MessageEncryptor): _AttachmentDef(filename="f2.pdf"), ], ) - encrypted = message_encryptor.encrypt(plaintext) - return encrypted, plaintext + return message_encryptor.encrypt(plaintext), plaintext + + +# Sentinel used in `test_able_to_run` parametrization to request the real +# GPG home from the session-scoped `message_encryptor` fixture at runtime. +_VALID_GPG_HOME = object() class TestMailMessageDecryptorAbleToRun: - """`MailMessageDecryptor.able_to_run()` configuration matrix — no DB needed.""" + """`MailMessageDecryptor.able_to_run()` configuration matrix.""" @pytest.mark.parametrize( ("settings_overrides", "expected"), [ pytest.param( { - "EMAIL_GNUPG_HOME": "_gpg_home_marker", + "EMAIL_GNUPG_HOME": _VALID_GPG_HOME, "EMAIL_ENABLE_GPG_DECRYPTOR": True, }, True, @@ -185,14 +194,14 @@ class TestMailMessageDecryptorAbleToRun: ) def test_able_to_run( self, - settings, + settings: SettingsWrapper, message_encryptor: MessageEncryptor, settings_overrides: dict, *, expected: bool, ) -> None: for key, value in settings_overrides.items(): - if value == "_gpg_home_marker": + if value is _VALID_GPG_HOME: value = message_encryptor.gpg_home setattr(settings, key, value) assert MailMessageDecryptor.able_to_run() is expected @@ -202,7 +211,11 @@ class TestMailMessageDecryptorAbleToRun: class TestMailMessageDecryptor: """End-to-end decrypt and consumption flow with a real GPG keyring.""" - def test_fails_at_initialization(self, settings, mocker) -> None: + def test_fails_at_initialization( + self, + settings: SettingsWrapper, + mocker: MockerFixture, + ) -> None: settings.EMAIL_ENABLE_GPG_DECRYPTOR = True mocker.patch( "gnupg.GPG.__init__", @@ -213,7 +226,12 @@ class TestMailMessageDecryptor: assert len(handler._message_preprocessors) == 0 - def test_decrypt_fails(self, settings, encrypted_pair, tmp_path: Path) -> None: + def test_decrypt_fails( + self, + settings: SettingsWrapper, + encrypted_pair: tuple[MailMessage, MailMessage], + tmp_path: Path, + ) -> None: """ A decryptor pointed at a fresh empty GPG home cannot decrypt the message — ensure it surfaces an exception rather than silently passing @@ -226,30 +244,22 @@ class TestMailMessageDecryptor: settings.EMAIL_ENABLE_GPG_DECRYPTOR = True settings.EMAIL_GNUPG_HOME = str(empty_gpg_home) - decryptor = MailMessageDecryptor() try: with pytest.raises(Exception): - decryptor.run(encrypted_message) + MailMessageDecryptor().run(encrypted_message) finally: - try: - subprocess.run( - ["gpgconf", "--kill", "gpg-agent"], - env={"GNUPGHOME": str(empty_gpg_home)}, - check=False, - capture_output=True, - timeout=5, - ) - except (FileNotFoundError, subprocess.TimeoutExpired): - pass + _kill_gpg_agent(str(empty_gpg_home)) - def test_decrypt_encrypted_mail(self, gpg_settings, encrypted_pair) -> None: + def test_decrypt_encrypted_mail( + self, + gpg_settings: SettingsWrapper, + encrypted_pair: tuple[MailMessage, MailMessage], + ) -> None: """ Creates a mail with attachments. Then encrypts it with a new key. Verifies that this encrypted message can be decrypted with attachments intact. """ encrypted_message, plaintext = encrypted_pair - headers = plaintext.headers - text = plaintext.text assert len(encrypted_message.attachments) == 1 assert encrypted_message.attachments[0].filename == "encrypted.asc" @@ -262,14 +272,14 @@ class TestMailMessageDecryptor: assert len(decrypted.attachments) == 2 assert decrypted.attachments[0].filename == "f1.pdf" assert decrypted.attachments[1].filename == "f2.pdf" - assert decrypted.headers == headers - assert decrypted.text == text + assert decrypted.headers == plaintext.headers + assert decrypted.text == plaintext.text assert decrypted.uid == plaintext.uid def test_handle_encrypted_message( self, - gpg_settings, - mail_mocker, + gpg_settings: SettingsWrapper, + mail_mocker: MailMocker, message_encryptor: MessageEncryptor, ) -> None: plaintext = mail_mocker.messageBuilder.create_message( @@ -280,16 +290,13 @@ class TestMailMessageDecryptor: ) encrypted = message_encryptor.encrypt(plaintext) - account = MailAccountFactory() - rule = MailRule( + rule = MailRule.objects.create( assign_title_from=MailRule.TitleSource.FROM_FILENAME, consumption_scope=MailRule.ConsumptionScope.EVERYTHING, - account=account, + account=MailAccountFactory(), ) - rule.save() - handler = MailAccountHandler() - result = handler._handle_message(encrypted, rule) + result = MailAccountHandler()._handle_message(encrypted, rule) assert result == 3 mail_mocker._queue_consumption_tasks_mock.assert_called()