From 1e692e3372e980c32a3477b17337253fbe9a7fbc Mon Sep 17 00:00:00 2001
From: Jens Scheffler <jens.scheffler@de.bosch.com>
Date: Sun, 28 May 2023 11:09:57 +0200
Subject: [PATCH] Trigger Button - Implement Part 2 of AIP-50

---
 airflow/www/templates/airflow/dag.html     | 43 +++++++---------------
 airflow/www/templates/airflow/dags.html    | 38 +++++++------------
 airflow/www/templates/airflow/trigger.html |  5 ++-
 airflow/www/views.py                       | 15 +++++---
 tests/www/views/test_views_tasks.py        | 12 +++---
 tests/www/views/test_views_trigger_dag.py  | 37 ++++++++++---------
 6 files changed, 66 insertions(+), 84 deletions(-)

diff --git a/airflow/www/templates/airflow/dag.html b/airflow/www/templates/airflow/dag.html
index 93acbcbb04d05..3a2c448f200c6 100644
--- a/airflow/www/templates/airflow/dag.html
+++ b/airflow/www/templates/airflow/dag.html
@@ -219,36 +219,21 @@ <h4 class="pull-right js-dataset-triggered" style="user-select: none;-moz-user-s
       </div>
       <div class="col-md-2">
         <div class="btn-group pull-right">
-          <div class="dropdown">
-            <a aria-label="Trigger DAG" class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_trigger }} trigger-dropdown-btn" data-toggle="dropdown">
+          {% if 'dag_id' in request.args %}
+            <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, **request.args), unpause=True) }}"
+          {% else %}
+            <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, dag_id=dag.dag_id, **request.args), unpause=True) }}"
+          {% endif %}
+              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>
-            </a>
-            <ul class="dropdown-menu trigger-dropdown-menu">
-              <li>
-                <form method="POST" action="{{ url_for('Airflow.trigger') }}">
-                  <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
-                  <input type="hidden" name="dag_id" value="{{ dag.dag_id }}">
-                  <input type="hidden" name="unpause" value="True">
-                  <!-- for task instance detail pages, dag_id is still a query param -->
-                  {% if 'dag_id' in request.args %}
-                    <input type="hidden" name="origin" value="{{ url_for(request.endpoint, **request.args) }}">
-                  {% else %}
-                    <input type="hidden" name="origin" value="{{ url_for(request.endpoint, dag_id=dag.dag_id, **request.args) }}">
-                  {% endif %}
-                  <button type="submit" class="dropdown-form-btn">Trigger DAG</button>
-                </form>
-              </li>
-              <li>
-                {% if 'dag_id' in request.args %}
-                  <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, origin=url_for(request.endpoint, **request.args)) }}">
-                {% 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 %}
-                Trigger DAG w/ config</a></li>
-            </ul>
-          </div>
-          <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint, dag_id=dag.dag_id)) }}" title="Delete&nbsp;DAG" class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_delete }}"
-            onclick="return confirmDeleteDag(this, '{{ dag.safe_dag_id }}')" aria-label="Delete DAG">
+          </a>
+          <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint, dag_id=dag.dag_id)) }}"
+            title="Delete&nbsp;DAG"
+            aria-label="Delete DAG"
+            class="btn btn-default btn-icon-only{{ ' disabled' if not dag.can_delete }}"
+            onclick="return confirmDeleteDag(this, '{{ dag.safe_dag_id }}')">
             <span class="material-icons text-danger" aria-hidden="true">delete_outline</span>
           </a>
         </div>
diff --git a/airflow/www/templates/airflow/dags.html b/airflow/www/templates/airflow/dags.html
index 99e59808814aa..9da9bafd614e9 100644
--- a/airflow/www/templates/airflow/dags.html
+++ b/airflow/www/templates/airflow/dags.html
@@ -362,33 +362,21 @@ <h2>{{ page_title }}</h2>
             </td>
             <td class="text-center">
               <div class="btn-group">
+                {# Use dag_id instead of dag.dag_id, because the DAG might not exist in the webserver's DagBag #}
                 {% if dag %}
-                  <div class="dropdown">
-                    <a aria-label="Trigger DAG"
-                       class="btn btn-sm btn-default btn-icon-only{{ ' disabled' if not dag.can_trigger }} trigger-dropdown-btn"
-                       data-toggle="dropdown">
-                      <span class="material-icons" aria-hidden="true">play_arrow</span>
-                    </a>
-                    <ul class="dropdown-menu trigger-dropdown-menu">
-                      <li>
-                        <form method="POST" action="{{ url_for('Airflow.trigger') }}">
-                          <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
-                          <input type="hidden" name="dag_id" value="{{ dag.dag_id }}">
-                          <input type="hidden" name="unpause" value="True">
-                          <button type="submit" class="dropdown-form-btn">Trigger DAG</button>
-                        </form>
-                      </li>
-                      <li><a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id) }}">Trigger DAG w/ config</a></li>
-                    </ul>
-                  </div>
+                  <a href="{{ url_for('Airflow.trigger', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint), unpause=True) }}"
+                    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>
+                  </a>
+                  <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint)) }}"
+                    onclick="return confirmDeleteDag(this, '{{ dag.dag_id }}')" title="Delete&nbsp;DAG"
+                    aria-label="Delete DAG"
+                    class="btn btn-sm btn-default btn-icon-only{{ ' disabled' if not dag.can_delete }}">
+                    <span class="material-icons text-danger" aria-hidden="true">delete_outline</span>
+                  </a>
                 {% endif %}
-                {# Use dag_id instead of dag.dag_id, because the DAG might not exist in the webserver's DagBag #}
-                <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id, redirect_url=url_for(request.endpoint)) }}"
-                   onclick="return confirmDeleteDag(this, '{{ dag.dag_id }}')" title="Delete&nbsp;DAG"
-                   aria-label="Delete DAG"
-                   class="btn btn-sm btn-default btn-icon-only {{ ' disabled' if not dag.can_delete }}">
-                  <span class="material-icons text-danger" aria-hidden="true">delete_outline</span>
-                </a>
               </div>
             </td>
             <td class="dags-table-more">
diff --git a/airflow/www/templates/airflow/trigger.html b/airflow/www/templates/airflow/trigger.html
index 56ad656883660..d0b4eac2c58c5 100644
--- a/airflow/www/templates/airflow/trigger.html
+++ b/airflow/www/templates/airflow/trigger.html
@@ -146,7 +146,10 @@
 
 {% block content %}
   {{ super() }}
-  <h2>Trigger DAG: {{ dag_id }}</h2>
+  <h2>
+    Trigger DAG: <a href="{{ url_for('Airflow.'+ dag.get_default_view(), dag_id=dag.dag_id) }}">{{ dag.dag_id }}</a>
+    <small class="text-muted">{{ dag.description[0:150] + '…' if dag.description and dag.description|length > 150 else dag.description|default('', true) }}</small>
+  </h2>
   {{ dag_docs(doc_md) }}
   <form method="POST" id="trigger_form">
     <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
diff --git a/airflow/www/views.py b/airflow/www/views.py
index a43c5db7c2dde..4ccb0971f7280 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1916,7 +1916,7 @@ def delete(self):
         # Upon success return to origin.
         return redirect(origin)
 
-    @expose("/trigger", methods=["POST", "GET"])
+    @expose("/dags/<string:dag_id>/trigger", methods=["POST", "GET"])
     @auth.has_access(
         [
             (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG),
@@ -1925,9 +1925,8 @@ def delete(self):
     )
     @action_logging
     @provide_session
-    def trigger(self, session: Session = NEW_SESSION):
+    def trigger(self, dag_id: str, session: Session = NEW_SESSION):
         """Triggers DAG Run."""
-        dag_id = request.values["dag_id"]
         run_id = request.values.get("run_id", "")
         origin = get_safe_url(request.values.get("origin"))
         unpause = request.values.get("unpause")
@@ -1963,6 +1962,7 @@ def trigger(self, session: Session = NEW_SESSION):
                 form_fields[k]["schema"]["custom_html_form"] = Markup(
                     form_fields[k]["schema"]["custom_html_form"]
                 )
+        ui_fields_defined = any("const" not in f["schema"] for f in form_fields.values())
 
         if not dag_orm:
             flash(f"Cannot find dag {dag_id}")
@@ -1993,7 +1993,7 @@ def trigger(self, session: Session = NEW_SESSION):
             if isinstance(recent_conf, dict) and any(recent_conf):
                 recent_confs[getattr(run, "run_id")] = json.dumps(recent_conf)
 
-        if request.method == "GET":
+        if request.method == "GET" and ui_fields_defined:
             # Populate conf textarea with conf requests parameter, or dag.params
             default_conf = ""
 
@@ -2014,6 +2014,7 @@ def trigger(self, session: Session = NEW_SESSION):
             return self.render_template(
                 "airflow/trigger.html",
                 form_fields=form_fields,
+                dag=dag,
                 dag_id=dag_id,
                 origin=origin,
                 conf=default_conf,
@@ -2031,9 +2032,10 @@ def trigger(self, session: Session = NEW_SESSION):
             return self.render_template(
                 "airflow/trigger.html",
                 form_fields=form_fields,
+                dag=dag,
                 dag_id=dag_id,
                 origin=origin,
-                conf=request_conf,
+                conf=request_conf if request_conf else {},
                 form=form,
                 is_dag_run_conf_overrides_params=is_dag_run_conf_overrides_params,
                 recent_confs=recent_confs,
@@ -2066,6 +2068,7 @@ def trigger(self, session: Session = NEW_SESSION):
                     return self.render_template(
                         "airflow/trigger.html",
                         form_fields=form_fields,
+                        dag=dag,
                         dag_id=dag_id,
                         origin=origin,
                         conf=request_conf,
@@ -2079,6 +2082,7 @@ def trigger(self, session: Session = NEW_SESSION):
                 return self.render_template(
                     "airflow/trigger.html",
                     form_fields=form_fields,
+                    dag=dag,
                     dag_id=dag_id,
                     origin=origin,
                     conf=request_conf,
@@ -2112,6 +2116,7 @@ def trigger(self, session: Session = NEW_SESSION):
             return self.render_template(
                 "airflow/trigger.html",
                 form_fields=form_fields,
+                dag=dag,
                 dag_id=dag_id,
                 origin=origin,
                 conf=request_conf,
diff --git a/tests/www/views/test_views_tasks.py b/tests/www/views/test_views_tasks.py
index 35538e5bf923b..34b58af8016bd 100644
--- a/tests/www/views/test_views_tasks.py
+++ b/tests/www/views/test_views_tasks.py
@@ -348,8 +348,8 @@ def test_tree_trigger_origin_tree_view(app, admin_client):
 
     url = "tree?dag_id=test_tree_view"
     resp = admin_client.get(url, follow_redirects=True)
-    params = {"dag_id": "test_tree_view", "origin": "/dags/test_tree_view/grid"}
-    href = f"/trigger?{html.escape(urllib.parse.urlencode(params))}"
+    params = {"origin": "/dags/test_tree_view/grid"}
+    href = f"/dags/test_tree_view/trigger?{html.escape(urllib.parse.urlencode(params))}"
     check_content_in_response(href, resp)
 
 
@@ -364,8 +364,8 @@ def test_graph_trigger_origin_graph_view(app, admin_client):
 
     url = "/dags/test_tree_view/graph"
     resp = admin_client.get(url, follow_redirects=True)
-    params = {"dag_id": "test_tree_view", "origin": "/dags/test_tree_view/graph"}
-    href = f"/trigger?{html.escape(urllib.parse.urlencode(params))}"
+    params = {"origin": "/dags/test_tree_view/graph"}
+    href = f"/dags/test_tree_view/trigger?{html.escape(urllib.parse.urlencode(params))}"
     check_content_in_response(href, resp)
 
 
@@ -380,8 +380,8 @@ def test_dag_details_trigger_origin_dag_details_view(app, admin_client):
 
     url = "/dags/test_graph_view/details"
     resp = admin_client.get(url, follow_redirects=True)
-    params = {"dag_id": "test_graph_view", "origin": "/dags/test_graph_view/details"}
-    href = f"/trigger?{html.escape(urllib.parse.urlencode(params))}"
+    params = {"origin": "/dags/test_graph_view/details"}
+    href = f"/dags/test_graph_view/trigger?{html.escape(urllib.parse.urlencode(params))}"
     check_content_in_response(href, resp)
 
 
diff --git a/tests/www/views/test_views_trigger_dag.py b/tests/www/views/test_views_trigger_dag.py
index 1c7cc5827ee49..7cf140ed7ca67 100644
--- a/tests/www/views/test_views_trigger_dag.py
+++ b/tests/www/views/test_views_trigger_dag.py
@@ -44,7 +44,7 @@ def initialize_one_dag():
 
 def test_trigger_dag_button_normal_exist(admin_client):
     resp = admin_client.get("/", follow_redirects=True)
-    assert "/trigger?dag_id=example_bash_operator" in resp.data.decode("utf-8")
+    assert "/dags/example_bash_operator/trigger" in resp.data.decode("utf-8")
     assert "return confirmDeleteDag(this, 'example_bash_operator')" in resp.data.decode("utf-8")
 
 
@@ -54,7 +54,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"trigger?dag_id={test_dag_id}{req}")
+    admin_client.post(f"dags/{test_dag_id}/trigger?{req}")
     with create_session() as session:
         run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
     assert run is not None
@@ -65,8 +65,8 @@ 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"trigger?dag_id={test_dag_id}&run_id={run_id}", follow_redirects=True)
-    response = admin_client.post(f"trigger?dag_id={test_dag_id}&run_id={run_id}", follow_redirects=True)
+    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)
     check_content_in_response(f"The run ID {run_id} already exists", response)
 
 
@@ -74,7 +74,7 @@ def test_trigger_dag_conf(admin_client):
     test_dag_id = "example_bash_operator"
     conf_dict = {"string": "Hello, World!"}
 
-    admin_client.post(f"trigger?dag_id={test_dag_id}", data={"conf": json.dumps(conf_dict)})
+    admin_client.post(f"dags/{test_dag_id}/trigger", data={"conf": json.dumps(conf_dict)})
 
     with create_session() as session:
         run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
@@ -87,7 +87,7 @@ def test_trigger_dag_conf(admin_client):
 def test_trigger_dag_conf_malformed(admin_client):
     test_dag_id = "example_bash_operator"
 
-    response = admin_client.post(f"trigger?dag_id={test_dag_id}", data={"conf": '{"a": "b"'})
+    response = admin_client.post(f"dags/{test_dag_id}/trigger", data={"conf": '{"a": "b"'})
     check_content_in_response("Invalid JSON configuration", response)
 
     with create_session() as session:
@@ -98,7 +98,7 @@ def test_trigger_dag_conf_malformed(admin_client):
 def test_trigger_dag_conf_not_dict(admin_client):
     test_dag_id = "example_bash_operator"
 
-    response = admin_client.post(f"trigger?dag_id={test_dag_id}", data={"conf": "string and not a dict"})
+    response = admin_client.post(f"dags/{test_dag_id}/trigger", data={"conf": "string and not a dict"})
     check_content_in_response("Invalid JSON configuration", response)
 
     with create_session() as session:
@@ -109,7 +109,7 @@ 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"trigger?dag_id={test_dag_id}", data={"execution_date": "not_a_date"})
+    response = admin_client.post(f"dags/{test_dag_id}/trigger", data={"execution_date": "not_a_date"})
     check_content_in_response("Invalid execution date", response)
 
     with create_session() as session:
@@ -121,7 +121,7 @@ def test_trigger_dag_execution_date_data_interval(admin_client):
     test_dag_id = "example_bash_operator"
     exec_date = timezone.utcnow()
 
-    admin_client.post(f"trigger?dag_id={test_dag_id}", data={"execution_date": exec_date.isoformat()})
+    admin_client.post(f"dags/{test_dag_id}/trigger", data={"execution_date": exec_date.isoformat()})
 
     with create_session() as session:
         run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
@@ -139,8 +139,9 @@ def test_trigger_dag_execution_date_data_interval(admin_client):
 
 def test_trigger_dag_form(admin_client):
     test_dag_id = "example_bash_operator"
-    resp = admin_client.get(f"trigger?dag_id={test_dag_id}")
-    check_content_in_response(f"Trigger DAG: {test_dag_id}", resp)
+    resp = admin_client.get(f"dags/{test_dag_id}/trigger")
+    check_content_in_response("Trigger DAG: <a href", resp)
+    check_content_in_response(f">{test_dag_id}</a>", resp)
 
 
 @pytest.mark.parametrize(
@@ -164,7 +165,7 @@ def test_trigger_dag_form(admin_client):
 def test_trigger_dag_form_origin_url(admin_client, test_origin, expected_origin):
     test_dag_id = "example_bash_operator"
 
-    resp = admin_client.get(f"trigger?dag_id={test_dag_id}&origin={test_origin}")
+    resp = admin_client.get(f"dags/{test_dag_id}/trigger?origin={test_origin}")
     check_content_in_response(f'<a class="btn" href="{expected_origin}">Cancel</a>', resp)
 
 
@@ -188,10 +189,10 @@ def test_trigger_dag_params_conf(admin_client, request_conf, expected_conf):
     doc_md = "Example Bash Operator"
 
     if not request_conf:
-        resp = admin_client.get(f"trigger?dag_id={test_dag_id}")
+        resp = admin_client.get(f"dags/{test_dag_id}/trigger")
     else:
         test_request_conf = json.dumps(request_conf, indent=4)
-        resp = admin_client.get(f"trigger?dag_id={test_dag_id}&conf={test_request_conf}&doc_md={doc_md}")
+        resp = admin_client.get(f"dags/{test_dag_id}/trigger?conf={test_request_conf}&doc_md={doc_md}")
     for key in expected_conf.keys():
         check_content_in_response(key, resp)
         check_content_in_response(str(expected_conf[key]), resp)
@@ -224,7 +225,7 @@ def test_trigger_dag_params_render(admin_client, dag_maker, session, app, monkey
             EmptyOperator(task_id="task1")
 
         m.setattr(app, "dag_bag", dag_maker.dagbag)
-        resp = admin_client.get(f"trigger?dag_id={DAG_ID}")
+        resp = admin_client.get(f"dags/{DAG_ID}/trigger")
 
     check_content_in_response(
         f'<textarea style="display: none;" id="json_start" name="json_start">{expected_dag_conf}</textarea>',
@@ -237,7 +238,7 @@ def test_trigger_endpoint_uses_existing_dagbag(admin_client):
     Test that Trigger Endpoint uses the DagBag already created in views.py
     instead of creating a new one.
     """
-    url = "trigger?dag_id=example_bash_operator"
+    url = "dags/example_bash_operator/trigger"
     resp = admin_client.post(url, data={}, follow_redirects=True)
     check_content_in_response("example_bash_operator", resp)
 
@@ -256,7 +257,7 @@ def test_viewer_cant_trigger_dag(app):
             (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_DAG_RUN),
         ],
     ) as client:
-        url = "trigger?dag_id=example_bash_operator"
+        url = "dags/example_bash_operator/trigger"
         resp = client.get(url, follow_redirects=True)
         response_data = resp.data.decode()
         assert "Access is Denied" in response_data
@@ -280,7 +281,7 @@ def test_trigger_dag_params_array_value_none_render(admin_client, dag_maker, ses
             EmptyOperator(task_id="task1")
 
         m.setattr(app, "dag_bag", dag_maker.dagbag)
-        resp = admin_client.get(f"trigger?dag_id={DAG_ID}")
+        resp = admin_client.get(f"dags/{DAG_ID}/trigger")
 
     check_content_in_response(
         f'<textarea style="display: none;" id="json_start" name="json_start">{expected_dag_conf}</textarea>',
