lots of configuration progress.
This commit is contained in:
@@ -40,7 +40,7 @@ def mock_user(app):
|
||||
first_name='Test',
|
||||
last_name='User',
|
||||
email='test@example.com',
|
||||
password_hash=b'hashed_password' # Will be properly hashed in real tests
|
||||
password_hash='hashed_password' # Will be properly hashed in real tests
|
||||
)
|
||||
|
||||
db.session.add(user)
|
||||
|
||||
78
tests/test_folder_deletion.py
Normal file
78
tests/test_folder_deletion.py
Normal file
@@ -0,0 +1,78 @@
|
||||
import pytest
|
||||
from app.models import User, Folder, ProcessedEmail, db
|
||||
from app import create_app
|
||||
|
||||
|
||||
class TestFolderDeletion:
|
||||
def test_delete_folder_with_emails(self, app, mock_user, authenticated_client):
|
||||
"""Test that deleting a folder also deletes its associated emails."""
|
||||
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 email records associated with this folder
|
||||
email_records = [
|
||||
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(email_records)
|
||||
db.session.commit()
|
||||
|
||||
# Verify emails were created
|
||||
assert ProcessedEmail.query.count() == 2
|
||||
|
||||
# Delete the folder
|
||||
response = authenticated_client.delete(f'/api/folders/{folder.id}')
|
||||
|
||||
# Verify the response is successful
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify the folder is deleted
|
||||
deleted_folder = Folder.query.filter_by(id=folder.id).first()
|
||||
assert deleted_folder is None
|
||||
|
||||
# Verify the associated emails are also deleted (cascade delete)
|
||||
assert ProcessedEmail.query.count() == 0
|
||||
|
||||
def test_delete_folder_with_no_emails(self, app, mock_user, authenticated_client):
|
||||
"""Test that deleting a folder with no associated emails works normally."""
|
||||
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()
|
||||
|
||||
# Verify no emails are associated with this folder
|
||||
assert ProcessedEmail.query.filter_by(folder_id=folder.id).count() == 0
|
||||
|
||||
# Delete the folder
|
||||
response = authenticated_client.delete(f'/api/folders/{folder.id}')
|
||||
|
||||
# Verify the response is successful
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify the folder is deleted
|
||||
deleted_folder = Folder.query.filter_by(id=folder.id).first()
|
||||
assert deleted_folder is None
|
||||
@@ -102,6 +102,10 @@ class TestIMAPService:
|
||||
with patch('app.imap_service.imaplib.IMAP4_SSL') as mock_imap_ssl:
|
||||
mock_connection = mock_imap_ssl.return_value
|
||||
mock_connection.list.return_value = ('OK', [])
|
||||
mock_connection.login.return_value = None
|
||||
mock_connection.select.return_value = ('OK', [b'1'])
|
||||
mock_connection.close.return_value = None
|
||||
mock_connection.logout.return_value = None
|
||||
|
||||
user = User(email='test@example.com', first_name='Test', last_name='User')
|
||||
user.imap_config = {'server': 'test.com', 'port': 993, 'username': 'user', 'password': 'pass'}
|
||||
@@ -119,6 +123,7 @@ class TestIMAPService:
|
||||
# Mock the IMAP connection to raise an exception
|
||||
with patch('app.imap_service.imaplib.IMAP4_SSL') as mock_imap_ssl:
|
||||
mock_connection = mock_imap_ssl.return_value
|
||||
mock_connection.login.return_value = None
|
||||
mock_connection.list.side_effect = Exception("IMAP error")
|
||||
|
||||
folders = imap_service.get_folders()
|
||||
@@ -193,22 +198,19 @@ class TestIMAPService:
|
||||
|
||||
def test_sync_folders_no_folders(self, app):
|
||||
with app.app_context():
|
||||
# Mock the IMAP connection to return an empty list of folders
|
||||
with patch('app.imap_service.imaplib.IMAP4_SSL') as mock_imap_ssl:
|
||||
mock_connection = mock_imap_ssl.return_value
|
||||
mock_connection.list.return_value = ('OK', [])
|
||||
user = User(email='test@example.com', first_name='Test', last_name='User')
|
||||
user.set_password('testpassword')
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
user.imap_config = {'server': 'localhost', 'port': 5143, 'username': 'user1@example.com', 'password': 'password1', 'use_ssl': False}
|
||||
imap_service = IMAPService(user)
|
||||
|
||||
# Mock the get_folders method to return an empty list
|
||||
with patch.object(imap_service, 'get_folders', return_value=[]):
|
||||
success, message = imap_service.sync_folders()
|
||||
|
||||
user = User(email='test@example.com', first_name='Test', last_name='User')
|
||||
user.set_password('testpassword')
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
user.imap_config = {'server': 'localhost', 'port': 5143, 'username': 'user1@example.com', 'password': 'password1', 'use_ssl': False}
|
||||
imap_service = IMAPService(user)
|
||||
|
||||
success, message = imap_service.sync_folders()
|
||||
|
||||
assert success is False
|
||||
assert message == "No folders found on IMAP server"
|
||||
assert success is False
|
||||
assert message == "No folders found on IMAP server"
|
||||
|
||||
def test_sync_folders_exception(self, app):
|
||||
with app.app_context():
|
||||
@@ -220,12 +222,9 @@ class TestIMAPService:
|
||||
user.imap_config = {'server': 'localhost', 'port': 5143, 'username': 'user1@example.com', 'password': 'password1', 'use_ssl': False }
|
||||
imap_service = IMAPService(user)
|
||||
|
||||
# Mock the IMAP connection to raise an exception
|
||||
with patch('app.imap_service.imaplib.IMAP4_SSL') as mock_imap_ssl:
|
||||
mock_connection = mock_imap_ssl.return_value
|
||||
mock_connection.list.side_effect = Exception("IMAP server error")
|
||||
|
||||
# Mock the get_folders method to raise an exception
|
||||
with patch.object(imap_service, 'get_folders', side_effect=Exception("IMAP server error")):
|
||||
success, message = imap_service.sync_folders()
|
||||
|
||||
assert success is False
|
||||
assert "No folders found on IMAP server" in message
|
||||
assert "Sync error: IMAP server error" in message
|
||||
267
tests/test_processed_emails_routes.py
Normal file
267
tests/test_processed_emails_routes.py
Normal file
@@ -0,0 +1,267 @@
|
||||
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.routes.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
|
||||
assert 'Test Subject 1' in response.get_data(as_text=True)
|
||||
assert 'Test Subject 2' in response.get_data(as_text=True)
|
||||
|
||||
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.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']
|
||||
|
||||
# Login and make request
|
||||
response = authenticated_client.post(f'/api/folders/{folder.id}/sync-emails')
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
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
|
||||
|
||||
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.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 == 404
|
||||
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']
|
||||
321
tests/test_processed_emails_service.py
Normal file
321
tests/test_processed_emails_service.py
Normal file
@@ -0,0 +1,321 @@
|
||||
import pytest
|
||||
from app.processed_emails_service import ProcessedEmailsService
|
||||
from app.models import User, Folder, ProcessedEmail, db
|
||||
from app import create_app
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class TestProcessedEmailsService:
|
||||
def test_init_with_user(self, app, mock_user):
|
||||
"""Test ProcessedEmailsService initialization with a user."""
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
assert service.user == mock_user
|
||||
|
||||
def test_get_pending_emails_empty(self, app, mock_user):
|
||||
"""Test get_pending_emails when no emails are pending."""
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock a folder
|
||||
folder = Folder(
|
||||
user_id=mock_user.id,
|
||||
name='Test Folder',
|
||||
rule_text='Test rule'
|
||||
)
|
||||
db.session.add(folder)
|
||||
db.session.commit()
|
||||
|
||||
# Test with no pending emails
|
||||
pending_uids = service.get_pending_emails('Test Folder')
|
||||
assert len(pending_uids) == 0
|
||||
|
||||
def test_get_pending_emails_with_data(self, app, mock_user):
|
||||
"""Test get_pending_emails when there are pending emails."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock 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()
|
||||
|
||||
# Test getting pending emails
|
||||
pending_uids = service.get_pending_emails('Test Folder')
|
||||
assert len(pending_uids) == 2
|
||||
assert '123' in pending_uids
|
||||
assert '456' in pending_uids
|
||||
|
||||
def test_mark_email_processed_existing(self, app, mock_user):
|
||||
"""Test mark_email_processed for an existing pending email."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock 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()
|
||||
|
||||
# Mark email as processed
|
||||
success = service.mark_email_processed('Test Folder', '123')
|
||||
assert success is 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
|
||||
assert updated_email.processed_at is not None
|
||||
|
||||
def test_mark_email_processed_new(self, app, mock_user):
|
||||
"""Test mark_email_processed for a new email record."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock a folder
|
||||
folder = Folder(
|
||||
user_id=mock_user.id,
|
||||
name='Test Folder',
|
||||
rule_text='Test rule'
|
||||
)
|
||||
db.session.add(folder)
|
||||
db.session.commit()
|
||||
|
||||
# Mark email as processed (create new record)
|
||||
success = service.mark_email_processed('Test Folder', '789')
|
||||
assert success is True
|
||||
|
||||
# Verify the record was created
|
||||
new_email = ProcessedEmail.query.filter_by(
|
||||
user_id=mock_user.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='789'
|
||||
).first()
|
||||
assert new_email is not None
|
||||
assert new_email.is_processed is True
|
||||
assert new_email.processed_at is not None
|
||||
|
||||
def test_mark_emails_processed(self, app, mock_user):
|
||||
"""Test mark_emails_processed for multiple emails."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock 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()
|
||||
|
||||
# Mark multiple emails as processed
|
||||
processed_count = service.mark_emails_processed('Test Folder', ['123', '456', '789'])
|
||||
assert processed_count == 3
|
||||
|
||||
# Verify all emails are marked as processed
|
||||
processed_uids = set()
|
||||
for email in ProcessedEmail.query.filter_by(
|
||||
user_id=mock_user.id,
|
||||
folder_name='Test Folder'
|
||||
).all():
|
||||
processed_uids.add(email.email_uid)
|
||||
assert email.is_processed is True
|
||||
|
||||
assert '123' in processed_uids
|
||||
assert '456' in processed_uids
|
||||
assert '789' in processed_uids
|
||||
|
||||
def test_sync_folder_emails(self, app, mock_user):
|
||||
"""Test sync_folder_emails for a folder."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock a folder
|
||||
folder = Folder(
|
||||
user_id=mock_user.id,
|
||||
name='Test Folder',
|
||||
rule_text='Test rule'
|
||||
)
|
||||
db.session.add(folder)
|
||||
db.session.commit()
|
||||
|
||||
# Sync emails
|
||||
email_uids = ['123', '456', '789']
|
||||
new_emails_count = service.sync_folder_emails('Test Folder', email_uids)
|
||||
assert new_emails_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
|
||||
|
||||
# Verify folder counts
|
||||
updated_folder = Folder.query.filter_by(id=folder.id).first()
|
||||
assert updated_folder.total_count == 3
|
||||
assert updated_folder.pending_count == 3
|
||||
|
||||
def test_get_pending_count(self, app, mock_user):
|
||||
"""Test get_pending_count for a folder."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock 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 and processed emails
|
||||
emails = [
|
||||
ProcessedEmail(
|
||||
user_id=mock_user.id,
|
||||
folder_id=folder.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='1',
|
||||
is_processed=False
|
||||
),
|
||||
ProcessedEmail(
|
||||
user_id=mock_user.id,
|
||||
folder_id=folder.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='2',
|
||||
is_processed=False
|
||||
),
|
||||
ProcessedEmail(
|
||||
user_id=mock_user.id,
|
||||
folder_id=folder.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='3',
|
||||
is_processed=True
|
||||
)
|
||||
]
|
||||
db.session.bulk_save_objects(emails)
|
||||
db.session.commit()
|
||||
|
||||
# Test pending count
|
||||
pending_count = service.get_pending_count('Test Folder')
|
||||
assert pending_count == 2
|
||||
|
||||
def test_cleanup_old_records(self, app, mock_user):
|
||||
"""Test cleanup_old_records for a folder."""
|
||||
with app.app_context():
|
||||
service = ProcessedEmailsService(mock_user)
|
||||
|
||||
# Mock 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 email records
|
||||
emails = [
|
||||
ProcessedEmail(
|
||||
user_id=mock_user.id,
|
||||
folder_id=folder.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='1',
|
||||
is_processed=False
|
||||
),
|
||||
ProcessedEmail(
|
||||
user_id=mock_user.id,
|
||||
folder_id=folder.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='2',
|
||||
is_processed=False
|
||||
),
|
||||
ProcessedEmail(
|
||||
user_id=mock_user.id,
|
||||
folder_id=folder.id,
|
||||
folder_name='Test Folder',
|
||||
email_uid='3',
|
||||
is_processed=True
|
||||
)
|
||||
]
|
||||
db.session.bulk_save_objects(emails)
|
||||
db.session.commit()
|
||||
|
||||
# Clean up records that don't exist in current UIDs
|
||||
current_uids = ['1', '2']
|
||||
deleted_count = service.cleanup_old_records('Test Folder', current_uids)
|
||||
assert deleted_count == 1
|
||||
|
||||
# Verify only existing records remain
|
||||
remaining_records = ProcessedEmail.query.filter_by(
|
||||
user_id=mock_user.id,
|
||||
folder_name='Test Folder'
|
||||
).all()
|
||||
assert len(remaining_records) == 2
|
||||
assert all(email.email_uid in current_uids for email in remaining_records)
|
||||
|
||||
# Verify folder counts
|
||||
updated_folder = Folder.query.filter_by(id=folder.id).first()
|
||||
assert updated_folder.total_count == 2
|
||||
assert updated_folder.pending_count == 2
|
||||
Reference in New Issue
Block a user