progress.

tests/unit/test_imap_service.py

@@ -0,0 +1,230 @@
+import pytest
+import imaplib
+from unittest.mock import patch, Mock
+from app.imap_service import IMAPService
+from app.models import User, db
+from app import create_app
+
+class TestIMAPService:
+    def test_init_with_user(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        user.imap_config = {'server': 'test.com', 'port': 993}
+        imap_service = IMAPService(user)
+        assert imap_service.user == user
+        assert imap_service.config == {'server': 'test.com', 'port': 993}
+    
+    def test_init_without_config(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        imap_service = IMAPService(user)
+        assert imap_service.user == user
+        assert imap_service.config == {}
+    
+    def test_test_connection_success(self,  app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        user.imap_config = {'server': 'localhost', 'port': 5143, 'username': 'user1@example.com', 'password': 'password1', 'use_ssl': True}
+        imap_service = IMAPService(user)
+        
+        # Mock the IMAP connection
+        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.select.return_value = ('OK', [b'1'])
+            mock_connection.close.return_value = None
+            mock_connection.logout.return_value = None
+            
+            success, message = imap_service.test_connection()
+            
+            assert message == "Connection successful"
+            assert success is True
+    
+    def test_test_connection_failure(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        user.imap_config = {'server': 'test.com', 'port': 993, 'username': 'user', 'password': 'pass'}
+        imap_service = IMAPService(user)
+        
+        # Mock the IMAP connection to raise a generic exception
+        with patch('app.imap_service.imaplib.IMAP4_SSL') as mock_imap_ssl:
+            mock_connection = mock_imap_ssl.return_value
+            mock_connection.login.side_effect = Exception("Connection failed")
+            
+            success, message = imap_service.test_connection()
+            
+            assert success is False
+            assert "Connection error: Connection failed" in message
+    
+    def test_test_connection_imap_error(self, app):
+        # Mock the IMAP connection to raise an IMAP error
+        with patch('app.imap_service.imaplib.IMAP4_SSL') as mock_imap_ssl:
+            mock_connection = mock_imap_ssl.return_value
+            mock_connection.login.side_effect = imaplib.IMAP4.error("IMAP login failed")
+            
+            user = User(email='test@example.com', first_name='Test', last_name='User')
+            user.imap_config = {'server': 'test.com', 'port': 993, 'username': 'user', 'password': 'pass'}
+            imap_service = IMAPService(user)
+            
+            success, message = imap_service.test_connection()
+            
+            assert success is False
+            assert "IMAP connection error: IMAP login failed" in message
+    
+    def test_test_connection_no_config(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        imap_service = IMAPService(user)
+        
+        success, message = imap_service.test_connection()
+        
+        assert success is False
+        assert message == "No IMAP configuration found"
+    
+    def test_get_folders_success(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        user.imap_config = {'server': 'test.com', 'port': 993, 'username': 'user', 'password': 'pass'}
+        imap_service = IMAPService(user)
+        
+        # Mock the IMAP connection
+        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', [
+                b'(\\HasNoChildren) "INBOX"',
+                b'(\\HasNoChildren) "Sent"',
+                b'(\\HasNoChildren) "Drafts"'
+            ])
+            
+            folders = imap_service.get_folders()
+            
+            assert len(folders) == 3
+            assert folders[0]['name'] == 'INBOX'
+            assert folders[1]['name'] == 'Sent'
+            assert folders[2]['name'] == 'Drafts'
+    
+    def test_get_folders_empty(self, app):
+        # 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', [])
+            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'}
+            imap_service = IMAPService(user)
+            
+            folders = imap_service.get_folders()
+            
+            assert len(folders) == 0
+    
+    def test_get_folders_error(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        user.imap_config = {'server': 'test.com', 'port': 993, 'username': 'user', 'password': 'pass'}
+        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.login.return_value = None
+            mock_connection.list.side_effect = Exception("IMAP error")
+            
+            folders = imap_service.get_folders()
+            
+            assert len(folders) == 0
+    
+    def test_get_folders_no_config(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        imap_service = IMAPService(user)
+        
+        folders = imap_service.get_folders()
+        
+        assert len(folders) == 0
+    
+    def test_sync_folders_success(self, app):
+        with app.app_context():
+            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'}
+            imap_service = IMAPService(user)
+        
+        # Mock the IMAP connection
+        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', [
+                b'(\\HasNoChildren) "CustomFolder1"',
+                b'(\\HasNoChildren) "CustomFolder2"'
+            ])
+            
+            # Mock the get_folder_email_count method
+            with patch.object(imap_service, 'get_folder_email_count') as mock_count:
+                mock_count.return_value = 5
+                
+                # Mock the get_recent_emails method
+                with patch.object(imap_service, 'get_recent_emails') as mock_recent:
+                    mock_recent.return_value = [
+                        {'subject': 'Test Email 1', 'date': '2023-01-01T12:00:00'},
+                        {'subject': 'Test Email 2', 'date': '2023-01-02T12:00:00'}
+                    ]
+                    
+                    # Get fresh user from database and create new IMAPService to fix DetachedInstanceError
+                    fresh_user = User.query.filter_by(email='test@example.com').first()
+                    fresh_user.imap_config = {'server': 'localhost', 'port': 5143, 'username': 'user1@example.com', 'password': 'password1'}
+                    fresh_imap_service = IMAPService(fresh_user)
+                    
+                    # Mock the get_folder_email_count method on the new service
+                    with patch.object(fresh_imap_service, 'get_folder_email_count') as mock_count:
+                        mock_count.return_value = 5
+                        
+                        # Mock the get_recent_emails method on the new service
+                        with patch.object(fresh_imap_service, 'get_recent_emails') as mock_recent:
+                            mock_recent.return_value = [
+                                {'subject': 'Test Email 1', 'date': '2023-01-01T12:00:00'},
+                                {'subject': 'Test Email 2', 'date': '2023-01-02T12:00:00'}
+                            ]
+                            
+                            success, message = fresh_imap_service.sync_folders()
+                            
+                            assert success is True
+                            assert "Successfully synced 2 folders" in message
+    
+    def test_sync_folders_no_config(self, app):
+        user = User(email='test@example.com', first_name='Test', last_name='User')
+        imap_service = IMAPService(user)
+        
+        success, message = imap_service.sync_folders()
+        
+        assert success is False
+        assert message == "No IMAP configuration found"
+    
+    def test_sync_folders_no_folders(self, app):
+        with app.app_context():
+            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()
+                
+                assert success is False
+                assert message == "No folders found on IMAP server"
+    
+    def test_sync_folders_exception(self, app):
+        with app.app_context():
+             
+            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 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 "Sync error: IMAP server error" in message

tests/unit/test_models.py

@@ -0,0 +1,32 @@
+import pytest
+from app.models import User, Folder
+from app import db
+import uuid
+import conftest
+
+def test_user_model(app,  mock_user):
+    """Test User model creation and properties."""
+    # Test user was created by fixture
+    assert mock_user.email == 'test@example.com'
+    
+    # Test querying user
+    user_from_db = User.query.filter_by(email='test@example.com').first()
+    assert user_from_db is not None
+    assert user_from_db.id == mock_user.id
+
+def test_folder_model(app, mock_folder, mock_user):
+    """Test Folder model creation and properties."""
+    # Test folder was created by fixture
+    assert mock_folder.user_id == mock_user.id
+    assert mock_folder.name == 'Test Folder'
+    assert mock_folder.rule_text == 'Test rule'
+    assert mock_folder.priority == 1
+    
+    # Test relationship
+    assert len(mock_user.folders) == 1
+    assert mock_user.folders[0].id == mock_folder.id
+    
+    # Test querying folder
+    folder_from_db = Folder.query.filter_by(name='Test Folder').first()
+    assert folder_from_db is not None
+    assert folder_from_db.user_id == mock_user.id

tests/unit/test_processed_emails_service.py

@@ -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