fix(providers/fab): alias is_authorized_dataset to is_authorized_asset (

apache#43469)

* fix(providers/fab): alias is_authorized_dataset to is_authorized_asset

* test(providers/fab): add test case to include deprecation warning

* test(providers/fab): replace record with recwarn

providers/src/airflow/providers/fab/auth_manager/fab_auth_manager.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import argparse
+import warnings
 from functools import cached_property
 from pathlib import Path
 from typing import TYPE_CHECKING, Container
@@ -46,7 +47,7 @@
     GroupCommand,
 )
 from airflow.configuration import conf
-from airflow.exceptions import AirflowConfigException, AirflowException
+from airflow.exceptions import AirflowConfigException, AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import DagModel
 from airflow.providers.fab.auth_manager.cli_commands.definition import (
     DB_COMMANDS,
@@ -271,6 +272,16 @@ def is_authorized_asset(
     ) -> bool:
         return self._is_authorized(method=method, resource_type=RESOURCE_ASSET, user=user)
 
+    def is_authorized_dataset(
+        self, *, method: ResourceMethod, details: AssetDetails | None = None, user: BaseUser | None = None
+    ) -> bool:
+        warnings.warn(
+            "is_authorized_dataset will be renamed as is_authorized_asset in Airflow 3 and will be removed when the minimum Airflow version is set to 3.0 for the fab provider",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
+        return self.is_authorized_asset(method=method, user=user)
+
     def is_authorized_pool(
         self, *, method: ResourceMethod, details: PoolDetails | None = None, user: BaseUser | None = None
     ) -> bool:

providers/tests/fab/auth_manager/views/__init__.py

@@ -15,3 +15,20 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
+from airflow import __version__ as airflow_version
+from airflow.exceptions import AirflowProviderDeprecationWarning
+
+
+def _assert_dataset_deprecation_warning(recwarn) -> None:
+    if airflow_version.startswith("2"):
+        warning = recwarn.pop(AirflowProviderDeprecationWarning)
+        assert warning.category == AirflowProviderDeprecationWarning
+        assert (
+            str(warning.message)
+            == "is_authorized_dataset will be renamed as is_authorized_asset in Airflow 3 and will be removed when the minimum Airflow version is set to 3.0 for the fab provider"
+        )
+
+
+__all__ = ["_assert_dataset_deprecation_warning"]

providers/tests/fab/auth_manager/views/test_permissions.py

@@ -23,6 +23,7 @@
 from airflow.www import app as application
 
 from providers.tests.fab.auth_manager.api_endpoints.api_connexion_utils import create_user, delete_user
+from providers.tests.fab.auth_manager.views import _assert_dataset_deprecation_warning
 from tests_common.test_utils.compat import AIRFLOW_V_2_9_PLUS
 from tests_common.test_utils.www import client_with_login
 
@@ -65,14 +66,20 @@ def client_permissions_reader(fab_app, user_permissions_reader):
 
 @pytest.mark.db_test
 class TestPermissionsView:
-    def test_action_model_view(self, client_permissions_reader):
+    def test_action_model_view(self, client_permissions_reader, recwarn):
         resp = client_permissions_reader.get("/actions/list/", follow_redirects=True)
+
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200
 
-    def test_permission_pair_model_view(self, client_permissions_reader):
+    def test_permission_pair_model_view(self, client_permissions_reader, recwarn):
         resp = client_permissions_reader.get("/permissions/list/", follow_redirects=True)
+
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200
 
-    def test_resource_model_view(self, client_permissions_reader):
+    def test_resource_model_view(self, client_permissions_reader, recwarn):
         resp = client_permissions_reader.get("/resources/list/", follow_redirects=True)
+
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200

providers/tests/fab/auth_manager/views/test_roles_list.py

@@ -23,6 +23,7 @@
 from airflow.www import app as application
 
 from providers.tests.fab.auth_manager.api_endpoints.api_connexion_utils import create_user, delete_user
+from providers.tests.fab.auth_manager.views import _assert_dataset_deprecation_warning
 from tests_common.test_utils.compat import AIRFLOW_V_2_9_PLUS
 from tests_common.test_utils.www import client_with_login
 
@@ -63,6 +64,8 @@ def client_roles_reader(fab_app, user_roles_reader):
 
 @pytest.mark.db_test
 class TestRolesListView:
-    def test_role_model_view(self, client_roles_reader):
+    def test_role_model_view(self, client_roles_reader, recwarn):
         resp = client_roles_reader.get("/roles/list/", follow_redirects=True)
+
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200

providers/tests/fab/auth_manager/views/test_user.py

@@ -23,6 +23,7 @@
 from airflow.www import app as application
 
 from providers.tests.fab.auth_manager.api_endpoints.api_connexion_utils import create_user, delete_user
+from providers.tests.fab.auth_manager.views import _assert_dataset_deprecation_warning
 from tests_common.test_utils.compat import AIRFLOW_V_2_9_PLUS
 from tests_common.test_utils.www import client_with_login
 
@@ -63,6 +64,8 @@ def client_user_reader(fab_app, user_user_reader):
 
 @pytest.mark.db_test
 class TestUserView:
-    def test_user_model_view(self, client_user_reader):
+    def test_user_model_view(self, client_user_reader, recwarn):
         resp = client_user_reader.get("/users/list/", follow_redirects=True)
+
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200

providers/tests/fab/auth_manager/views/test_user_edit.py

@@ -23,6 +23,7 @@
 from airflow.www import app as application
 
 from providers.tests.fab.auth_manager.api_endpoints.api_connexion_utils import create_user, delete_user
+from providers.tests.fab.auth_manager.views import _assert_dataset_deprecation_warning
 from tests_common.test_utils.compat import AIRFLOW_V_2_9_PLUS
 from tests_common.test_utils.www import client_with_login
 
@@ -63,6 +64,7 @@ def client_user_reader(fab_app, user_user_reader):
 
 @pytest.mark.db_test
 class TestUserEditView:
-    def test_reset_my_password_view(self, client_user_reader):
+    def test_reset_my_password_view(self, client_user_reader, recwarn):
         resp = client_user_reader.get("/resetmypassword/form", follow_redirects=True)
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200

providers/tests/fab/auth_manager/views/test_user_stats.py

@@ -23,6 +23,7 @@
 from airflow.www import app as application
 
 from providers.tests.fab.auth_manager.api_endpoints.api_connexion_utils import create_user, delete_user
+from providers.tests.fab.auth_manager.views import _assert_dataset_deprecation_warning
 from tests_common.test_utils.compat import AIRFLOW_V_2_9_PLUS
 from tests_common.test_utils.www import client_with_login
 
@@ -63,6 +64,7 @@ def client_user_stats_reader(fab_app, user_user_stats_reader):
 
 @pytest.mark.db_test
 class TestUserStats:
-    def test_user_stats(self, client_user_stats_reader):
+    def test_user_stats(self, client_user_stats_reader, recwarn):
         resp = client_user_stats_reader.get("/userstatschartview/chart", follow_redirects=True)
+        _assert_dataset_deprecation_warning(recwarn)
         assert resp.status_code == 200