Change Trigger UI to use HTTP POST in web ui (#36026)

* Change Trigger UI to use HTTP POST in web ui, GET always shows trigger form
* Adjust tests to changed behavior of trigger handling, expects data submitted in POST

airflow/www/templates/airflow/dag.html

@@ -254,7 +254,7 @@ <h4 class="pull-right js-dataset-triggered" style="user-select: none;-moz-user-s
           {% else %}
             <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, dag_id=dag.dag_id, **request.args)) }}"
           {% endif %}
-              title="Trigger&nbsp;DAG"
+              onclick="return triggerDag(this, '{{ dag.dag_id }}')" title="Trigger&nbsp;DAG"
               aria-label="Trigger DAG"
               class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_trigger }} trigger-dropdown-btn">
               <span class="material-icons" aria-hidden="true">play_arrow</span>
@@ -289,5 +289,10 @@ <h4 class="pull-right js-dataset-triggered" style="user-select: none;-moz-user-s
       }
       return false;
     }
+
+    function triggerDag(link, dagId) {
+      postAsForm(link.href, {});
+      return false;
+    }
   </script>
 {% endblock %}

airflow/www/templates/airflow/dags.html

@@ -385,7 +385,7 @@ <h2>{{ page_title }}</h2>
                   </div>
                 {% else %}
                   <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint)) }}"
-                    title="Trigger&nbsp;DAG"
+                  onclick="return triggerDag(this, '{{ dag.dag_id }}')" title="Trigger&nbsp;DAG"
                     aria-label="Trigger DAG"
                     class="btn btn-sm btn-default btn-icon-only{{ ' disabled' if not dag.can_trigger }} trigger-dropdown-btn">
                     <span class="material-icons" aria-hidden="true">play_arrow</span>
@@ -483,5 +483,10 @@ <h2>{{ page_title }}</h2>
       }
       return false;
     }
+
+    function triggerDag(link, dagId) {
+      postAsForm(link.href, {});
+      return false;
+    }
   </script>
 {% endblock %}

airflow/www/views.py

@@ -2047,7 +2047,9 @@ def trigger(self, dag_id: str, session: Session = NEW_SESSION):
             if isinstance(run_conf, dict) and any(run_conf)
         }
 
-        if request.method == "GET" and (ui_fields_defined or show_trigger_form_if_no_params):
+        if request.method == "GET" or (
+            not request_conf and (ui_fields_defined or show_trigger_form_if_no_params)
+        ):
             # Populate conf textarea with conf requests parameter, or dag.params
             default_conf = ""
 

tests/www/views/test_views_trigger_dag.py

@@ -57,7 +57,7 @@ def test_trigger_dag_button_normal_exist(admin_client):
 )
 def test_trigger_dag_button(admin_client, req, expected_run_id):
     test_dag_id = "example_bash_operator"
-    admin_client.post(f"dags/{test_dag_id}/trigger?{req}")
+    admin_client.post(f"dags/{test_dag_id}/trigger?{req}", data={"conf": "{}"})
     with create_session() as session:
         run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
     assert run is not None
@@ -68,8 +68,12 @@ def test_trigger_dag_button(admin_client, req, expected_run_id):
 def test_duplicate_run_id(admin_client):
     test_dag_id = "example_bash_operator"
     run_id = "test_run"
-    admin_client.post(f"dags/{test_dag_id}/trigger?run_id={run_id}", follow_redirects=True)
-    response = admin_client.post(f"dags/{test_dag_id}/trigger?run_id={run_id}", follow_redirects=True)
+    admin_client.post(
+        f"dags/{test_dag_id}/trigger?run_id={run_id}", data={"conf": "{}"}, follow_redirects=True
+    )
+    response = admin_client.post(
+        f"dags/{test_dag_id}/trigger?run_id={run_id}", data={"conf": "{}"}, follow_redirects=True
+    )
     check_content_in_response(f"The run ID {run_id} already exists", response)
 
 
@@ -112,7 +116,9 @@ def test_trigger_dag_conf_not_dict(admin_client):
 def test_trigger_dag_wrong_execution_date(admin_client):
     test_dag_id = "example_bash_operator"
 
-    response = admin_client.post(f"dags/{test_dag_id}/trigger", data={"execution_date": "not_a_date"})
+    response = admin_client.post(
+        f"dags/{test_dag_id}/trigger", data={"conf": "{}", "execution_date": "not_a_date"}
+    )
     check_content_in_response("Invalid execution date", response)
 
     with create_session() as session:
@@ -124,7 +130,9 @@ def test_trigger_dag_execution_date_data_interval(admin_client):
     test_dag_id = "example_bash_operator"
     exec_date = timezone.utcnow()
 
-    admin_client.post(f"dags/{test_dag_id}/trigger", data={"execution_date": exec_date.isoformat()})
+    admin_client.post(
+        f"dags/{test_dag_id}/trigger", data={"conf": "{}", "execution_date": exec_date.isoformat()}
+    )
 
     with create_session() as session:
         run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
@@ -361,7 +369,7 @@ def test_trigger_dag_params_array_value_none_render(admin_client, dag_maker, ses
 def test_dag_run_id_pattern(session, admin_client, pattern, run_id, result):
     with conf_vars({("scheduler", "allowed_run_id_pattern"): pattern}):
         test_dag_id = "example_bash_operator"
-        admin_client.post(f"dags/{test_dag_id}/trigger?&run_id={run_id}")
+        admin_client.post(f"dags/{test_dag_id}/trigger?run_id={run_id}", data={"conf": "{}"})
         run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
         if result:
             assert run is not None