[v2-10-test] Allow fetching XCom with forward slash from the API and escape it in the UI (#45134)

airflow/api_connexion/openapi/v1.yaml

@@ -5733,6 +5733,7 @@ components:
       name: xcom_key
       schema:
         type: string
+        format: path
       required: true
       description: The XCom key.
 

airflow/www/static/js/api/useTaskXcom.ts

@@ -57,14 +57,16 @@ export const useTaskXcomEntry = ({
 }: TaskXcomProps) =>
   useQuery(
     ["taskXcom", dagId, dagRunId, taskId, mapIndex, xcomKey, tryNumber],
-    () =>
-      axios.get<AxiosResponse, API.XCom>(
-        getMetaValue("task_xcom_entry_api")
-          .replace("_DAG_RUN_ID_", dagRunId)
-          .replace("_TASK_ID_", taskId)
-          .replace("_XCOM_KEY_", xcomKey),
-        { params: { map_index: mapIndex, stringify: false } }
-      ),
+    () => {
+      const taskXcomEntryApiUrl = getMetaValue("task_xcom_entry_api")
+        .replace("_DAG_RUN_ID_", dagRunId)
+        .replace("_TASK_ID_", taskId)
+        .replace("_XCOM_KEY_", encodeURIComponent(xcomKey));
+
+      return axios.get<AxiosResponse, API.XCom>(taskXcomEntryApiUrl, {
+        params: { map_index: mapIndex, stringify: false },
+      });
+    },
     {
       enabled: !!xcomKey,
     }

newsfragments/45134.bugfix.rst

@@ -0,0 +1 @@
+(v2 API & UI) Allow fetching XCom with forward slash from the API and escape it in the UI

tests/api_connexion/endpoints/test_xcom_endpoint.py

@@ -272,52 +272,67 @@ def _create_xcom_entry(
         )
 
     @pytest.mark.parametrize(
-        "allowed, query, expected_status_or_value",
+        "allowed, query, expected_status_or_value, key",
         [
             pytest.param(
                 True,
                 "?deserialize=true",
                 "real deserialized TEST_VALUE",
+                "key",
                 id="true",
             ),
             pytest.param(
                 False,
                 "?deserialize=true",
                 400,
+                "key",
                 id="disallowed",
             ),
             pytest.param(
                 True,
                 "?deserialize=false",
                 "orm deserialized TEST_VALUE",
+                "key",
                 id="false-irrelevant",
             ),
             pytest.param(
                 False,
                 "?deserialize=false",
                 "orm deserialized TEST_VALUE",
+                "key",
                 id="false",
             ),
             pytest.param(
                 True,
                 "",
                 "orm deserialized TEST_VALUE",
+                "key",
                 id="default-irrelevant",
             ),
             pytest.param(
                 False,
                 "",
                 "orm deserialized TEST_VALUE",
+                "key",
                 id="default",
             ),
+            pytest.param(
+                False,
+                "",
+                "orm deserialized TEST_VALUE",
+                "key/with/slashes",
+                id="key-with-slashes",
+            ),
         ],
     )
     @conf_vars({("core", "xcom_backend"): "tests.api_connexion.endpoints.test_xcom_endpoint.CustomXCom"})
-    def test_custom_xcom_deserialize(self, allowed: bool, query: str, expected_status_or_value: int | str):
+    def test_custom_xcom_deserialize(
+        self, allowed: bool, query: str, expected_status_or_value: int | str, key: str
+    ):
         XCom = resolve_xcom_backend()
-        self._create_xcom_entry("dag", "run", utcnow(), "task", "key", backend=XCom)
+        self._create_xcom_entry("dag", "run", utcnow(), "task", key, backend=XCom)
 
-        url = f"/api/v1/dags/dag/dagRuns/run/taskInstances/task/xcomEntries/key{query}"
+        url = f"/api/v1/dags/dag/dagRuns/run/taskInstances/task/xcomEntries/{key}{query}"
         with mock.patch("airflow.api_connexion.endpoints.xcom_endpoint.XCom", XCom):
             with conf_vars({("api", "enable_xcom_deserialize_support"): str(allowed)}):
                 response = self.client.get(url, environ_overrides={"REMOTE_USER": "test"})