Test that execute_tasks_new_python_interpreter only affects the task process (#73149)

The existing fork+exec wiring tests patch the platform gate and check it flows
through to WatchedSubprocess.start; none set the option itself. These tests set
[core] execute_tasks_new_python_interpreter and assert the Dag processor, the
triggerer and the callback supervisor still start their child with a bare fork.

Author: kaxil

airflow-core/tests/unit/dag_processing/test_processor.py

@@ -512,6 +512,32 @@ def test_start_opts_into_fork_exec(monkeypatch, mocker, platform_uses_exec, targ
     assert base_start.call_args.kwargs["use_exec"] is expected_use_exec
 
 
+@pytest.mark.parametrize("option_value", ["True", "False"])
+def test_start_ignores_execute_tasks_new_python_interpreter(mocker, option_value):
+    """
+    ``[core] execute_tasks_new_python_interpreter`` is a task-process opt-in and must not reach the
+    parsing child, which only follows the platform gate (pinned to bare fork by ``_force_bare_fork``).
+    """
+    base_start = mocker.patch(
+        "airflow.sdk.execution_time.supervisor.WatchedSubprocess.start", return_value=MagicMock()
+    )
+    mocker.patch("airflow.dag_processing.processor._pre_import_airflow_modules")
+
+    with conf_vars({("core", "execute_tasks_new_python_interpreter"): option_value}):
+        DagFileProcessorProcess.start(
+            path="some_dag.py",
+            bundle_path=pathlib.Path("/tmp/bundle"),
+            bundle_name="testing",
+            dag_file_rel_path="some_dag.py",
+            callbacks=[],
+            client=MagicMock(spec=Client),
+            target=_parse_file_entrypoint,
+            logger=MagicMock(),
+        )
+
+    assert base_start.call_args.kwargs["use_exec"] is False
+
+
 def write_dag_in_a_fn_to_file(fn: Callable[[], None], folder: pathlib.Path) -> pathlib.Path:
     # Create the dag in a fn, and use inspect.getsource to write it to a file so that
     # a) the test dag is directly viewable here in the tests

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -298,6 +298,22 @@ def test_start_opts_into_fork_exec(monkeypatch, mocker, platform_uses_exec):
     assert base_start.call_args.kwargs["target"] == TriggerRunnerSupervisor.run_in_process
 
 
+@pytest.mark.parametrize("option_value", ["True", "False"])
+def test_start_ignores_execute_tasks_new_python_interpreter(mocker, option_value):
+    """
+    ``[core] execute_tasks_new_python_interpreter`` is a task-process opt-in and must not reach the
+    runner child, which only follows the platform gate (pinned to bare fork by ``_force_bare_fork``).
+    """
+    base_start = mocker.patch(
+        "airflow.sdk.execution_time.supervisor.WatchedSubprocess.start", return_value=MagicMock()
+    )
+
+    with conf_vars({("core", "execute_tasks_new_python_interpreter"): option_value}):
+        TriggerRunnerSupervisor.start(job=Job(id=999), capacity=10)
+
+    assert base_start.call_args.kwargs["use_exec"] is False
+
+
 @pytest.fixture
 def supervisor_builder(mocker, session):
     def builder(job=None):

task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py

@@ -48,6 +48,8 @@
     _RequestFrame,
 )
 
+from tests_common.test_utils.config import conf_vars
+
 
 def callback_no_args():
     """A simple callback that takes no arguments."""
@@ -553,6 +555,17 @@ def test_callback_supervisor_should_exit_on_error(self, base_start_kwargs):
 
         assert exc_info.value.code == 1
 
+    @pytest.mark.parametrize("option_value", ["True", "False"])
+    def test_start_keeps_bare_fork_regardless_of_exec_option(self, base_start_kwargs, option_value):
+        """
+        ``[core] execute_tasks_new_python_interpreter`` is a task-process opt-in and must not reach the
+        callback child: its target is a closure, which only a bare fork can run.
+        """
+        with conf_vars({("core", "execute_tasks_new_python_interpreter"): option_value}):
+            CallbackSubprocess.start(**base_start_kwargs)
+
+        assert self.mock_super_start.call_args.kwargs.get("use_exec", False) is False
+
 
 class TestSuperviseCallbackExchangesTokenFirst:
     """No callback code may run until the single-use token has been exchanged."""