239 lines
8.7 KiB
Python
239 lines
8.7 KiB
Python
import pytest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from datetime import datetime, timedelta
|
|
from app.email_processor import EmailProcessor
|
|
from app.models import User, Folder, ProcessedEmail
|
|
|
|
def test_email_processor_initialization():
|
|
"""Test that EmailProcessor initializes correctly."""
|
|
# Create a mock user
|
|
mock_user = Mock(spec=User)
|
|
mock_user.id = 1
|
|
mock_user.email = 'test@example.com'
|
|
mock_user.imap_config = {'username': 'user', 'password': 'pass'}
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
|
|
# Verify initialization
|
|
assert processor.user == mock_user
|
|
assert processor.logger is not None
|
|
|
|
@patch('app.email_processor.EmailProcessor._process_email_batch')
|
|
def test_process_folder_emails_no_pending(mock_batch):
|
|
"""Test processing a folder with no pending emails."""
|
|
# Create mocks
|
|
mock_user = Mock(spec=User)
|
|
mock_user.id = 1
|
|
|
|
mock_folder = Mock(spec=Folder)
|
|
mock_folder.id = 1
|
|
mock_folder.name = 'Test Folder'
|
|
mock_folder.rule_text = 'move to Archive'
|
|
|
|
# Mock the processed emails service
|
|
with patch('app.email_processor.ProcessedEmailsService') as mock_service:
|
|
mock_service_instance = mock_service.return_value
|
|
mock_service_instance.get_pending_emails.return_value = []
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
result = processor.process_folder_emails(mock_folder)
|
|
|
|
# Verify results
|
|
assert result['processed_count'] == 0
|
|
assert result['error_count'] == 0
|
|
assert mock_batch.called is False
|
|
|
|
@patch('app.email_processor.EmailProcessor._process_email_batch')
|
|
def test_process_folder_emails_with_pending(mock_batch):
|
|
"""Test processing a folder with pending emails."""
|
|
# Create mocks
|
|
mock_user = Mock(spec=User)
|
|
mock_user.id = 1
|
|
|
|
mock_folder = Mock(spec=Folder)
|
|
mock_folder.id = 1
|
|
mock_folder.name = 'Test Folder'
|
|
mock_folder.rule_text = 'move to Archive'
|
|
|
|
# Mock the processed emails service
|
|
with patch('app.email_processor.ProcessedEmailsService') as mock_service:
|
|
mock_service_instance = mock_service.return_value
|
|
mock_service_instance.get_pending_emails.return_value = ['1', '2', '3']
|
|
|
|
# Setup batch processing mock
|
|
mock_batch.return_value = {'processed_count': 3, 'error_count': 0}
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
result = processor.process_folder_emails(mock_folder)
|
|
|
|
# Verify results
|
|
assert result['processed_count'] == 3
|
|
assert result['error_count'] == 0
|
|
mock_batch.assert_called_once()
|
|
|
|
@patch('app.email_processor.EmailProcessor._update_folder_counts')
|
|
def test_process_user_emails_no_folders(mock_update):
|
|
"""Test processing user emails with no folders to process."""
|
|
# Create mock user
|
|
mock_user = Mock(spec=User)
|
|
mock_user.id = 1
|
|
mock_user.email = 'test@example.com'
|
|
|
|
# Mock the database query
|
|
with patch('app.email_processor.Folder') as mock_folder:
|
|
mock_folder.query.filter_by.return_value.order_by.return_value.all.return_value = []
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
result = processor.process_user_emails()
|
|
|
|
# Verify results
|
|
assert result['success_count'] == 0
|
|
assert result['error_count'] == 0
|
|
assert len(result['processed_folders']) == 0
|
|
mock_update.assert_not_called()
|
|
|
|
@patch('app.email_processor.EmailProcessor._update_folder_counts')
|
|
def test_process_user_emails_with_folders(mock_update):
|
|
"""Test processing user emails with folders to process."""
|
|
# Create mock user
|
|
mock_user = Mock(spec=User)
|
|
mock_user.id = 1
|
|
mock_user.email = 'test@example.com'
|
|
|
|
# Create mock folder
|
|
mock_folder = Mock(spec=Folder)
|
|
mock_folder.id = 1
|
|
mock_folder.name = 'Test Folder'
|
|
mock_folder.rule_text = 'move to Archive'
|
|
mock_folder.priority = 1
|
|
|
|
# Mock the database query
|
|
with patch('app.email_processor.Folder') as mock_folder_class:
|
|
mock_folder_class.query.filter_by.return_value.order_by.return_value.all.return_value = [mock_folder]
|
|
|
|
# Mock the process_folder_emails method
|
|
with patch('app.email_processor.EmailProcessor.process_folder_emails') as mock_process:
|
|
mock_process.return_value = {
|
|
'processed_count': 5,
|
|
'error_count': 0
|
|
}
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
result = processor.process_user_emails()
|
|
|
|
# Verify results
|
|
assert result['success_count'] == 5
|
|
assert result['error_count'] == 0
|
|
assert len(result['processed_folders']) == 1
|
|
mock_process.assert_called_once()
|
|
|
|
@patch('app.email_processor.EmailProcessor._move_email')
|
|
def test_process_email_batch_success(mock_move):
|
|
"""Test processing an email batch successfully."""
|
|
# Create mocks
|
|
mock_user = Mock(spec=User)
|
|
mock_user.id = 1
|
|
mock_user.imap_config = {'username': 'user', 'password': 'pass'}
|
|
|
|
mock_folder = Mock(spec=Folder)
|
|
mock_folder.id = 1
|
|
mock_folder.name = 'Source'
|
|
mock_folder.rule_text = 'move to Archive'
|
|
|
|
# Mock IMAP service
|
|
with patch('app.email_processor.IMAPService') as mock_imap:
|
|
mock_imap_instance = mock_imap.return_value
|
|
mock_imap_instance._connect.return_value = None
|
|
mock_imap_instance.connection.login.return_value = ('OK', [])
|
|
mock_imap_instance.connection.select.return_value = ('OK', [])
|
|
mock_imap_instance.get_email_headers.return_value = {
|
|
'subject': 'Test Email',
|
|
'from': 'sender@example.com'
|
|
}
|
|
|
|
# Mock rule evaluation
|
|
with patch('app.email_processor.EmailProcessor._evaluate_rules') as mock_evaluate:
|
|
mock_evaluate.return_value = 'Archive'
|
|
mock_move.return_value = True
|
|
|
|
# Mock processed emails service
|
|
with patch('app.email_processor.ProcessedEmailsService') as mock_service:
|
|
mock_service_instance = mock_service.return_value
|
|
mock_service_instance.mark_emails_processed.return_value = 1
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
result = processor._process_email_batch(mock_folder, ['1'])
|
|
|
|
# Verify results
|
|
assert result['processed_count'] == 1
|
|
assert result['error_count'] == 0
|
|
mock_move.assert_called_once()
|
|
|
|
def test_evaluate_rules_no_rule_text():
|
|
"""Test rule evaluation with no rule text."""
|
|
# Create mocks
|
|
mock_user = Mock(spec=User)
|
|
mock_folder = Mock(spec=Folder)
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
|
|
# Test with None rule text
|
|
result = processor._evaluate_rules({'subject': 'Test'}, None)
|
|
assert result is None
|
|
|
|
# Test with empty rule text
|
|
result = processor._evaluate_rules({'subject': 'Test'}, '')
|
|
assert result is None
|
|
|
|
def test_evaluate_rules_with_move_to():
|
|
"""Test rule evaluation with 'move to' directive."""
|
|
# Create mocks
|
|
mock_user = Mock(spec=User)
|
|
mock_folder = Mock(spec=Folder)
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
|
|
# Test with simple move to
|
|
result = processor._evaluate_rules({'subject': 'Test'}, 'move to Archive')
|
|
assert result == 'Archive'
|
|
|
|
# Test with punctuation
|
|
result = processor._evaluate_rules({'subject': 'Test'}, 'move to Archive.')
|
|
assert result == 'Archive'
|
|
|
|
# Test with extra spaces
|
|
result = processor._evaluate_rules({'subject': 'Test'}, 'move to Archive ')
|
|
assert result == 'Archive'
|
|
|
|
@patch('app.email_processor.ProcessedEmailsService')
|
|
def test_update_folder_counts(mock_service):
|
|
"""Test updating folder counts after processing."""
|
|
# Create mocks
|
|
mock_user = Mock(spec=User)
|
|
mock_folder = Mock(spec=Folder)
|
|
mock_folder.name = 'Test Folder'
|
|
mock_folder.pending_count = 5
|
|
mock_folder.total_count = 10
|
|
|
|
# Mock service methods
|
|
mock_service_instance = mock_service.return_value
|
|
mock_service_instance.get_pending_count.return_value = 3
|
|
|
|
with patch('app.email_processor.EmailProcessor._get_imap_connection') as mock_imap:
|
|
mock_imap.return_value.get_folder_email_count.return_value = 12
|
|
|
|
# Initialize processor
|
|
processor = EmailProcessor(mock_user)
|
|
processor._update_folder_counts(mock_folder)
|
|
|
|
# Verify counts were updated
|
|
assert mock_folder.pending_count == 3
|
|
assert mock_folder.total_count == 12 |