Don't overwrite connection extra with invalid json (#27142)

If invalid json is provided for a connection's extra, don't actually
overwrite it with the invalid data. This matches the intended behavior,
based on the flash message we show the user.

(cherry picked from commit fece7f0)

airflow/www/forms.py

@@ -180,6 +180,7 @@ class ConnectionForm(DynamicForm):
     conn_id = StringField(
         lazy_gettext('Connection Id'), validators=[InputRequired()], widget=BS3TextFieldWidget()
     )
+    # conn_type is added later via lazy_add_provider_discovered_options_to_connection_form
     description = StringField(lazy_gettext('Description'), widget=BS3TextAreaFieldWidget())
     host = StringField(lazy_gettext('Host'), widget=BS3TextFieldWidget())
     schema = StringField(lazy_gettext('Schema'), widget=BS3TextFieldWidget())

airflow/www/views.py

@@ -4243,14 +4243,15 @@ def process_form(self, form, is_created):
                 flash(
                     Markup(
                         "<p>The <em>Extra</em> connection field contained an invalid value for Conn ID: "
-                        f"<q>{conn_id}</q>.</p>"
+                        "<q>{conn_id}</q>.</p>"
                         "<p>If connection parameters need to be added to <em>Extra</em>, "
                         "please make sure they are in the form of a single, valid JSON object.</p><br>"
                         "The following <em>Extra</em> parameters were <b>not</b> added to the connection:<br>"
-                        f"{extra_json}",
-                    ),
+                        "{extra_json}"
+                    ).format(conn_id=conn_id, extra_json=extra_json),
                     category="error",
                 )
+                del form.extra
         del extra_json
 
         for key in self.extra_fields:

tests/www/views/test_views_connection.py

@@ -324,3 +324,29 @@ def test_connection_form_widgets_testable_types(mock_pm_hooks, admin_client):
     }
 
     assert ["first"] == ConnectionFormWidget().testable_connection_types
+
+
+def test_process_form_invalid_extra_removed(admin_client):
+    """
+    Test that when an invalid json `extra` is passed in the form, it is removed and _not_
+    saved over the existing extras.
+    """
+    from airflow.www.views import lazy_add_provider_discovered_options_to_connection_form
+
+    lazy_add_provider_discovered_options_to_connection_form()
+
+    conn_details = {"conn_id": "test_conn", "conn_type": "http"}
+    conn = Connection(**conn_details, extra='{"foo": "bar"}')
+    conn.id = 1
+
+    with create_session() as session:
+        session.add(conn)
+
+    data = {**conn_details, "extra": "Invalid"}
+    resp = admin_client.post('/connection/edit/1', data=data, follow_redirects=True)
+
+    assert resp.status_code == 200
+    with create_session() as session:
+        conn = session.query(Connection).get(1)
+
+    assert conn.extra == '{"foo": "bar"}'