Use WebEncoder to encode DagRun.conf in DagRun's list view (#32385)

* Use WebEncoder to encode DagRun.conf in DagRun's list view

* test case for json_f with WebEncoder

* Always use WebEncoder in web views

airflow/www/utils.py

@@ -49,6 +49,7 @@
 from airflow.utils import timezone
 from airflow.utils.code_utils import get_python_source
 from airflow.utils.helpers import alchemy_to_dict
+from airflow.utils.json import WebEncoder
 from airflow.utils.state import State, TaskInstanceState
 from airflow.www.forms import DateTimeWithTimezoneField
 from airflow.www.widgets import AirflowDateTimePickerWidget
@@ -481,7 +482,7 @@ def json_f(attr_name):
 
     def json_(attr):
         f = attr.get(attr_name)
-        serialized = json.dumps(f)
+        serialized = json.dumps(f, cls=WebEncoder)
         return Markup("<nobr>{}</nobr>").format(serialized)
 
     return json_

tests/www/test_utils.py

@@ -19,13 +19,15 @@
 
 import re
 from datetime import datetime
+from unittest.mock import Mock
 from urllib.parse import parse_qs
 
 from bs4 import BeautifulSoup
+from markupsafe import Markup
 
 from airflow.utils import json as utils_json
 from airflow.www import utils
-from airflow.www.utils import wrapped_markdown
+from airflow.www.utils import json_f, wrapped_markdown
 
 
 class TestUtils:
@@ -243,6 +245,26 @@ def test_get_dag_run_conf(self):
         )
         assert expected_encoded_dag_run_conf == encoded_dag_run_conf
 
+    def test_json_f_webencoder(self):
+        dag_run_conf = {
+            "1": "string",
+            "2": b"bytes",
+            "3": 123,
+            "4": "à".encode("latin"),
+            "5": datetime(2023, 1, 1),
+        }
+        expected_encoded_dag_run_conf = (
+            # HTML sanitization is insane
+            '{"1": "string", "2": "bytes", "3": 123, "4": "\\u00e0", "5": "2023-01-01T00:00:00+00:00"}'
+        )
+        expected_markup = Markup("<nobr>{}</nobr>").format(expected_encoded_dag_run_conf)
+
+        formatter = json_f("conf")
+        dagrun = Mock()
+        dagrun.get = Mock(return_value=dag_run_conf)
+
+        assert formatter(dagrun) == expected_markup
+
 
 class TestWrappedMarkdown:
     def test_wrapped_markdown_with_docstring_curly_braces(self):