Simplifications and improvements

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:

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"):

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()