import imaplib from email.message import Message from unittest.mock import MagicMock, patch from sqlalchemy.orm import Session from app.crud.newsletters import create_newsletter from app.crud.settings import create_or_update_settings from app.models.newsletters import Newsletter from app.schemas.newsletters import NewsletterCreate from app.schemas.settings import SettingsCreate from app.services.email_processor import _process_single_email def _setup_test_email_processing( db_session: Session, newsletter_create_data: NewsletterCreate, settings_create_data: SettingsCreate, ) -> tuple[MagicMock, Newsletter, SettingsCreate]: """Help to set up mocks and data for email processing tests.""" settings = create_or_update_settings(db_session, settings_create_data) newsletter = create_newsletter(db_session, newsletter_create_data) mock_mail = MagicMock(spec=imaplib.IMAP4_SSL) msg = Message() msg["From"] = newsletter_create_data.sender_emails[0] msg["Subject"] = "Test Email" msg["Message-ID"] = "" msg.set_payload("

Original Body

", "utf-8") mock_mail.fetch.return_value = ("OK", [(b"1 (RFC822)", msg.as_bytes())]) return mock_mail, newsletter, settings def test_process_single_email_with_newsletter_move_folder(db_session: Session): """Test that the per-newsletter move_to_folder is used, overriding the global setting.""" # 1. ARRANGE settings_data = SettingsCreate( imap_server="test.com", imap_username="test", imap_password="password", move_to_folder="GlobalArchive", ) newsletter_data = NewsletterCreate( name="Test Newsletter", sender_emails=["test@example.com"], move_to_folder="NewsletterArchive", ) mock_mail, newsletter, settings = _setup_test_email_processing( db_session, newsletter_data, settings_data ) sender_map = {newsletter.senders[0].email: newsletter} # 2. ACT _process_single_email("1", mock_mail, db_session, sender_map, settings) # 3. ASSERT mock_mail.copy.assert_called_once_with("1", "NewsletterArchive") mock_mail.store.assert_any_call("1", "+FLAGS", "\\Deleted") def test_process_single_email_with_global_move_folder(db_session: Session): """Test that the global move_to_folder is used when the per-newsletter one is not set.""" # 1. ARRANGE settings_data = SettingsCreate( imap_server="test.com", imap_username="test", imap_password="password", move_to_folder="GlobalArchive", ) newsletter_data = NewsletterCreate( name="Test Newsletter", sender_emails=["test@example.com"] ) mock_mail, newsletter, settings = _setup_test_email_processing( db_session, newsletter_data, settings_data ) sender_map = {newsletter.senders[0].email: newsletter} # 2. ACT _process_single_email("1", mock_mail, db_session, sender_map, settings) # 3. ASSERT mock_mail.copy.assert_called_once_with("1", "GlobalArchive") mock_mail.store.assert_any_call("1", "+FLAGS", "\\Deleted") @patch("app.services.email_processor._extract_and_clean_html") def test_process_single_email_with_content_extraction( mock_extract_clean, db_session: Session, ): """Test that the cleaning function is called when extract_content is True.""" # 1. ARRANGE mock_extract_clean.return_value = { "title": "Extracted Title", "body": "Extracted Body", } settings_data = SettingsCreate( imap_server="test.com", imap_username="test", imap_password="password" ) newsletter_data = NewsletterCreate( name="Test Newsletter", sender_emails=["test@example.com"], extract_content=True, ) mock_mail, newsletter, settings = _setup_test_email_processing( db_session, newsletter_data, settings_data ) sender_map = {newsletter.senders[0].email: newsletter} # 2. ACT with patch("app.services.email_processor.create_entry") as mock_create_entry: _process_single_email("1", mock_mail, db_session, sender_map, settings) # 3. ASSERT mock_extract_clean.assert_called_once() # Check that create_entry was called with the extracted body mock_create_entry.assert_called_once() entry_create_arg = mock_create_entry.call_args[0][1] assert entry_create_arg.body == "Extracted Body" # Subject should still come from the email, not the extracted title assert entry_create_arg.subject == "Test Email"