Merge pull request #108 from christianlouis/copilot/fix-s3-upload-error

Fix upload failures: standardize task signatures to accept file_id parameter

app/tasks/upload_to_email.py

@@ -15,6 +15,7 @@ from jinja2 import Environment, FileSystemLoader, select_autoescape
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 
 logger = logging.getLogger(__name__)
 
@@ -159,14 +160,32 @@ def _send_email_with_smtp(msg, filename, recipients):
         logger.error(error_msg)
         return {"status": "Failed", "reason": error_msg, "error": str(e)}
 
-@celery.task(base=BaseTaskWithRetry)
-def upload_to_email(file_path: str, recipients=None, subject=None, message=None, template_name="default.html", include_metadata=True):
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def upload_to_email(self, file_path: str, recipients=None, subject=None, message=None, template_name="default.html", include_metadata=True, file_id: int = None):
     """
     Sends a file via email to the specified recipients.
     If recipients is None, uses the configured default email recipient.
+    
+    Args:
+        file_path: Path to the file to send
+        recipients: Optional list of recipient email addresses
+        subject: Optional email subject
+        message: Optional custom message
+        template_name: Email template to use
+        include_metadata: Whether to include metadata in the email
+        file_id: Optional file ID to associate with logs
     """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting email send: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_email", "in_progress", f"Sending via email: {os.path.basename(file_path)}", file_id=file_id
+    )
+    
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_email", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename
     filename = os.path.basename(file_path)
@@ -174,16 +193,19 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None,
     # Check if email settings are configured
     if not settings.email_host:
         error_msg = "Email host is not configured"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_email", "skipped", error_msg, file_id=file_id)
         return {"status": "Skipped", "reason": error_msg}
         
     # Log email configuration for debugging
-    logger.debug(f"Email config - Host: {settings.email_host}, Port: {settings.email_port}, " 
+    logger.debug(f"[{task_id}] Email config - Host: {settings.email_host}, Port: {settings.email_port}, " 
                 f"Username: {settings.email_username}, TLS: {settings.email_use_tls}")
 
     # Process recipients
     recipients, error = _prepare_recipients(recipients)
     if error:
+        logger.error(f"[{task_id}] {error}")
+        log_task_progress(task_id, "upload_to_email", "skipped", error, file_id=file_id)
         return {"status": "Skipped", "reason": error}
 
     # Use provided subject or create default
@@ -237,8 +259,13 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None,
         # Send the email through SMTP
         error_result = _send_email_with_smtp(msg, filename, recipients)
         if error_result:
+            logger.error(f"[{task_id}] Failed to send email: {error_result.get('reason')}")
+            log_task_progress(task_id, "upload_to_email", "failure", error_result.get('reason'), file_id=file_id)
             return error_result
 
+        logger.info(f"[{task_id}] Successfully sent {filename} via email to {len(recipients)} recipients")
+        log_task_progress(task_id, "upload_to_email", "success", f"Sent via email: {filename}", file_id=file_id)
+        
         return {
             "status": "Completed",
             "file": file_path,
@@ -250,5 +277,6 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None,
         
     except Exception as e:
         error_msg = f"Failed to send {filename} via email: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_email", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_ftp.py

@@ -5,16 +5,31 @@ import ftplib
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 import logging
 
 logger = logging.getLogger(__name__)
 
-@celery.task(base=BaseTaskWithRetry)
-def upload_to_ftp(file_path: str):
-    """Uploads a file to an FTP server in the configured folder."""
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def upload_to_ftp(self, file_path: str, file_id: int = None):
+    """
+    Uploads a file to an FTP server in the configured folder.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting FTP upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_ftp", "in_progress", f"Uploading to FTP: {os.path.basename(file_path)}", file_id=file_id
+    )
 
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename
     filename = os.path.basename(file_path)
@@ -22,7 +37,8 @@ def upload_to_ftp(file_path: str):
     # Check if FTP settings are configured
     if not settings.ftp_host:
         error_msg = "FTP host is not configured"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id)
         raise ValueError(error_msg)
 
     try:
@@ -123,7 +139,8 @@ def upload_to_ftp(file_path: str):
         # Close FTP connection
         ftp.quit()
         
-        logger.info(f"Successfully uploaded {filename} to FTP server at {settings.ftp_host}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to FTP server at {settings.ftp_host}")
+        log_task_progress(task_id, "upload_to_ftp", "success", f"Uploaded to FTP: {filename}", file_id=file_id)
         return {
             "status": "Completed", 
             "file": file_path, 
@@ -134,5 +151,6 @@ def upload_to_ftp(file_path: str):
     
     except Exception as e:
         error_msg = f"Failed to upload {filename} to FTP server: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_google_drive.py

@@ -16,6 +16,7 @@ from google.auth.exceptions import RefreshError
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 
 logger = logging.getLogger(__name__)
 
@@ -142,12 +143,27 @@ def truncate_property_value(key, value, max_bytes=100):
         
     return str_value
 
-@celery.task(base=BaseTaskWithRetry)
-def upload_to_google_drive(file_path: str, include_metadata=True):
-    """Uploads a file to Google Drive in the configured folder with optional metadata."""
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def upload_to_google_drive(self, file_path: str, include_metadata=True, file_id: int = None):
+    """
+    Uploads a file to Google Drive in the configured folder with optional metadata.
+    
+    Args:
+        file_path: Path to the file to upload
+        include_metadata: Whether to include metadata in the upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting Google Drive upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_google_drive", "in_progress", f"Uploading to Google Drive: {os.path.basename(file_path)}", file_id=file_id
+    )
 
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename from path
     filename = os.path.basename(file_path)
@@ -161,7 +177,10 @@ def upload_to_google_drive(file_path: str, include_metadata=True):
         # Get Google Drive service
         service = get_google_drive_service()
         if not service:
-            raise Exception("Failed to initialize Google Drive service")
+            error_msg = "Failed to initialize Google Drive service"
+            logger.error(f"[{task_id}] {error_msg}")
+            log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id)
+            raise Exception(error_msg)
 
         # Prepare the file metadata
         file_metadata = {
@@ -221,16 +240,17 @@ def upload_to_google_drive(file_path: str, include_metadata=True):
         ).execute()
         
         # Log success details
-        file_id = file.get('id')
+        google_drive_file_id = file.get('id')
         web_view_link = file.get('webViewLink')
         
-        logger.info(f"Successfully uploaded {filename} to Google Drive with ID: {file_id}")
-        logger.info(f"File accessible at: {web_view_link}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to Google Drive with ID: {google_drive_file_id}")
+        logger.info(f"[{task_id}] File accessible at: {web_view_link}")
+        log_task_progress(task_id, "upload_to_google_drive", "success", f"Uploaded to Google Drive: {filename}", file_id=file_id)
         
         result = {
             "status": "Completed", 
             "file_path": file_path,
-            "google_drive_file_id": file_id,
+            "google_drive_file_id": google_drive_file_id,
             "google_drive_web_link": web_view_link
         }
         
@@ -242,5 +262,6 @@ def upload_to_google_drive(file_path: str, include_metadata=True):
 
     except Exception as e:
         error_msg = f"Failed to upload {filename} to Google Drive: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_sftp.py

@@ -8,21 +8,35 @@ from app.config import settings
 from app.celery_app import celery
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path
+from app.utils import log_task_progress
 
 logger = logging.getLogger(__name__)
 
-@celery.task(base=BaseTaskWithRetry)
-def upload_to_sftp(file_path: str):
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def upload_to_sftp(self, file_path: str, file_id: int = None):
     """
     Upload a file to an SFTP server.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
     """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting SFTP upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_sftp", "in_progress", f"Uploading to SFTP: {os.path.basename(file_path)}", file_id=file_id
+    )
+    
     if not os.path.exists(file_path):
         error_msg = f"File not found: {file_path}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_sftp", "failure", error_msg, file_id=file_id)
         raise FileNotFoundError(error_msg)
     
     if not (settings.sftp_host and settings.sftp_port and settings.sftp_username):
-        logger.info("SFTP upload skipped: Missing configuration")
+        error_msg = "SFTP upload skipped: Missing configuration"
+        logger.info(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_sftp", "skipped", error_msg, file_id=file_id)
         return {"status": "Skipped", "reason": "SFTP settings not configured"}
     
     filename = os.path.basename(file_path)
@@ -102,9 +116,10 @@ def upload_to_sftp(file_path: str):
                 logger.warning(f"Failed to create directory structure {remote_dir}: {str(e)}")
         
         # Upload the file
-        logger.info(f"Uploading {filename} to SFTP at {remote_path}")
+        logger.info(f"[{task_id}] Uploading {filename} to SFTP at {remote_path}")
         sftp.put(file_path, remote_path)
-        logger.info(f"Successfully uploaded {filename} to SFTP at {remote_path}")
+        logger.info(f"[{task_id}] Successfully uploaded {filename} to SFTP at {remote_path}")
+        log_task_progress(task_id, "upload_to_sftp", "success", f"Uploaded to SFTP: {filename}", file_id=file_id)
         
         # Close connections
         sftp.close()
@@ -126,5 +141,6 @@ def upload_to_sftp(file_path: str):
             pass
             
         error_msg = f"Failed to upload {filename} to SFTP server: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_sftp", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

app/tasks/upload_to_webdav.py

@@ -6,16 +6,31 @@ from urllib.parse import urljoin
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
 from app.celery_app import celery
+from app.utils import log_task_progress
 import logging
 
 logger = logging.getLogger(__name__)
 
-@celery.task(base=BaseTaskWithRetry)
-def upload_to_webdav(file_path: str):
-    """Uploads a file to a WebDAV server in the configured folder."""
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def upload_to_webdav(self, file_path: str, file_id: int = None):
+    """
+    Uploads a file to a WebDAV server in the configured folder.
+    
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting WebDAV upload: {file_path}")
+    log_task_progress(
+        task_id, "upload_to_webdav", "in_progress", f"Uploading to WebDAV: {os.path.basename(file_path)}", file_id=file_id
+    )
 
     if not os.path.exists(file_path):
-        raise FileNotFoundError(f"File not found: {file_path}")
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
 
     # Extract filename
     filename = os.path.basename(file_path)
@@ -23,7 +38,8 @@ def upload_to_webdav(file_path: str):
     # Check if WebDAV settings are configured
     if not settings.webdav_url:
         error_msg = "WebDAV URL is not configured"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
         raise ValueError(error_msg)
 
     # Construct the full upload URL
@@ -53,14 +69,17 @@ def upload_to_webdav(file_path: str):
 
         # Check if upload was successful
         if response.status_code in (200, 201, 204):
-            logger.info(f"Successfully uploaded {filename} to WebDAV at {webdav_url}.")
+            logger.info(f"[{task_id}] Successfully uploaded {filename} to WebDAV at {webdav_url}.")
+            log_task_progress(task_id, "upload_to_webdav", "success", f"Uploaded to WebDAV: {filename}", file_id=file_id)
             return {"status": "Completed", "file": file_path, "url": webdav_url}
         else:
             error_msg = f"Failed to upload {filename} to WebDAV: {response.status_code} - {response.text}"
-            logger.error(error_msg)
+            logger.error(f"[{task_id}] {error_msg}")
+            log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
             raise Exception(error_msg)
     
     except Exception as e:
         error_msg = f"Error uploading {filename} to WebDAV: {str(e)}"
-        logger.error(error_msg)
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id)
         raise Exception(error_msg)

tests/test_upload_tasks.py

@@ -1,5 +1,5 @@
 """
-Tests for upload tasks including OneDrive and S3.
+Tests for upload tasks including OneDrive, S3, FTP, SFTP, WebDAV, Google Drive, and Email.
 """
 
 import os
@@ -7,6 +7,11 @@ import pytest
 from unittest.mock import Mock, patch, MagicMock
 from app.tasks.upload_to_onedrive import upload_to_onedrive
 from app.tasks.upload_to_s3 import upload_to_s3
+from app.tasks.upload_to_ftp import upload_to_ftp
+from app.tasks.upload_to_sftp import upload_to_sftp
+from app.tasks.upload_to_webdav import upload_to_webdav
+from app.tasks.upload_to_google_drive import upload_to_google_drive
+from app.tasks.upload_to_email import upload_to_email
 
 
 @pytest.fixture
@@ -182,3 +187,267 @@ def test_upload_to_s3_logs_with_file_id(sample_text_file, mock_settings):
             call for call in mock_log.call_args_list if "file_id" in call[1] and call[1]["file_id"] == 456
         ]
         assert len(calls_with_file_id) > 0
+
+
+# Tests for newly standardized upload tasks
+
+@pytest.mark.unit
+def test_upload_to_ftp_accepts_file_id(sample_text_file):
+    """Test that upload_to_ftp accepts file_id parameter."""
+    with patch("app.tasks.upload_to_ftp.settings") as mock_settings, \
+         patch("app.tasks.upload_to_ftp.ftplib.FTP") as mock_ftp, \
+         patch("app.tasks.upload_to_ftp.log_task_progress"):
+        
+        # Setup settings
+        mock_settings.ftp_host = "ftp.example.com"
+        mock_settings.ftp_port = 21
+        mock_settings.ftp_username = "test_user"
+        mock_settings.ftp_password = "test_pass"
+        mock_settings.ftp_folder = "uploads"
+        mock_settings.ftp_use_tls = False
+        mock_settings.ftp_allow_plaintext = True
+        
+        # Setup mock FTP
+        mock_ftp_instance = Mock()
+        mock_ftp.return_value = mock_ftp_instance
+        
+        # Call with file_id parameter
+        result = upload_to_ftp.apply(args=[sample_text_file], kwargs={"file_id": 100}).get()
+        
+        assert result["status"] == "Completed"
+        assert result["file"] == sample_text_file
+        assert result["ftp_host"] == "ftp.example.com"
+
+
+@pytest.mark.unit
+def test_upload_to_ftp_without_file_id(sample_text_file):
+    """Test that upload_to_ftp works without file_id parameter."""
+    with patch("app.tasks.upload_to_ftp.settings") as mock_settings, \
+         patch("app.tasks.upload_to_ftp.ftplib.FTP") as mock_ftp, \
+         patch("app.tasks.upload_to_ftp.log_task_progress"):
+        
+        # Setup settings
+        mock_settings.ftp_host = "ftp.example.com"
+        mock_settings.ftp_username = "test_user"
+        mock_settings.ftp_password = "test_pass"
+        mock_settings.ftp_folder = None
+        mock_settings.ftp_use_tls = False
+        mock_settings.ftp_allow_plaintext = True
+        
+        # Setup mock FTP
+        mock_ftp_instance = Mock()
+        mock_ftp.return_value = mock_ftp_instance
+        
+        # Call without file_id parameter
+        result = upload_to_ftp.apply(args=[sample_text_file]).get()
+        
+        assert result["status"] == "Completed"
+
+
+@pytest.mark.unit
+def test_upload_to_sftp_accepts_file_id(sample_text_file):
+    """Test that upload_to_sftp accepts file_id parameter."""
+    with patch("app.tasks.upload_to_sftp.settings") as mock_settings, \
+         patch("app.tasks.upload_to_sftp.paramiko.SSHClient") as mock_ssh, \
+         patch("app.tasks.upload_to_sftp.log_task_progress"), \
+         patch("app.tasks.upload_to_sftp.extract_remote_path") as mock_extract, \
+         patch("app.tasks.upload_to_sftp.get_unique_filename") as mock_unique:
+        
+        # Setup settings
+        mock_settings.sftp_host = "sftp.example.com"
+        mock_settings.sftp_port = 22
+        mock_settings.sftp_username = "test_user"
+        mock_settings.sftp_password = "test_pass"
+        mock_settings.sftp_folder = "/uploads"
+        mock_settings.workdir = "/tmp"
+        
+        # Setup mocks
+        mock_ssh_instance = Mock()
+        mock_sftp = Mock()
+        mock_ssh.return_value = mock_ssh_instance
+        mock_ssh_instance.open_sftp.return_value = mock_sftp
+        mock_extract.return_value = "/uploads/test.txt"
+        mock_unique.return_value = "/uploads/test.txt"
+        
+        # Call with file_id parameter
+        result = upload_to_sftp.apply(args=[sample_text_file], kwargs={"file_id": 200}).get()
+        
+        assert result["status"] == "Completed"
+        assert result["file_path"] == sample_text_file
+        assert "sftp_path" in result
+
+
+@pytest.mark.unit
+def test_upload_to_webdav_accepts_file_id(sample_text_file):
+    """Test that upload_to_webdav accepts file_id parameter."""
+    with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
+         patch("app.tasks.upload_to_webdav.requests.put") as mock_put, \
+         patch("app.tasks.upload_to_webdav.log_task_progress"):
+        
+        # Setup settings
+        mock_settings.webdav_url = "https://webdav.example.com/"
+        mock_settings.webdav_username = "test_user"
+        mock_settings.webdav_password = "test_pass"
+        mock_settings.webdav_folder = "uploads"
+        mock_settings.webdav_verify_ssl = True
+        
+        # Setup mock response
+        mock_response = Mock()
+        mock_response.status_code = 201
+        mock_put.return_value = mock_response
+        
+        # Call with file_id parameter
+        result = upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 300}).get()
+        
+        assert result["status"] == "Completed"
+        assert result["file"] == sample_text_file
+        assert "url" in result
+
+
+@pytest.mark.unit
+def test_upload_to_google_drive_accepts_file_id(sample_text_file):
+    """Test that upload_to_google_drive accepts file_id parameter."""
+    with patch("app.tasks.upload_to_google_drive.get_google_drive_service") as mock_service, \
+         patch("app.tasks.upload_to_google_drive.MediaFileUpload") as mock_media, \
+         patch("app.tasks.upload_to_google_drive.extract_metadata_from_file") as mock_metadata, \
+         patch("app.tasks.upload_to_google_drive.settings") as mock_settings, \
+         patch("app.tasks.upload_to_google_drive.log_task_progress"):
+        
+        # Setup settings
+        mock_settings.google_drive_folder_id = "test_folder_id"
+        
+        # Setup mocks
+        mock_drive_service = Mock()
+        mock_service.return_value = mock_drive_service
+        mock_metadata.return_value = {}
+        
+        mock_files = Mock()
+        mock_drive_service.files.return_value = mock_files
+        mock_create = Mock()
+        mock_files.create.return_value = mock_create
+        mock_create.execute.return_value = {
+            "id": "file123",
+            "name": "test.txt",
+            "webViewLink": "https://drive.google.com/file/d/file123"
+        }
+        
+        # Call with file_id parameter
+        result = upload_to_google_drive.apply(args=[sample_text_file], kwargs={"file_id": 400}).get()
+        
+        assert result["status"] == "Completed"
+        assert result["file_path"] == sample_text_file
+        assert "google_drive_file_id" in result
+
+
+@pytest.mark.unit
+def test_upload_to_email_accepts_file_id(sample_text_file):
+    """Test that upload_to_email accepts file_id parameter."""
+    with patch("app.tasks.upload_to_email.settings") as mock_settings, \
+         patch("app.tasks.upload_to_email.smtplib.SMTP") as mock_smtp, \
+         patch("app.tasks.upload_to_email.get_email_template") as mock_template, \
+         patch("app.tasks.upload_to_email.extract_metadata_from_file") as mock_metadata, \
+         patch("app.tasks.upload_to_email.log_task_progress"), \
+         patch("app.tasks.upload_to_email._prepare_recipients") as mock_recipients, \
+         patch("app.tasks.upload_to_email._send_email_with_smtp") as mock_send, \
+         patch("app.tasks.upload_to_email.attach_logo") as mock_logo:
+        
+        # Setup settings
+        mock_settings.email_host = "smtp.example.com"
+        mock_settings.email_port = 587
+        mock_settings.email_username = "test@example.com"
+        mock_settings.email_password = "test_pass"
+        mock_settings.email_use_tls = True
+        mock_settings.email_sender = "sender@example.com"
+        mock_settings.external_hostname = "docuelevate.example.com"
+        
+        # Setup mocks
+        mock_recipients.return_value = (["recipient@example.com"], None)
+        mock_send.return_value = None
+        mock_metadata.return_value = {}
+        mock_logo.return_value = False
+        
+        mock_template_obj = Mock()
+        mock_template_obj.render.return_value = "<html>Test email</html>"
+        mock_template.return_value = mock_template_obj
+        
+        # Call with file_id parameter
+        result = upload_to_email.apply(args=[sample_text_file], kwargs={"file_id": 500}).get()
+        
+        assert result["status"] == "Completed"
+        assert result["file"] == sample_text_file
+        assert "recipients" in result
+
+
+@pytest.mark.unit
+def test_upload_to_ftp_file_not_found():
+    """Test that upload_to_ftp raises error for missing file."""
+    with patch("app.tasks.upload_to_ftp.settings") as mock_settings, \
+         patch("app.tasks.upload_to_ftp.log_task_progress"):
+        
+        mock_settings.ftp_host = "ftp.example.com"
+        
+        with pytest.raises(FileNotFoundError):
+            upload_to_ftp.apply(args=["/nonexistent/file.pdf"], kwargs={"file_id": 1}).get()
+
+
+@pytest.mark.unit
+def test_upload_to_sftp_file_not_found():
+    """Test that upload_to_sftp raises error for missing file."""
+    with patch("app.tasks.upload_to_sftp.settings") as mock_settings, \
+         patch("app.tasks.upload_to_sftp.log_task_progress"):
+        
+        mock_settings.sftp_host = "sftp.example.com"
+        mock_settings.sftp_port = 22
+        mock_settings.sftp_username = "test_user"
+        
+        with pytest.raises(FileNotFoundError):
+            upload_to_sftp.apply(args=["/nonexistent/file.pdf"], kwargs={"file_id": 1}).get()
+
+
+@pytest.mark.unit
+def test_upload_to_webdav_file_not_found():
+    """Test that upload_to_webdav raises error for missing file."""
+    with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
+         patch("app.tasks.upload_to_webdav.log_task_progress"):
+        
+        mock_settings.webdav_url = "https://webdav.example.com/"
+        
+        with pytest.raises(FileNotFoundError):
+            upload_to_webdav.apply(args=["/nonexistent/file.pdf"], kwargs={"file_id": 1}).get()
+
+
+@pytest.mark.unit
+def test_all_upload_tasks_have_consistent_signature(sample_text_file):
+    """Test that all upload tasks accept file_id as a keyword parameter.
+    
+    This test verifies that all upload tasks can be called with the same signature
+    as used in send_to_all.py: task.delay(file_path, file_id)
+    
+    Note: We use task.run to inspect the actual function signature because
+    Celery tasks wrap the original function, and .run provides access to
+    the unwrapped callable's signature.
+    """
+    
+    upload_tasks = [
+        (upload_to_s3, "app.tasks.upload_to_s3"),
+        (upload_to_ftp, "app.tasks.upload_to_ftp"),
+        (upload_to_sftp, "app.tasks.upload_to_sftp"),
+        (upload_to_webdav, "app.tasks.upload_to_webdav"),
+        (upload_to_google_drive, "app.tasks.upload_to_google_drive"),
+        (upload_to_email, "app.tasks.upload_to_email"),
+    ]
+    
+    import inspect
+    
+    for task, module_path in upload_tasks:
+        # Use task.run to inspect the actual wrapped function's signature
+        # This is necessary because Celery's task decorator wraps the original function
+        sig = inspect.signature(task.run)
+        params = list(sig.parameters.keys())
+        
+        # Should have at least file_path and file_id parameters
+        assert "file_path" in params, f"{task.name} missing file_path parameter"
+        assert "file_id" in params, f"{task.name} missing file_id parameter"
+        
+        # file_id should have a default value (None)
+        assert sig.parameters["file_id"].default is None, f"{task.name} file_id should default to None"