Resolve AzureFileShareToGCSOperator directory_name alias after rendering

providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py

@@ -69,8 +69,8 @@ class AzureFileShareToGCSOperator(BaseOperator):
     :param return_gcs_uris: If True, return a list of GCS URIs. If False (default), return the legacy
         list of Azure FileShare filenames and emit a deprecation warning.
 
-    Note that ``share_name``, ``directory_path``, ``prefix``, and ``dest_gcs`` are
-    templated, so you can use variables in them if you wish.
+    Note that ``share_name``, ``directory_name``, ``directory_path``, ``prefix``, and
+    ``dest_gcs`` are templated, so you can use variables in them if you wish.
     """
 
     template_fields: Sequence[str] = (
@@ -102,8 +102,11 @@ def __init__(
         self.share_name = share_name
         self.directory_path = directory_path
         self.directory_name = directory_name
-        if self.directory_path is None and self.directory_name is not None:
-            self.directory_path = self.directory_name
+        # The deprecated directory_name->directory_path alias is decided here on the un-rendered
+        # values (native rendering can turn a supplied directory_path into None) and applied in
+        # execute(), which runs for mapped tasks too — render_template_fields overrides do not.
+        self._use_directory_name = directory_path is None and directory_name is not None
+        if self._use_directory_name:
             warnings.warn(
                 "Use 'directory_path' instead of 'directory_name'. Planned removal date: October 5, 2026.",
                 AirflowProviderDeprecationWarning,
@@ -139,6 +142,8 @@ def _check_inputs(self) -> None:
             )
 
     def execute(self, context: Context) -> list[str]:
+        if self._use_directory_name:
+            self.directory_path = self.directory_name
         self._check_inputs()
         azure_fileshare_hook = AzureFileShareHook(
             share_name=self.share_name,

providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py

@@ -16,14 +16,18 @@
 # under the License.
 from __future__ import annotations
 
+import datetime
 from unittest import mock
 
 import pytest
 
+from airflow import DAG
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs import AzureFileShareToGCSOperator
 
 pytestmark = pytest.mark.filterwarnings("ignore::FutureWarning")
 
+DEFAULT_DATE = datetime.datetime(2024, 1, 1)
 TASK_ID = "test-azure-fileshare-to-gcs"
 AZURE_FILESHARE_SHARE = "test-share"
 AZURE_FILESHARE_DIRECTORY_PATH = "/path/to/dir"
@@ -56,6 +60,104 @@ def test_init(self):
         assert operator.dest_gcs == GCS_PATH_PREFIX
         assert operator.google_impersonation_chain == IMPERSONATION_CHAIN
 
+    @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook")
+    @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.GCSHook")
+    def test_directory_name_alias_uses_rendered_value(self, gcs_mock_hook, azure_fileshare_mock_hook):
+        """A templated directory_name is aliased to directory_path using its rendered value, not the Jinja expression."""
+        dag = DAG("test_azure_fileshare_alias", schedule=None, start_date=DEFAULT_DATE)
+        with pytest.warns(AirflowProviderDeprecationWarning, match="Use 'directory_path' instead"):
+            operator = AzureFileShareToGCSOperator(
+                task_id=TASK_ID,
+                share_name=AZURE_FILESHARE_SHARE,
+                directory_name="{{ params.legacy_dir }}",
+                params={"legacy_dir": "rendered/dir"},
+                azure_fileshare_conn_id=AZURE_FILESHARE_CONN_ID,
+                gcp_conn_id=GCS_CONN_ID,
+                dest_gcs=GCS_PATH_PREFIX,
+                return_gcs_uris=True,
+                dag=dag,
+            )
+        assert operator.directory_path is None
+
+        operator.render_template_fields({"params": {"legacy_dir": "rendered/dir"}})
+        assert operator.directory_path is None
+
+        azure_fileshare_mock_hook.return_value.list_files.return_value = MOCK_FILES
+        operator.execute(None)
+        azure_fileshare_mock_hook.assert_any_call(
+            share_name=AZURE_FILESHARE_SHARE,
+            azure_fileshare_conn_id=AZURE_FILESHARE_CONN_ID,
+            directory_path="rendered/dir",
+        )
+
+    @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook")
+    @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.GCSHook")
+    def test_native_directory_path_rendering_to_none_is_not_aliased(
+        self, gcs_mock_hook, azure_fileshare_mock_hook
+    ):
+        """
+        Behaviour-preservation guard: the alias decision must come from the un-rendered values.
+
+        With render_template_as_native_obj an explicitly supplied directory_path can render to None,
+        and re-deciding the alias after rendering would wrongly fall back to the deprecated
+        directory_name. Deciding in __init__ (pre-render) keeps this case an explicit error.
+        """
+        dag = DAG(
+            "test_azure_fileshare_native",
+            schedule=None,
+            start_date=DEFAULT_DATE,
+            render_template_as_native_obj=True,
+        )
+        operator = AzureFileShareToGCSOperator(
+            task_id=TASK_ID,
+            share_name=AZURE_FILESHARE_SHARE,
+            directory_path="{{ params.p }}",
+            directory_name="legacy",
+            params={"p": None},
+            azure_fileshare_conn_id=AZURE_FILESHARE_CONN_ID,
+            gcp_conn_id=GCS_CONN_ID,
+            dest_gcs=GCS_PATH_PREFIX,
+            return_gcs_uris=True,
+            dag=dag,
+        )
+
+        operator.render_template_fields({"params": {"p": None}})
+        assert operator.directory_path is None
+
+        # The deprecated alias must not silently substitute directory_name; a genuinely unset
+        # directory surfaces as the operator's own error instead of listing the wrong directory.
+        azure_fileshare_mock_hook.return_value.list_files.return_value = MOCK_FILES
+        with pytest.raises(RuntimeError, match="directory_name must be set"):
+            operator.execute(None)
+
+    @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook")
+    @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.GCSHook")
+    def test_mapped_task_aliases_directory_name(self, gcs_mock_hook, azure_fileshare_mock_hook):
+        """
+        Mapped tasks reach execute() through MappedOperator.unmap(), never through
+        render_template_fields overrides, so the alias must not depend on one.
+        """
+        mapped = AzureFileShareToGCSOperator.partial(
+            task_id=TASK_ID,
+            share_name=AZURE_FILESHARE_SHARE,
+            directory_name=AZURE_FILESHARE_DIRECTORY_PATH,
+            azure_fileshare_conn_id=AZURE_FILESHARE_CONN_ID,
+            gcp_conn_id=GCS_CONN_ID,
+            dest_gcs=GCS_PATH_PREFIX,
+            return_gcs_uris=True,
+        ).expand(prefix=["sub/a/", "sub/b/"])
+
+        with pytest.warns(AirflowProviderDeprecationWarning, match="Use 'directory_path' instead"):
+            operator = mapped.unmap({"prefix": "sub/a/"})
+
+        azure_fileshare_mock_hook.return_value.list_files.return_value = MOCK_FILES
+        operator.execute(None)
+        azure_fileshare_mock_hook.assert_any_call(
+            share_name=AZURE_FILESHARE_SHARE,
+            azure_fileshare_conn_id=AZURE_FILESHARE_CONN_ID,
+            directory_path=AZURE_FILESHARE_DIRECTORY_PATH,
+        )
+
     @pytest.mark.parametrize("return_gcs_uris", [True, False])
     @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook")
     @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.GCSHook")

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -18,7 +18,6 @@ providers/google/src/airflow/providers/google/cloud/operators/functions.py::Clou
 providers/google/src/airflow/providers/google/cloud/operators/gcs.py::GCSFileTransformOperator
 providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py::BigQueryDataTransferServiceTransferRunSensor
 providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::CloudComposerExternalTaskSensor
-providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py::AzureFileShareToGCSOperator
 providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py::GCSToBigQueryOperator
 providers/google/src/airflow/providers/google/marketing_platform/operators/campaign_manager.py::GoogleCampaignManagerDeleteReportOperator
 providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py::GCSToAzureBlobStorageOperator