Fix: Update unit tests to handle new DAG authorization behavior

Author: Will Yang

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

@@ -564,9 +564,8 @@ def _is_authorized_dag(
             # Check whether the user has permissions to access a specific DAG
             resource_dag_name = permissions.resource_name(details.id, RESOURCE_DAG)
             return self._is_authorized(method=method, resource_type=resource_dag_name, user=user)
-        else:
-            authorized_dags = self.get_authorized_dag_ids(user=user, method=method)
-            return len(authorized_dags) > 0
+        authorized_dags = self.get_authorized_dag_ids(user=user, method=method)
+        return len(authorized_dags) > 0
 
     def _is_authorized_dag_run(
         self,
@@ -591,9 +590,8 @@ def _is_authorized_dag_run(
             # Check whether the user has permissions to access a specific DAG Run permission on a DAG Level
             resource_dag_name = permissions.resource_name(details.id, RESOURCE_DAG_RUN)
             return self._is_authorized(method=method, resource_type=resource_dag_name, user=user)
-        else:
-            authorized_dags = self.get_authorized_dag_ids(user=user, method=method)
-            return len(authorized_dags) > 0
+        authorized_dags = self.get_authorized_dag_ids(user=user, method=method)
+        return len(authorized_dags) > 0
 
     @staticmethod
     def _get_fab_action(method: ResourceMethod) -> str:

providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py

@@ -279,7 +279,7 @@ def test_is_authorized(self, api_name, method, user_permissions, expected_result
                 [(ACTION_CAN_READ, "resource_test")],
                 False,
             ),
-             # With specific DAG permissions but no specific DAG requested
+            # With specific DAG permissions but no specific DAG requested
             (
                 "GET",
                 None,
@@ -301,7 +301,7 @@ def test_is_authorized(self, api_name, method, user_permissions, expected_result
                 None,
                 None,
                 [(ACTION_CAN_READ, "DAG:test_dag_id")],
-                False,
+                True,
             ),
             # With correct method permissions for specific DAG
             (
@@ -418,7 +418,7 @@ def test_is_authorized(self, api_name, method, user_permissions, expected_result
                 DagAccessEntity.RUN,
                 None,
                 [(ACTION_CAN_READ, RESOURCE_DAG_RUN)],
-                False,
+                True,
             ),
             # DAG sub-entity with specific DAG permissions but missing sub-entity permission
             (
@@ -442,21 +442,28 @@ def test_is_authorized(self, api_name, method, user_permissions, expected_result
                 DagAccessEntity.RUN,
                 None,
                 [(ACTION_CAN_READ, "DAG:test_dag_id"), (ACTION_CAN_READ, RESOURCE_DAG_RUN)],
-                False,
+                True,
             ),
         ],
     )
+    @mock.patch.object(FabAuthManager, "get_authorized_dag_ids")
     def test_is_authorized_dag(
         self,
+        mock_get_authorized_dag_ids,
         method,
         dag_access_entity,
         dag_details,
         user_permissions,
         expected_result,
         auth_manager_with_appbuilder,
     ):
+        dag_permissions = [perm[1] for perm in user_permissions if perm[1].startswith("DAG:")]
+        dag_ids = {perm.replace("DAG:", "") for perm in dag_permissions}
+        mock_get_authorized_dag_ids.return_value = dag_ids
+
         user = Mock()
         user.perms = user_permissions
+        user.id = 1
         result = auth_manager_with_appbuilder.is_authorized_dag(
             method=method, access_entity=dag_access_entity, details=dag_details, user=user
         )