Revert "Disable _sync_dag_perms until we can fix its implementation (a…

…pache#45087)" (apache#45091)

This reverts commit 80c5f35.

airflow/dag_processing/collection.py

@@ -188,17 +188,11 @@ def _serialize_dag_capturing_errors(dag: MaybeSerializedDAG, session: Session, p
             session=session,
             processor_subdir=processor_subdir,
         )
-        # TODO: Currently _sync_dag_perms is FAB specific. We need to re-implement it using the auth manager
-        #  interface. In the short term, we are going to disable it but this should be fixed and re-enabled
-        #  before AF3 is released
-        # if dag_was_updated:
-        #     _sync_dag_perms(dag, session=session)
-        # else:
-        #     # Check and update DagCode
-        if not dag_was_updated:
+        if dag_was_updated:
+            _sync_dag_perms(dag, session=session)
+        else:
+            # Check and update DagCode
             DagCode.update_source_code(dag.dag_id, dag.fileloc)
-
-
         return []
     except OperationalError:
         raise

tests/dag_processing/test_collection.py

@@ -177,7 +177,6 @@ def _dag_import_error_listener(self):
         get_listener_manager().clear()
         dag_import_error_listener.clear()
 
-    @pytest.mark.skip("Skipping until we fix the implementation to not be FAB-specific")
     @pytest.mark.usefixtures("clean_db")  # sync_perms in fab has bad session commit hygiene
     def test_sync_perms_syncs_dag_specific_perms_on_update(
         self, monkeypatch, spy_agency: SpyAgency, session, time_machine
@@ -393,7 +392,6 @@ def test_remove_error_clears_import_error(self, session):
 
         assert import_errors == {"def.py"}
 
-    @pytest.mark.skip("Skipping until we fix the implementation to not be FAB-specific")
     def test_sync_perm_for_dag_with_dict_access_control(self, session, spy_agency: SpyAgency):
         """
         Test that dagbag._sync_perm_for_dag will call ApplessAirflowSecurityManager.sync_perm_for_dag