[v3-3-test] Remove a Dag Run or Task Instance note when its content is cleared (#69033)

airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py

@@ -229,8 +229,10 @@ def patch_dag_run_state(
 
 
 def patch_dag_run_note(*, dag_run: DagRun, note: str | None, user: BaseUser) -> None:
-    """Set or update a Dag Run's note."""
-    if dag_run.dag_run_note is None:
+    """Set, update, or clear a Dag Run's note. An empty note removes it so the run is left without a note."""
+    if note == "":
+        dag_run.dag_run_note = None
+    elif dag_run.dag_run_note is None:
         dag_run.note = (note, user.get_id())
     else:
         dag_run.dag_run_note.content = note

airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py

@@ -313,7 +313,9 @@ def _patch_task_instance_note(
 ) -> None:
     for ti in tis:
         if update_mask or task_instance_body.note is not None:
-            if ti.task_instance_note is None:
+            if task_instance_body.note == "":
+                ti.task_instance_note = None
+            elif ti.task_instance_note is None:
                 ti.note = (task_instance_body.note, user.get_id())
             else:
                 ti.task_instance_note.content = task_instance_body.note

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py

@@ -1504,6 +1504,13 @@ class TestPatchDagRun:
                 {"state": DagRunState.SUCCESS, "note": "updated note"},
                 {"user_id": "test", "content": "updated note"},
             ),
+            (
+                DAG1_ID,
+                DAG1_RUN1_ID,
+                {"note": ""},
+                {"state": DagRunState.SUCCESS, "note": None},
+                None,
+            ),
             (
                 DAG1_ID,
                 DAG1_RUN2_ID,
@@ -1955,11 +1962,16 @@ def test_should_respond_403(self, unauthorized_test_client):
         ("body", "expected_note"),
         [
             ({"dry_run": False, "note": "cleared by test"}, "cleared by test"),
-            ({"dry_run": False, "note": ""}, ""),
+            ({"dry_run": False, "note": ""}, None),
             ({"dry_run": False, "note": None}, "test_note"),
             ({"dry_run": False}, "test_note"),
         ],
-        ids=["set-new-note", "set-empty-note", "explicit-null-leaves-existing", "omit-leaves-existing"],
+        ids=[
+            "set-new-note",
+            "empty-note-removes-existing",
+            "explicit-null-leaves-existing",
+            "omit-leaves-existing",
+        ],
     )
     @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
     def test_clear_dag_run_applies_note(self, test_client, session, body, expected_note):

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -5144,6 +5144,20 @@ def test_set_note_should_respond_200(self, test_client, session):
             session, response_data["task_instances"][0]["id"], {"content": new_note_value, "user_id": "test"}
         )
 
+    def test_set_empty_note_removes_existing_note(self, test_client, session):
+        self.create_task_instances(session)
+        url = "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context"
+
+        set_response = test_client.patch(url, json={"note": "a note to remove"})
+        assert set_response.status_code == 200, set_response.text
+        ti_id = set_response.json()["task_instances"][0]["id"]
+        _check_task_instance_note(session, ti_id, {"content": "a note to remove", "user_id": "test"})
+
+        clear_response = test_client.patch(url, json={"note": ""})
+        assert clear_response.status_code == 200, clear_response.text
+        assert clear_response.json()["task_instances"][0]["note"] is None
+        _check_task_instance_note(session, ti_id, None)
+
     def test_set_note_should_respond_200_mapped_task_with_rtif(self, test_client, session):
         """Verify we don't duplicate rows through join to RTIF"""
         tis = self.create_task_instances(session)