diff --git a/tests/test_upload_tasks.py b/tests/test_upload_tasks.py index 3ab315ce..cdf82382 100644 --- a/tests/test_upload_tasks.py +++ b/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,259 @@ 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 = "Test email" + 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) + + 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"), + ] + + for task, module_path in upload_tasks: + # Verify that the task has the expected signature by inspecting its function + import inspect + 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"