Correctly restore upstream_task_ids when deserializing Operators (#8775)

This test exposed a bug in one of the example dags, that wasn't caught
by #6549. That will be a fixed in a separate issue, but it caused the
round-trip tests to fail here

Fixes #8720

(cherry picked from commit 280f1f0)

airflow/serialization/serialized_objects.py

@@ -591,7 +591,7 @@ def deserialize_dag(cls, encoded_dag):
             for task_id in serializable_task.downstream_task_ids:
                 # Bypass set_upstream etc here - it does more than we want
                 # noinspection PyProtectedMember
-                dag.task_dict[task_id]._upstream_task_ids.add(task_id)  # pylint: disable=protected-access
+                dag.task_dict[task_id]._upstream_task_ids.add(serializable_task.task_id)  # noqa: E501 # pylint: disable=protected-access
 
         return dag
 

tests/serialization/test_dag_serialization.py

@@ -358,6 +358,9 @@ def validate_deserialized_task(self, serialized_task, task,):
         assert serialized_task.task_type == task.task_type
         assert set(serialized_task.template_fields) == set(task.template_fields)
 
+        assert serialized_task.upstream_task_ids == task.upstream_task_ids
+        assert serialized_task.downstream_task_ids == task.downstream_task_ids
+
         for field in fields_to_check:
             assert getattr(serialized_task, field) == getattr(task, field), \
                 '{}.{}.{} does not match'.format(task.dag.dag_id, task.task_id, field)