This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 62e3b1a1dce [v3-4-test] Stop a commented executable_resumable from
marking a module durable (#73607) (#75660)
62e3b1a1dce is described below
commit 62e3b1a1dce0cd68fdc8ee250630967e2b0a98f7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 26 14:46:07 2026 +0000
[v3-4-test] Stop a commented executable_resumable from marking a module
durable (#73607) (#75660)
(cherry picked from commit 77da6dac565e9e8968b446a54f9c3734130b6f90)
Co-authored-by: Amogh Desai <[email protected]>
---
dev/registry/extract_parameters.py | 12 ++-
dev/registry/tests/test_extract_parameters.py | 97 ++++++++++++++++++++++
.../tests/test_registry_contract_models.py | 10 ++-
3 files changed, 113 insertions(+), 6 deletions(-)
diff --git a/dev/registry/extract_parameters.py
b/dev/registry/extract_parameters.py
index df9f344ab00..4536a5a48e9 100644
--- a/dev/registry/extract_parameters.py
+++ b/dev/registry/extract_parameters.py
@@ -529,7 +529,9 @@ def _execute_chain_calls_resumable(cls: type, depth: int)
-> bool:
"""Return True if some class along `cls`'s resolved `execute()` chain
calls execute_resumable().
Same walk as `_delegates_execute_to`: a delegating override's own source
may not mention
- `execute_resumable` even though the class it hands off to does.
+ `execute_resumable` even though the class it hands off to does. Only a
real call counts,
+ so the match is made on the parsed tree: a comment, docstring or other
string literal
+ naming the call is not the call.
"""
owner = _find_owner_of_execute(cls.__mro__, 0)
remaining = depth
@@ -537,8 +539,14 @@ def _execute_chain_calls_resumable(cls: type, depth: int)
-> bool:
source = _get_method_source(owner, "execute")
if source is None:
return False
- if "execute_resumable" in source:
+ if any(
+ isinstance(node, ast.Call)
+ and isinstance(node.func, ast.Attribute)
+ and node.func.attr == "execute_resumable"
+ for node in ast.walk(ast.parse(textwrap.dedent(source)))
+ ):
return True
+ source = _strip_comment_lines(source)
if remaining <= 0:
return False
owner = _next_execute_hop(cls, owner, source)
diff --git a/dev/registry/tests/test_extract_parameters.py
b/dev/registry/tests/test_extract_parameters.py
index b751ae64f08..b46bd7aafe1 100644
--- a/dev/registry/tests/test_extract_parameters.py
+++ b/dev/registry/tests/test_extract_parameters.py
@@ -383,6 +383,27 @@ class TestIsDurableCapable:
assert is_durable_capable(CommentedNonDelegatingSubclass,
FakeResumableJobMixin) is False
+ def test_mentioning_execute_resumable_is_not_mistaken_for_a_call(self):
+ """Only a real `execute_resumable()` call counts: a whole-line
comment, a trailing
+ comment or a docstring naming it would otherwise report the subclass
capable."""
+
+ class WholeLineComment(FullyImplementedResumableOperator):
+ def execute(self, context):
+ # this path deliberately skips execute_resumable(context)
+ return None
+
+ class TrailingComment(FullyImplementedResumableOperator):
+ def execute(self, context):
+ return None # deliberately skips execute_resumable(context)
+
+ class Docstring(FullyImplementedResumableOperator):
+ def execute(self, context):
+ """Deliberately skips execute_resumable()."""
+ return None
+
+ for operator_cls in (WholeLineComment, TrailingComment, Docstring):
+ assert is_durable_capable(operator_cls, FakeResumableJobMixin) is
False, operator_cls.__name__
+
def test_non_class_execute_call_does_not_shadow_real_delegation(self):
"""A local name that is not a class in the MRO (e.g. a DB cursor) is
skipped rather
than dead-ending the walk before the real delegation below it."""
@@ -619,6 +640,70 @@ def _wrap_in_foreign_module(func):
return foreign.__dict__["wrapper"]
+class GrandparentCallingOverridableHelper:
+ """Its execute() dispatches to a helper a subclass is free to replace."""
+
+ def execute(self, context):
+ return self.run_job(context)
+
+ def run_job(self, context):
+ return None
+
+
+class MiddleDelegatingToSuperExecute(GrandparentCallingOverridableHelper):
+ def execute(self, context):
+ return super().execute(context)
+
+
+class SubclassOverridingHelperToDefer(MiddleDelegatingToSuperExecute):
+ """Only the subclass's copy of `run_job` defers, and the call to it lives
two hops up
+ the chain. Resolving `self.<name>()` against the class the hop landed on
instead of the
+ class being asked about finds the grandparent's inert copy and reports not
deferrable.
+ """
+
+ def run_job(self, context):
+ return self.defer()
+
+ def defer(self, *args, **kwargs):
+ return None
+
+
+class SharedHelperReachedAtTwoBudgets:
+ """`check_state` is reached twice: once down the long leg with too little
depth budget
+ left to walk out to the defer, and once straight from execute() with
budget to spare.
+ A `visited` key without the remaining depth in it skips the second visit
and reports
+ not deferrable.
+ """
+
+ def execute(self, context):
+ self.first_leg(context)
+ return self.check_state(context)
+
+ def first_leg(self, context):
+ return self.second_leg(context)
+
+ def second_leg(self, context):
+ return self.third_leg(context)
+
+ def third_leg(self, context):
+ return self.check_state(context)
+
+ def check_state(self, context):
+ return self.poll(context)
+
+ def poll(self, context):
+ return self.wait_for_job(context)
+
+ def wait_for_job(self, context):
+ return self.submit(context)
+
+ def submit(self, context):
+ return self.defer()
+
+ def defer(self, *args, **kwargs):
+ return None
+
+
class DeferralCommentOnlyOperator:
"""A comment naming self.defer() is not a deferral, the same way it is not
delegation."""
@@ -707,6 +792,18 @@ class TestSupportsDeferrable:
def test_comment_mentioning_defer_is_not_mistaken_for_deferral(self):
assert supports_deferrable(DeferralCommentOnlyOperator) is False
+ @pytest.mark.parametrize(
+ "operator_cls",
+ [
+ pytest.param(SubclassOverridingHelperToDefer,
id="helper-resolved-against-class-under-test"),
+ pytest.param(SharedHelperReachedAtTwoBudgets,
id="helper-rewalked-with-more-depth-budget"),
+ ],
+ )
+ # SharedHelperReachedAtTwoBudgets only exercises the depth-aware `visited`
key at this budget.
+ @patch("extract_parameters._MAX_DEFERRAL_WALK_DEPTH", 6)
+ def test_helper_call_is_walked_to_the_deferral(self, operator_cls):
+ assert supports_deferrable(operator_cls) is True
+
# ---------------------------------------------------------------------------
# Module dataclass
diff --git a/dev/registry/tests/test_registry_contract_models.py
b/dev/registry/tests/test_registry_contract_models.py
index 0e29b7aa2c4..d5bd9d8f2ec 100644
--- a/dev/registry/tests/test_registry_contract_models.py
+++ b/dev/registry/tests/test_registry_contract_models.py
@@ -99,8 +99,9 @@ def
test_module_contract_accepts_legacy_modules_without_supports_durable_executi
def test_module_contract_preserves_supports_durable_execution_true():
- validated = validate_modules_catalog({"modules":
[_module_payload(supports_durable_execution=True)]})
- assert validated["modules"][0]["supports_durable_execution"] is True
+ payload = _module_payload(supports_durable_execution=True)
+ validate_modules_catalog({"modules": [payload]})
+ assert ModuleContract.model_validate(payload).supports_durable_execution
is True
def test_module_contract_accepts_legacy_modules_without_supports_deferrable():
@@ -110,8 +111,9 @@ def
test_module_contract_accepts_legacy_modules_without_supports_deferrable():
def test_module_contract_preserves_supports_deferrable_true():
- validated = validate_modules_catalog({"modules":
[_module_payload(supports_deferrable=True)]})
- assert validated["modules"][0]["supports_deferrable"] is True
+ payload = _module_payload(supports_deferrable=True)
+ validate_modules_catalog({"modules": [payload]})
+ assert ModuleContract.model_validate(payload).supports_deferrable is True
def test_module_contract_omits_guide_url_for_undocumented_classes():
