From e27691c88cd76a3d776f04af9308a542de007ea7 Mon Sep 17 00:00:00 2001
From: Jarek Potiuk <jarek.potiuk@polidea.com>
Date: Fri, 17 Jan 2020 13:01:54 +0100
Subject: [PATCH] [AIRFLOW-6557] Add test for newly added fields in
 BaseOperator (#7162)

Adding new field in BaseOperator requires some manual updates
in serialization code. This test detects new fields and informs
what should be done in case new field is added.

(cherry picked from commit 5abce471e0690c6b8d06ca25685b0845c5fd270f)
---
 tests/serialization/test_dag_serialization.py | 61 +++++++++++++++++++
 1 file changed, 61 insertions(+)

diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index 558253aabd415..172c0165c533f 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -551,6 +551,67 @@ def test_dag_serialized_fields_with_schema(self):
         dag_params = set(dag_schema.keys()) - ignored_keys
         self.assertEqual(set(DAG.get_serialized_fields()), dag_params)
 
+    def test_no_new_fields_added_to_base_operator(self):
+        """
+        This test verifies that there are no new fields added to BaseOperator. And reminds that
+        tests should be added for it.
+        """
+        base_operator = BaseOperator(task_id="10")
+        fields = base_operator.__dict__
+        self.assertEqual({'_downstream_task_ids': set(),
+                          '_inlets': {"auto": False, "datasets": [], "task_ids": []},
+                          '_log': base_operator.log,
+                          '_outlets': {'datasets': []},
+                          '_schedule_interval': None,
+                          '_upstream_task_ids': set(),
+                          'depends_on_past': False,
+                          'do_xcom_push': True,
+                          'email': None,
+                          'email_on_failure': True,
+                          'email_on_retry': True,
+                          'end_date': None,
+                          'execution_timeout': None,
+                          'executor_config': {},
+                          'inlets': [],
+                          'lineage_data': None,
+                          'max_retry_delay': None,
+                          'on_failure_callback': None,
+                          'on_retry_callback': None,
+                          'on_success_callback': None,
+                          'outlets': [],
+                          'owner': 'airflow',
+                          'params': {},
+                          'pool': 'default_pool',
+                          'priority_weight': 1,
+                          'queue': 'default',
+                          'resources': None,
+                          'retries': 0,
+                          'retry_delay': timedelta(0, 300),
+                          'retry_exponential_backoff': False,
+                          'run_as_user': None,
+                          'sla': None,
+                          'start_date': None,
+                          'subdag': None,
+                          'task_concurrency': None,
+                          'task_id': '10',
+                          'trigger_rule': 'all_success',
+                          'wait_for_downstream': False,
+                          'weight_rule': 'downstream'}, fields,
+                         """
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+
+     ACTION NEEDED! PLEASE READ THIS CAREFULLY AND CORRECT TESTS CAREFULLY
+
+ Some fields were added to the BaseOperator! Please add them to the list above and make sure that
+ you add support for DAG serialization - you should add the field to
+ `airflow/serialization/schema.json` - they should have correct type defined there.
+
+ Note that we do not support versioning yet so you should only add optional fields to BaseOperator.
+
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+                         """
+                         )
+
 
 if __name__ == '__main__':
     unittest.main()