Fix BashOperator script detection after templating

providers/standard/src/airflow/providers/standard/decorators/bash.py

@@ -88,7 +88,6 @@ def execute(self, context: Context) -> Any:
         if not isinstance(self.bash_command, str) or self.bash_command.strip() == "":
             raise TypeError("The returned value from the TaskFlow callable must be a non-empty string.")
 
-        self._is_inline_cmd = self._is_inline_command(bash_command=self.bash_command)
         self.render_template_fields(context)
         return super().execute(context)
 

providers/standard/src/airflow/providers/standard/operators/bash.py

@@ -178,9 +178,7 @@ def __init__(
         self.cwd = cwd
         self.append_env = append_env
         self.output_processor = output_processor
-        self._is_inline_cmd = None
-        if isinstance(bash_command, str):
-            self._is_inline_cmd = self._is_inline_command(bash_command=bash_command)
+        self._is_inline_cmd: bool | None = None
 
     @cached_property
     def subprocess_hook(self):
@@ -215,6 +213,7 @@ def execute(self, context: Context):
                 raise AirflowException(f"The cwd {self.cwd} must be a directory")
         env = self.get_env(context)
 
+        self._is_inline_cmd = self._is_inline_command(bash_command=cast("str", self.bash_command))
         if self._is_inline_cmd:
             result = self._run_inline_command(bash_path=bash_path, env=env)
         else:

providers/standard/tests/unit/standard/operators/test_bash.py

@@ -33,6 +33,7 @@
     AirflowTaskTimeout,
     timezone,
 )
+from airflow.providers.standard.hooks.subprocess import SubprocessResult
 from airflow.providers.standard.operators.bash import BashOperator
 from airflow.utils.state import State
 from airflow.utils.types import DagRunType
@@ -286,6 +287,22 @@ def test_templated_fields(self, dag_maker, create_task_instance_of_operator):
         assert task.bash_command == 'echo "test_templated_fields_dag"'
         assert task.cwd == Path(__file__).absolute().parent.as_posix()
 
+    @mock.patch.object(BashOperator, "_run_inline_command")
+    @mock.patch.object(
+        BashOperator, "_run_rendered_script_file", return_value=SubprocessResult(exit_code=0, output="ok")
+    )
+    def test_execute_detects_script_after_bash_command_is_rendered(
+        self, mock_run_rendered_script_file, mock_run_inline_command, context
+    ):
+        op = BashOperator(task_id="abc", bash_command="{{ bash_script }}")
+        op.bash_command = "sample.sh"
+
+        result = op.execute(context)
+
+        assert result == "ok"
+        mock_run_rendered_script_file.assert_called_once()
+        mock_run_inline_command.assert_not_called()
+
     @pytest.mark.db_test
     def test_templated_bash_script(self, dag_maker, create_task_instance_of_operator, tmp_path, session):
         """

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -30,4 +30,3 @@ providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py::GCS
 providers/google/src/airflow/providers/google/marketing_platform/operators/campaign_manager.py::GoogleCampaignManagerDeleteReportOperator
 providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py::GCSToAzureBlobStorageOperator
 providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator
-providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator