update code and tests

airflow/serialization/helpers.py

@@ -43,16 +43,20 @@ def is_jsonable(x):
     max_length = conf.getint("core", "max_templated_field_length")
 
     if not is_jsonable(template_field):
-        if len(str(template_field)) > max_length:
-            rendered = redact(str(template_field), name)
+        serialized = str(template_field)
+        if len(serialized) > max_length:
+            rendered = redact(serialized, name)
             return (
                 "Truncated. You can change this behaviour in [core]max_templated_field_length. "
                 f"{rendered[:max_length - 79]!r}... "
             )
         return str(template_field)
     else:
-        if template_field and len(str(template_field)) > max_length:
-            rendered = redact(str(template_field), name)
+        if not template_field:
+            return template_field
+        serialized = str(template_field)
+        if len(serialized) > max_length:
+            rendered = redact(serialized, name)
             return (
                 "Truncated. You can change this behaviour in [core]max_templated_field_length. "
                 f"{rendered[:max_length - 79]!r}... "

tests/models/test_renderedtifields.py

@@ -174,24 +174,20 @@ def test_secrets_are_masked_when_large_string(self, dag_maker):
         Test that secrets are masked when the templated field is a large string
         """
         Variable.set(
-            key="api_key",
-            value="Api key secret should be masked as before"
-            * 5000,
+            key="name",
+            value="A field exceeding the max length of the templated field length" * 5000,
         )
         with dag_maker("test_serialized_rendered_fields"):
-            task = BashOperator(task_id="test", bash_command="echo {{ var.value.api_key }}")
+            task = BashOperator(task_id="test", bash_command="echo {{ var.value.name }}")
         dr = dag_maker.create_dagrun()
         ti = dr.task_instances[0]
         ti.task = task
         rtif = RTIF(ti=ti)
-
-        assert ti.dag_id == rtif.dag_id
-        assert ti.task_id == rtif.task_id
-        assert ti.run_id == rtif.run_id
         assert "***" in rtif.rendered_fields.get("bash_command")
 
     @mock.patch("airflow.models.BaseOperator.render_template")
     def test_pandas_dataframes_works_with_the_string_compare(self, render_mock, dag_maker):
+        """Test that rendered dataframe gets passed through the serialized template fields."""
         import pandas
 
         render_mock.return_value = pandas.DataFrame({"a": [1, 2, 3]})