Allow creation of empty files on GCS

providers/google/src/airflow/providers/google/cloud/hooks/gcs.py

@@ -549,13 +549,13 @@ def _call_with_retry(f: Callable[[], None]) -> None:
         if cache_control:
             blob.cache_control = cache_control
 
-        if filename and data:
+        if filename is not None and data is not None:
             raise ValueError(
                 "'filename' and 'data' parameter provided. Please "
                 "specify a single parameter, either 'filename' for "
                 "local file uploads or 'data' for file content uploads."
             )
-        if filename:
+        if filename is not None:
             if not mime_type:
                 mime_type = "application/octet-stream"
             if gzip:
@@ -575,7 +575,7 @@ def _call_with_retry(f: Callable[[], None]) -> None:
             if gzip:
                 os.remove(filename)
             self.log.info("File %s uploaded to %s in %s bucket", filename, object_name, bucket_name)
-        elif data:
+        elif data is not None:
             if not mime_type:
                 mime_type = "text/plain"
             if gzip:

providers/google/tests/unit/google/cloud/hooks/test_gcs.py

@@ -1199,6 +1199,30 @@ def test_upload_data_str(self, mock_service, testdata_string):
 
         upload_method.assert_called_once_with(testdata_string, content_type="text/plain", timeout=60)
 
+    @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
+    def test_upload_empty_filename(self, mock_service):
+        test_bucket = "test_bucket"
+        test_object = "test_object"
+
+        upload_method = mock_service.return_value.bucket.return_value.blob.return_value.upload_from_filename
+
+        self.gcs_hook.upload(test_bucket, test_object, filename="")
+
+        upload_method.assert_called_once_with(
+            filename="", content_type="application/octet-stream", timeout=60
+        )
+
+    @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
+    def test_upload_empty_data(self, mock_service):
+        test_bucket = "test_bucket"
+        test_object = "test_object"
+
+        upload_method = mock_service.return_value.bucket.return_value.blob.return_value.upload_from_string
+
+        self.gcs_hook.upload(test_bucket, test_object, data="")
+
+        upload_method.assert_called_once_with("", content_type="text/plain", timeout=60)
+
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
     def test_upload_data_bytes(self, mock_service, testdata_bytes):
         test_bucket = "test_bucket"