Fix issues related to access_control={} (#34114)

* allow empty access control on dags

* Add test which demonstrates loss of information during ser/de

* Keep empty value when serializing access_control dict

(cherry picked from commit a61b5e8)

airflow/models/dag.py

@@ -777,7 +777,7 @@ def _upgrade_outdated_dag_access_control(access_control=None):
         will be replaced with 'can_read', in {'role2': {'can_dag_read', 'can_dag_edit'}}
         'can_dag_edit' will be replaced with 'can_edit', etc.
         """
-        if not access_control:
+        if access_control is None:
             return None
         new_perm_mapping = {
             permissions.DEPRECATED_ACTION_CAN_DAG_READ: permissions.ACTION_CAN_READ,

airflow/serialization/serialized_objects.py

@@ -1385,6 +1385,14 @@ def deserialize_dag(cls, encoded_dag: dict[str, Any]) -> SerializedDAG:
 
         return dag
 
+    @classmethod
+    def _is_excluded(cls, var: Any, attrname: str, op: DAGNode):
+        # {} is explicitly different from None in the case of DAG-level access control
+        # and as a result we need to preserve empty dicts through serialization for this field
+        if attrname == "_access_control" and var is not None:
+            return False
+        return super()._is_excluded(var, attrname, op)
+
     @classmethod
     def to_dict(cls, var: Any) -> dict:
         """Stringifies DAGs and operators contained by var and returns a dict of var."""

airflow/www/security.py

@@ -626,7 +626,7 @@ def create_dag_specific_permissions(self) -> None:
                 if (action_name, dag_resource_name) not in perms:
                     self._merge_perm(action_name, dag_resource_name)
 
-            if dag.access_control:
+            if dag.access_control is not None:
                 self.sync_perm_for_dag(dag_resource_name, dag.access_control)
 
     def update_admin_permission(self) -> None:

tests/serialization/test_dag_serialization.py

@@ -427,6 +427,15 @@ def test_dag_serialization_to_timetable(self, timetable, serialized_timetable):
                 print(task["task_id"], k, v)
         assert actual == expected
 
+    def test_dag_serialization_preserves_empty_access_roles(self):
+        """Verify that an explicitly empty access_control dict is preserved."""
+        dag = collect_dags(["airflow/example_dags"])["simple_dag"]
+        dag.access_control = {}
+        serialized_dag = SerializedDAG.to_dict(dag)
+        SerializedDAG.validate_schema(serialized_dag)
+
+        assert serialized_dag["dag"]["_access_control"] == {"__type": "dict", "__var": {}}
+
     def test_dag_serialization_unregistered_custom_timetable(self):
         """Verify serialization fails without timetable registration."""
         dag = get_timetable_based_simple_dag(CustomSerializationTimetable("bar"))