Follow up on comment

homeassistant/components/mqtt/client.py

@@ -71,7 +71,6 @@ from .const import (
     DEFAULT_QOS,
     DEFAULT_TRANSPORT,
     DEFAULT_WILL,
-    DEFAULT_WS_HEADERS,
     DEFAULT_WS_PATH,
     DOMAIN,
     MQTT_CONNECTION_STATE,
@@ -414,7 +413,7 @@ class MqttClientSetup:
         tls_insecure = config.get(CONF_TLS_INSECURE)
         if transport == TRANSPORT_WEBSOCKETS:
             ws_path: str = config.get(CONF_WS_PATH, DEFAULT_WS_PATH)
-            ws_headers: dict[str, str] = config.get(CONF_WS_HEADERS, DEFAULT_WS_HEADERS)
+            ws_headers: dict[str, str] = config.get(CONF_WS_HEADERS, {})
             self._client.ws_set_options(ws_path, ws_headers)
         if certificate is not None:
             self._client.tls_set(

homeassistant/components/mqtt/config_flow.py

@@ -373,7 +373,6 @@ from .const import (
     DEFAULT_CLIMATE_INITIAL_TEMPERATURE,
     DEFAULT_DISCOVERY,
     DEFAULT_ENCODING,
-    DEFAULT_KEEPALIVE,
     DEFAULT_ON_COMMAND_TYPE,
     DEFAULT_PAYLOAD_ARM_AWAY,
     DEFAULT_PAYLOAD_ARM_CUSTOM_BYPASS,
@@ -414,7 +413,6 @@ from .const import (
     DEFAULT_TILT_OPEN_POSITION,
     DEFAULT_TRANSPORT,
     DEFAULT_WILL,
-    DEFAULT_WS_PATH,
     DOMAIN,
     REMOTE_CODE,
     REMOTE_CODE_TEXT,
@@ -441,7 +439,7 @@ ADDON_SETUP_TIMEOUT_ROUNDS = 5
 
 CONF_CLIENT_KEY_PASSWORD = "client_key_password"
 
-ADVANCED_OPTIONS = "advanced_options"
+OTHER_SETTINGS = "other_settings"
 SET_CA_CERT = "set_ca_cert"
 SET_CLIENT_CERT = "set_client_cert"
 
@@ -4036,24 +4034,22 @@ def subentry_schema_default_data_from_fields(
 @callback
 def update_password_from_user_input(
     entry_password: str | None, user_input: dict[str, Any]
-) -> dict[str, Any]:
+) -> None:
     """Update the password if the entry has been updated.
 
     As we want to avoid reflecting the stored password in the UI,
     we replace the suggested value in the UI with a sentitel,
     and we change it back here if it was changed.
     """
-    substituted_used_data = dict(user_input)
     # Take out the password submitted
-    user_password: str | None = substituted_used_data.pop(CONF_PASSWORD, None)
+    user_password: str | None = user_input.pop(CONF_PASSWORD, None)
     # Only add the password if it has changed.
     # If the sentinel password is submitted, we replace that with our current
     # password from the config entry data.
     password_changed = user_password is not None and user_password != PWD_NOT_CHANGED
     password = user_password if password_changed else entry_password
     if password is not None:
-        substituted_used_data[CONF_PASSWORD] = password
-    return substituted_used_data
+        user_input[CONF_PASSWORD] = password
 
 
 REAUTH_SCHEMA = vol.Schema(
@@ -4063,6 +4059,35 @@ REAUTH_SCHEMA = vol.Schema(
     }
 )
 
+OTHER_SETTINGS_SCHEMA = vol.Schema(
+    {
+        vol.Optional(CONF_CLIENT_ID): TEXT_SELECTOR,
+        vol.Optional(CONF_KEEPALIVE): KEEPALIVE_SELECTOR,
+        vol.Required(SET_CLIENT_CERT): BOOLEAN_SELECTOR,
+        vol.Optional(CONF_CLIENT_CERT): CERT_UPLOAD_SELECTOR,
+        vol.Optional(CONF_CLIENT_KEY): CERT_KEY_UPLOAD_SELECTOR,
+        vol.Optional(CONF_CLIENT_KEY_PASSWORD): PASSWORD_SELECTOR,
+        vol.Required(SET_CA_CERT): BROKER_VERIFICATION_SELECTOR,
+        vol.Optional(CONF_CERTIFICATE): CA_CERT_UPLOAD_SELECTOR,
+        vol.Optional(CONF_TLS_INSECURE): BOOLEAN_SELECTOR,
+        vol.Required(CONF_TRANSPORT, default=DEFAULT_TRANSPORT): TRANSPORT_SELECTOR,
+        vol.Optional(CONF_WS_PATH): TEXT_SELECTOR,
+        vol.Optional(CONF_WS_HEADERS): WS_HEADERS_SELECTOR,
+    }
+)
+CONFIG_DATAFLOW_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_BROKER): TEXT_SELECTOR,
+        vol.Required(CONF_PORT, default=DEFAULT_PORT): PORT_SELECTOR,
+        vol.Required(CONF_PROTOCOL, default=DEFAULT_PROTOCOL): PROTOCOL_SELECTOR,
+        vol.Optional(CONF_USERNAME): TEXT_SELECTOR,
+        vol.Optional(CONF_PASSWORD): PASSWORD_SELECTOR,
+        vol.Optional(OTHER_SETTINGS): section(
+            OTHER_SETTINGS_SCHEMA, SectionConfig({"collapsed": True})
+        ),
+    }
+)
+
 
 class FlowHandler(ConfigFlow, domain=DOMAIN):
     """Handle a config flow."""
@@ -4072,11 +4097,13 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
 
     _hassio_discovery: dict[str, Any] | None = None
     _addon_manager: AddonManager
+    last_uploaded: dict[str, Any]
 
     def __init__(self) -> None:
         """Set up flow instance."""
         self.install_task: asyncio.Task | None = None
         self.start_task: asyncio.Task | None = None
+        self.last_uploaded = {}
 
     @classmethod
     @callback
@@ -4178,7 +4205,6 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
                 CONF_PROTOCOL: DEFAULT_PROTOCOL,
                 CONF_USERNAME: addon_discovery_config.get(CONF_USERNAME),
                 CONF_PASSWORD: addon_discovery_config.get(CONF_PASSWORD),
-                CONF_DISCOVERY: DEFAULT_DISCOVERY,
             }
         except AddonError:
             # We do not have discovery information yet
@@ -4308,8 +4334,9 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
 
         reauth_entry = self._get_reauth_entry()
         if user_input:
-            substituted_used_data = update_password_from_user_input(
-                reauth_entry.data.get(CONF_PASSWORD), user_input
+            substituted_used_data = deepcopy(user_input)
+            update_password_from_user_input(
+                reauth_entry.data.get(CONF_PASSWORD), substituted_used_data
             )
             new_entry_data = {**reauth_entry.data, **substituted_used_data}
             if await self.hass.async_add_executor_job(
@@ -4333,49 +4360,72 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
             errors=errors,
         )
 
+    @callback
+    def async_get_entry_defaults(self) -> dict[str, Any]:
+        """Load the default settings from the entry."""
+        data = self._get_reconfigure_entry().data
+        advanced_schema_settings: dict[str, Any] = {
+            key: data[key] for key in OTHER_SETTINGS_SCHEMA.schema if key in data
+        }
+        advanced_schema_settings[SET_CLIENT_CERT] = (
+            CONF_CLIENT_CERT in advanced_schema_settings
+        ) and (CONF_CLIENT_KEY in advanced_schema_settings)
+        advanced_schema_settings.pop(CONF_CLIENT_CERT, None)
+        advanced_schema_settings.pop(CONF_CLIENT_KEY, None)
+        conf_cert = advanced_schema_settings.pop(CONF_CERTIFICATE, None)
+        advanced_schema_settings[SET_CA_CERT] = (
+            "auto"
+            if conf_cert == "auto"
+            else "custom"
+            if conf_cert is not None
+            else "off"
+        )
+        if CONF_WS_HEADERS in advanced_schema_settings:
+            advanced_schema_settings[CONF_WS_HEADERS] = json_dumps(
+                advanced_schema_settings.pop(CONF_WS_HEADERS)
+            )
+
+        settings: dict[str, Any] = {
+            key: data[key] for key in CONFIG_DATAFLOW_SCHEMA.schema if key in data
+        }
+        settings[OTHER_SETTINGS] = advanced_schema_settings
+        if CONF_PASSWORD in settings:
+            # Hide entry password
+            settings[CONF_PASSWORD] = PWD_NOT_CHANGED
+        return settings
+
     async def async_step_broker(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Confirm the setup."""
         errors: dict[str, str] = {}
-        fields: OrderedDict[Any, Any] = OrderedDict()
-        validated_user_input: dict[str, Any] = {}
+        schema = CONFIG_DATAFLOW_SCHEMA
+        entry_config_update: dict[str, Any] = {}
+        entry_defaults: dict[str, Any] | None = None
         if is_reconfigure := (self.source == SOURCE_RECONFIGURE):
             reconfigure_entry = self._get_reconfigure_entry()
-        if await async_get_broker_settings(
+            entry_defaults = self.async_get_entry_defaults()
+        if await async_validate_broker_settings(
             self,
-            fields,
             reconfigure_entry.data if is_reconfigure else None,
             user_input,
-            validated_user_input,
+            entry_config_update,
             errors,
         ):
             if is_reconfigure:
-                validated_user_input = update_password_from_user_input(
-                    reconfigure_entry.data.get(CONF_PASSWORD), validated_user_input
+                return self.async_update_and_abort(
+                    reconfigure_entry,
+                    data=entry_config_update,
                 )
-
-            can_connect = await self.hass.async_add_executor_job(
-                try_connection,
-                validated_user_input,
+            return self.async_create_entry(
+                title=entry_config_update[CONF_BROKER],
+                data=entry_config_update,
             )
 
-            if can_connect:
-                if is_reconfigure:
-                    return self.async_update_and_abort(
-                        reconfigure_entry,
-                        data=validated_user_input,
-                    )
-                return self.async_create_entry(
-                    title=validated_user_input[CONF_BROKER],
-                    data=validated_user_input,
-                )
-
-            errors["base"] = "cannot_connect"
-
-        return self.async_show_form(
-            step_id="broker", data_schema=vol.Schema(fields), errors=errors
+        schema = self.add_suggested_values_to_schema(
+            schema, (entry_defaults or {}) | (user_input or {})
         )
+        return self.async_show_form(step_id="broker", data_schema=schema, errors=errors)
 
     async def async_step_reconfigure(
         self, user_input: dict[str, Any] | None = None
@@ -5248,331 +5298,162 @@ async def _get_uploaded_file(hass: HomeAssistant, id: str) -> bytes:
     return await hass.async_add_executor_job(_proces_uploaded_file)
 
 
-def _validate_pki_file(
-    file_id: str | None, pem_data: str | None, errors: dict[str, str], error: str
-) -> bool:
-    """Return False if uploaded file could not be converted to PEM format."""
-    if file_id and not pem_data:
-        errors["base"] = error
-        return False
-    return True
-
-
-async def async_get_broker_settings(
-    flow: ConfigFlow | OptionsFlow,
-    fields: OrderedDict[Any, Any],
+async def async_validate_broker_settings(
+    flow: FlowHandler,
     entry_config: MappingProxyType[str, Any] | None,
     user_input: dict[str, Any] | None,
-    validated_user_input: dict[str, Any],
+    entry_config_update: dict[str, Any],
     errors: dict[str, str],
 ) -> bool:
-    """Build the config flow schema to collect the broker settings.
+    """Validate the broker settings, and return the updated entry dataset."""
 
-    Shows advanced options if one or more are configured
-    or when the advanced_broker_options checkbox was selected.
-    Returns True when settings are collected successfully.
-    """
-    hass = flow.hass
-    advanced_broker_options: bool = False
-    user_input_basic: dict[str, Any] = {}
-    current_config: dict[str, Any] = (
-        entry_config.copy() if entry_config is not None else {}
-    )
-
-    async def _async_validate_broker_settings(
-        config: dict[str, Any],
-        user_input: dict[str, Any],
-        validated_user_input: dict[str, Any],
-        errors: dict[str, str],
+    async def _async_process_file_upload(
+        upload_id: str,
+        field: str,
+        pem_type: PEMType,
+        error_code: str,
+        password: str | None = None,
     ) -> bool:
-        """Additional validation on broker settings for better error messages."""
-
-        if CONF_PROTOCOL not in validated_user_input:
-            validated_user_input[CONF_PROTOCOL] = DEFAULT_PROTOCOL
-        # Get current certificate settings from config entry
-        certificate: str | None = (
-            "auto"
-            if user_input.get(SET_CA_CERT, "off") == "auto"
-            else config.get(CONF_CERTIFICATE)
-            if user_input.get(SET_CA_CERT, "off") == "custom"
-            else None
-        )
-        client_certificate: str | None = (
-            config.get(CONF_CLIENT_CERT) if user_input.get(SET_CLIENT_CERT) else None
-        )
-        client_key: str | None = (
-            config.get(CONF_CLIENT_KEY) if user_input.get(SET_CLIENT_CERT) else None
-        )
-
-        # Prepare entry update with uploaded files
-        validated_user_input.update(user_input)
-        client_certificate_id: str | None = user_input.get(CONF_CLIENT_CERT)
-        client_key_id: str | None = user_input.get(CONF_CLIENT_KEY)
-        # We do not store the private key password in the entry data
-        client_key_password: str | None = validated_user_input.pop(
-            CONF_CLIENT_KEY_PASSWORD, None
-        )
-        if (client_certificate_id and not client_key_id) or (
-            not client_certificate_id and client_key_id
-        ):
-            errors["base"] = "invalid_inclusion"
-            return False
-        certificate_id: str | None = user_input.get(CONF_CERTIFICATE)
-        if certificate_id:
-            certificate_data_raw = await _get_uploaded_file(hass, certificate_id)
-            certificate = async_convert_to_pem(
-                certificate_data_raw, PEMType.CERTIFICATE
-            )
-        if not _validate_pki_file(
-            certificate_id, certificate, errors, "bad_certificate"
-        ):
-            return False
-
-        # Return to form for file upload CA cert or client cert and key
-        if (
-            (
-                not client_certificate
-                and user_input.get(SET_CLIENT_CERT)
-                and not client_certificate_id
-            )
-            or (
-                not certificate
-                and user_input.get(SET_CA_CERT, "off") == "custom"
-                and not certificate_id
-            )
-            or (
-                user_input.get(CONF_TRANSPORT) == TRANSPORT_WEBSOCKETS
-                and CONF_WS_PATH not in user_input
-            )
-        ):
-            return False
-
-        if client_certificate_id:
-            client_certificate_data = await _get_uploaded_file(
-                hass, client_certificate_id
-            )
-            client_certificate = async_convert_to_pem(
-                client_certificate_data, PEMType.CERTIFICATE
-            )
-        if not _validate_pki_file(
-            client_certificate_id, client_certificate, errors, "bad_client_cert"
-        ):
-            return False
-
-        if client_key_id:
-            client_key_data = await _get_uploaded_file(hass, client_key_id)
-            client_key = async_convert_to_pem(
-                client_key_data, PEMType.PRIVATE_KEY, password=client_key_password
-            )
-        if not _validate_pki_file(
-            client_key_id, client_key, errors, "client_key_error"
-        ):
-            return False
-
-        certificate_data: dict[str, Any] = {}
-        if certificate:
-            certificate_data[CONF_CERTIFICATE] = certificate
-        if client_certificate:
-            certificate_data[CONF_CLIENT_CERT] = client_certificate
-            certificate_data[CONF_CLIENT_KEY] = client_key
-
-        validated_user_input.update(certificate_data)
-        await async_create_certificate_temp_files(hass, certificate_data)
-        if error := await hass.async_add_executor_job(
-            check_certicate_chain,
-        ):
-            errors["base"] = error
-            return False
-
-        validated_user_input.pop(SET_CA_CERT, None)
-        validated_user_input.pop(SET_CLIENT_CERT, None)
-        if validated_user_input.get(CONF_TRANSPORT, TRANSPORT_TCP) == TRANSPORT_TCP:
-            validated_user_input.pop(CONF_WS_PATH, None)
-            validated_user_input.pop(CONF_WS_HEADERS, None)
-            return True
+        """Get uploaded file, or a preserved copy, and convert to a PEM file."""
         try:
-            validated_user_input[CONF_WS_HEADERS] = json_loads(
-                validated_user_input.get(CONF_WS_HEADERS, "{}")
+            data_raw = await _get_uploaded_file(hass, upload_id)
+        except ValueError:
+            # Use preserved file if available.
+            # When an uploaded file was read, but an error occurs,
+            # the form will reload but the temporary file from the upload
+            # will not be available any more. If it was processed correctly,
+            # we can use the preserved copy.
+            if upload_id in flow.last_uploaded:
+                data_raw = flow.last_uploaded[upload_id]
+            else:
+                raise
+        else:
+            # Preserve a copy in case the validation fails,
+            # and we need it later
+            flow.last_uploaded[upload_id] = data_raw
+        pem_data = async_convert_to_pem(data_raw, pem_type, password)
+        if upload_id and not pem_data:
+            errors["base"] = error_code
+            return False
+        entry_config_update[field] = pem_data
+        return True
+
+    if user_input is None:
+        return False
+
+    hass = flow.hass
+
+    # Copy basic and other entry fields
+    entry_config_update |= user_input
+    entry_config_update.update(entry_config_update.pop(OTHER_SETTINGS))
+    # Pop incompatible fields for update
+    for key in (
+        SET_CA_CERT,
+        SET_CLIENT_CERT,
+        CONF_CERTIFICATE,
+        CONF_CLIENT_CERT,
+        CONF_CLIENT_KEY,
+        CONF_CLIENT_KEY_PASSWORD,
+    ):
+        entry_config_update.pop(key, None)
+
+    # Get current CA certificate settings from config entry
+    if (set_ca_cert := user_input[OTHER_SETTINGS][SET_CA_CERT]) == "auto":
+        entry_config_update[CONF_CERTIFICATE] = "auto"
+    elif (
+        entry_config is not None
+        and set_ca_cert == "custom"
+        and (current_cert := entry_config.get(CONF_CERTIFICATE))
+    ):
+        entry_config_update[CONF_CERTIFICATE] = current_cert
+
+    # Prepare entry update with uploaded certificate files
+    # converted to PEM format
+    new_client_certificate: str | None = user_input[OTHER_SETTINGS].get(
+        CONF_CLIENT_CERT
+    )
+    new_client_key: str | None = user_input[OTHER_SETTINGS].get(CONF_CLIENT_KEY)
+    set_client_cert = user_input[OTHER_SETTINGS][SET_CLIENT_CERT]
+
+    if (new_client_certificate and not new_client_key) or (
+        not new_client_certificate and new_client_key
+    ):
+        errors["base"] = "invalid_inclusion"
+        return False
+
+    if new_certificate := user_input[OTHER_SETTINGS].get(CONF_CERTIFICATE):
+        if not await _async_process_file_upload(
+            new_certificate, CONF_CERTIFICATE, PEMType.CERTIFICATE, "bad_certificate"
+        ):
+            return False
+
+    if new_client_certificate:
+        if not await _async_process_file_upload(
+            new_client_certificate,
+            CONF_CLIENT_CERT,
+            PEMType.CERTIFICATE,
+            "bad_client_cert",
+        ):
+            return False
+    elif (
+        entry_config is not None
+        and set_client_cert
+        and (client_cert := entry_config.get(CONF_CLIENT_CERT))
+    ):
+        entry_config_update[CONF_CLIENT_CERT] = client_cert
+
+    if new_client_key:
+        if not await _async_process_file_upload(
+            new_client_key,
+            CONF_CLIENT_KEY,
+            PEMType.PRIVATE_KEY,
+            "client_key_error",
+            password=user_input[OTHER_SETTINGS].get(CONF_CLIENT_KEY_PASSWORD),
+        ):
+            return False
+    elif (
+        entry_config is not None
+        and set_client_cert
+        and (client_key := entry_config.get(CONF_CLIENT_KEY))
+    ):
+        entry_config_update[CONF_CLIENT_KEY] = client_key
+
+    # We temporary create the current and new uploaded certificate files
+    # and we check the certificate chain.
+    await async_create_certificate_temp_files(hass, entry_config_update)
+    if error := await hass.async_add_executor_job(
+        check_certicate_chain,
+    ):
+        errors["base"] = error
+        return False
+
+    if user_input[OTHER_SETTINGS].get(CONF_TRANSPORT, TRANSPORT_TCP) == TRANSPORT_TCP:
+        entry_config_update.pop(CONF_WS_PATH, None)
+        entry_config_update.pop(CONF_WS_HEADERS, None)
+    else:
+        # Web socket transport
+        try:
+            entry_config_update[CONF_WS_HEADERS] = json_loads(
+                user_input[OTHER_SETTINGS].get(CONF_WS_HEADERS, "{}")
             )
-            schema = vol.Schema({cv.string: cv.template})
-            schema(validated_user_input[CONF_WS_HEADERS])
+            schema = vol.Schema({str: str})
+            schema(entry_config_update[CONF_WS_HEADERS])
         except (*JSON_DECODE_EXCEPTIONS, vol.MultipleInvalid):
             errors["base"] = "bad_ws_headers"
             return False
+
+    # Test the configuration
+    if entry_config is not None:
+        update_password_from_user_input(
+            entry_config.get(CONF_PASSWORD), entry_config_update
+        )
+    if await hass.async_add_executor_job(
+        try_connection,
+        entry_config_update,
+    ):
         return True
 
-    if user_input:
-        user_input_basic = user_input.copy()
-        advanced_broker_options = user_input_basic.get(ADVANCED_OPTIONS, False)
-        if ADVANCED_OPTIONS not in user_input or advanced_broker_options is False:
-            if await _async_validate_broker_settings(
-                current_config,
-                user_input_basic,
-                validated_user_input,
-                errors,
-            ):
-                return True
-        # Get defaults settings from previous post
-        current_broker = user_input_basic.get(CONF_BROKER)
-        current_port = user_input_basic.get(CONF_PORT, DEFAULT_PORT)
-        current_user = user_input_basic.get(CONF_USERNAME)
-        current_pass = user_input_basic.get(CONF_PASSWORD)
-    else:
-        # Get default settings from entry (if any)
-        current_broker = current_config.get(CONF_BROKER)
-        current_port = current_config.get(CONF_PORT, DEFAULT_PORT)
-        current_user = current_config.get(CONF_USERNAME)
-        # Return the sentinel password to avoid exposure
-        current_entry_pass = current_config.get(CONF_PASSWORD)
-        current_pass = PWD_NOT_CHANGED if current_entry_pass else None
-
-    # Treat the previous post as an update of the current settings
-    # (if there was a basic broker setup step)
-    current_config.update(user_input_basic)
-
-    # Get default settings for advanced broker options
-    current_client_id = current_config.get(CONF_CLIENT_ID)
-    current_keepalive = current_config.get(CONF_KEEPALIVE, DEFAULT_KEEPALIVE)
-    current_ca_certificate = current_config.get(CONF_CERTIFICATE)
-    current_client_certificate = current_config.get(CONF_CLIENT_CERT)
-    current_client_key = current_config.get(CONF_CLIENT_KEY)
-    current_tls_insecure = current_config.get(CONF_TLS_INSECURE, False)
-    current_protocol = current_config.get(CONF_PROTOCOL, DEFAULT_PROTOCOL)
-    current_transport = current_config.get(CONF_TRANSPORT, DEFAULT_TRANSPORT)
-    current_ws_path = current_config.get(CONF_WS_PATH, DEFAULT_WS_PATH)
-    current_ws_headers = (
-        json_dumps(current_config.get(CONF_WS_HEADERS))
-        if CONF_WS_HEADERS in current_config
-        else None
-    )
-    advanced_broker_options |= bool(
-        current_client_id
-        or current_keepalive != DEFAULT_KEEPALIVE
-        or current_ca_certificate
-        or current_client_certificate
-        or current_client_key
-        or current_tls_insecure
-        or current_config.get(SET_CA_CERT, "off") != "off"
-        or current_config.get(SET_CLIENT_CERT)
-        or current_transport == TRANSPORT_WEBSOCKETS
-    )
-
-    # Build form
-    fields[vol.Required(CONF_BROKER, default=current_broker)] = TEXT_SELECTOR
-    fields[vol.Required(CONF_PORT, default=current_port)] = PORT_SELECTOR
-    fields[
-        vol.Optional(
-            CONF_PROTOCOL,
-            description={"suggested_value": current_protocol},
-        )
-    ] = PROTOCOL_SELECTOR
-    fields[
-        vol.Optional(
-            CONF_USERNAME,
-            description={"suggested_value": current_user},
-        )
-    ] = TEXT_SELECTOR
-    fields[
-        vol.Optional(
-            CONF_PASSWORD,
-            description={"suggested_value": current_pass},
-        )
-    ] = PASSWORD_SELECTOR
-    # show advanced options checkbox if no defaults
-    # of the advanced options are overridden
-    if not advanced_broker_options:
-        fields[
-            vol.Optional(
-                ADVANCED_OPTIONS,
-            )
-        ] = BOOLEAN_SELECTOR
-        return False
-    fields[
-        vol.Optional(
-            CONF_CLIENT_ID,
-            description={"suggested_value": current_client_id},
-        )
-    ] = TEXT_SELECTOR
-    fields[
-        vol.Optional(
-            CONF_KEEPALIVE,
-            description={"suggested_value": current_keepalive},
-        )
-    ] = KEEPALIVE_SELECTOR
-    fields[
-        vol.Optional(
-            SET_CLIENT_CERT,
-            default=current_client_certificate is not None
-            or current_config.get(SET_CLIENT_CERT) is True,
-        )
-    ] = BOOLEAN_SELECTOR
-    if (
-        current_client_certificate is not None
-        or current_config.get(SET_CLIENT_CERT) is True
-    ):
-        fields[
-            vol.Optional(
-                CONF_CLIENT_CERT,
-                description={"suggested_value": user_input_basic.get(CONF_CLIENT_CERT)},
-            )
-        ] = CERT_UPLOAD_SELECTOR
-        fields[
-            vol.Optional(
-                CONF_CLIENT_KEY,
-                description={"suggested_value": user_input_basic.get(CONF_CLIENT_KEY)},
-            )
-        ] = CERT_KEY_UPLOAD_SELECTOR
-        fields[
-            vol.Optional(
-                CONF_CLIENT_KEY_PASSWORD,
-                description={
-                    "suggested_value": user_input_basic.get(CONF_CLIENT_KEY_PASSWORD)
-                },
-            )
-        ] = PASSWORD_SELECTOR
-    verification_mode = current_config.get(SET_CA_CERT) or (
-        "off"
-        if current_ca_certificate is None
-        else "auto"
-        if current_ca_certificate == "auto"
-        else "custom"
-    )
-    fields[
-        vol.Optional(
-            SET_CA_CERT,
-            default=verification_mode,
-        )
-    ] = BROKER_VERIFICATION_SELECTOR
-    if current_ca_certificate is not None or verification_mode == "custom":
-        fields[
-            vol.Optional(
-                CONF_CERTIFICATE,
-                user_input_basic.get(CONF_CERTIFICATE),
-            )
-        ] = CA_CERT_UPLOAD_SELECTOR
-    fields[
-        vol.Optional(
-            CONF_TLS_INSECURE,
-            description={"suggested_value": current_tls_insecure},
-        )
-    ] = BOOLEAN_SELECTOR
-    fields[
-        vol.Optional(
-            CONF_TRANSPORT,
-            description={"suggested_value": current_transport},
-        )
-    ] = TRANSPORT_SELECTOR
-    if current_transport == TRANSPORT_WEBSOCKETS:
-        fields[
-            vol.Optional(CONF_WS_PATH, description={"suggested_value": current_ws_path})
-        ] = TEXT_SELECTOR
-        fields[
-            vol.Optional(
-                CONF_WS_HEADERS, description={"suggested_value": current_ws_headers}
-            )
-        ] = WS_HEADERS_SELECTOR
-
-    # Show form
+    errors["base"] = "cannot_connect"
     return False
 
 

homeassistant/components/mqtt/const.py

@@ -315,7 +315,6 @@ DEFAULT_TILT_MAX = 100
 DEFAULT_TILT_MIN = 0
 DEFAULT_TILT_OPEN_POSITION = 100
 DEFAULT_TILT_OPTIMISTIC = False
-DEFAULT_WS_HEADERS: dict[str, str] = {}
 DEFAULT_WS_PATH = "/"
 DEFAULT_POSITION_CLOSED = 0
 DEFAULT_POSITION_OPEN = 100

homeassistant/components/mqtt/strings.json

@@ -26,46 +26,53 @@
     "step": {
       "broker": {
         "data": {
-          "advanced_options": "Advanced options",
           "broker": "Broker",
-          "certificate": "Upload custom CA certificate file",
-          "client_cert": "Upload client certificate file",
-          "client_id": "Client ID (leave empty to randomly generated one)",
-          "client_key": "Upload private key file",
-          "client_key_password": "[%key:common::config_flow::data::password%]",
-          "keepalive": "The time between sending keep alive messages",
           "password": "[%key:common::config_flow::data::password%]",
           "port": "[%key:common::config_flow::data::port%]",
           "protocol": "MQTT protocol",
-          "set_ca_cert": "Broker certificate validation",
-          "set_client_cert": "Use a client certificate",
-          "tls_insecure": "Ignore broker certificate validation",
-          "transport": "MQTT transport",
-          "username": "[%key:common::config_flow::data::username%]",
-          "ws_headers": "WebSocket headers in JSON format",
-          "ws_path": "WebSocket path"
+          "username": "[%key:common::config_flow::data::username%]"
         },
         "data_description": {
-          "advanced_options": "Enable and select **Submit** to set advanced options.",
           "broker": "The hostname or IP address of your MQTT broker.",
-          "certificate": "The custom CA certificate file to validate your MQTT broker's certificate.",
-          "client_cert": "The client certificate to authenticate against your MQTT broker.",
-          "client_id": "The unique ID to identify the Home Assistant MQTT API as MQTT client. It is recommended to leave this option blank.",
-          "client_key": "The private key file that belongs to your client certificate.",
-          "client_key_password": "The password for the private key file (if set).",
-          "keepalive": "A value less than 90 seconds is advised.",
           "password": "The password to log in to your MQTT broker.",
           "port": "The port your MQTT broker listens to. For example 1883.",
           "protocol": "The MQTT protocol version that is used. Note that Home Assistant will silently change to version 5 if your broker supports it.",
-          "set_ca_cert": "Select **Auto** for automatic CA validation, or **Custom** and select **Next** to set a custom CA certificate, to allow validating your MQTT broker's certificate.",
-          "set_client_cert": "Enable and select **Next** to set a client certificate and private key to authenticate against your MQTT broker.",
-          "tls_insecure": "Option to ignore validation of your MQTT broker's certificate.",
-          "transport": "The transport to be used for the connection to your MQTT broker.",
-          "username": "The username to log in to your MQTT broker.",
-          "ws_headers": "The WebSocket headers to pass through the WebSocket-based connection to your MQTT broker.",
-          "ws_path": "The WebSocket path to be used for the connection to your MQTT broker."
+          "username": "The username to log in to your MQTT broker."
         },
-        "description": "Please enter the connection information of your MQTT broker."
+        "description": "Please enter the connection information of your MQTT broker.",
+        "sections": {
+          "other_settings": {
+            "data": {
+              "certificate": "Upload custom CA certificate file",
+              "client_cert": "Upload client certificate file",
+              "client_id": "Client ID (leave empty to randomly generated one)",
+              "client_key": "Upload private key file",
+              "client_key_password": "[%key:common::config_flow::data::password%]",
+              "keepalive": "The time between sending keep alive messages",
+              "set_ca_cert": "Broker certificate validation",
+              "set_client_cert": "Use a client certificate",
+              "tls_insecure": "Ignore broker certificate validation",
+              "transport": "MQTT transport",
+              "ws_headers": "WebSocket headers in JSON format",
+              "ws_path": "WebSocket path"
+            },
+            "data_description": {
+              "certificate": "The custom CA certificate file to validate your MQTT broker's certificate.",
+              "client_cert": "The client certificate to authenticate against your MQTT broker.",
+              "client_id": "The unique ID to identify the Home Assistant MQTT API as MQTT client. It is recommended to leave this option blank.",
+              "client_key": "The private key file that belongs to your client certificate.",
+              "client_key_password": "The password for the private key file (if set).",
+              "keepalive": "A value less than 90 seconds is advised. Defaults to 60 seconds.",
+              "set_ca_cert": "When already set to **Custom**, a custom CA validation certificate is configured. Select **Auto** for automatic CA validation, or upload a custom CA certificate, to allow validating your MQTT broker's certificate.",
+              "set_client_cert": "When already selected, client certificate authentication is enabled. Upload a client certificate and key to enable.",
+              "tls_insecure": "Option to ignore validation of your MQTT broker's certificate.",
+              "transport": "The transport to be used for the connection to your MQTT broker.",
+              "ws_headers": "The WebSocket headers to pass through the WebSocket-based connection to your MQTT broker.",
+              "ws_path": "The WebSocket path to be used for the connection to your MQTT broker."
+            },
+            "name": "Other settings"
+          }
+        }
       },
       "hassio_confirm": {
         "description": "Do you want to configure Home Assistant to connect to the MQTT broker provided by the {addon} app?",
@@ -1178,48 +1185,6 @@
       "invalid_inclusion": "[%key:component::mqtt::config::error::invalid_inclusion%]"
     },
     "step": {
-      "broker": {
-        "data": {
-          "advanced_options": "[%key:component::mqtt::config::step::broker::data::advanced_options%]",
-          "broker": "[%key:component::mqtt::config::step::broker::data::broker%]",
-          "certificate": "[%key:component::mqtt::config::step::broker::data::certificate%]",
-          "client_cert": "[%key:component::mqtt::config::step::broker::data::client_cert%]",
-          "client_id": "[%key:component::mqtt::config::step::broker::data::client_id%]",
-          "client_key": "[%key:component::mqtt::config::step::broker::data::client_key%]",
-          "keepalive": "[%key:component::mqtt::config::step::broker::data::keepalive%]",
-          "password": "[%key:common::config_flow::data::password%]",
-          "port": "[%key:common::config_flow::data::port%]",
-          "protocol": "[%key:component::mqtt::config::step::broker::data::protocol%]",
-          "set_ca_cert": "[%key:component::mqtt::config::step::broker::data::set_ca_cert%]",
-          "set_client_cert": "[%key:component::mqtt::config::step::broker::data::set_client_cert%]",
-          "tls_insecure": "[%key:component::mqtt::config::step::broker::data::tls_insecure%]",
-          "transport": "[%key:component::mqtt::config::step::broker::data::transport%]",
-          "username": "[%key:common::config_flow::data::username%]",
-          "ws_headers": "[%key:component::mqtt::config::step::broker::data::ws_headers%]",
-          "ws_path": "[%key:component::mqtt::config::step::broker::data::ws_path%]"
-        },
-        "data_description": {
-          "advanced_options": "[%key:component::mqtt::config::step::broker::data_description::advanced_options%]",
-          "broker": "[%key:component::mqtt::config::step::broker::data_description::broker%]",
-          "certificate": "[%key:component::mqtt::config::step::broker::data_description::certificate%]",
-          "client_cert": "[%key:component::mqtt::config::step::broker::data_description::client_cert%]",
-          "client_id": "[%key:component::mqtt::config::step::broker::data_description::client_id%]",
-          "client_key": "[%key:component::mqtt::config::step::broker::data_description::client_key%]",
-          "keepalive": "[%key:component::mqtt::config::step::broker::data_description::keepalive%]",
-          "password": "[%key:component::mqtt::config::step::broker::data_description::password%]",
-          "port": "[%key:component::mqtt::config::step::broker::data_description::port%]",
-          "protocol": "[%key:component::mqtt::config::step::broker::data_description::protocol%]",
-          "set_ca_cert": "[%key:component::mqtt::config::step::broker::data_description::set_ca_cert%]",
-          "set_client_cert": "[%key:component::mqtt::config::step::broker::data_description::set_client_cert%]",
-          "tls_insecure": "[%key:component::mqtt::config::step::broker::data_description::tls_insecure%]",
-          "transport": "[%key:component::mqtt::config::step::broker::data_description::transport%]",
-          "username": "[%key:component::mqtt::config::step::broker::data_description::username%]",
-          "ws_headers": "[%key:component::mqtt::config::step::broker::data_description::ws_headers%]",
-          "ws_path": "[%key:component::mqtt::config::step::broker::data_description::ws_path%]"
-        },
-        "description": "[%key:component::mqtt::config::step::broker::description%]",
-        "title": "Broker options"
-      },
       "options": {
         "data": {
           "birth_enable": "Enable birth message",

tests/components/mqtt/test_config_flow.py

@@ -87,6 +87,23 @@ ADD_ON_DISCOVERY_INFO = {
     "ssl": False,
 }
 
+MOCK_BROKER_FORM_DATA = {
+    "broker": "127.0.0.1",
+    "port": "1883",
+    "protocol": "5",
+    "other_settings": {
+        "transport": "tcp",
+        "set_client_cert": False,
+        "set_ca_cert": "off",
+    },
+}
+MOCK_BROKER_ENTRY_DATA = {
+    "broker": "127.0.0.1",
+    "port": 1883,
+    "protocol": "5",
+    "transport": "tcp",
+}
+
 MOCK_CA_CERT = (
     b"-----BEGIN CERTIFICATE-----\n"
     b"## mock CA certificate file ##"
@@ -128,13 +145,6 @@ MOCK_CLIENT_KEY_DER = b"## mock DER formatted key file ##\n"
 MOCK_ENCRYPTED_CLIENT_KEY_DER = b"## mock DER formatted encrypted key file ##\n"
 
 
-MOCK_ENTRY_DATA = {
-    mqtt.CONF_BROKER: "test-broker",
-    CONF_PROTOCOL: "5",
-    CONF_PORT: 1234,
-    CONF_USERNAME: "user",
-    CONF_PASSWORD: "pass",
-}
 MOCK_ENTRY_OPTIONS = {
     mqtt.CONF_DISCOVERY: True,
     mqtt.CONF_BIRTH_MESSAGE: {
@@ -380,15 +390,11 @@ async def test_user_connection_works(
     assert result["type"] is FlowResultType.FORM
 
     result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"broker": "127.0.0.1"}
+        result["flow_id"], MOCK_BROKER_FORM_DATA
     )
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["result"].data == {
-        "broker": "127.0.0.1",
-        "protocol": "5",
-        "port": 1883,
-    }
+    assert result["result"].data == MOCK_BROKER_ENTRY_DATA
     # Check we have the latest Config Entry version
     assert result["result"].version == 2
     assert result["result"].minor_version == 1
@@ -423,15 +429,11 @@ async def test_user_connection_works_with_supervisor(
     assert result["type"] is FlowResultType.FORM
 
     result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"broker": "127.0.0.1"}
+        result["flow_id"], MOCK_BROKER_FORM_DATA
     )
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["result"].data == {
-        "broker": "127.0.0.1",
-        "protocol": "5",
-        "port": 1883,
-    }
+    assert result["result"].data == MOCK_BROKER_ENTRY_DATA
     # Check we tried the connection
     assert len(mock_try_connection.mock_calls) == 1
     # Check config entry got setup
@@ -453,27 +455,14 @@ async def test_user_v5_connection_works(
         context={"source": config_entries.SOURCE_USER},
     )
     assert result["type"] is FlowResultType.FORM
-
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"broker": "127.0.0.1", "advanced_options": True}
-    )
-
     assert result["step_id"] == "broker"
 
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "another-broker",
-            CONF_PORT: 2345,
-            CONF_PROTOCOL: "5",
-        },
+        user_input=MOCK_BROKER_FORM_DATA,
     )
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["result"].data == {
-        "broker": "another-broker",
-        "port": 2345,
-        "protocol": "5",
-    }
+    assert result["result"].data == MOCK_BROKER_ENTRY_DATA
     # Check we tried the connection
     assert len(mock_try_connection.mock_calls) == 1
     # Check config entry got setup
@@ -492,14 +481,16 @@ async def test_user_connection_fails(
     assert result["type"] is FlowResultType.FORM
 
     result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"broker": "127.0.0.1"}
+        result["flow_id"], MOCK_BROKER_FORM_DATA | {"broker": "127.0.0.1"}
     )
 
     assert result["type"] is FlowResultType.FORM
     assert result["errors"]["base"] == "cannot_connect"
 
     # Check we tried the connection
-    assert len(mock_try_connection_time_out.mock_calls)
+    mock_try_connection_time_out.connect_async.assert_called_once_with(
+        "127.0.0.1", 1883
+    )
     # Check config entry did not setup
     assert len(mock_finish_setup.mock_calls) == 0
 
@@ -510,7 +501,7 @@ async def test_manual_config_set(
     mock_try_connection: MqttMockPahoClient,
     mock_finish_setup: MagicMock,
 ) -> None:
-    """Test manual config does not create an entry, and entry can be setup late."""
+    """Test manual config does not create an entry, and entry can be set up late."""
     assert len(mock_finish_setup.mock_calls) == 0
 
     mock_try_connection.return_value = True
@@ -521,28 +512,19 @@ async def test_manual_config_set(
     )
     assert result["type"] is FlowResultType.FORM
 
+    # Submit form data
     result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"broker": "127.0.0.1", "port": "1883"}
+        result["flow_id"], MOCK_BROKER_FORM_DATA
     )
-
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["result"].data == {
-        "broker": "127.0.0.1",
-        "protocol": "5",
-        "port": 1883,
-    }
+    assert result["result"].data == MOCK_BROKER_ENTRY_DATA
+
     # Check we tried the connection, with precedence for config entry settings
-    mock_try_connection.assert_called_once_with(
-        {
-            "broker": "127.0.0.1",
-            "protocol": "5",
-            "port": 1883,
-        },
-    )
+    mock_try_connection.assert_called_once_with(MOCK_BROKER_ENTRY_DATA)
     # Check config entry got setup
     assert len(mock_finish_setup.mock_calls) == 1
     config_entry = hass.config_entries.async_entries(DOMAIN)[0]
-    assert config_entry.title == "127.0.0.1"
+    assert config_entry.title == MOCK_BROKER_ENTRY_DATA["broker"]
 
 
 async def test_user_single_instance(hass: HomeAssistant) -> None:
@@ -732,7 +714,6 @@ async def test_addon_flow_with_supervisor_addon_running(
         "port": 1883,
         "username": "mock-user",
         "password": "mock-pass",
-        "discovery": True,
     }
     # Check we tried the connection
     assert len(mock_try_connection_success.mock_calls)
@@ -800,7 +781,6 @@ async def test_addon_flow_with_supervisor_addon_installed(
         "port": 1883,
         "username": "mock-user",
         "password": "mock-pass",
-        "discovery": True,
     }
     # Check we tried the connection
     assert len(mock_try_connection_success.mock_calls)
@@ -1040,7 +1020,6 @@ async def test_addon_flow_with_supervisor_addon_not_installed(
         "port": 1883,
         "username": "mock-user",
         "password": "mock-pass",
-        "discovery": True,
     }
     # Check we tried the connection
     assert len(mock_try_connection_success.mock_calls)
@@ -1159,6 +1138,7 @@ async def test_option_flow(
     assert yaml_mock.await_count
 
 
+@pytest.mark.usefixtures("mock_ca_cert")
 @pytest.mark.parametrize(
     ("mock_ca_cert", "mock_client_cert", "mock_client_key", "client_key_password"),
     [
@@ -1195,7 +1175,7 @@ async def test_option_flow(
         None,
     ],
 )
-async def test_bad_certificate(
+async def test_bad_certificate_validation(
     hass: HomeAssistant,
     mqtt_mock_entry: MqttMockHAClientGenerator,
     mock_try_connection_success: MqttMockPahoClient,
@@ -1203,9 +1183,8 @@ async def test_bad_certificate(
     mock_process_uploaded_file: MagicMock,
     test_error: str | None,
     client_key_password: str,
-    mock_ca_cert: bytes,
 ) -> None:
-    """Test bad certificate tests."""
+    """Test bad certificate validation in config and reconfig flow."""
 
     def _side_effect_on_client_cert(data: bytes) -> MagicMock:
         """Raise on client cert only.
@@ -1227,12 +1206,16 @@ async def test_bad_certificate(
     test_input = {
         mqtt.CONF_BROKER: "another-broker",
         CONF_PORT: 2345,
-        mqtt.CONF_CERTIFICATE: file_id[mqtt.CONF_CERTIFICATE],
-        mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
-        mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
-        "client_key_password": client_key_password,
-        "set_ca_cert": set_ca_cert,
-        "set_client_cert": True,
+        mqtt.CONF_PROTOCOL: "5",
+        "other_settings": {
+            mqtt.CONF_CERTIFICATE: file_id[mqtt.CONF_CERTIFICATE],
+            mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
+            mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
+            "client_key_password": client_key_password,
+            "set_ca_cert": set_ca_cert,
+            "set_client_cert": True,
+            mqtt.CONF_TRANSPORT: "tcp",
+        },
     }
     if test_error == "bad_certificate":
         # CA chain is not loading
@@ -1255,22 +1238,16 @@ async def test_bad_certificate(
         mock_ssl_context["context"]().load_cert_chain.side_effect = SSLError
     elif test_error == "invalid_inclusion":
         # Client key file without client cert, client cert without key file
-        test_input.pop(mqtt.CONF_CLIENT_KEY)
+        test_input["other_settings"].pop(mqtt.CONF_CLIENT_KEY)
 
+    test_input["other_settings"]["set_client_cert"] = set_client_cert
+    test_input["other_settings"]["set_ca_cert"] = set_ca_cert
+    test_input["other_settings"]["tls_insecure"] = tls_insecure
+
+    # Test errors in reconfigure flow
     mqtt_mock = await mqtt_mock_entry()
     config_entry: MockConfigEntry = hass.config_entries.async_entries(DOMAIN)[0]
-    # Add at least one advanced option to get the full form
-    hass.config_entries.async_update_entry(
-        config_entry,
-        data={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 1234,
-            CONF_CLIENT_ID: "custom1234",
-            mqtt.CONF_KEEPALIVE: 60,
-            mqtt.CONF_TLS_INSECURE: False,
-            CONF_PROTOCOL: "3.1.1",
-        },
-    )
+    hass.config_entries.async_update_entry(config_entry, data=MOCK_BROKER_ENTRY_DATA)
     await hass.async_block_till_done()
 
     mqtt_mock.async_connect.reset_mock()
@@ -1279,23 +1256,6 @@ async def test_bad_certificate(
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "broker"
 
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "another-broker",
-            CONF_PORT: 2345,
-            mqtt.CONF_KEEPALIVE: 60,
-            "set_client_cert": set_client_cert,
-            "set_ca_cert": set_ca_cert,
-            mqtt.CONF_TLS_INSECURE: tls_insecure,
-            CONF_PROTOCOL: "3.1.1",
-            CONF_CLIENT_ID: "custom1234",
-        },
-    )
-    test_input["set_client_cert"] = set_client_cert
-    test_input["set_ca_cert"] = set_ca_cert
-    test_input["tls_insecure"] = tls_insecure
-
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input=test_input,
@@ -1305,6 +1265,23 @@ async def test_bad_certificate(
         return
     assert "errors" not in result
 
+    # Remove existing MQTT config entry
+    await hass.config_entries.async_remove(config_entry.entry_id)
+    await hass.async_block_till_done()
+    # Test config flow
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.FORM
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input=test_input
+    )
+    await hass.async_block_till_done()
+    if test_error is not None:
+        assert result["errors"]["base"] == test_error
+        return
+    assert "errors" not in result
+
 
 @pytest.mark.parametrize(
     ("input_value", "error"),
@@ -1327,25 +1304,12 @@ async def test_keepalive_validation(
 ) -> None:
     """Test validation of the keep alive option."""
 
-    test_input = {
-        mqtt.CONF_BROKER: "another-broker",
-        CONF_PORT: 2345,
-        mqtt.CONF_KEEPALIVE: input_value,
-    }
+    test_input = deepcopy(MOCK_BROKER_FORM_DATA)
+    test_input["other_settings"][mqtt.CONF_KEEPALIVE] = input_value
 
     mqtt_mock = await mqtt_mock_entry()
     mock_try_connection.return_value = True
     config_entry: MockConfigEntry = hass.config_entries.async_entries(DOMAIN)[0]
-    # Add at least one advanced option to get the full form
-    hass.config_entries.async_update_entry(
-        config_entry,
-        data={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 1234,
-            CONF_CLIENT_ID: "custom1234",
-        },
-    )
-
     mqtt_mock.async_connect.reset_mock()
 
     result = await config_entry.start_reconfigure_flow(hass)
@@ -1860,10 +1824,7 @@ async def test_reconfigure_user_connection_fails(
     config_entry.add_to_hass(hass)
     hass.config_entries.async_update_entry(
         config_entry,
-        data={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 1234,
-        },
+        data=MOCK_BROKER_ENTRY_DATA,
     )
     result = await config_entry.start_reconfigure_flow(hass)
     assert result["type"] is FlowResultType.FORM
@@ -1871,19 +1832,19 @@ async def test_reconfigure_user_connection_fails(
     mock_try_connection_time_out.reset_mock()
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
-        user_input={mqtt.CONF_BROKER: "bad-broker", CONF_PORT: 2345},
+        user_input=MOCK_BROKER_FORM_DATA
+        | {mqtt.CONF_BROKER: "bad-broker", CONF_PORT: 2345},
     )
 
     assert result["type"] is FlowResultType.FORM
     assert result["errors"]["base"] == "cannot_connect"
 
     # Check we tried the connection
-    assert len(mock_try_connection_time_out.mock_calls)
+    mock_try_connection_time_out.connect_async.assert_called_once_with(
+        "bad-broker", 2345
+    )
     # Check config entry did not update
-    assert config_entry.data == {
-        mqtt.CONF_BROKER: "test-broker",
-        CONF_PORT: 1234,
-    }
+    assert config_entry.data == MOCK_BROKER_ENTRY_DATA
 
 
 async def test_options_bad_birth_message_fails(
@@ -1967,12 +1928,15 @@ async def test_options_bad_will_message_fails(
     [MOCK_CLIENT_KEY, MOCK_EC_CLIENT_KEY, MOCK_RSA_CLIENT_KEY],
 )
 @pytest.mark.usefixtures("mock_ssl_context", "mock_process_uploaded_file")
-async def test_try_connection_with_advanced_parameters(
+async def test_reconfigure_with_tls_client_key_formats(
     hass: HomeAssistant,
     mock_try_connection_success: MqttMockPahoClient,
     mock_context_client_key: bytes,
 ) -> None:
-    """Test config flow with advanced parameters from config."""
+    """Test config flow with different PEM client keys.
+
+    Also test if default and suggested values are loaded correctly.
+    """
     config_entry = MockConfigEntry(
         domain=DOMAIN,
         version=mqtt.CONFIG_ENTRY_VERSION,
@@ -1995,19 +1959,6 @@ async def test_try_connection_with_advanced_parameters(
             mqtt.CONF_WS_PATH: "/path/",
             mqtt.CONF_WS_HEADERS: {"h1": "v1", "h2": "v2"},
             mqtt.CONF_KEEPALIVE: 30,
-            mqtt.CONF_DISCOVERY: True,
-            mqtt.CONF_BIRTH_MESSAGE: {
-                mqtt.ATTR_TOPIC: "ha_state/online",
-                mqtt.ATTR_PAYLOAD: "online",
-                mqtt.ATTR_QOS: 1,
-                mqtt.ATTR_RETAIN: True,
-            },
-            mqtt.CONF_WILL_MESSAGE: {
-                mqtt.ATTR_TOPIC: "ha_state/offline",
-                mqtt.ATTR_PAYLOAD: "offline",
-                mqtt.ATTR_QOS: 2,
-                mqtt.ATTR_RETAIN: False,
-            },
         },
     )
 
@@ -2015,25 +1966,32 @@ async def test_try_connection_with_advanced_parameters(
     result = await config_entry.start_reconfigure_flow(hass)
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "broker"
-    defaults = {
-        mqtt.CONF_BROKER: "test-broker",
-        CONF_PORT: 1234,
-        "set_client_cert": True,
-        "set_ca_cert": "auto",
-    }
+    defaults = {CONF_PORT: 1883}
     suggested = {
         CONF_USERNAME: "user",
         CONF_PASSWORD: PWD_NOT_CHANGED,
-        mqtt.CONF_TLS_INSECURE: True,
+        mqtt.CONF_TLS_INSECURE: None,
         CONF_PROTOCOL: "5",
-        mqtt.CONF_TRANSPORT: "websockets",
-        mqtt.CONF_WS_PATH: "/path/",
-        mqtt.CONF_WS_HEADERS: '{"h1":"v1","h2":"v2"}',
     }
+    suggested_other_settings = {
+        "username": None,
+        "password": None,
+        "tls_insecure": True,
+        "protocol": None,
+        "other_settings": None,
+    }
+
     for k, v in defaults.items():
         assert get_default(result["data_schema"].schema, k) == v
     for k, v in suggested.items():
         assert get_schema_suggested_value(result["data_schema"].schema, k) == v
+    for k, v in suggested_other_settings.items():
+        assert (
+            get_schema_suggested_value(
+                result["data_schema"].schema["other_settings"].schema.schema, k
+            )
+            == v
+        )
 
     # test we can change username and password
     mock_try_connection_success.reset_mock()
@@ -2042,14 +2000,17 @@ async def test_try_connection_with_advanced_parameters(
         user_input={
             mqtt.CONF_BROKER: "another-broker",
             CONF_PORT: 2345,
+            CONF_PROTOCOL: "5",
             CONF_USERNAME: "us3r",
             CONF_PASSWORD: "p4ss",
-            "set_ca_cert": "auto",
-            "set_client_cert": True,
-            mqtt.CONF_TLS_INSECURE: True,
-            mqtt.CONF_TRANSPORT: "websockets",
-            mqtt.CONF_WS_PATH: "/new/path",
-            mqtt.CONF_WS_HEADERS: '{"h3": "v3"}',
+            "other_settings": {
+                "set_ca_cert": "auto",
+                "set_client_cert": True,
+                mqtt.CONF_TLS_INSECURE: True,
+                mqtt.CONF_TRANSPORT: "websockets",
+                mqtt.CONF_WS_PATH: "/new/path",
+                mqtt.CONF_WS_HEADERS: '{"h3": "v3"}',
+            },
         },
     )
     assert result["type"] is FlowResultType.ABORT
@@ -2117,23 +2078,14 @@ async def test_setup_with_advanced_settings(
         minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
     )
     config_entry.add_to_hass(hass)
-    hass.config_entries.async_update_entry(
-        config_entry,
-        data={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PROTOCOL: "5",
-            CONF_PORT: 1234,
-        },
-    )
+    hass.config_entries.async_update_entry(config_entry, data=MOCK_BROKER_ENTRY_DATA)
 
     mock_try_connection.return_value = True
 
     result = await config_entry.start_reconfigure_flow(hass)
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "broker"
-    assert result["data_schema"].schema["advanced_options"]
 
-    # first iteration, basic settings
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
@@ -2141,100 +2093,66 @@ async def test_setup_with_advanced_settings(
             CONF_PORT: 2345,
             CONF_USERNAME: "user",
             CONF_PASSWORD: "secret",
-            "advanced_options": True,
+            CONF_PROTOCOL: "5",
+            "other_settings": {
+                mqtt.CONF_KEEPALIVE: 30,
+                "set_ca_cert": "auto",
+                "set_client_cert": True,
+                mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
+                mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
+                mqtt.CONF_TLS_INSECURE: True,
+                mqtt.CONF_TRANSPORT: "websockets",
+                mqtt.CONF_WS_PATH: "/custom_path/",
+                mqtt.CONF_WS_HEADERS: (
+                    '{"header_1": "content_header_1", "header_2": "content_header_2"'
+                ),
+            },
         },
     )
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "broker"
-    assert "advanced_options" not in result["data_schema"].schema
-    assert result["data_schema"].schema[CONF_CLIENT_ID]
-    assert result["data_schema"].schema[mqtt.CONF_KEEPALIVE]
-    assert result["data_schema"].schema["set_client_cert"]
-    assert result["data_schema"].schema["set_ca_cert"]
-    assert result["data_schema"].schema[mqtt.CONF_TLS_INSECURE]
+
     assert result["data_schema"].schema[CONF_PROTOCOL]
-    assert result["data_schema"].schema[mqtt.CONF_TRANSPORT]
-    assert mqtt.CONF_CLIENT_CERT not in result["data_schema"].schema
-    assert mqtt.CONF_CLIENT_KEY not in result["data_schema"].schema
 
-    # second iteration, advanced settings with request for client cert
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 2345,
-            CONF_USERNAME: "user",
-            CONF_PASSWORD: "secret",
-            mqtt.CONF_KEEPALIVE: 30,
-            "set_ca_cert": "auto",
-            "set_client_cert": True,
-            mqtt.CONF_TLS_INSECURE: True,
-            CONF_PROTOCOL: "3.1.1",
-            mqtt.CONF_TRANSPORT: "websockets",
-        },
-    )
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "broker"
-    assert "advanced_options" not in result["data_schema"].schema
-    assert result["data_schema"].schema[CONF_CLIENT_ID]
-    assert result["data_schema"].schema[mqtt.CONF_KEEPALIVE]
-    assert result["data_schema"].schema["set_client_cert"]
-    assert result["data_schema"].schema["set_ca_cert"]
-    assert result["data_schema"].schema[mqtt.CONF_TLS_INSECURE]
-    assert result["data_schema"].schema[CONF_PROTOCOL]
-    assert result["data_schema"].schema[mqtt.CONF_CLIENT_CERT]
-    assert result["data_schema"].schema[mqtt.CONF_CLIENT_KEY]
-    assert result["data_schema"].schema[mqtt.CONF_TRANSPORT]
-    assert result["data_schema"].schema[mqtt.CONF_WS_PATH]
-    assert result["data_schema"].schema[mqtt.CONF_WS_HEADERS]
-
-    # third iteration, advanced settings with client cert and key
-    # set and bad json payload
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 2345,
-            CONF_USERNAME: "user",
-            CONF_PASSWORD: "secret",
-            mqtt.CONF_KEEPALIVE: 30,
-            "set_ca_cert": "auto",
-            "set_client_cert": True,
-            mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
-            mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
-            mqtt.CONF_TLS_INSECURE: True,
-            mqtt.CONF_TRANSPORT: "websockets",
-            mqtt.CONF_WS_PATH: "/custom_path/",
-            mqtt.CONF_WS_HEADERS: (
-                '{"header_1": "content_header_1", "header_2": "content_header_2"'
-            ),
-        },
-    )
+    for key in (
+        CONF_CLIENT_ID,
+        mqtt.CONF_KEEPALIVE,
+        "set_client_cert",
+        "set_ca_cert",
+        mqtt.CONF_TLS_INSECURE,
+        mqtt.CONF_CLIENT_CERT,
+        mqtt.CONF_CLIENT_KEY,
+        mqtt.CONF_TRANSPORT,
+        mqtt.CONF_WS_PATH,
+        mqtt.CONF_WS_HEADERS,
+    ):
+        assert result["data_schema"].schema["other_settings"].schema.schema[key]
 
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "broker"
     assert result["errors"]["base"] == "bad_ws_headers"
 
-    # fourth iteration, advanced settings with client cert and key set
+    # next iteration, with client cert and key set
     # and correct json payload for ws_headers
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
             mqtt.CONF_BROKER: "test-broker",
             CONF_PORT: 2345,
+            CONF_PROTOCOL: "5",
             CONF_USERNAME: "user",
             CONF_PASSWORD: "secret",
-            mqtt.CONF_KEEPALIVE: 30,
-            "set_ca_cert": "auto",
-            "set_client_cert": True,
-            mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
-            mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
-            mqtt.CONF_TLS_INSECURE: True,
-            mqtt.CONF_TRANSPORT: "websockets",
-            mqtt.CONF_WS_PATH: "/custom_path/",
-            mqtt.CONF_WS_HEADERS: (
-                '{"header_1": "content_header_1", "header_2": "content_header_2"}'
-            ),
+            "other_settings": {
+                mqtt.CONF_KEEPALIVE: 30,
+                "set_ca_cert": "auto",
+                "set_client_cert": True,
+                mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
+                mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
+                mqtt.CONF_TLS_INSECURE: True,
+                mqtt.CONF_TRANSPORT: "websockets",
+                mqtt.CONF_WS_PATH: "/custom_path/",
+                mqtt.CONF_WS_HEADERS: (
+                    '{"header_1": "content_header_1", "header_2": "content_header_2"}'
+                ),
+            },
         },
     )
 
@@ -2297,111 +2215,103 @@ async def test_setup_with_certificates(
     """Test config flow setup with PEM and DER encoded certificates."""
     file_id = mock_process_uploaded_file.file_id
 
-    config_entry = MockConfigEntry(
-        domain=DOMAIN,
-        version=mqtt.CONFIG_ENTRY_VERSION,
-        minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
-    config_entry.add_to_hass(hass)
-    hass.config_entries.async_update_entry(
-        config_entry,
-        data={
+    assert result["type"] is FlowResultType.FORM
+    mock_try_connection.return_value = False
+
+    # Flow raises an error with stale file id's
+    # This test is just for coverage purpose
+    with (
+        patch(
+            "homeassistant.components.mqtt.config_flow.process_uploaded_file",
+            side_effect=ValueError("File does not exist"),
+        ),
+        pytest.raises(ValueError),
+    ):
+        await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                mqtt.CONF_BROKER: "test-broker",
+                CONF_PORT: 2345,
+                CONF_PROTOCOL: "5",
+                CONF_USERNAME: "user",
+                CONF_PASSWORD: "secret",
+                "other_settings": {
+                    mqtt.CONF_KEEPALIVE: 30,
+                    "set_ca_cert": "custom",
+                    "set_client_cert": True,
+                    "client_key_password": client_key_password,
+                    mqtt.CONF_CERTIFICATE: str(uuid4()),
+                    mqtt.CONF_CLIENT_CERT: str(uuid4()),
+                    mqtt.CONF_CLIENT_KEY: str(uuid4()),
+                    mqtt.CONF_TLS_INSECURE: False,
+                    mqtt.CONF_TRANSPORT: "tcp",
+                },
+            },
+        )
+
+    # Repeat the test with valid files, but connection fails
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
             mqtt.CONF_BROKER: "test-broker",
+            CONF_PORT: 2345,
             CONF_PROTOCOL: "5",
-            CONF_PORT: 1234,
+            CONF_USERNAME: "user",
+            CONF_PASSWORD: "secret",
+            "other_settings": {
+                mqtt.CONF_KEEPALIVE: 30,
+                "set_ca_cert": "custom",
+                "set_client_cert": True,
+                "client_key_password": client_key_password,
+                mqtt.CONF_CERTIFICATE: file_id[mqtt.CONF_CERTIFICATE],
+                mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
+                mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
+                mqtt.CONF_TLS_INSECURE: False,
+                mqtt.CONF_TRANSPORT: "tcp",
+            },
         },
     )
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"] == {"base": "cannot_connect"}
 
+    # Now retry, but using the preserved uploaded files
     mock_try_connection.return_value = True
+    with patch(
+        "homeassistant.components.mqtt.config_flow.process_uploaded_file",
+        side_effect=ValueError("File does not exist"),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                mqtt.CONF_BROKER: "test-broker",
+                CONF_PORT: 1234,
+                CONF_PROTOCOL: "5",
+                CONF_USERNAME: "user",
+                CONF_PASSWORD: "secret",
+                "other_settings": {
+                    mqtt.CONF_KEEPALIVE: 30,
+                    "set_ca_cert": "custom",
+                    "set_client_cert": True,
+                    "client_key_password": client_key_password,
+                    mqtt.CONF_CERTIFICATE: file_id[mqtt.CONF_CERTIFICATE],
+                    mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
+                    mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
+                    mqtt.CONF_TLS_INSECURE: False,
+                    mqtt.CONF_TRANSPORT: "tcp",
+                },
+            },
+        )
 
-    result = await config_entry.start_reconfigure_flow(hass)
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "broker"
-    assert result["data_schema"].schema["advanced_options"]
+    assert result["type"] is FlowResultType.CREATE_ENTRY
 
-    # first iteration, basic settings
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 2345,
-            CONF_USERNAME: "user",
-            CONF_PASSWORD: "secret",
-            "advanced_options": True,
-        },
-    )
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "broker"
-    assert "advanced_options" not in result["data_schema"].schema
-    assert result["data_schema"].schema[CONF_CLIENT_ID]
-    assert result["data_schema"].schema[mqtt.CONF_KEEPALIVE]
-    assert result["data_schema"].schema["set_client_cert"]
-    assert result["data_schema"].schema["set_ca_cert"]
-    assert result["data_schema"].schema[mqtt.CONF_TLS_INSECURE]
-    assert result["data_schema"].schema[CONF_PROTOCOL]
-    assert result["data_schema"].schema[mqtt.CONF_TRANSPORT]
-    assert mqtt.CONF_CLIENT_CERT not in result["data_schema"].schema
-    assert mqtt.CONF_CLIENT_KEY not in result["data_schema"].schema
-
-    # second iteration, advanced settings with request for client cert
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 2345,
-            CONF_USERNAME: "user",
-            CONF_PASSWORD: "secret",
-            mqtt.CONF_KEEPALIVE: 30,
-            "set_ca_cert": "custom",
-            "set_client_cert": True,
-            mqtt.CONF_TLS_INSECURE: False,
-            CONF_PROTOCOL: "5",
-            mqtt.CONF_TRANSPORT: "tcp",
-        },
-    )
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "broker"
-    assert "advanced_options" not in result["data_schema"].schema
-    assert result["data_schema"].schema[CONF_CLIENT_ID]
-    assert result["data_schema"].schema[mqtt.CONF_KEEPALIVE]
-    assert result["data_schema"].schema["set_client_cert"]
-    assert result["data_schema"].schema["set_ca_cert"]
-    assert result["data_schema"].schema["client_key_password"]
-    assert result["data_schema"].schema[mqtt.CONF_TLS_INSECURE]
-    assert result["data_schema"].schema[CONF_PROTOCOL]
-    assert result["data_schema"].schema[mqtt.CONF_CERTIFICATE]
-    assert result["data_schema"].schema[mqtt.CONF_CLIENT_CERT]
-    assert result["data_schema"].schema[mqtt.CONF_CLIENT_KEY]
-    assert result["data_schema"].schema[mqtt.CONF_TRANSPORT]
-
-    # third iteration, advanced settings with client cert and key and CA certificate
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"],
-        user_input={
-            mqtt.CONF_BROKER: "test-broker",
-            CONF_PORT: 2345,
-            CONF_USERNAME: "user",
-            CONF_PASSWORD: "secret",
-            mqtt.CONF_KEEPALIVE: 30,
-            "set_ca_cert": "custom",
-            "set_client_cert": True,
-            "client_key_password": client_key_password,
-            mqtt.CONF_CERTIFICATE: file_id[mqtt.CONF_CERTIFICATE],
-            mqtt.CONF_CLIENT_CERT: file_id[mqtt.CONF_CLIENT_CERT],
-            mqtt.CONF_CLIENT_KEY: file_id[mqtt.CONF_CLIENT_KEY],
-            mqtt.CONF_TLS_INSECURE: False,
-            mqtt.CONF_TRANSPORT: "tcp",
-        },
-    )
-
-    assert result["type"] is FlowResultType.ABORT
-    assert result["reason"] == "reconfigure_successful"
-
-    # Check config entry result
-    assert config_entry.data == {
+    # Check config entry data is set
+    assert result["data"] == {
         mqtt.CONF_BROKER: "test-broker",
         CONF_PROTOCOL: "5",
-        CONF_PORT: 2345,
+        CONF_PORT: 1234,
         CONF_USERNAME: "user",
         CONF_PASSWORD: "secret",
         mqtt.CONF_KEEPALIVE: 30,
@@ -2441,9 +2351,6 @@ async def test_change_websockets_transport_to_tcp(
     result = await config_entry.start_reconfigure_flow(hass)
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "broker"
-    assert result["data_schema"].schema["transport"]
-    assert result["data_schema"].schema["ws_path"]
-    assert result["data_schema"].schema["ws_headers"]
 
     # Change transport to tcp
     result = await hass.config_entries.flow.async_configure(
@@ -2451,9 +2358,14 @@ async def test_change_websockets_transport_to_tcp(
         user_input={
             mqtt.CONF_BROKER: "test-broker",
             CONF_PORT: 1234,
-            mqtt.CONF_TRANSPORT: "tcp",
-            mqtt.CONF_WS_HEADERS: '{"header_1": "custom_header1"}',
-            mqtt.CONF_WS_PATH: "/some_path",
+            CONF_PROTOCOL: "5",
+            "other_settings": {
+                "set_ca_cert": "off",
+                "set_client_cert": False,
+                mqtt.CONF_TRANSPORT: "tcp",
+                mqtt.CONF_WS_HEADERS: '{"header_1": "custom_header1"}',
+                mqtt.CONF_WS_PATH: "/some_path",
+            },
         },
     )
     assert result["type"] is FlowResultType.ABORT
@@ -2468,7 +2380,7 @@ async def test_change_websockets_transport_to_tcp(
     }
 
 
-@pytest.mark.usefixtures("mock_ssl_context", "mock_process_uploaded_file")
+@pytest.mark.usefixtures("mock_ssl_context")
 @pytest.mark.parametrize(
     "mqtt_config_entry_data",
     [
@@ -2485,24 +2397,53 @@ async def test_reconfigure_flow_form(
     hass: HomeAssistant,
     mock_try_connection: MagicMock,
     mqtt_mock_entry: MqttMockHAClientGenerator,
+    mock_process_uploaded_file: MagicMock,
 ) -> None:
-    """Test reconfigure flow."""
+    """Test reconfigure flow with existing certificates set in the config entry."""
     await mqtt_mock_entry()
     entry: MockConfigEntry = hass.config_entries.async_entries(DOMAIN)[0]
+
+    # Add certificates to the current entry
+    entry_data = MOCK_BROKER_ENTRY_DATA | {
+        mqtt.CONF_CERTIFICATE: MOCK_GENERIC_CERT.decode(encoding="utf-8"),
+        mqtt.CONF_CLIENT_CERT: MOCK_GENERIC_CERT.decode(encoding="utf-8"),
+        mqtt.CONF_CLIENT_KEY: MOCK_CLIENT_KEY.decode(encoding="utf-8"),
+    }
+    hass.config_entries.async_update_entry(entry, data=entry_data)
+    await hass.async_block_till_done()
+
     result = await entry.start_reconfigure_flow(hass)
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "broker"
     assert result["errors"] == {}
+    assert (
+        get_schema_suggested_value(
+            result["data_schema"].schema["other_settings"].schema.schema,
+            "set_client_cert",
+        )
+        is True
+    )
+    assert (
+        get_schema_suggested_value(
+            result["data_schema"].schema["other_settings"].schema.schema, "set_ca_cert"
+        )
+        == "custom"
+    )
 
+    # Keep current certificate files
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
             mqtt.CONF_BROKER: "10.10.10,10",
             CONF_PORT: 1234,
             CONF_PROTOCOL: "5",
-            mqtt.CONF_TRANSPORT: "websockets",
-            mqtt.CONF_WS_HEADERS: '{"header_1": "custom_header1"}',
-            mqtt.CONF_WS_PATH: "/some_new_path",
+            "other_settings": {
+                "set_ca_cert": "custom",
+                "set_client_cert": True,
+                mqtt.CONF_TRANSPORT: "websockets",
+                mqtt.CONF_WS_HEADERS: '{"header_1": "custom_header1"}',
+                mqtt.CONF_WS_PATH: "/some_new_path",
+            },
         },
     )
 
@@ -2515,6 +2456,9 @@ async def test_reconfigure_flow_form(
         mqtt.CONF_TRANSPORT: "websockets",
         mqtt.CONF_WS_HEADERS: {"header_1": "custom_header1"},
         mqtt.CONF_WS_PATH: "/some_new_path",
+        mqtt.CONF_CERTIFICATE: MOCK_GENERIC_CERT.decode(encoding="utf-8"),
+        mqtt.CONF_CLIENT_CERT: MOCK_GENERIC_CERT.decode(encoding="utf-8"),
+        mqtt.CONF_CLIENT_KEY: MOCK_CLIENT_KEY.decode(encoding="utf-8"),
     }
     await hass.async_block_till_done(wait_background_tasks=True)
 
@@ -2555,9 +2499,13 @@ async def test_reconfigure_no_changed_password(
             CONF_USERNAME: "mqtt-user",
             CONF_PASSWORD: PWD_NOT_CHANGED,
             CONF_PORT: 1234,
-            mqtt.CONF_TRANSPORT: "websockets",
-            mqtt.CONF_WS_HEADERS: '{"header_1": "custom_header1"}',
-            mqtt.CONF_WS_PATH: "/some_new_path",
+            "other_settings": {
+                "set_ca_cert": "auto",
+                "set_client_cert": False,
+                mqtt.CONF_TRANSPORT: "websockets",
+                mqtt.CONF_WS_HEADERS: '{"header_1": "custom_header1"}',
+                mqtt.CONF_WS_PATH: "/some_new_path",
+            },
         },
     )
 
@@ -2569,6 +2517,7 @@ async def test_reconfigure_no_changed_password(
         CONF_PASSWORD: "mqtt-password",
         CONF_PORT: 1234,
         CONF_PROTOCOL: "5",
+        mqtt.CONF_CERTIFICATE: "auto",
         mqtt.CONF_TRANSPORT: "websockets",
         mqtt.CONF_WS_HEADERS: {"header_1": "custom_header1"},
         mqtt.CONF_WS_PATH: "/some_new_path",
@@ -2586,9 +2535,9 @@ async def test_reconfigure_no_changed_password(
         "expected_minor_version",
     ),
     [
-        (1, 1, MOCK_ENTRY_DATA | MOCK_ENTRY_OPTIONS, {}, 2, 1),
-        (1, 2, MOCK_ENTRY_DATA, MOCK_ENTRY_OPTIONS, 2, 1),
-        (2, 1, MOCK_ENTRY_DATA, MOCK_ENTRY_OPTIONS, 2, 1),
+        (1, 1, MOCK_BROKER_ENTRY_DATA | MOCK_ENTRY_OPTIONS, {}, 2, 1),
+        (1, 2, MOCK_BROKER_ENTRY_DATA, MOCK_ENTRY_OPTIONS, 2, 1),
+        (2, 1, MOCK_BROKER_ENTRY_DATA, MOCK_ENTRY_OPTIONS, 2, 1),
     ],
 )
 @pytest.mark.usefixtures("mock_reload_after_entry_update")
@@ -2617,7 +2566,8 @@ async def test_migrate_config_entry(
     await mqtt_mock_entry()
     await hass.async_block_till_done()
     assert (
-        config_entry.data | config_entry.options == MOCK_ENTRY_DATA | MOCK_ENTRY_OPTIONS
+        config_entry.data | config_entry.options
+        == MOCK_BROKER_ENTRY_DATA | MOCK_ENTRY_OPTIONS
     )
     assert config_entry.version == expected_version
     assert config_entry.minor_version == expected_minor_version