[airflow] 03/08: Fix url generation for TriggerDagRunOperatorLink (#14990)

[ash]

This is an automated email from the ASF dual-hosted git repository.

ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit dc9bf44a59392af0c59bcee45343ae7073cbd139
Author: Alan Ma <alanma0...@gmail.com>
AuthorDate: Sun Apr 11 04:51:59 2021 -0700

    Fix url generation for TriggerDagRunOperatorLink (#14990)
    
    Fixes: #14675
    
    Instead of building the relative url manually, we can leverage flask's url 
generation to account for differing airflow base URL and HTML base URL.
    
    (cherry picked from commit aaa3bf6b44238241bd61178426b692df53770c22)
---
 airflow/utils/helpers.py    |  4 +++-
 tests/utils/test_helpers.py | 13 ++++++++++---
 2 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index 69ac5a0..7fce177 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -24,6 +24,7 @@ from itertools import filterfalse, tee
 from typing import Any, Callable, Dict, Generator, Iterable, List, Optional, 
TypeVar
 from urllib import parse
 
+from flask import url_for
 from jinja2 import Template
 
 from airflow.configuration import conf
@@ -213,4 +214,5 @@ def build_airflow_url_with_query(query: Dict[str, Any]) -> 
str:
     
'http://0.0.0.0:8000/base/graph?dag_id=my-task&root=&execution_date=2020-10-27T10%3A59%3A25.615587
     """
     view = conf.get('webserver', 'dag_default_view').lower()
-    return f"/{view}?{parse.urlencode(query)}"
+    url = url_for(f"Airflow.{view}")
+    return f"{url}?{parse.urlencode(query)}"
diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py
index fffa2d4..bb7b453 100644
--- a/tests/utils/test_helpers.py
+++ b/tests/utils/test_helpers.py
@@ -142,10 +142,17 @@ class TestHelpers(unittest.TestCase):
 
     @conf_vars(
         {
-            ("webserver", "dag_default_view"): "custom",
+            ("webserver", "dag_default_view"): "graph",
         }
     )
     def test_build_airflow_url_with_query(self):
+        """
+        Test query generated with dag_id and params
+        """
         query = {"dag_id": "test_dag", "param": "key/to.encode"}
-        url = build_airflow_url_with_query(query)
-        assert url == "/custom?dag_id=test_dag&param=key%2Fto.encode"
+        expected_url = "/graph?dag_id=test_dag&param=key%2Fto.encode"
+
+        from airflow.www.app import cached_app
+
+        with cached_app(testing=True).test_request_context():
+            assert build_airflow_url_with_query(query) == expected_url