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 1787ef7cbba Run the deferred AWS hook configuration check as a prek
hook (#73681)
1787ef7cbba is described below
commit 1787ef7cbbab44e2df726387614ecb916e03d797
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Fri Oct 2 04:29:34 2026 +0530
Run the deferred AWS hook configuration check as a prek hook (#73681)
The check is a static sweep over the provider source, but it lived in the
test
suite, so it only ran where the amazon provider and its optional peers were
installed. A missing optional dependency turned it into a collection error
rather than a result, and a contributor editing a call site only learned
about a
dropped parameter once the suite ran.
Keying the hand-built hook exceptions by path rather than by basename
closes a
collision that would silently widen an exception the day two trigger modules
share a name.
Follow-up to apache/airflow#72171, where moving this was agreed in review.
---
providers/.pre-commit-config.yaml | 7 +
.../amazon/aws/test_deferred_hook_configuration.py | 242 +------------
.../ci/prek/check_deferred_hook_configuration.py | 294 ++++++++++++++++
.../prek/test_check_deferred_hook_configuration.py | 373 +++++++++++++++++++++
4 files changed, 688 insertions(+), 228 deletions(-)
diff --git a/providers/.pre-commit-config.yaml
b/providers/.pre-commit-config.yaml
index 66ef2b5be0b..ae234bd20a3 100644
--- a/providers/.pre-commit-config.yaml
+++ b/providers/.pre-commit-config.yaml
@@ -37,6 +37,13 @@ repos:
entry: ../scripts/ci/prek/check_trigger_serialize_init.py
pass_filenames: true
files: ^.*/src/airflow/providers/.*/triggers/[^/]+\.py$
+ - id: check-deferred-hook-configuration
+ name: Check deferred AWS tasks pass their hook configuration
+ description: An operator that defers must hand region_name, verify and
botocore_config to its trigger
+ language: python
+ entry: ../scripts/ci/prek/check_deferred_hook_configuration.py
+ pass_filenames: false
+ files: ^amazon/src/airflow/providers/amazon/aws/.*\.py$
- id: check-migration-patterns
name: Check migration files for anti-patterns and style (MIG001-MIG004)
entry: ../scripts/ci/prek/check_migration_patterns.py
diff --git
a/providers/amazon/tests/unit/amazon/aws/test_deferred_hook_configuration.py
b/providers/amazon/tests/unit/amazon/aws/test_deferred_hook_configuration.py
index c3b8dd6da66..1c92a1f9ab7 100644
--- a/providers/amazon/tests/unit/amazon/aws/test_deferred_hook_configuration.py
+++ b/providers/amazon/tests/unit/amazon/aws/test_deferred_hook_configuration.py
@@ -15,192 +15,35 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+"""
+The half of the deferred hook configuration invariant that needs the classes
themselves.
+
+The static half -- that every ``self.defer()`` hands its hook configuration to
the trigger -- is a
+prek hook, ``scripts/ci/prek/check_deferred_hook_configuration.py``, because
it only needs to parse
+the provider. This one has to import it, so it stays a test.
+"""
+
from __future__ import annotations
-import ast
import importlib
import inspect
import pkgutil
-from collections.abc import Iterator
-from pathlib import Path
import pytest
-import airflow.providers.amazon.aws as aws_module
import airflow.providers.amazon.aws.triggers as triggers_module
from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
-AWS_ROOT = Path(inspect.getfile(aws_module)).parent
-HOOK_CONFIGURATION = ("region_name", "verify", "botocore_config")
-
-# A deferrable task builds its hook twice: once in the worker, once in the
triggerer. Unless the
-# operator hands its hook configuration to the trigger, the triggerer silently
falls back to boto3
-# defaults -- a different region, different SSL verification, different
timeouts.
-UNCONFIGURABLE_TRIGGERS = frozenset(
- {
- # Not an AwsBaseWaiterTrigger: its hook is addressed by execution
name, and takes no
- # connection parameters at all.
- "SageMakerNotebookJobTrigger",
- # A KubernetesPodTrigger; it reaches the pod through a kubeconfig, not
a boto3 client.
- "EksPodTrigger",
- }
-)
-
-# Sites whose trigger is built elsewhere and only referenced here, so the
class cannot be read off
-# the call. Kept explicit so that a new unreadable site fails the suite
instead of being skipped.
-UNREADABLE_DEFER_SITES = frozenset({("operators/eks.py", "trigger")})
-
-# Services carved out as Contributors Workshop tasks, so their triggers are
still unmigrated. Each
-# entry is one self-contained contribution: widen the trigger's __init__, set
aws_hook_class, pass
-# the parameters at the call site, then delete the entry here. The test
asserts an entry is still
-# needed, so the allowlist cannot outlive the work it tracks.
-PENDING_MIGRATION = frozenset(
- {
- ("sensors/batch.py", "BatchJobTrigger"),
- }
-)
-
-
-def read_trigger_name(call: ast.Call) -> str | None:
- """The trigger class a construction names, or ``None`` when the callee
cannot be read."""
- if isinstance(call.func, ast.Name):
- return call.func.id
- if isinstance(call.func, ast.Attribute):
- return call.func.attr
- return None
-
-
-def trigger_constructions(expr: ast.expr) -> list[tuple[ast.Call, str]] | None:
- """
- Resolve a ``trigger=`` expression to the constructions it can evaluate to,
each with its name.
-
- ``None`` means the expression cannot be read statically. Returning that
rather than an empty
- list is what keeps a site from disappearing: a bare reference, a
subscript, or a conditional
- with one unreadable branch all have to be acknowledged in
``UNREADABLE_DEFER_SITES`` instead of
- quietly contributing nothing to the sweep.
- """
- if isinstance(expr, ast.Call):
- # A construction whose callee cannot be named is no more readable than
a bare reference:
- # the allowlists key on the class name, so an unnamed one could never
match them.
- name = read_trigger_name(expr)
- return [(expr, name)] if name is not None else None
- if isinstance(expr, ast.IfExp):
- constructions: list[tuple[ast.Call, str]] = []
- for branch in (expr.body, expr.orelse):
- resolved = trigger_constructions(branch)
- if resolved is None:
- return None
- constructions.extend(resolved)
- return constructions
- return None
-
-
-def walk_defer_sites() -> Iterator[tuple[Path, ast.expr]]:
- """Yield the ``trigger=`` expression of every ``self.defer(...)`` in the
provider."""
- # Every file, not just operators/ and sensors/: ``defer`` is a
BaseOperator method, so a site
- # can appear anywhere, and a directory filter would drop a nested
subpackage without saying so.
- for path in sorted(AWS_ROOT.rglob("*.py")):
- for node in ast.walk(ast.parse(path.read_text())):
- if not isinstance(node, ast.Call):
- continue
- func = node.func
- if not (
- isinstance(func, ast.Attribute)
- and func.attr == "defer"
- and isinstance(func.value, ast.Name)
- and func.value.id == "self"
- ):
- continue
- trigger = next((kw.value for kw in node.keywords if kw.arg ==
"trigger"), None)
- if trigger is not None:
- yield path, trigger
-
-
-def find_defer_sites() -> list[tuple[str, int, str, list[str]]]:
- """Collect every ``self.defer(trigger=SomeTrigger(...))`` in the
provider."""
- sites: list[tuple[str, int, str, list[str]]] = []
- for path, trigger in walk_defer_sites():
- # The trigger may be built inline, or picked between in a conditional
expression, so take
- # every construction the expression can yield rather than assuming a
single call.
- for call, name in trigger_constructions(trigger) or ():
- if name in UNCONFIGURABLE_TRIGGERS:
- continue
- passed = {kw.arg for kw in call.keywords if kw.arg}
- sites.append(
- (
- path.relative_to(AWS_ROOT).as_posix(),
- call.lineno,
- name,
- [p for p in HOOK_CONFIGURATION if p not in passed],
- )
- )
- return sites
-
-
-def find_unreadable_defer_sites() -> set[tuple[str, str]]:
- """Defer sites whose trigger expression cannot be resolved to the
constructions it yields."""
- return {
- (path.relative_to(AWS_ROOT).as_posix(), ast.unparse(trigger))
- for path, trigger in walk_defer_sites()
- if trigger_constructions(trigger) is None
- }
-
-
-DEFER_SITES = find_defer_sites()
-
-
-def test_defer_sites_are_discovered():
- assert DEFER_SITES, f"no self.defer(trigger=...) calls found under
{AWS_ROOT}"
-
-
-def test_no_defer_site_escapes_the_check():
- """A defer site whose trigger cannot be read statically must be
acknowledged, not skipped."""
- assert find_unreadable_defer_sites() == UNREADABLE_DEFER_SITES
-
-
[email protected](
- ("expression", "expected"),
- [
- pytest.param("SomeTrigger(x=1)", 1, id="call"),
- pytest.param("A() if flag else B()", 2,
id="conditional-both-readable"),
- pytest.param("trigger", None, id="bare-name"),
- pytest.param("self._trigger", None, id="attribute"),
- pytest.param("triggers[kind]", None, id="subscript"),
- pytest.param("A() if flag else self._trigger", None,
id="conditional-one-unreadable"),
- pytest.param("TRIGGERS[kind](x=1)", None, id="unnameable-callee"),
- pytest.param("module.SomeTrigger(x=1)", 1,
id="module-qualified-callee"),
- ],
-)
-def test_unreadable_trigger_expressions_resolve_to_none(expression, expected):
- """Anything the sweep cannot resolve must report None so the site is
forced onto the allowlist."""
- constructions = trigger_constructions(ast.parse(expression,
mode="eval").body)
-
- assert (constructions if constructions is None else len(constructions)) ==
expected
-
-
[email protected](
- ("source", "line", "trigger", "missing"),
- DEFER_SITES,
- ids=[f"{source}:{line}" for source, line, _, _ in DEFER_SITES],
-)
-def test_deferred_trigger_receives_hook_configuration(source, line, trigger,
missing):
- if (source, trigger) in PENDING_MIGRATION:
- assert missing, (
- f"{source}:{line} now passes its hook configuration to {trigger}. "
- f"Drop it from PENDING_MIGRATION so the site stays covered."
- )
- pytest.skip(f"{source} is a Contributors Workshop task; see
PENDING_MIGRATION")
-
- assert not missing, (
- f"{source}:{line} defers to {trigger} without passing {',
'.join(missing)}. "
- f"The triggerer builds its own hook, so anything not passed here is
lost."
- )
-
def find_waiter_triggers() -> list[type[AwsBaseWaiterTrigger]]:
"""Import every trigger module, then walk the subclass tree."""
for module in pkgutil.iter_modules(triggers_module.__path__):
- importlib.import_module(f"{triggers_module.__name__}.{module.name}")
+ try:
+
importlib.import_module(f"{triggers_module.__name__}.{module.name}")
+ except ImportError:
+ # triggers/eks.py reaches into cncf.kubernetes. Installing amazon
without its optional
+ # peers is supported, and must not turn collection of this module
into an error.
+ continue
found: set[type[AwsBaseWaiterTrigger]] = set()
pending = [AwsBaseWaiterTrigger]
@@ -224,60 +67,3 @@ def test_waiter_trigger_can_build_a_hook(trigger_class):
inspect.signature(trigger_class.aws_hook_class).bind_partial(
aws_conn_id=None, region_name=None, verify=None, config=None
)
-
-
-# A trigger may build a second hook by hand for a side channel -- streaming
CloudWatch logs, most
-# often -- alongside the one ``aws_hook_class`` gives it. That hook talks to
AWS too, so it needs
-# the same configuration; a trigger whose job client verifies TLS while its
log client does not is
-# the same bug in miniature.
-HAND_BUILT_HOOK_EXCEPTIONS = frozenset(
- {
- # Addressed by execution name; takes no connection parameters at all.
- ("sagemaker_unified_studio.py", "SageMakerNotebookHook"),
- # EksPodOperator is a KubernetesPodOperator: it carries no verify or
botocore_config to pass.
- ("eks.py", "EksHook"),
- }
-)
-
-
-def find_hand_built_hooks() -> list[tuple[str, int, str, list[str]]]:
- """Collect every hook constructed directly inside a trigger module."""
- sites = []
- for path in sorted((AWS_ROOT / "triggers").rglob("*.py")):
- for node in ast.walk(ast.parse(path.read_text())):
- if not isinstance(node, ast.Call):
- continue
- # Match a module-qualified ``module.SomeHook(...)`` as well as a
bare name.
- hook = node.func.attr if isinstance(node.func, ast.Attribute) else
getattr(node.func, "id", "")
- if not hook.endswith("Hook"):
- continue
- passed = {keyword.arg for keyword in node.keywords if keyword.arg}
- # AwsGenericHook names the botocore config ``config``.
- if "config" in passed:
- passed.add("botocore_config")
- missing = [name for name in HOOK_CONFIGURATION if name not in
passed]
- sites.append((path.name, node.lineno, hook, missing))
- return sites
-
-
-HAND_BUILT_HOOKS = find_hand_built_hooks()
-
-
[email protected](
- ("source", "line", "hook", "missing"),
- HAND_BUILT_HOOKS,
- ids=[f"{source}:{line}" for source, line, _, _ in HAND_BUILT_HOOKS],
-)
-def test_hand_built_trigger_hook_receives_configuration(source, line, hook,
missing):
- """A hook a trigger builds itself must carry the same configuration as its
main hook."""
- if (source, hook) in HAND_BUILT_HOOK_EXCEPTIONS:
- assert missing, (
- f"{source}:{line} now configures {hook}. "
- f"Drop it from HAND_BUILT_HOOK_EXCEPTIONS so the site stays
covered."
- )
- pytest.skip(f"{source} builds {hook} with nothing to configure")
-
- assert not missing, (
- f"{source}:{line} builds {hook} without {', '.join(missing)}. "
- f"It reaches AWS with boto3 defaults while the trigger's own hook does
not."
- )
diff --git a/scripts/ci/prek/check_deferred_hook_configuration.py
b/scripts/ci/prek/check_deferred_hook_configuration.py
new file mode 100755
index 00000000000..ef28acc2e39
--- /dev/null
+++ b/scripts/ci/prek/check_deferred_hook_configuration.py
@@ -0,0 +1,294 @@
+#!/usr/bin/env python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# /// script
+# requires-python = ">=3.10,<3.11"
+# dependencies = [
+# "rich>=13.6.0",
+# ]
+# ///
+"""
+Check that deferrable Amazon tasks hand their hook configuration to the
trigger they defer to.
+
+A deferrable task builds its hook twice: once in the worker, once in the
triggerer. The triggerer
+half is built from the arguments the operator passed at the ``self.defer()``
call site, so anything
+not passed there is lost, and the deferred half silently reaches AWS with
boto3 defaults: a
+different region, different TLS verification, different timeouts.
+
+Two shapes are checked, both by parsing the provider with ``ast``:
+
+1. Every ``self.defer(trigger=SomeTrigger(...))`` passes ``region_name``,
``verify`` and
+ ``botocore_config``.
+2. Every hook a trigger module builds by hand, alongside the one
``aws_hook_class`` gives it,
+ carries the same configuration. A trigger whose job client verifies TLS
while its log client
+ does not is the same bug in miniature.
+
+A ``trigger=`` expression that cannot be resolved statically is never skipped
silently: it has to
+be acknowledged in ``UNREADABLE_DEFER_SITES`` instead. The allowlists below
are checked in both
+directions, so an entry that is no longer needed fails the check and has to be
removed.
+"""
+
+from __future__ import annotations
+
+import ast
+import sys
+from collections.abc import Iterator
+from pathlib import Path
+
+from common_prek_utils import AIRFLOW_PROVIDERS_ROOT_PATH, console
+
+AWS_ROOT = AIRFLOW_PROVIDERS_ROOT_PATH / "amazon" / "src" / "airflow" /
"providers" / "amazon" / "aws"
+
+HOOK_CONFIGURATION = ("region_name", "verify", "botocore_config")
+
+# Triggers that take no boto3 client configuration at all, so there is nothing
to hand them.
+UNCONFIGURABLE_TRIGGERS = frozenset(
+ {
+ # Not an AwsBaseWaiterTrigger: its hook is addressed by execution name.
+ "SageMakerNotebookJobTrigger",
+ # A KubernetesPodTrigger; it reaches the pod through a kubeconfig, not
a boto3 client.
+ "EksPodTrigger",
+ }
+)
+
+# Sites whose trigger is built elsewhere and only referenced at the call, so
the class cannot be
+# read off it.
+UNREADABLE_DEFER_SITES = frozenset({("operators/eks.py", "trigger")})
+
+# Services carved out as Contributors Workshop tasks, so their triggers are
still unmigrated. Each
+# entry is one self-contained contribution: widen the trigger's __init__, set
aws_hook_class, pass
+# the parameters at the call site, then delete the entry here.
+PENDING_MIGRATION = frozenset(
+ {
+ ("sensors/batch.py", "BatchJobTrigger"),
+ }
+)
+
+# Hand-built hooks that take no connection parameters, keyed by path relative
to the aws package so
+# that two trigger modules sharing a basename cannot share an entry.
+HAND_BUILT_HOOK_EXCEPTIONS = frozenset(
+ {
+ # Addressed by execution name; takes no connection parameters at all.
+ ("triggers/sagemaker_unified_studio.py", "SageMakerNotebookHook"),
+ # EksPodOperator is a KubernetesPodOperator: it carries no verify or
botocore_config to pass.
+ ("triggers/eks.py", "EksHook"),
+ }
+)
+
+
+def read_trigger_name(call: ast.Call) -> str | None:
+ """Return the trigger class a construction names, or ``None`` when the
callee cannot be read."""
+ if isinstance(call.func, ast.Name):
+ return call.func.id
+ if isinstance(call.func, ast.Attribute):
+ return call.func.attr
+ return None
+
+
+def resolve_trigger_constructions(expr: ast.expr) -> list[tuple[ast.Call,
str]] | None:
+ """
+ Resolve a ``trigger=`` expression to the constructions it can evaluate to,
each with its name.
+
+ ``None`` means the expression cannot be read statically. Returning that
rather than an empty
+ list is what keeps a site from disappearing: a bare reference, a
subscript, or a conditional
+ with one unreadable branch all have to be acknowledged in
``UNREADABLE_DEFER_SITES`` instead of
+ quietly contributing nothing to the sweep.
+ """
+ if isinstance(expr, ast.Call):
+ # A construction whose callee cannot be named is no more readable than
a bare reference:
+ # the allowlists key on the class name, so an unnamed one could never
match them.
+ name = read_trigger_name(expr)
+ return [(expr, name)] if name is not None else None
+ if isinstance(expr, ast.IfExp):
+ constructions: list[tuple[ast.Call, str]] = []
+ for branch in (expr.body, expr.orelse):
+ resolved = resolve_trigger_constructions(branch)
+ if resolved is None:
+ return None
+ constructions.extend(resolved)
+ return constructions
+ return None
+
+
+def read_module(path: Path) -> ast.Module | None:
+ """Parse *path*, or return ``None`` when it cannot be parsed."""
+ try:
+ return ast.parse(path.read_text())
+ except SyntaxError:
+ return None
+
+
+def find_unparseable_modules(root: Path) -> list[str]:
+ """Return the modules the sweep could not parse, so they are reported
rather than skipped."""
+ return [
+ path.relative_to(root).as_posix() for path in
sorted(root.rglob("*.py")) if read_module(path) is None
+ ]
+
+
+def walk_defer_sites(root: Path) -> Iterator[tuple[Path, ast.expr]]:
+ """Yield the ``trigger=`` expression of every ``self.defer(...)`` under
*root*."""
+ # Every file, not just operators/ and sensors/: ``defer`` is a
BaseOperator method, so a site
+ # can appear anywhere, and a directory filter would drop a nested
subpackage without saying so.
+ for path in sorted(root.rglob("*.py")):
+ tree = read_module(path)
+ if tree is None:
+ continue
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Call):
+ continue
+ func = node.func
+ if not (
+ isinstance(func, ast.Attribute)
+ and func.attr == "defer"
+ and isinstance(func.value, ast.Name)
+ and func.value.id == "self"
+ ):
+ continue
+ trigger = next((kw.value for kw in node.keywords if kw.arg ==
"trigger"), None)
+ if trigger is not None:
+ yield path, trigger
+
+
+def find_defer_sites(root: Path) -> list[tuple[str, int, str, list[str]]]:
+ """Collect every ``self.defer(trigger=SomeTrigger(...))`` under *root*."""
+ sites: list[tuple[str, int, str, list[str]]] = []
+ for path, trigger in walk_defer_sites(root):
+ # The trigger may be built inline, or picked between in a conditional
expression, so take
+ # every construction the expression can yield rather than assuming a
single call.
+ for call, name in resolve_trigger_constructions(trigger) or ():
+ if name in UNCONFIGURABLE_TRIGGERS:
+ continue
+ passed = {kw.arg for kw in call.keywords if kw.arg}
+ sites.append(
+ (
+ path.relative_to(root).as_posix(),
+ call.lineno,
+ name,
+ [parameter for parameter in HOOK_CONFIGURATION if
parameter not in passed],
+ )
+ )
+ return sites
+
+
+def find_unreadable_defer_sites(root: Path) -> set[tuple[str, str]]:
+ """Return defer sites whose trigger expression cannot be resolved to the
constructions it yields."""
+ return {
+ (path.relative_to(root).as_posix(), ast.unparse(trigger))
+ for path, trigger in walk_defer_sites(root)
+ if resolve_trigger_constructions(trigger) is None
+ }
+
+
+def find_hand_built_hooks(root: Path) -> list[tuple[str, int, str, list[str]]]:
+ """Collect every hook constructed directly inside a trigger module under
*root*."""
+ sites: list[tuple[str, int, str, list[str]]] = []
+ for path in sorted((root / "triggers").rglob("*.py")):
+ tree = read_module(path)
+ if tree is None:
+ continue
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Call):
+ continue
+ # Match a module-qualified ``module.SomeHook(...)`` as well as a
bare name.
+ hook = node.func.attr if isinstance(node.func, ast.Attribute) else
getattr(node.func, "id", "")
+ if not hook.endswith("Hook"):
+ continue
+ passed = {keyword.arg for keyword in node.keywords if keyword.arg}
+ # AwsGenericHook names the botocore config ``config``.
+ if "config" in passed:
+ passed.add("botocore_config")
+ sites.append(
+ (
+ path.relative_to(root).as_posix(),
+ node.lineno,
+ hook,
+ [parameter for parameter in HOOK_CONFIGURATION if
parameter not in passed],
+ )
+ )
+ return sites
+
+
+def collect_errors(root: Path) -> list[str]:
+ """Return one message per violation, empty when the provider is clean."""
+ errors: list[str] = []
+
+ for source in find_unparseable_modules(root):
+ errors.append(f"{source} could not be parsed, so the check cannot see
what it defers to.")
+
+ defer_sites = find_defer_sites(root)
+ if not defer_sites:
+ errors.append(f"No self.defer(trigger=...) calls found under {root};
the check is not looking at it.")
+
+ unreadable = find_unreadable_defer_sites(root)
+ for source, expression in sorted(unreadable - UNREADABLE_DEFER_SITES):
+ errors.append(
+ f"{source} defers to '{expression}', which cannot be read
statically. "
+ f"Add it to UNREADABLE_DEFER_SITES so the site is acknowledged
rather than skipped."
+ )
+ for source, expression in sorted(UNREADABLE_DEFER_SITES - unreadable):
+ errors.append(
+ f"{source} no longer defers to the unreadable '{expression}'. "
+ f"Drop it from UNREADABLE_DEFER_SITES."
+ )
+
+ for source, line, trigger, missing in defer_sites:
+ if (source, trigger) in PENDING_MIGRATION:
+ if not missing:
+ errors.append(
+ f"{source}:{line} now passes its hook configuration to
{trigger}. "
+ f"Drop it from PENDING_MIGRATION so the site stays
covered."
+ )
+ continue
+ if missing:
+ errors.append(
+ f"{source}:{line} defers to {trigger} without passing {',
'.join(missing)}. "
+ f"The triggerer builds its own hook, so anything not passed
here is lost."
+ )
+
+ for source, line, hook, missing in find_hand_built_hooks(root):
+ if (source, hook) in HAND_BUILT_HOOK_EXCEPTIONS:
+ if not missing:
+ errors.append(
+ f"{source}:{line} now configures {hook}. "
+ f"Drop it from HAND_BUILT_HOOK_EXCEPTIONS so the site
stays covered."
+ )
+ continue
+ if missing:
+ errors.append(
+ f"{source}:{line} builds {hook} without {', '.join(missing)}. "
+ f"It reaches AWS with boto3 defaults while the trigger's own
hook does not."
+ )
+
+ return errors
+
+
+def main() -> int:
+ errors = collect_errors(AWS_ROOT)
+ for error in errors:
+ console.print(f"[red]{error}[/]")
+ if errors:
+ console.print(
+ f"\n[red]Found {len(errors)} deferred hook configuration
problem(s).[/]\n"
+ "A deferrable task builds its hook again in the triggerer, from
what the operator "
+ "passed to the trigger. Anything not passed there is lost.\n"
+ )
+ return 1 if errors else 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/tests/ci/prek/test_check_deferred_hook_configuration.py
b/scripts/tests/ci/prek/test_check_deferred_hook_configuration.py
new file mode 100644
index 00000000000..89e7fc90385
--- /dev/null
+++ b/scripts/tests/ci/prek/test_check_deferred_hook_configuration.py
@@ -0,0 +1,373 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import ast
+import textwrap
+from pathlib import Path
+
+import pytest
+from check_deferred_hook_configuration import (
+ collect_errors,
+ find_defer_sites,
+ find_hand_built_hooks,
+ find_unparseable_modules,
+ find_unreadable_defer_sites,
+ resolve_trigger_constructions,
+)
+
+CONFIGURED = "region_name=self.region_name, verify=self.verify,
botocore_config=self.botocore_config"
+
+
[email protected]
+def aws_tree(tmp_path):
+ """Factory fixture: write modules into an aws-package-shaped tree and
return its root."""
+
+ def _write(modules: dict[str, str]) -> Path:
+ root = tmp_path / "aws"
+ root.mkdir(parents=True, exist_ok=True)
+ for relative_path, source in modules.items():
+ path = root / relative_path
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(textwrap.dedent(source))
+ return root
+
+ return _write
+
+
[email protected](autouse=True)
+def empty_allowlists(monkeypatch):
+ """Start every case from empty allowlists; the real entries describe the
real tree, not a fixture."""
+ for name in ("UNREADABLE_DEFER_SITES", "PENDING_MIGRATION",
"HAND_BUILT_HOOK_EXCEPTIONS"):
+ monkeypatch.setattr(f"check_deferred_hook_configuration.{name}",
frozenset())
+
+
[email protected](
+ ("expression", "expected"),
+ [
+ pytest.param("SomeTrigger(x=1)", 1, id="call"),
+ pytest.param("A() if flag else B()", 2,
id="conditional-both-readable"),
+ pytest.param("trigger", None, id="bare-name"),
+ pytest.param("self._trigger", None, id="attribute"),
+ pytest.param("triggers[kind]", None, id="subscript"),
+ pytest.param("A() if flag else self._trigger", None,
id="conditional-one-unreadable"),
+ pytest.param("TRIGGERS[kind](x=1)", None, id="unnameable-callee"),
+ pytest.param("module.SomeTrigger(x=1)", 1,
id="module-qualified-callee"),
+ ],
+)
+def test_unreadable_trigger_expressions_resolve_to_none(expression, expected):
+ """Anything the sweep cannot resolve must report None so the site is
forced onto the allowlist."""
+ constructions = resolve_trigger_constructions(ast.parse(expression,
mode="eval").body)
+
+ assert (constructions if constructions is None else len(constructions)) ==
expected
+
+
+def test_find_defer_sites_reports_the_parameters_not_passed(aws_tree):
+ root = aws_tree(
+ {
+ "operators/glue.py": """
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger(job_name=self.job_name,
region_name=self.region_name))
+ """,
+ }
+ )
+
+ assert find_defer_sites(root) == [("operators/glue.py", 4, "GlueTrigger",
["verify", "botocore_config"])]
+
+
+def test_find_defer_sites_is_satisfied_by_a_fully_configured_site(aws_tree):
+ root = aws_tree(
+ {
+ "operators/glue.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger({CONFIGURED}))
+ """,
+ }
+ )
+
+ assert find_defer_sites(root) == [("operators/glue.py", 4, "GlueTrigger",
[])]
+
+
+def test_find_defer_sites_covers_both_branches_of_a_conditional(aws_tree):
+ root = aws_tree(
+ {
+ "operators/emr.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=A({CONFIGURED}) if self.x else
B(waiter_delay=1))
+ """,
+ }
+ )
+
+ assert [(name, missing) for _, _, name, missing in find_defer_sites(root)]
== [
+ ("A", []),
+ ("B", ["region_name", "verify", "botocore_config"]),
+ ]
+
+
+def test_find_defer_sites_skips_triggers_that_take_no_configuration(aws_tree):
+ root = aws_tree(
+ {
+ "operators/eks.py": """
+ class O:
+ def execute(self, context):
+ self.defer(trigger=EksPodTrigger(pod_name=self.pod_name))
+ """,
+ }
+ )
+
+ assert find_defer_sites(root) == []
+
+
+def test_walk_finds_defer_sites_outside_operators_and_sensors(aws_tree):
+ """``defer`` is a BaseOperator method, so a nested subpackage must not be
skipped."""
+ root = aws_tree(
+ {
+ "nested/deeper/thing.py": """
+ class O:
+ def execute(self, context):
+ self.defer(trigger=NestedTrigger(job_id=1))
+ """,
+ }
+ )
+
+ assert [source for source, _, _, _ in find_defer_sites(root)] ==
["nested/deeper/thing.py"]
+
+
+def test_find_unreadable_defer_sites_records_the_expression(aws_tree):
+ root = aws_tree(
+ {
+ "operators/eks.py": """
+ class O:
+ def execute(self, context):
+ self.defer(trigger=self._trigger)
+ """,
+ }
+ )
+
+ assert find_unreadable_defer_sites(root) == {("operators/eks.py",
"self._trigger")}
+
+
+def test_find_hand_built_hooks_accepts_the_config_spelling(aws_tree):
+ """``AwsGenericHook`` names the botocore config ``config``, not
``botocore_config``."""
+ root = aws_tree(
+ {
+ "triggers/glue.py": """
+ class T:
+ def logs(self):
+ return AwsLogsHook(region_name=self.region_name,
verify=self.verify, config=self.config)
+ """,
+ }
+ )
+
+ assert find_hand_built_hooks(root) == [("triggers/glue.py", 4,
"AwsLogsHook", [])]
+
+
+def test_find_hand_built_hooks_keys_on_the_path_not_the_basename(aws_tree):
+ """Two trigger modules may share a basename, so an exception must not
cover both."""
+ root = aws_tree(
+ {
+ "triggers/eks.py": """
+ class T:
+ def logs(self):
+ return EksHook(aws_conn_id=self.aws_conn_id)
+ """,
+ "triggers/nested/eks.py": """
+ class T:
+ def logs(self):
+ return EksHook(aws_conn_id=self.aws_conn_id)
+ """,
+ }
+ )
+
+ assert sorted(source for source, _, _, _ in find_hand_built_hooks(root))
== [
+ "triggers/eks.py",
+ "triggers/nested/eks.py",
+ ]
+
+
+def test_collect_errors_is_empty_for_a_fully_configured_tree(aws_tree):
+ root = aws_tree(
+ {
+ "operators/glue.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger({CONFIGURED}))
+ """,
+ }
+ )
+
+ assert collect_errors(root) == []
+
+
+def test_collect_errors_flags_a_tree_with_no_defer_sites(aws_tree):
+ root = aws_tree({"operators/glue.py": "class O:\n pass\n"})
+
+ assert "the check is not looking at it" in "\n".join(collect_errors(root))
+
+
+def test_collect_errors_flags_an_unconfigured_defer_site(aws_tree):
+ root = aws_tree(
+ {
+ "operators/glue.py": """
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger(job_name=self.job_name))
+ """,
+ }
+ )
+
+ (error,) = collect_errors(root)
+ assert "defers to GlueTrigger without passing region_name, verify,
botocore_config" in error
+
+
+def test_collect_errors_flags_an_unacknowledged_unreadable_site(aws_tree):
+ root = aws_tree(
+ {
+ "operators/glue.py": """
+ class O:
+ def execute(self, context):
+ self.defer(trigger=self._trigger)
+ """,
+ }
+ )
+
+ errors = "\n".join(collect_errors(root))
+ assert "cannot be read statically" in errors
+ assert "Add it to UNREADABLE_DEFER_SITES" in errors
+
+
+def test_collect_errors_flags_a_stale_unreadable_entry(aws_tree, monkeypatch):
+ monkeypatch.setattr(
+ "check_deferred_hook_configuration.UNREADABLE_DEFER_SITES",
+ frozenset({("operators/glue.py", "self._gone")}),
+ )
+ root = aws_tree(
+ {
+ "operators/glue.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger({CONFIGURED}))
+ """,
+ }
+ )
+
+ assert "Drop it from UNREADABLE_DEFER_SITES" in
"\n".join(collect_errors(root))
+
+
+def
test_collect_errors_flags_a_pending_migration_entry_that_is_now_fixed(aws_tree,
monkeypatch):
+ """The allowlist cannot outlive the work it tracks."""
+ monkeypatch.setattr(
+ "check_deferred_hook_configuration.PENDING_MIGRATION",
+ frozenset({("sensors/batch.py", "BatchJobTrigger")}),
+ )
+ root = aws_tree(
+ {
+ "sensors/batch.py": f"""
+ class S:
+ def execute(self, context):
+ self.defer(trigger=BatchJobTrigger({CONFIGURED}))
+ """,
+ }
+ )
+
+ (error,) = collect_errors(root)
+ assert "Drop it from PENDING_MIGRATION" in error
+
+
+def
test_collect_errors_stays_quiet_for_a_pending_migration_entry_still_needed(aws_tree,
monkeypatch):
+ monkeypatch.setattr(
+ "check_deferred_hook_configuration.PENDING_MIGRATION",
+ frozenset({("sensors/batch.py", "BatchJobTrigger")}),
+ )
+ root = aws_tree(
+ {
+ "sensors/batch.py": """
+ class S:
+ def execute(self, context):
+ self.defer(trigger=BatchJobTrigger(job_id=self.job_id))
+ """,
+ }
+ )
+
+ assert collect_errors(root) == []
+
+
+def test_collect_errors_flags_an_unconfigured_hand_built_hook(aws_tree):
+ root = aws_tree(
+ {
+ "operators/glue.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger({CONFIGURED}))
+ """,
+ "triggers/glue.py": """
+ class T:
+ def logs(self):
+ return AwsLogsHook(aws_conn_id=self.aws_conn_id)
+ """,
+ }
+ )
+
+ (error,) = collect_errors(root)
+ assert "builds AwsLogsHook without region_name, verify, botocore_config"
in error
+
+
+def
test_collect_errors_flags_a_hand_built_exception_that_is_now_fixed(aws_tree,
monkeypatch):
+ monkeypatch.setattr(
+ "check_deferred_hook_configuration.HAND_BUILT_HOOK_EXCEPTIONS",
+ frozenset({("triggers/glue.py", "AwsLogsHook")}),
+ )
+ root = aws_tree(
+ {
+ "operators/glue.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger({CONFIGURED}))
+ """,
+ "triggers/glue.py": f"""
+ class T:
+ def logs(self):
+ return AwsLogsHook({CONFIGURED})
+ """,
+ }
+ )
+
+ (error,) = collect_errors(root)
+ assert "Drop it from HAND_BUILT_HOOK_EXCEPTIONS" in error
+
+
+def test_a_module_that_cannot_be_parsed_is_reported_not_skipped(aws_tree):
+ """prek does not guarantee ruff runs first, so the sweep must survive a
half-edited file."""
+ root = aws_tree(
+ {
+ "operators/broken.py": "class O:\n def execute(self:\n",
+ "operators/glue.py": f"""
+ class O:
+ def execute(self, context):
+ self.defer(trigger=GlueTrigger({CONFIGURED}))
+ """,
+ }
+ )
+
+ assert find_unparseable_modules(root) == ["operators/broken.py"]
+ # The readable file is still swept rather than the whole run dying on its
neighbour.
+ assert [name for _, _, name, _ in find_defer_sites(root)] ==
["GlueTrigger"]
+ (error,) = collect_errors(root)
+ assert "operators/broken.py could not be parsed" in error
