[AIRFLOW-4524] Fix incorrect field names in view for Mark Success/Fai…

…lure (apache#5486)

Another mistake that wasn't caught from apache#5039 - we renamed the fields
in the template (to be unique) but didn't update the view

(cherry picked from commit 6afb12f)

airflow/www/views.py

@@ -1435,10 +1435,10 @@ def failed(self):
         execution_date = request.form.get('execution_date')
 
         confirmed = request.form.get('confirmed') == "true"
-        upstream = request.form.get('upstream') == "true"
-        downstream = request.form.get('downstream') == "true"
-        future = request.form.get('future') == "true"
-        past = request.form.get('past') == "true"
+        upstream = request.form.get('failed_upstream') == "true"
+        downstream = request.form.get('failed_downstream') == "true"
+        future = request.form.get('failed_future') == "true"
+        past = request.form.get('failed_past') == "true"
 
         return self._mark_task_instance_state(dag_id, task_id, origin, execution_date,
                                               confirmed, upstream, downstream,
@@ -1455,10 +1455,10 @@ def success(self):
         execution_date = request.form.get('execution_date')
 
         confirmed = request.form.get('confirmed') == "true"
-        upstream = request.form.get('upstream') == "true"
-        downstream = request.form.get('downstream') == "true"
-        future = request.form.get('future') == "true"
-        past = request.form.get('past') == "true"
+        upstream = request.form.get('success_upstream') == "true"
+        downstream = request.form.get('success_downstream') == "true"
+        future = request.form.get('success_future') == "true"
+        past = request.form.get('success_past') == "true"
 
         return self._mark_task_instance_state(dag_id, task_id, origin, execution_date,
                                               confirmed, upstream, downstream,

airflow/www_rbac/views.py

@@ -1166,10 +1166,10 @@ def failed(self):
         execution_date = request.form.get('execution_date')
 
         confirmed = request.form.get('confirmed') == "true"
-        upstream = request.form.get('upstream') == "true"
-        downstream = request.form.get('downstream') == "true"
-        future = request.form.get('future') == "true"
-        past = request.form.get('past') == "true"
+        upstream = request.form.get('failed_upstream') == "true"
+        downstream = request.form.get('failed_downstream') == "true"
+        future = request.form.get('failed_future') == "true"
+        past = request.form.get('failed_past') == "true"
 
         return self._mark_task_instance_state(dag_id, task_id, origin, execution_date,
                                               confirmed, upstream, downstream,
@@ -1186,10 +1186,10 @@ def success(self):
         execution_date = request.form.get('execution_date')
 
         confirmed = request.form.get('confirmed') == "true"
-        upstream = request.form.get('upstream') == "true"
-        downstream = request.form.get('downstream') == "true"
-        future = request.form.get('future') == "true"
-        past = request.form.get('past') == "true"
+        upstream = request.form.get('success_upstream') == "true"
+        downstream = request.form.get('success_downstream') == "true"
+        future = request.form.get('success_future') == "true"
+        past = request.form.get('success_past') == "true"
 
         return self._mark_task_instance_state(dag_id, task_id, origin, execution_date,
                                               confirmed, upstream, downstream,

tests/core.py

@@ -1835,6 +1835,12 @@ def setUp(self):
         self.runme_0 = self.dag_bash.get_task('runme_0')
         self.example_xcom = self.dagbag.dags['example_xcom']
 
+        session = Session()
+        session.query(models.DagRun).delete()
+        session.query(models.TaskInstance).delete()
+        session.commit()
+        session.close()
+
         self.dagrun_python = self.dag_python.create_dagrun(
             run_id="test_{}".format(models.DagRun.id_for_date(timezone.utcnow())),
             execution_date=EXAMPLE_DAG_DEFAULT_DATE,
@@ -2020,10 +2026,10 @@ def test_dag_views(self):
         response = self.app.post("/admin/airflow/success", data=dict(
             task_id="print_the_context",
             dag_id="example_python_operator",
-            upstream="false",
-            downstream="false",
-            future="false",
-            past="false",
+            success_upstream="false",
+            success_downstream="false",
+            success_future="false",
+            success_past="false",
             execution_date=EXAMPLE_DAG_DEFAULT_DATE,
             origin="/admin"))
         self.assertIn("Wait a minute", response.data.decode('utf-8'))
@@ -2042,10 +2048,10 @@ def test_dag_views(self):
         form = dict(
             task_id="section-1",
             dag_id="example_subdag_operator",
-            upstream="true",
-            downstream="true",
-            future="false",
-            past="false",
+            success_upstream="true",
+            success_downstream="true",
+            success_future="false",
+            success_past="false",
             execution_date=EXAMPLE_DAG_DEFAULT_DATE,
             origin="/admin")
         response = self.app.post("/admin/airflow/success", data=form)