Stop deserializing pickle when enable_xcom_pickling is False

airflow/models/xcom.py

@@ -685,10 +685,8 @@ def _deserialize_value(result: XCom, orm: bool) -> Any:
             except pickle.UnpicklingError:
                 return json.loads(result.value.decode("UTF-8"), cls=XComDecoder, object_hook=object_hook)
         else:
-            try:
-                return json.loads(result.value.decode("UTF-8"), cls=XComDecoder, object_hook=object_hook)
-            except (json.JSONDecodeError, UnicodeDecodeError):
-                return pickle.loads(result.value)
+            # Since xcom_pickling is disabled, we should only try to deserialize with JSON
+            return json.loads(result.value.decode("UTF-8"), cls=XComDecoder, object_hook=object_hook)
 
     @staticmethod
     def deserialize_value(result: XCom) -> Any:

tests/api_connexion/schemas/test_xcom_schema.py

@@ -30,6 +30,7 @@
 from airflow.models import DagRun, XCom
 from airflow.utils.dates import parse_execution_date
 from airflow.utils.session import create_session
+from tests.test_utils.config import conf_vars
 
 pytestmark = pytest.mark.db_test
 
@@ -188,6 +189,7 @@ class TestXComSchema:
     default_time = "2016-04-02T21:00:00+00:00"
     default_time_parsed = parse_execution_date(default_time)
 
+    @conf_vars({("core", "enable_xcom_pickling"): "True"})
     def test_serialize(self, create_xcom, session):
         create_xcom(
             dag_id="test_dag",
@@ -208,6 +210,7 @@ def test_serialize(self, create_xcom, session):
             "map_index": -1,
         }
 
+    @conf_vars({("core", "enable_xcom_pickling"): "True"})
     def test_deserialize(self):
         xcom_dump = {
             "key": "test_key",

tests/models/test_xcom.py

@@ -140,7 +140,7 @@ def test_xcom_deserialize_with_json_to_pickle_switch(self, task_instance, sessio
             ret_value = XCom.get_value(key="xcom_test3", ti_key=ti_key, session=session)
         assert ret_value == {"key": "value"}
 
-    def test_xcom_deserialize_with_pickle_to_json_switch(self, task_instance, session):
+    def test_xcom_deserialize_pickle_when_xcom_pickling_is_disabled(self, task_instance, session):
         with conf_vars({("core", "enable_xcom_pickling"): "True"}):
             XCom.set(
                 key="xcom_test3",
@@ -151,14 +151,14 @@ def test_xcom_deserialize_with_pickle_to_json_switch(self, task_instance, sessio
                 session=session,
             )
         with conf_vars({("core", "enable_xcom_pickling"): "False"}):
-            ret_value = XCom.get_one(
-                key="xcom_test3",
-                dag_id=task_instance.dag_id,
-                task_id=task_instance.task_id,
-                run_id=task_instance.run_id,
-                session=session,
-            )
-        assert ret_value == {"key": "value"}
+            with pytest.raises(UnicodeDecodeError):
+                XCom.get_one(
+                    key="xcom_test3",
+                    dag_id=task_instance.dag_id,
+                    task_id=task_instance.task_id,
+                    run_id=task_instance.run_id,
+                    session=session,
+                )
 
     @conf_vars({("core", "xcom_enable_pickling"): "False"})
     def test_xcom_disable_pickle_type_fail_on_non_json(self, task_instance, session):