diff --git a/homeassistant/components/victron_gx/config_flow.py b/homeassistant/components/victron_gx/config_flow.py index a74d8d721207..f4b0361b24f7 100644 --- a/homeassistant/components/victron_gx/config_flow.py +++ b/homeassistant/components/victron_gx/config_flow.py @@ -5,10 +5,22 @@ import logging from typing import Any, override from urllib.parse import urlparse -from victron_mqtt import AuthenticationError, CannotConnectError, Hub as VictronVenusHub +from victron_mqtt import ( + AuthenticationError, + CannotConnectError, + Hub as VictronVenusHub, + PairingError, + PairingToken, + request_pairing_token, +) import voluptuous as vol -from homeassistant.config_entries import SOURCE_IGNORE, ConfigFlow, ConfigFlowResult +from homeassistant.config_entries import ( + SOURCE_IGNORE, + SOURCE_REAUTH, + ConfigFlow, + ConfigFlowResult, +) from homeassistant.const import ( CONF_HOST, CONF_MODEL, @@ -21,10 +33,11 @@ from homeassistant.helpers import selector from homeassistant.helpers.redact import async_redact_data from homeassistant.helpers.service_info.ssdp import SsdpServiceInfo -from .const import CONF_INSTALLATION_ID, CONF_SERIAL, DOMAIN +from .const import CONF_INSTALLATION_ID, CONF_MQTT_TOKEN_PAIRING, CONF_SERIAL, DOMAIN DEFAULT_HOST = "venus.local" DEFAULT_PORT = 1883 +DEFAULT_SSL_PORT = 8883 _LOGGER = logging.getLogger(__name__) @@ -46,11 +59,10 @@ STEP_USER_DATA_SCHEMA = vol.Schema( STEP_SSDP_AUTH_DATA_SCHEMA = vol.Schema( { - vol.Optional(CONF_USERNAME, default=""): selector.TextSelector(), vol.Optional(CONF_PASSWORD, default=""): selector.TextSelector( selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD) ), - vol.Optional(CONF_SSL, default=False): selector.BooleanSelector(), + vol.Optional(CONF_SSL): selector.BooleanSelector(), } ) @@ -110,6 +122,8 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): self.installation_id: str | None = None self.friendly_name: str | None = None self.model_name: str | None = None + self.mqtt_token_pairing = False + self.ssdp_use_ssl = True @override async def async_step_user( @@ -177,6 +191,7 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): self.installation_id = discovery_info.upnp["X_VrmPortalId"] self.model_name = discovery_info.upnp["modelName"] self.friendly_name = discovery_info.upnp["friendlyName"] + self.mqtt_token_pairing = discovery_info.upnp.get("X_MqttTokenPairing") == "1" await self.async_set_unique_id(self.installation_id) @@ -207,25 +222,6 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): "name": self.friendly_name or self.hostname } - # Verify connectivity before showing the confirmation dialog - try: - ssdp_conf = { - CONF_HOST: self.hostname, - CONF_PORT: DEFAULT_PORT, - CONF_SERIAL: self.serial, - CONF_INSTALLATION_ID: self.installation_id, - } - await validate_input(ssdp_conf) - except AuthenticationError: - return await self.async_step_ssdp_auth() - except CannotConnectError: - return self.async_abort(reason="cannot_connect") - except Exception: - _LOGGER.exception( - "Unexpected error validating SSDP discovery for Victron GX" - ) - return self.async_abort(reason="unknown") - return await self.async_step_ssdp_confirm() async def async_step_ssdp_confirm( @@ -236,19 +232,52 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): assert self.installation_id is not None if user_input is not None: + data: dict[str, Any] = { + CONF_HOST: self.hostname, + CONF_PORT: DEFAULT_SSL_PORT, + CONF_SERIAL: self.serial, + CONF_INSTALLATION_ID: self.installation_id, + CONF_MODEL: self.model_name, + CONF_SSL: True, + } + try: + await validate_input(data) + except AuthenticationError: + if self.mqtt_token_pairing: + return await self.async_step_ssdp_token_pairing() + return await self.async_step_ssdp_auth() + except CannotConnectError: + data[CONF_PORT] = DEFAULT_PORT + data[CONF_SSL] = False + self.ssdp_use_ssl = False + try: + await validate_input(data) + except AuthenticationError: + if self.mqtt_token_pairing: + return await self.async_step_ssdp_token_pairing() + return await self.async_step_ssdp_auth() + except CannotConnectError: + if self.mqtt_token_pairing: + return await self.async_step_ssdp_token_pairing() + return self.async_abort(reason="cannot_connect") + except Exception: + _LOGGER.exception( + "Unexpected error validating SSDP discovery for Victron GX" + ) + return self.async_abort(reason="unknown") + except Exception: + _LOGGER.exception( + "Unexpected error validating SSDP discovery for Victron GX" + ) + return self.async_abort(reason="unknown") + return self.async_create_entry( title=ENTRY_TITLE_FORMAT.format( installation_id=self.installation_id, host=self.hostname, - port=DEFAULT_PORT, + port=data[CONF_PORT], ), - data={ - CONF_HOST: self.hostname, - CONF_PORT: DEFAULT_PORT, - CONF_SERIAL: self.serial, - CONF_INSTALLATION_ID: self.installation_id, - CONF_MODEL: self.model_name, - }, + data=data, ) self._set_confirm_only() @@ -257,6 +286,66 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): description_placeholders={"name": self.friendly_name or self.hostname}, ) + async def async_step_ssdp_token_pairing( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle automatic token pairing with the GX device.""" + assert self.hostname is not None + assert self.installation_id is not None + + errors: dict[str, str] = {} + if user_input is not None: + try: + credentials: PairingToken = await request_pairing_token( + self.hostname, self.installation_id + ) + except PairingError: + errors["base"] = "pairing_failed" + except Exception: + _LOGGER.exception("Failed to connect to GX device for token pairing") + errors["base"] = "cannot_connect" + else: + data: dict[str, Any] = { + CONF_HOST: self.hostname, + CONF_PORT: DEFAULT_SSL_PORT, + CONF_SERIAL: self.serial, + CONF_INSTALLATION_ID: self.installation_id, + CONF_MODEL: self.model_name, + CONF_MQTT_TOKEN_PAIRING: True, + CONF_USERNAME: credentials.token_name, + CONF_PASSWORD: credentials.password, + CONF_SSL: True, + } + try: + await validate_input(data) + except AuthenticationError: + errors["base"] = "invalid_auth" + except CannotConnectError: + errors["base"] = "cannot_connect" + except Exception: + _LOGGER.exception("Unexpected error validating paired credentials") + errors["base"] = "unknown" + else: + if self.source == SOURCE_REAUTH: + return self.async_update_reload_and_abort( + self._get_reauth_entry(), data_updates=data + ) + return self.async_create_entry( + title=ENTRY_TITLE_FORMAT.format( + installation_id=self.installation_id, + host=self.hostname, + port=DEFAULT_SSL_PORT, + ), + data=data, + ) + + self._set_confirm_only() + return self.async_show_form( + step_id="ssdp_token_pairing", + errors=errors, + description_placeholders={CONF_HOST: self.hostname}, + ) + async def async_step_ssdp_auth( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -271,14 +360,16 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): "SSDP auth user input received: %s", async_redact_data(user_input, TO_REDACT), ) + use_ssl = user_input.get(CONF_SSL, self.ssdp_use_ssl) data: dict[str, Any] = { CONF_HOST: self.hostname, - CONF_PORT: DEFAULT_PORT, + CONF_PORT: DEFAULT_SSL_PORT if use_ssl else DEFAULT_PORT, CONF_SERIAL: self.serial, CONF_INSTALLATION_ID: self.installation_id, - CONF_USERNAME: user_input.get(CONF_USERNAME), - CONF_PASSWORD: user_input.get(CONF_PASSWORD), - CONF_SSL: user_input.get(CONF_SSL), + CONF_MODEL: self.model_name, + CONF_USERNAME: "remoteconsole", + CONF_PASSWORD: user_input.get(CONF_PASSWORD) or None, + CONF_SSL: use_ssl, } try: @@ -298,7 +389,7 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): title=ENTRY_TITLE_FORMAT.format( installation_id=self.installation_id, host=self.hostname, - port=DEFAULT_PORT, + port=data[CONF_PORT], ), data=data, ) @@ -306,7 +397,8 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): return self.async_show_form( step_id="ssdp_auth", data_schema=self.add_suggested_values_to_schema( - STEP_SSDP_AUTH_DATA_SCHEMA, user_input + STEP_SSDP_AUTH_DATA_SCHEMA, + user_input or {CONF_SSL: self.ssdp_use_ssl}, ), errors=errors, description_placeholders={CONF_HOST: self.hostname}, @@ -368,6 +460,13 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN): async def async_step_reauth(self, _: Mapping[str, Any]) -> ConfigFlowResult: """Handle reauthentication.""" + reauth_entry = self._get_reauth_entry() + if reauth_entry.data.get(CONF_MQTT_TOKEN_PAIRING): + self.hostname = reauth_entry.data[CONF_HOST] + self.serial = reauth_entry.data.get(CONF_SERIAL) + self.installation_id = reauth_entry.data[CONF_INSTALLATION_ID] + self.model_name = reauth_entry.data.get(CONF_MODEL) + return await self.async_step_ssdp_token_pairing() return await self.async_step_reauth_confirm() async def async_step_reauth_confirm( diff --git a/homeassistant/components/victron_gx/const.py b/homeassistant/components/victron_gx/const.py index b97e67e05876..6b51c92c58ba 100644 --- a/homeassistant/components/victron_gx/const.py +++ b/homeassistant/components/victron_gx/const.py @@ -3,6 +3,7 @@ DOMAIN = "victron_gx" CONF_INSTALLATION_ID = "installation_id" +CONF_MQTT_TOKEN_PAIRING = "mqtt_token_pairing" CONF_SERIAL = "serial" # Binary sensor enum ids must be "on" for on and "off" for off. diff --git a/homeassistant/components/victron_gx/manifest.json b/homeassistant/components/victron_gx/manifest.json index fbfbf52675f9..d1832f469e87 100644 --- a/homeassistant/components/victron_gx/manifest.json +++ b/homeassistant/components/victron_gx/manifest.json @@ -12,6 +12,10 @@ { "X_MqttOnLan": "1", "manufacturer": "Victron Energy" + }, + { + "X_MqttTokenPairing": "1", + "manufacturer": "Victron Energy" } ] } diff --git a/homeassistant/components/victron_gx/strings.json b/homeassistant/components/victron_gx/strings.json index e5f5e14660de..c81ec9e17a73 100644 --- a/homeassistant/components/victron_gx/strings.json +++ b/homeassistant/components/victron_gx/strings.json @@ -118,6 +118,7 @@ "error": { "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", + "pairing_failed": "Token pairing failed. Ensure pairing mode is active on the GX device and try again.", "unknown": "[%key:common::config_flow::error::unknown%]" }, "step": { @@ -153,21 +154,23 @@ }, "ssdp_auth": { "data": { - "password": "[%key:common::config_flow::data::password%]", - "ssl": "[%key:common::config_flow::data::ssl%]", - "username": "[%key:common::config_flow::data::username%]" + "password": "GX password", + "ssl": "[%key:common::config_flow::data::ssl%]" }, "data_description": { "password": "[%key:component::victron_gx::config::step::user::data_description::password%]", - "ssl": "[%key:component::victron_gx::config::step::user::data_description::ssl%]", - "username": "[%key:component::victron_gx::config::step::user::data_description::username%]" + "ssl": "[%key:component::victron_gx::config::step::user::data_description::ssl%]" }, - "description": "Authentication is required to connect to {host}.", + "description": "Enter the GX password for the GX device at {host}.", "title": "Authenticate Victron GX" }, "ssdp_confirm": { "description": "Do you want to set up the Victron GX device {name}?" }, + "ssdp_token_pairing": { + "description": "Enable pairing mode on the GX device before submitting:\n\n- Via the GX user interface: **Settings** > **Integrations** > **MQTT Devices** > **Pairing mode**\n- On GX devices without a built-in screen: Quickly double-press the built-in button\n\nPairing mode is active for 120 seconds.", + "title": "Pair with GX device" + }, "user": { "data": { "host": "[%key:common::config_flow::data::host%]", diff --git a/homeassistant/generated/ssdp.py b/homeassistant/generated/ssdp.py index 8409d1d8f436..cdf5317a2725 100644 --- a/homeassistant/generated/ssdp.py +++ b/homeassistant/generated/ssdp.py @@ -389,6 +389,10 @@ SSDP = { "X_MqttOnLan": "1", "manufacturer": "Victron Energy", }, + { + "X_MqttTokenPairing": "1", + "manufacturer": "Victron Energy", + }, ], "webostv": [ { diff --git a/tests/components/victron_gx/test_config_flow.py b/tests/components/victron_gx/test_config_flow.py index c3a31e0b4478..0a7a2c039852 100644 --- a/tests/components/victron_gx/test_config_flow.py +++ b/tests/components/victron_gx/test_config_flow.py @@ -3,11 +3,20 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from victron_mqtt import AuthenticationError, CannotConnectError +from victron_mqtt import ( + AuthenticationError, + CannotConnectError, + PairingError, + PairingToken, +) -from homeassistant.components.victron_gx.config_flow import DEFAULT_PORT +from homeassistant.components.victron_gx.config_flow import ( + DEFAULT_PORT, + DEFAULT_SSL_PORT, +) from homeassistant.components.victron_gx.const import ( CONF_INSTALLATION_ID, + CONF_MQTT_TOKEN_PAIRING, CONF_SERIAL, DOMAIN, ) @@ -47,6 +56,25 @@ def assert_entry_title( assert result["title"] == f"Victron OS {installation_id} ({host}:{port})" +def ssdp_discovery_info(token_pairing: str = "0") -> SsdpServiceInfo: + """Return Victron GX SSDP discovery data.""" + upnp = { + "serialNumber": MOCK_SERIAL, + "X_VrmPortalId": MOCK_INSTALLATION_ID, + "modelName": MOCK_MODEL, + "friendlyName": MOCK_FRIENDLY_NAME, + "X_MqttOnLan": "1", + "X_MqttTokenPairing": token_pairing, + "manufacturer": "Victron Energy", + } + return SsdpServiceInfo( + ssdp_usn="mock_usn", + ssdp_st="upnp:rootdevice", + ssdp_location="http://192.168.1.100:80/", + upnp=upnp, + ) + + @pytest.fixture def mock_victron_hub(): """Mock the Victron Hub.""" @@ -234,13 +262,14 @@ async def test_ssdp_flow_success(hass: HomeAssistant) -> None: assert result["type"] is FlowResultType.CREATE_ENTRY assert result["result"].unique_id == MOCK_INSTALLATION_ID - assert_entry_title(result) + assert_entry_title(result, port=DEFAULT_SSL_PORT) assert result["data"] == { CONF_HOST: MOCK_HOST, - CONF_PORT: DEFAULT_PORT, + CONF_PORT: DEFAULT_SSL_PORT, CONF_SERIAL: MOCK_SERIAL, CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID, CONF_MODEL: MOCK_MODEL, + CONF_SSL: True, } @@ -280,6 +309,13 @@ async def test_ssdp_discovery_error( data=discovery_info, ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.ABORT assert result["reason"] == reason @@ -342,6 +378,13 @@ async def test_ssdp_flow_auth_required( data=discovery_info, ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "ssdp_auth" @@ -352,22 +395,22 @@ async def test_ssdp_flow_auth_required( result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ - CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password", }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["result"].unique_id == MOCK_INSTALLATION_ID - assert_entry_title(result) + assert_entry_title(result, port=DEFAULT_SSL_PORT) assert result["data"] == { CONF_HOST: MOCK_HOST, - CONF_PORT: DEFAULT_PORT, + CONF_PORT: DEFAULT_SSL_PORT, CONF_SERIAL: MOCK_SERIAL, CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID, - CONF_USERNAME: "test-username", + CONF_MODEL: MOCK_MODEL, + CONF_USERNAME: "remoteconsole", CONF_PASSWORD: "test-password", - CONF_SSL: False, + CONF_SSL: True, } @@ -399,6 +442,13 @@ async def test_ssdp_auth_invalid_credentials( data=discovery_info, ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "ssdp_auth" @@ -410,8 +460,8 @@ async def test_ssdp_auth_invalid_credentials( result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ - CONF_USERNAME: "wrong-user", CONF_PASSWORD: "wrong-password", + CONF_SSL: False, }, ) @@ -425,8 +475,8 @@ async def test_ssdp_auth_invalid_credentials( result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ - CONF_USERNAME: "test-user", CONF_PASSWORD: "test-password", + CONF_SSL: False, }, ) @@ -438,7 +488,8 @@ async def test_ssdp_auth_invalid_credentials( CONF_PORT: DEFAULT_PORT, CONF_SERIAL: MOCK_SERIAL, CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID, - CONF_USERNAME: "test-user", + CONF_MODEL: MOCK_MODEL, + CONF_USERNAME: "remoteconsole", CONF_PASSWORD: "test-password", CONF_SSL: False, } @@ -482,6 +533,13 @@ async def test_ssdp_auth_error( data=discovery_info, ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "ssdp_auth" @@ -490,8 +548,8 @@ async def test_ssdp_auth_error( result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ - CONF_USERNAME: "test-user", CONF_PASSWORD: "test-password", + CONF_SSL: False, }, ) @@ -641,6 +699,48 @@ async def test_reauth_flow_clears_credentials( assert mock_config_entry.data[CONF_SSL] is False +async def test_reauth_flow_renews_pairing_token( + hass: HomeAssistant, mock_victron_hub: MagicMock +) -> None: + """Test reauthentication renews generated pairing credentials.""" + token_entry = MockConfigEntry( + domain=DOMAIN, + unique_id=MOCK_INSTALLATION_ID, + data={ + CONF_HOST: MOCK_HOST, + CONF_PORT: DEFAULT_SSL_PORT, + CONF_USERNAME: "token/homeassistant/old", + CONF_PASSWORD: "old-password", + CONF_SSL: True, + CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID, + CONF_MQTT_TOKEN_PAIRING: True, + CONF_MODEL: MOCK_MODEL, + CONF_SERIAL: MOCK_SERIAL, + }, + title=(f"Victron OS {MOCK_INSTALLATION_ID} ({MOCK_HOST}:{DEFAULT_SSL_PORT})"), + ) + token_entry.add_to_hass(hass) + + result = await token_entry.start_reauth_flow(hass) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_token_pairing" + + with patch( + "homeassistant.components.victron_gx.config_flow.request_pairing_token", + return_value=PairingToken("token/homeassistant/new", "new-generated-password"), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reauth_successful" + assert token_entry.data[CONF_USERNAME] == "token/homeassistant/new" + assert token_entry.data[CONF_PASSWORD] == "new-generated-password" + assert token_entry.data[CONF_MQTT_TOKEN_PAIRING] is True + mock_victron_hub.return_value.connect.assert_awaited_once() + + @pytest.mark.parametrize( ("exception", "error"), [ @@ -923,3 +1023,270 @@ async def test_ssdp_flow_abort_on_invalid_hostname( assert result["type"] is FlowResultType.ABORT assert result["reason"] == "cannot_connect" + + +async def test_ssdp_flow_falls_back_to_plain_mqtt( + hass: HomeAssistant, mock_victron_hub: MagicMock +) -> None: + """Test SSDP setup falls back to plain MQTT when TLS is unavailable.""" + mock_victron_hub.return_value.connect.side_effect = [ + CannotConnectError("TLS unavailable"), + None, + ] + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_SSDP}, + data=ssdp_discovery_info(), + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert_entry_title(result) + assert result["data"][CONF_PORT] == DEFAULT_PORT + assert result["data"][CONF_SSL] is False + + +@pytest.mark.parametrize( + ("second_exception", "result_type", "result_key", "result_value"), + [ + pytest.param( + AuthenticationError("Authentication required"), + FlowResultType.FORM, + "step_id", + "ssdp_auth", + id="authentication-required", + ), + pytest.param( + Exception("Unexpected error"), + FlowResultType.ABORT, + "reason", + "unknown", + id="unexpected-error", + ), + ], +) +async def test_ssdp_flow_tls_fallback_error( + hass: HomeAssistant, + mock_victron_hub: MagicMock, + second_exception: Exception, + result_type: FlowResultType, + result_key: str, + result_value: str, +) -> None: + """Test errors while falling back from TLS during SSDP setup.""" + mock_victron_hub.return_value.connect.side_effect = [ + CannotConnectError("TLS unavailable"), + second_exception, + ] + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_SSDP}, + data=ssdp_discovery_info(), + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is result_type + assert result[result_key] == result_value + + +async def test_ssdp_flow_preserves_plain_mqtt_for_authentication( + hass: HomeAssistant, mock_victron_hub: MagicMock +) -> None: + """Test SSDP authentication preserves the discovered plaintext transport.""" + mock_victron_hub.return_value.connect.side_effect = [ + CannotConnectError("TLS unavailable"), + AuthenticationError("Authentication required"), + None, + ] + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_SSDP}, + data=ssdp_discovery_info(), + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_auth" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={CONF_PASSWORD: "test-password"} + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert_entry_title(result) + assert result["data"][CONF_PORT] == DEFAULT_PORT + assert result["data"][CONF_SSL] is False + + +@pytest.mark.parametrize( + "initial_exceptions", + [ + pytest.param( + [AuthenticationError("Authentication required"), None], + id="authentication-required", + ), + pytest.param( + [ + CannotConnectError("TLS unavailable"), + AuthenticationError("Authentication required"), + None, + ], + id="plain-mqtt-authentication-required", + ), + pytest.param( + [ + CannotConnectError("TLS unavailable"), + CannotConnectError("Plain MQTT unavailable"), + None, + ], + id="mqtt-unavailable", + ), + ], +) +async def test_ssdp_token_pairing_success( + hass: HomeAssistant, + mock_victron_hub: MagicMock, + initial_exceptions: list[Exception | None], +) -> None: + """Test SSDP token pairing creates an entry with generated credentials.""" + mock_victron_hub.return_value.connect.side_effect = initial_exceptions + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_SSDP}, + data=ssdp_discovery_info("1"), + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_token_pairing" + + with patch( + "homeassistant.components.victron_gx.config_flow.request_pairing_token", + return_value=PairingToken( + token_name="token/homeassistant/homeassistant_abc123", + password="generated-password", + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert_entry_title(result, port=DEFAULT_SSL_PORT) + assert result["data"] == { + CONF_HOST: MOCK_HOST, + CONF_PORT: DEFAULT_SSL_PORT, + CONF_SERIAL: MOCK_SERIAL, + CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID, + CONF_MODEL: MOCK_MODEL, + CONF_MQTT_TOKEN_PAIRING: True, + CONF_USERNAME: "token/homeassistant/homeassistant_abc123", + CONF_PASSWORD: "generated-password", + CONF_SSL: True, + } + + +@pytest.mark.parametrize( + ("exception", "error"), + [ + pytest.param( + PairingError("Pairing mode inactive"), + "pairing_failed", + id="pairing-rejected", + ), + pytest.param( + ConnectionError("Connection refused"), + "cannot_connect", + id="connection-error", + ), + ], +) +async def test_ssdp_token_pairing_request_error( + hass: HomeAssistant, + mock_victron_hub: MagicMock, + exception: Exception, + error: str, +) -> None: + """Test errors while requesting token pairing credentials.""" + mock_victron_hub.return_value.connect.side_effect = AuthenticationError( + "Authentication required" + ) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_SSDP}, + data=ssdp_discovery_info("1"), + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + with patch( + "homeassistant.components.victron_gx.config_flow.request_pairing_token", + side_effect=exception, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_token_pairing" + assert result["errors"] == {"base": error} + + +@pytest.mark.parametrize( + ("exception", "error"), + [ + pytest.param( + AuthenticationError("Invalid credentials"), + "invalid_auth", + id="invalid-auth", + ), + pytest.param( + CannotConnectError("Cannot connect"), + "cannot_connect", + id="cannot-connect", + ), + pytest.param(Exception("Unexpected error"), "unknown", id="unknown"), + ], +) +async def test_ssdp_token_pairing_validation_error( + hass: HomeAssistant, + mock_victron_hub: MagicMock, + exception: Exception, + error: str, +) -> None: + """Test errors while validating generated pairing credentials.""" + mock_victron_hub.return_value.connect.side_effect = [ + AuthenticationError("Authentication required"), + exception, + ] + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_SSDP}, + data=ssdp_discovery_info("1"), + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + with patch( + "homeassistant.components.victron_gx.config_flow.request_pairing_token", + return_value=PairingToken("token/homeassistant/test", "password"), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "ssdp_token_pairing" + assert result["errors"] == {"base": error}