Bugfix to correct GCSHook being called even when not required with BeamRunPythonPipelineOperator

airflow/providers/apache/beam/operators/beam.py

@@ -364,11 +364,12 @@ def execute(self, context: Context):
 
     def execute_sync(self, context: Context):
         with ExitStack() as exit_stack:
-            gcs_hook = GCSHook(gcp_conn_id=self.gcp_conn_id)
             if self.py_file.lower().startswith("gs://"):
+                gcs_hook = GCSHook(gcp_conn_id=self.gcp_conn_id)
                 tmp_gcs_file = exit_stack.enter_context(gcs_hook.provide_file(object_url=self.py_file))
                 self.py_file = tmp_gcs_file.name
             if self.snake_case_pipeline_options.get("requirements_file", "").startswith("gs://"):
+                gcs_hook = GCSHook(gcp_conn_id=self.gcp_conn_id)    
                 tmp_req_file = exit_stack.enter_context(
                     gcs_hook.provide_file(object_url=self.snake_case_pipeline_options["requirements_file"])
                 )

tests/providers/apache/beam/operators/test_beam.py

@@ -256,6 +256,59 @@ def test_on_kill_direct_runner(self, _, dataflow_mock, __):
         op.on_kill()
         dataflow_cancel_job.assert_not_called()
 
+    @mock.patch(BEAM_OPERATOR_PATH.format("BeamHook"))
+    @mock.patch(BEAM_OPERATOR_PATH.format("GCSHook"))
+    def test_execute_gcs_hook_called_only_with_gs_prefix(self, mock_gcs_hook, _):
+        local_test_op_args = {
+            "task_id": TASK_ID,
+            "py_file": 'local_file.py',
+            "py_options": ['-m'],
+            "default_pipeline_options": {
+                "project": TEST_PROJECT,
+                'requirements_file': 'local_requirements.txt'
+            },
+            "pipeline_options": {
+                "output": 'test_local/output', "labels": {"foo": "bar"}
+            }
+        }
+
+        """
+        Test that execute method does not call GCSHook when neither py_file nor requirements_file
+        starts with 'gs://'.
+        """
+        test_kwargs_local = copy.deepcopy(local_test_op_args)
+        op = BeamRunPythonPipelineOperator(**test_kwargs_local)
+        context_mock = mock.MagicMock()
+
+        op.execute(context_mock)
+        mock_gcs_hook.assert_not_called()
+        mock_gcs_hook.reset_mock()
+
+        """
+        Test that execute method calls GCSHook when only 'py_file' starts with 'gs://'.
+        """
+        test_kwargs_local = copy.deepcopy(local_test_op_args)
+        test_kwargs_local['py_file'] = 'gs://gcs_file.py'
+        op = BeamRunPythonPipelineOperator(**test_kwargs_local)
+        context_mock = mock.MagicMock()
+        op.execute(context_mock)
+        mock_gcs_hook.assert_called_once()
+        mock_gcs_hook.reset_mock()
+
+        """
+        Test that execute calls GCSHook when only pipeline_options 'requirements_file' starts with 'gs://'.
+        Note: "pipeline_options" is merged with and overrides keys in "default_pipeline_options" when 
+              BeamRunPythonPipelineOperator is instantiated, so testing 'requirements_file' specified 
+              in "pipeline_options"
+        """
+        test_kwargs_local = copy.deepcopy(local_test_op_args)
+        test_kwargs_local['pipeline_options']['requirements_file'] = 'gs://gcs_requirements.txt'
+        op = BeamRunPythonPipelineOperator(**test_kwargs_local)
+        context_mock = mock.MagicMock()
+        op.execute(context_mock)
+        mock_gcs_hook.assert_called_once()
+        mock_gcs_hook.reset_mock()
+
 
 class TestBeamRunJavaPipelineOperator:
     @pytest.fixture(autouse=True)