292 lines
12 KiB
Python
292 lines
12 KiB
Python
import pytest
|
|
from flask import url_for
|
|
from unittest.mock import patch
|
|
from app.models import User, Folder, ProcessedEmail, db
|
|
from app import create_app
|
|
from app.imap_service import IMAPService
|
|
|
|
|
|
class TestProcessedEmailsRoutes:
|
|
def test_get_pending_emails_success(self, app, mock_user, authenticated_client):
|
|
"""Test get_pending_emails endpoint successfully."""
|
|
with app.app_context():
|
|
# Create a folder
|
|
folder = Folder(
|
|
user_id=mock_user.id,
|
|
name='Test Folder',
|
|
rule_text='Test rule'
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
|
|
# Create some pending email records
|
|
pending_emails = [
|
|
ProcessedEmail(
|
|
user_id=mock_user.id,
|
|
folder_id=folder.id,
|
|
folder_name='Test Folder',
|
|
email_uid='123',
|
|
is_processed=False
|
|
),
|
|
ProcessedEmail(
|
|
user_id=mock_user.id,
|
|
folder_id=folder.id,
|
|
folder_name='Test Folder',
|
|
email_uid='456',
|
|
is_processed=False
|
|
)
|
|
]
|
|
db.session.bulk_save_objects(pending_emails)
|
|
db.session.commit()
|
|
|
|
# Mock IMAP service to return email headers
|
|
with patch('app.imap_service.IMAPService') as mock_imap_service:
|
|
mock_imap_instance = mock_imap_service.return_value
|
|
mock_imap_instance.get_email_headers.side_effect = [
|
|
{
|
|
'subject': 'Test Subject 1',
|
|
'date': '2023-01-01T12:00:00',
|
|
'from': 'sender1@example.com',
|
|
'to': 'recipient@example.com',
|
|
'message_id': '<msg1@example.com>'
|
|
},
|
|
{
|
|
'subject': 'Test Subject 2',
|
|
'date': '2023-01-02T12:00:00',
|
|
'from': 'sender2@example.com',
|
|
'to': 'recipient@example.com',
|
|
'message_id': '<msg2@example.com>'
|
|
}
|
|
]
|
|
|
|
response = authenticated_client.get(f'/api/folders/{folder.id}/pending-emails')
|
|
|
|
assert response.status_code == 200
|
|
# The response should be HTML with the pending emails dialog
|
|
response_text = response.get_data(as_text=True)
|
|
# Check that the response contains the expected content
|
|
assert 'Pending Emails in Test Folder' in response_text
|
|
assert 'Total Emails' in response_text
|
|
assert 'Pending' in response_text
|
|
assert 'Processed' in response_text
|
|
|
|
def test_get_pending_emails_folder_not_found(self, app, mock_user, authenticated_client):
|
|
"""Test get_pending_emails endpoint with non-existent folder."""
|
|
with app.app_context():
|
|
response = authenticated_client.get('/api/folders/999/pending-emails')
|
|
|
|
assert response.status_code == 404
|
|
assert 'Folder not found' in response.get_json()['error']
|
|
|
|
def test_mark_email_processed_success(self, app, mock_user, authenticated_client):
|
|
"""Test mark_email_processed endpoint successfully."""
|
|
with app.app_context():
|
|
# Create a folder
|
|
folder = Folder(
|
|
user_id=mock_user.id,
|
|
name='Test Folder',
|
|
rule_text='Test rule'
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
|
|
# Create a pending email record
|
|
pending_email = ProcessedEmail(
|
|
user_id=mock_user.id,
|
|
folder_id=folder.id,
|
|
folder_name='Test Folder',
|
|
email_uid='123',
|
|
is_processed=False
|
|
)
|
|
db.session.add(pending_email)
|
|
db.session.commit()
|
|
|
|
response = authenticated_client.post(f'/api/folders/{folder.id}/emails/123/process')
|
|
|
|
assert response.status_code == 200
|
|
# The response should be HTML with the updated dialog
|
|
assert 'has been marked as processed successfully' in response.get_data(as_text=True)
|
|
|
|
# Verify the email is marked as processed
|
|
updated_email = ProcessedEmail.query.filter_by(
|
|
user_id=mock_user.id,
|
|
folder_name='Test Folder',
|
|
email_uid='123'
|
|
).first()
|
|
assert updated_email.is_processed is True
|
|
|
|
def test_mark_email_processed_folder_not_found(self, app, mock_user, authenticated_client):
|
|
"""Test mark_email_processed endpoint with non-existent folder."""
|
|
with app.app_context():
|
|
# Login and make request
|
|
login_response = response = authenticated_client.post('/api/folders/999/emails/123/process')
|
|
|
|
assert response.status_code == 404
|
|
assert 'Folder not found' in response.get_json()['error']
|
|
|
|
def test_sync_folder_emails_success(self, app, mock_user, authenticated_client):
|
|
"""Test sync_folder_emails endpoint successfully."""
|
|
with app.app_context():
|
|
# Create a folder
|
|
folder = Folder(
|
|
user_id=mock_user.id,
|
|
name='Test Folder',
|
|
rule_text='Test rule'
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
|
|
# Mock IMAP service to return email UIDs
|
|
with patch('app.routes.emails.IMAPService') as mock_imap_service:
|
|
mock_imap_instance = mock_imap_service.return_value
|
|
mock_imap_instance.get_folder_email_uids.return_value = ['123', '456', '789']
|
|
|
|
# Set up the user's IMAP config
|
|
mock_user.imap_config = {
|
|
'server': 'localhost',
|
|
'port': 5143,
|
|
'username': 'user1@example.com',
|
|
'password': 'password1',
|
|
'use_ssl': False
|
|
}
|
|
db.session.commit()
|
|
|
|
# Login and make request
|
|
response = authenticated_client.post(f'/api/folders/{folder.id}/sync-emails')
|
|
|
|
assert response.status_code == 200
|
|
|
|
data = response.get_json()
|
|
print(data)
|
|
assert data['success'] is True
|
|
assert 'Synced 3 new emails' in data['message']
|
|
assert data['pending_count'] == 3
|
|
assert data['total_count'] == 3
|
|
|
|
# Verify records were created
|
|
records = ProcessedEmail.query.filter_by(
|
|
user_id=mock_user.id,
|
|
folder_name='Test Folder'
|
|
).all()
|
|
assert len(records) == 3
|
|
assert data['pending_count'] == 3
|
|
assert data['total_count'] == 3
|
|
|
|
# Verify records were created
|
|
records = ProcessedEmail.query.filter_by(
|
|
user_id=mock_user.id,
|
|
folder_name='Test Folder'
|
|
).all()
|
|
assert len(records) == 3
|
|
|
|
def test_sync_folder_emails_no_emails(self, app, mock_user, authenticated_client):
|
|
"""Test sync_folder_emails endpoint with no emails found."""
|
|
with app.app_context():
|
|
# Create a folder
|
|
folder = Folder(
|
|
user_id=mock_user.id,
|
|
name='Test Folder',
|
|
rule_text='Test rule'
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
|
|
# Mock IMAP service to return no email UIDs
|
|
with patch('app.routes.emails.IMAPService') as mock_imap_service:
|
|
mock_imap_instance = mock_imap_service.return_value
|
|
mock_imap_instance.get_folder_email_uids.return_value = []
|
|
|
|
# Login and make request
|
|
response = authenticated_client.post(f'/api/folders/{folder.id}/sync-emails')
|
|
|
|
assert response.status_code == 200
|
|
assert 'No emails found in folder' in response.get_json()['error']
|
|
|
|
def test_sync_folder_emails_folder_not_found(self, app, mock_user, authenticated_client):
|
|
"""Test sync_folder_emails endpoint with non-existent folder."""
|
|
with app.app_context():
|
|
# Login and make request
|
|
login_response = response = authenticated_client.post('/api/folders/999/sync-emails')
|
|
|
|
assert response.status_code == 404
|
|
assert 'Folder not found' in response.get_json()['error']
|
|
|
|
def test_process_folder_emails_success(self, app, mock_user, authenticated_client):
|
|
"""Test process_folder_emails endpoint successfully."""
|
|
with app.app_context():
|
|
# Create a folder
|
|
folder = Folder(
|
|
user_id=mock_user.id,
|
|
name='Test Folder',
|
|
rule_text='Test rule'
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
|
|
# Create some pending email records
|
|
pending_emails = [
|
|
ProcessedEmail(
|
|
user_id=mock_user.id,
|
|
folder_id=folder.id,
|
|
folder_name='Test Folder',
|
|
email_uid='123',
|
|
is_processed=False
|
|
),
|
|
ProcessedEmail(
|
|
user_id=mock_user.id,
|
|
folder_id=folder.id,
|
|
folder_name='Test Folder',
|
|
email_uid='456',
|
|
is_processed=False
|
|
)
|
|
]
|
|
db.session.bulk_save_objects(pending_emails)
|
|
db.session.commit()
|
|
|
|
# Login and make request
|
|
login_response = response = authenticated_client.post(f'/api/folders/{folder.id}/process-emails', data={
|
|
'email_uids': ['123', '456']
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert 'Processed 2 emails' in data['message']
|
|
assert data['pending_count'] == 0
|
|
|
|
# Verify emails are marked as processed
|
|
processed_emails = ProcessedEmail.query.filter_by(
|
|
user_id=mock_user.id,
|
|
folder_name='Test Folder',
|
|
is_processed=True
|
|
).all()
|
|
assert len(processed_emails) == 2
|
|
|
|
def test_process_folder_emails_no_uids(self, app, mock_user, authenticated_client):
|
|
"""Test process_folder_emails endpoint with no email UIDs provided."""
|
|
with app.app_context():
|
|
# Create a folder
|
|
folder = Folder(
|
|
user_id=mock_user.id,
|
|
name='Test Folder',
|
|
rule_text='Test rule'
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
|
|
# Login and make request
|
|
login_response = response = authenticated_client.post(f'/api/folders/{folder.id}/process-emails', data={})
|
|
|
|
assert response.status_code == 400
|
|
assert 'No email UIDs provided' in response.get_json()['error']
|
|
|
|
def test_process_folder_emails_folder_not_found(self, app, mock_user, authenticated_client):
|
|
"""Test process_folder_emails endpoint with non-existent folder."""
|
|
with app.app_context():
|
|
# Login and make request
|
|
login_response = response = authenticated_client.post('/api/folders/999/process-emails', data={
|
|
'email_uids': ['123', '456']
|
|
})
|
|
|
|
assert response.status_code == 404
|
|
assert 'Folder not found' in response.get_json()['error'] |