Distinguish invalid Dag access control from FAB backend failures

providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py

@@ -163,6 +163,10 @@ class FabException(Exception):
     """Custom exception for FAB security manager."""
 
 
+class DagAccessControlValidationError(FabException, ValueError):
+    """The Dag's access control names an invalid role, resource, or permission."""
+
+
 class AzureTenantResolutionError(FabException):
     """Raised when an Azure AD tenant identifier cannot be resolved to a canonical GUID."""
 
@@ -1129,7 +1133,7 @@ def _get_or_create_dag_permission(action_name: str, dag_resource_name: str) -> P
         for rolename, resource_actions_raw in access_control.items():
             role = self.find_role(rolename)
             if not role:
-                raise FabException(
+                raise DagAccessControlValidationError(
                     f"The access_control mapping for DAG '{dag_id}' includes a role named "
                     f"'{rolename}', but that role does not exist"
                 )
@@ -1143,7 +1147,7 @@ def _get_or_create_dag_permission(action_name: str, dag_resource_name: str) -> P
 
             for resource_name, actions in resource_actions.items():
                 if resource_name not in self.RESOURCE_DETAILS_MAP:
-                    raise FabException(
+                    raise DagAccessControlValidationError(
                         f"The access_control map for DAG '{dag_id}' includes the following invalid "
                         f"resource name: '{resource_name}'; "
                         f"The set of valid resource names is: {self.RESOURCE_DETAILS_MAP.keys()}"
@@ -1155,7 +1159,7 @@ def _get_or_create_dag_permission(action_name: str, dag_resource_name: str) -> P
                 invalid_actions = set(actions) - self.RESOURCE_DETAILS_MAP[resource_name]["actions"]
 
                 if invalid_actions:
-                    raise FabException(
+                    raise DagAccessControlValidationError(
                         f"The access_control map for DAG '{dag_resource_name}' includes "
                         f"the following invalid permissions: {invalid_actions}; "
                         f"The set of valid permissions is: {self.RESOURCE_DETAILS_MAP[resource_name]['actions']}"

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

@@ -55,6 +55,7 @@ class _TestBase(DeclarativeBase):
 from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
 from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
 from airflow.providers.fab.auth_manager.security_manager.override import (
+    DagAccessControlValidationError,
     FabAirflowSecurityManagerOverride,
     FabException,
 )
@@ -866,18 +867,20 @@ def test_has_all_dag_access(app, security_manager):
 
 
 def test_access_control_with_non_existent_role(security_manager):
-    with pytest.raises(FabException) as ctx:
+    with pytest.raises(DagAccessControlValidationError) as ctx:
         security_manager._sync_dag_view_permissions(
             dag_id="access-control-test",
             access_control={
                 "this-role-does-not-exist": [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]
             },
         )
     assert "role does not exist" in str(ctx.value)
+    assert isinstance(ctx.value, FabException)
+    assert isinstance(ctx.value, ValueError)
 
 
 def test_access_control_with_non_allowed_resource(security_manager):
-    with pytest.raises(FabException) as ctx:
+    with pytest.raises(DagAccessControlValidationError) as ctx:
         security_manager._sync_dag_view_permissions(
             dag_id="access-control-test",
             access_control={
@@ -908,30 +911,32 @@ def test_all_dag_access_doesnt_give_non_dag_access(app, security_manager):
             )
 
 
-def test_access_control_with_invalid_permission(app, security_manager):
-    invalid_actions = [
+@pytest.mark.parametrize(
+    "action",
+    [
         "can_varimport",  # a real action, but not a member of DAG_ACTIONS
         "can_eat_pudding",  # clearly not a real action
-    ]
+    ],
+)
+def test_access_control_with_invalid_permission(app, security_manager, action):
     username = "LaUser"
     rolename = "team-a"
     with create_user_scope(
         app,
         username=username,
         role_name=rolename,
     ):
-        for action in invalid_actions:
-            with pytest.raises(FabException) as ctx:
-                security_manager._sync_dag_view_permissions(
-                    "access_control_test",
-                    access_control={rolename: {action}},
-                )
-            assert "invalid permissions" in str(ctx.value)
-            if hasattr(permissions, "resource_name"):
-                assert "invalid permission" in str(ctx.value)
-            else:
-                # Test with old airflow running new FAB
-                assert "invalid resource name" in str(ctx.value)
+        with pytest.raises(DagAccessControlValidationError) as ctx:
+            security_manager._sync_dag_view_permissions(
+                "access_control_test",
+                access_control={rolename: {action}},
+            )
+        assert "invalid permissions" in str(ctx.value)
+        if hasattr(permissions, "resource_name"):
+            assert "invalid permission" in str(ctx.value)
+        else:
+            # Test with old airflow running new FAB
+            assert "invalid resource name" in str(ctx.value)
 
 
 def test_access_control_is_set_on_init(