Return list of destination URIs from HttpToGCSOperator

providers/google/src/airflow/providers/google/cloud/transfers/http_to_gcs.py

@@ -170,7 +170,8 @@ def gcs_hook(self) -> GCSHook:
         """Create and return an GCSHook."""
         return GCSHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
 
-    def execute(self, context: Context):
+    def execute(self, context: Context) -> list[str]:
+        """Return List of destination URIs (gs://bucket_name/object_name) for uploaded file."""
         self.log.info("Calling HTTP method")
         response = self.http_hook.run(
             endpoint=self.endpoint, data=self.data, headers=self.headers, extra_options=self.extra_options
@@ -191,3 +192,5 @@ def execute(self, context: Context):
             cache_control=self.cache_control,
             user_project=self.user_project,
         )
+
+        return [f"gs://{self.bucket_name}/{self.object_name}"]

providers/google/tests/unit/google/cloud/transfers/test_http_to_gcs.py

@@ -68,7 +68,7 @@ def test_execute_copy_single_file(self, http_hook, gcs_hook):
             gcp_conn_id=GCP_CONN_ID,
             impersonation_chain=IMPERSONATION_CHAIN,
         )
-        task.execute(None)
+        result = task.execute(None)
 
         # GCS
         gcs_hook.assert_called_once_with(gcp_conn_id=GCP_CONN_ID, impersonation_chain=IMPERSONATION_CHAIN)
@@ -100,3 +100,28 @@ def test_execute_copy_single_file(self, http_hook, gcs_hook):
         task.http_hook.run.assert_called_once_with(
             endpoint=ENDPOINT, headers=HEADERS, data=DATA, extra_options=EXTRA_OPTIONS
         )
+
+        # Return value: list of destination GCS URIs (per issue #11323 / PR #61306)
+        expected_uri = f"gs://{TEST_BUCKET}/{DESTINATION_PATH_FILE}"
+        assert result == [expected_uri]
+
+    @mock.patch("airflow.providers.google.cloud.transfers.http_to_gcs.GCSHook")
+    @mock.patch("airflow.providers.google.cloud.transfers.http_to_gcs.HttpHook")
+    def test_execute_returns_destination_uris(self, http_hook, gcs_hook):
+        """Test that execute() returns a list of destination GCS URIs (gs://bucket/object)."""
+        task = HttpToGCSOperator(
+            task_id="http_to_gcs_operator",
+            http_conn_id=HTTP_CONN_ID,
+            endpoint=ENDPOINT,
+            headers=HEADERS,
+            data=DATA,
+            extra_options=EXTRA_OPTIONS,
+            object_name=DESTINATION_PATH_FILE,
+            bucket_name=TEST_BUCKET,
+            gcp_conn_id=GCP_CONN_ID,
+            impersonation_chain=IMPERSONATION_CHAIN,
+        )
+        result = task.execute(None)
+
+        expected_uris = f"gs://{TEST_BUCKET}/{DESTINATION_PATH_FILE}"
+        assert result == [expected_uris]