This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ebd339fa620 Test that execute_tasks_new_python_interpreter only
affects the task process (#73149)
ebd339fa620 is described below
commit ebd339fa6202a8d100c18f0fc7b21876c992e515
Author: Kaxil Naik <[email protected]>
AuthorDate: Wed Sep 16 11:59:47 2026 +0100
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.
---
.../tests/unit/dag_processing/test_processor.py | 26 ++++++++++++++++++++++
airflow-core/tests/unit/jobs/test_triggerer_job.py | 16 +++++++++++++
.../execution_time/test_callback_supervisor.py | 13 +++++++++++
3 files changed, 55 insertions(+)
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index f54b82fef82..d51118f2b61 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/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
[email protected]("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
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 8231dc2cbec..438bc5f68ec 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/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
[email protected]("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):
diff --git a/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
index 7ccbb8c0dfa..8acd3feeddf 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_callback_supervisor.py
@@ -48,6 +48,8 @@ from airflow.sdk.execution_time.comms import (
_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 @@ class TestCallbackSubprocessStart:
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."""
