diff --git a/app/tasks/upload_to_google_drive.py b/app/tasks/upload_to_google_drive.py index 7dd87937..a5c4eded 100644 --- a/app/tasks/upload_to_google_drive.py +++ b/app/tasks/upload_to_google_drive.py @@ -144,14 +144,14 @@ def truncate_property_value(key, value, max_bytes=100): return str_value @celery.task(base=BaseTaskWithRetry, bind=True) -def upload_to_google_drive(self, file_path: str, file_id: int = None, include_metadata=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 - file_id: Optional file ID to associate with logs 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}") diff --git a/tests/test_upload_tasks.py b/tests/test_upload_tasks.py index cdf82382..89419ee2 100644 --- a/tests/test_upload_tasks.py +++ b/tests/test_upload_tasks.py @@ -418,9 +418,15 @@ def test_upload_to_webdav_file_not_found(): @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) + """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"), @@ -431,9 +437,11 @@ def test_all_upload_tasks_have_consistent_signature(sample_text_file): (upload_to_email, "app.tasks.upload_to_email"), ] + import inspect + for task, module_path in upload_tasks: - # Verify that the task has the expected signature by inspecting its function - import inspect + # 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())