"""Comprehensive tests for upload_to_webdav task.""" import os import pytest from unittest.mock import patch, Mock, MagicMock from requests.exceptions import ConnectionError, Timeout, RequestException from app.tasks.upload_to_webdav import upload_to_webdav @pytest.mark.unit class TestUploadToWebDAV: """Comprehensive tests for WebDAV upload task.""" def test_upload_success_with_file_id(self, sample_text_file): """Test successful upload with 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") as mock_log: # 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 mock_settings.http_request_timeout = 30 # Setup mock response - 201 Created mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response # Execute upload result = upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 100}).get() # Verify result assert result["status"] == "Completed" assert result["file"] == sample_text_file assert "url" in result assert "webdav.example.com" in result["url"] # Verify requests.put was called correctly assert mock_put.called call_args = mock_put.call_args assert call_args[1]["auth"] == ("test_user", "test_pass") assert call_args[1]["verify"] is True assert call_args[1]["timeout"] == 30 # Verify logging was called with file_id assert mock_log.called log_calls_with_file_id = [ call for call in mock_log.call_args_list if call[1].get("file_id") == 100 ] assert len(log_calls_with_file_id) > 0 def test_upload_success_without_file_id(self, sample_text_file): """Test successful upload without 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"): mock_settings.webdav_url = "https://webdav.example.com/" mock_settings.webdav_username = "test_user" mock_settings.webdav_password = "test_pass" mock_settings.webdav_folder = "" mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 200 # 200 OK is also valid mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() assert result["status"] == "Completed" assert result["file"] == sample_text_file def test_upload_success_status_204(self, sample_text_file): """Test successful upload with 204 No Content status.""" 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"): mock_settings.webdav_url = "https://webdav.example.com/" mock_settings.webdav_username = "test_user" mock_settings.webdav_password = "test_pass" mock_settings.webdav_folder = None mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 204 # 204 No Content mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() assert result["status"] == "Completed" def test_missing_webdav_url(self, sample_text_file): """Test that missing WebDAV URL raises ValueError.""" with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \ patch("app.tasks.upload_to_webdav.log_task_progress"): mock_settings.webdav_url = None mock_settings.webdav_username = "test_user" mock_settings.webdav_password = "test_pass" with pytest.raises(ValueError, match="WebDAV URL is not configured"): upload_to_webdav.apply(args=[sample_text_file]).get() def test_file_not_found(self): """Test that missing file raises FileNotFoundError.""" 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, match="File not found"): upload_to_webdav.apply(args=["/nonexistent/file.pdf"]).get() def test_http_error_response(self, sample_text_file): """Test handling of HTTP error responses (4xx, 5xx).""" 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"): 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 mock_settings.http_request_timeout = 30 # Simulate 401 Unauthorized mock_response = Mock() mock_response.status_code = 401 mock_response.text = "Unauthorized" mock_put.return_value = mock_response with pytest.raises(Exception, match="Failed to upload.*401"): upload_to_webdav.apply(args=[sample_text_file]).get() def test_http_404_not_found(self, sample_text_file): """Test handling of 404 Not Found response.""" 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"): mock_settings.webdav_url = "https://webdav.example.com/" mock_settings.webdav_username = "test_user" mock_settings.webdav_password = "test_pass" mock_settings.webdav_folder = "nonexistent" mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 404 mock_response.text = "Not Found" mock_put.return_value = mock_response with pytest.raises(Exception, match="Failed to upload.*404"): upload_to_webdav.apply(args=[sample_text_file]).get() def test_http_500_server_error(self, sample_text_file): """Test handling of 500 Internal Server Error.""" 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"): 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 500 mock_response.text = "Internal Server Error" mock_put.return_value = mock_response with pytest.raises(Exception, match="Failed to upload.*500"): upload_to_webdav.apply(args=[sample_text_file]).get() def test_connection_error(self, sample_text_file): """Test handling of connection errors.""" 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"): 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 mock_settings.http_request_timeout = 30 # Simulate connection error mock_put.side_effect = ConnectionError("Connection refused") with pytest.raises(Exception, match="Error uploading.*Connection refused"): upload_to_webdav.apply(args=[sample_text_file]).get() def test_timeout_error(self, sample_text_file): """Test handling of timeout errors.""" 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"): 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 mock_settings.http_request_timeout = 30 # Simulate timeout mock_put.side_effect = Timeout("Request timed out") with pytest.raises(Exception, match="Error uploading.*timed out"): upload_to_webdav.apply(args=[sample_text_file]).get() def test_url_construction_with_trailing_slash(self, sample_text_file): """Test URL construction when base URL has trailing slash.""" 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"): 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() # Verify URL construction called_url = mock_put.call_args[0][0] assert called_url.startswith("https://webdav.example.com/") assert "uploads" in called_url assert os.path.basename(sample_text_file) in called_url def test_url_construction_without_trailing_slash(self, sample_text_file): """Test URL construction when base URL has no trailing slash.""" 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"): mock_settings.webdav_url = "https://webdav.example.com" mock_settings.webdav_username = "test_user" mock_settings.webdav_password = "test_pass" mock_settings.webdav_folder = "documents" mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() # Verify URL construction called_url = mock_put.call_args[0][0] assert "webdav.example.com" in called_url assert "documents" in called_url def test_folder_path_with_leading_slash(self, sample_text_file): """Test folder path normalization when it starts with /.""" 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"): 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/documents" # Leading slash mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() # Verify the leading slash was removed in URL construction called_url = mock_put.call_args[0][0] # Should not have double slashes like //uploads assert "//" not in called_url.replace("https://", "") def test_empty_folder_path(self, sample_text_file): """Test upload with empty folder path (root directory).""" 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"): mock_settings.webdav_url = "https://webdav.example.com/" mock_settings.webdav_username = "test_user" mock_settings.webdav_password = "test_pass" mock_settings.webdav_folder = "" # Empty folder mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() assert result["status"] == "Completed" def test_ssl_verification_enabled(self, sample_text_file): """Test that SSL verification is enabled when configured.""" 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"): 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response upload_to_webdav.apply(args=[sample_text_file]).get() # Verify SSL verification was enabled call_kwargs = mock_put.call_args[1] assert call_kwargs["verify"] is True def test_ssl_verification_disabled(self, sample_text_file): """Test that SSL verification can be disabled.""" 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"): 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 = False mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response upload_to_webdav.apply(args=[sample_text_file]).get() # Verify SSL verification was disabled call_kwargs = mock_put.call_args[1] assert call_kwargs["verify"] is False def test_authentication_credentials(self, sample_text_file): """Test that authentication credentials are properly passed.""" 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"): mock_settings.webdav_url = "https://webdav.example.com/" mock_settings.webdav_username = "custom_user" mock_settings.webdav_password = "custom_password123" mock_settings.webdav_folder = "uploads" mock_settings.webdav_verify_ssl = True mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response upload_to_webdav.apply(args=[sample_text_file]).get() # Verify correct credentials were used call_kwargs = mock_put.call_args[1] assert call_kwargs["auth"] == ("custom_user", "custom_password123") def test_logging_on_success(self, sample_text_file): """Test that progress is logged on successful upload.""" 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") as mock_log: 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 42}).get() # Verify logging calls assert mock_log.call_count >= 2 # At least in_progress and success # Check for success log success_calls = [ call for call in mock_log.call_args_list if call[0][2] == "success" ] assert len(success_calls) >= 1 def test_logging_on_failure(self, sample_text_file): """Test that progress is logged on failed upload.""" 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") as mock_log: 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 403 mock_response.text = "Forbidden" mock_put.return_value = mock_response with pytest.raises(Exception): upload_to_webdav.apply(args=[sample_text_file], kwargs={"file_id": 42}).get() # Check for failure log failure_calls = [ call for call in mock_log.call_args_list if call[0][2] == "failure" ] assert len(failure_calls) >= 1 def test_file_content_uploaded(self, sample_text_file): """Test that file content is actually read and uploaded.""" 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"): 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response upload_to_webdav.apply(args=[sample_text_file]).get() # Verify data was passed to PUT request call_kwargs = mock_put.call_args[1] assert "data" in call_kwargs # Data should be a file-like object from open() assert call_kwargs["data"] is not None def test_return_value_structure(self, sample_text_file): """Test that return value has correct structure.""" 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"): 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 mock_settings.http_request_timeout = 30 mock_response = Mock() mock_response.status_code = 201 mock_put.return_value = mock_response result = upload_to_webdav.apply(args=[sample_text_file]).get() # Check return value structure assert isinstance(result, dict) assert "status" in result assert "file" in result assert "url" in result assert result["status"] == "Completed" assert result["file"] == sample_text_file def test_module_importable(self): """Test that upload_to_webdav module is importable.""" from app.tasks.upload_to_webdav import upload_to_webdav assert callable(upload_to_webdav) def test_task_has_retry_configuration(self): """Test that the task has retry configuration from BaseTaskWithRetry.""" from app.tasks.upload_to_webdav import upload_to_webdav # BaseTaskWithRetry should provide retry configuration assert hasattr(upload_to_webdav, 'max_retries') # BaseTaskWithRetry configures 3 retries assert upload_to_webdav.max_retries == 3