Fix FAB role deletion foreign key constraint violation

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

@@ -78,7 +78,9 @@
     Resource,
     Role,
     User,
+    assoc_group_role,
     assoc_permission_role,
+    assoc_user_role,
 )
 from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
 from airflow.providers.fab.auth_manager.security_manager.constants import EXISTING_ROLES
@@ -1332,12 +1334,21 @@ def delete_role(self, role_name: str) -> None:
         """
         Delete the given Role.
 
+        Cleans up association table rows (permission-role, user-role, group-role) before
+        deleting the role itself, so that databases whose FK constraints lack CASCADE
+        (e.g. databases migrated from older Airflow versions) do not raise IntegrityError.
+
         :param role_name: the name of a role in the ab_role table
         """
-        role = self.session.scalars(select(Role).where(Role.name == role_name)).first()
+        role = self.session.scalars(select(self.role_model).where(self.role_model.name == role_name)).first()
         if role:
             log.info("Deleting role '%s'", role_name)
-            self.session.execute(delete(Role).where(Role.name == role_name))
+            self.session.execute(
+                delete(assoc_permission_role).where(assoc_permission_role.c.role_id == role.id)
+            )
+            self.session.execute(delete(assoc_user_role).where(assoc_user_role.c.role_id == role.id))
+            self.session.execute(delete(assoc_group_role).where(assoc_group_role.c.role_id == role.id))
+            self.session.execute(delete(self.role_model).where(self.role_model.id == role.id))
             self.session.commit()
         else:
             raise FabException(f"Role named '{role_name}' does not exist")

providers/fab/tests/unit/fab/auth_manager/security_manager/test_override.py

@@ -32,7 +32,10 @@
     Role,
     User,
 )
-from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
+from airflow.providers.fab.auth_manager.security_manager.override import (
+    FabAirflowSecurityManagerOverride,
+    FabException,
+)
 
 
 class EmptySecurityManager(FabAirflowSecurityManagerOverride):
@@ -43,6 +46,41 @@ def __init__(self):
 
 
 class TestFabAirflowSecurityManagerOverride:
+    @mock.patch("airflow.providers.fab.auth_manager.security_manager.override.log")
+    def test_delete_role_cleans_up_associations_before_delete(self, mock_log):
+        """delete_role must remove association rows before the role row itself."""
+        sm = EmptySecurityManager()
+        role = Mock(spec=Role, id=42, name="TestRole")
+
+        mock_session = Mock(spec=Session)
+        mock_scalars = Mock()
+        mock_scalars.first.return_value = role
+        mock_session.scalars.return_value = mock_scalars
+
+        with mock.patch.object(EmptySecurityManager, "session", mock_session):
+            sm.delete_role("TestRole")
+
+        # 4 deletes: permission-role, user-role, group-role, then the role itself
+        assert mock_session.execute.call_count == 4
+        mock_session.commit.assert_called_once()
+        mock_log.info.assert_called_once_with("Deleting role '%s'", "TestRole")
+
+    def test_delete_role_raises_for_missing_role(self):
+        """delete_role must raise FabException when the role does not exist."""
+        sm = EmptySecurityManager()
+
+        mock_session = Mock(spec=Session)
+        mock_scalars = Mock()
+        mock_scalars.first.return_value = None
+        mock_session.scalars.return_value = mock_scalars
+
+        with mock.patch.object(EmptySecurityManager, "session", mock_session):
+            with pytest.raises(FabException, match="Role named 'NoSuchRole' does not exist"):
+                sm.delete_role("NoSuchRole")
+
+        mock_session.execute.assert_not_called()
+        mock_session.commit.assert_not_called()
+
     @mock.patch("airflow.providers.fab.auth_manager.security_manager.override.log")
     def test_add_permission_to_role_ignores_duplicate_from_concurrent_worker(self, mock_log):
         sm = EmptySecurityManager()