potiuk commented on issue #76568: URL: https://github.com/apache/airflow/issues/76568#issuecomment-6740395067
Tested my changes against the RC distributions in Breeze. Every check passed. ✅ ### Setup - **Distributions:** all 64 RC files (32 providers × wheel + sdist) from https://dist.apache.org/repos/dist/dev/airflow/providers/2026-11-03/. All signatures and SHA-512 checksums verified, and all of them rebuilt byte-identical from tag `providers/2026-11-03`. - **Airflow:** `apache-airflow` / `apache-airflow-core` 3.4.0 wheels from PyPI. - **Breeze:** run from a clean clone at the tag, with the files in `dist/`: ``` breeze start-airflow --mount-sources remove --use-distributions-from-dist \ --use-airflow-version wheel --python 3.11 --backend postgres --load-default-connections ``` - **Test Dag:** one Dag, `verify_potiuk_providers_2026_11_03`, triggered from the UI. Each task: - prints the installed provider version, so you can see the RC was picked up; - raises an error if the old behaviour comes back. The Dag has `params={"literal": "{{ dag.dag_id }}"}`. A field that references that param should be rendered once, by the task runner, and so end up holding the literal text `{{ dag.dag_id }}`. The old decorators rendered it a second time, which turned it into the Dag id. ### Results **#76244 Render only the returned value in TaskFlow decorators that template it** - **common.sql 3.0.0rc1:** `@task.sql(conn_id="postgres_default", parameters={"v": "{{ params.literal }}"})` returns `SELECT %(v)s AS v, '{{ ti.task_id }}' AS t`. It ran a real query against Postgres, and the downstream task checked the returned row: - `v == "{{ dag.dag_id }}"`, so `parameters` was not rendered a second time. - `t == "sql_returned_only"`, so the returned SQL was rendered. - In the UI, *Rendered Templates* for that task shows `parameters` as the literal `{"v": "{{ dag.dag_id }}"}`. - **cncf.kubernetes 11.0.0rc1:** `_KubernetesCmdDecoratedOperator` with `env_vars={"MESSAGE": "{{ dag.dag_id }}"}`, whose callable returns `["echo", "{{ ti.task_id }}"]`. `KubernetesPodOperator.execute` was mocked with `autospec`, so no cluster was needed. Task log: ``` cncf.kubernetes 11.0.0: cmds=['echo', 'kubernetes_cmd_returned_only'] env={'MESSAGE': '{{ dag.dag_id }}'} ``` The returned command is rendered, and `env_vars` is left as given. - **standard 2.0.0rc1** (excluded from this wave, tested anyway): `@task.bash(env={"MSG": "{{ params.literal }}"})` returns `echo "$MSG|{{ ti.task_id }}"`. It ran for real, and the output was `{{ dag.dag_id }}|bash_returned_only`. **#76231 fab 3.11.0rc1: Use only the configured auth type when creating tokens in the FAB auth manager** I called `FabAuthManager().create_token()` with an autospecced `FabAirflowSecurityManagerOverride` in place of the real security manager, once per `AUTH_TYPE`: | `AUTH_TYPE` | Result | |---|---| | `AUTH_OAUTH` | No token; neither `auth_user_db` nor `auth_user_ldap` called | | `AUTH_REMOTE_USER` | No token; neither `auth_user_db` nor `auth_user_ldap` called | | `AUTH_LDAP` | Only `auth_user_ldap("admin", "secret", rotate_session_id=False)`. When LDAP refuses, there is no fallback to the DB password check | | `AUTH_DB` | Only `auth_user_db("admin", "secret", rotate_session_id=False)`; LDAP is never consulted | | `AUTH_DB` with a NUL-only password, or with a non-string username | `ValueError` | **#76222 microsoft.azure 15.3.1rc1: Verify S3 TLS certificates by default in S3ToAzureBlobStorageOperator** `S3Hook` was patched with `autospec`: | Operator argument | `s3_verify` | `S3Hook` called with | |---|---|---| | none (default) | `None` | `verify=None`, so TLS certificates are verified | | `s3_verify=False` | `False` | `verify=False` (still passed through) | <details> <summary>Test Dag source</summary> ```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. """ Verify Jarek's changes in the providers wave 2026-11-03 (testing issue #76568). Each task exercises one changed code path in the RC distribution and raises on the old behaviour. The param ``literal`` holds Jinja text. A field that references it renders once (by the task runner) to the literal ``{{ dag.dag_id }}``; the old decorators rendered it a second time into the Dag id. """ from __future__ import annotations import importlib.util from unittest import mock from airflow.sdk import dag, get_current_context, task from airflow.sdk.exceptions import AirflowSkipException LITERAL = "{{ dag.dag_id }}" def require(module: str) -> None: if importlib.util.find_spec(module) is None: raise AirflowSkipException(f"{module} is not installed") def get_provider_version(dist: str) -> str: from importlib.metadata import version return version(f"apache-airflow-providers-{dist}") @dag(schedule=None, catchup=False, params={"literal": LITERAL}, tags=["rc-verification"]) def verify_potiuk_providers_2026_11_03(): @task.sql(conn_id="postgres_default", parameters={"v": "{{ params.literal }}"}) def sql_returned_only(): """#76244 common.sql: only the returned SQL is rendered in execute().""" return "SELECT %(v)s AS v, '{{ ti.task_id }}' AS t" @task def check_sql(rows): print(f"common.sql {get_provider_version('common-sql')} returned rows: {rows}") value, task_id = rows[0] if value != LITERAL: raise RuntimeError(f"parameters were rendered twice: got {value!r}, expected {LITERAL!r}") if task_id != "sql_returned_only": raise RuntimeError(f"returned SQL was not rendered: got {task_id!r}") print("OK #76244 common.sql: parameters rendered once, returned SQL rendered") @task.bash(env={"MSG": "{{ params.literal }}"}) def bash_returned_only(): """#76244 standard: only the returned command is rendered in execute().""" return 'echo "$MSG|{{ ti.task_id }}"' @task def check_bash(output): print(f"standard {get_provider_version('standard')} bash output: {output!r}") expected = f"{LITERAL}|bash_returned_only" if output != expected: raise RuntimeError(f"got {output!r}, expected {expected!r}") print("OK #76244 standard: env rendered once, returned command rendered") @task def kubernetes_cmd_returned_only(): """#76244 cncf.kubernetes: @task.kubernetes_cmd renders only the returned command (pod launch mocked).""" require("airflow.providers.cncf.kubernetes") from airflow.providers.cncf.kubernetes.decorators.kubernetes_cmd import ( _KubernetesCmdDecoratedOperator, ) from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator def hello(): return ["echo", "{{ ti.task_id }}"] context = get_current_context() op = _KubernetesCmdDecoratedOperator( task_id="kubernetes_cmd_probe", python_callable=hello, image="busybox", # Stands for a value the task runner already rendered: it still contains Jinja text. env_vars={"MESSAGE": LITERAL}, ) with mock.patch.object(KubernetesPodOperator, "execute", autospec=True, return_value=None) as launch: op.execute(context) launch.assert_called_once() print(f"cncf.kubernetes {get_provider_version('cncf-kubernetes')}: cmds={op.cmds} env={op.env_vars}") if op.cmds != ["echo", context["ti"].task_id]: raise RuntimeError(f"returned command not rendered: {op.cmds}") env = ( sorted(op.env_vars.items()) if isinstance(op.env_vars, dict) else [(e.name, e.value) for e in op.env_vars] ) if env != [("MESSAGE", LITERAL)]: raise RuntimeError(f"env_vars rendered a second time: {env}") print("OK #76244 cncf.kubernetes: env_vars rendered once, returned command rendered") @task def fab_token_uses_configured_auth_type(): """#76231 fab: create_token checks the password only against the configured AUTH_TYPE.""" require("airflow.providers.fab") from flask_appbuilder.const import AUTH_DB, AUTH_LDAP, AUTH_OAUTH, AUTH_REMOTE_USER from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager from airflow.providers.fab.auth_manager.security_manager.override import ( FabAirflowSecurityManagerOverride, ) print(f"fab {get_provider_version('fab')}") body = {"username": "admin", "password": "secret"} def build_manager(auth_type): manager = FabAuthManager() security_manager = mock.create_autospec(FabAirflowSecurityManagerOverride, instance=True) security_manager.auth_type = auth_type security_manager.auth_user_ldap.return_value = None manager.__dict__["security_manager"] = security_manager return manager, security_manager for auth_type, name in ((AUTH_OAUTH, "AUTH_OAUTH"), (AUTH_REMOTE_USER, "AUTH_REMOTE_USER")): manager, sm = build_manager(auth_type) if manager.create_token(headers={}, body=body) is not None: raise RuntimeError(f"{name}: a token was issued") if sm.auth_user_db.called or sm.auth_user_ldap.called: raise RuntimeError(f"{name}: password was checked against DB/LDAP") print(f"OK {name}: no token, no password check") manager, sm = build_manager(AUTH_LDAP) manager.create_token(headers={}, body=body) if sm.auth_user_db.called: raise RuntimeError("AUTH_LDAP: fell back to the DB password check after LDAP refused") sm.auth_user_ldap.assert_called_once_with("admin", "secret", rotate_session_id=False) print("OK AUTH_LDAP: LDAP only, no DB fallback") manager, sm = build_manager(AUTH_DB) manager.create_token(headers={}, body=body) sm.auth_user_db.assert_called_once_with("admin", "secret", rotate_session_id=False) if sm.auth_user_ldap.called: raise RuntimeError("AUTH_DB: LDAP was consulted") print("OK AUTH_DB: DB only") for bad in ({"username": "admin", "password": "\x00\x00"}, {"username": ["admin"], "password": "x"}): manager, sm = build_manager(AUTH_DB) try: manager.create_token(headers={}, body=bad) except ValueError: print(f"OK rejected {bad!r}") else: raise RuntimeError(f"accepted invalid credentials {bad!r}") print("OK #76231 fab") @task def s3_to_wasb_verifies_tls_by_default(): """#76222 microsoft.azure: s3_verify defaults to None, so the S3 hook verifies certificates.""" require("airflow.providers.amazon") from airflow.providers.microsoft.azure.transfers import s3_to_wasb print(f"microsoft.azure {get_provider_version('microsoft-azure')}") for kwargs, expected in (({}, None), ({"s3_verify": False}, False)): op = s3_to_wasb.S3ToAzureBlobStorageOperator( task_id=f"s3_to_wasb_probe_{len(kwargs)}", s3_bucket="bucket", container_name="container", s3_prefix="prefix", **kwargs, ) if op.s3_verify is not expected: raise RuntimeError(f"s3_verify={op.s3_verify!r}, expected {expected!r}") with mock.patch.object(s3_to_wasb, "S3Hook", autospec=True) as hook: _ = op.s3_hook hook.assert_called_once_with(aws_conn_id=op.aws_conn_id, verify=expected) print(f"OK {kwargs or 'default'}: S3Hook(verify={expected!r})") print("OK #76222 microsoft.azure") check_sql(sql_returned_only()) check_bash(bash_returned_only()) kubernetes_cmd_returned_only() fab_token_uses_configured_auth_type() s3_to_wasb_verifies_tls_by_default() verify_potiuk_providers_2026_11_03() ``` </details> --- Drafted-by: Claude Opus 5.5; reviewed by @potiuk before posting -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
