Recommended installation option for Z-Wave (#145327)

Recommended installation option for ZWave
This commit is contained in:
Petar Petrov 2025-05-21 14:30:59 +03:00 committed by GitHub
parent 00a1d9d1b0
commit efa7fe0dc9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 467 additions and 45 deletions

View File

@ -197,6 +197,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
self._migrating = False
self._reconfigure_config_entry: ConfigEntry | None = None
self._usb_discovery = False
self._recommended_install = False
async def async_step_install_addon(
self, user_input: dict[str, Any] | None = None
@ -372,10 +373,22 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Handle the initial step."""
if is_hassio(self.hass):
return await self.async_step_on_supervisor()
return await self.async_step_installation_type()
return await self.async_step_manual()
async def async_step_installation_type(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the installation type step."""
return self.async_show_menu(
step_id="installation_type",
menu_options=[
"intent_recommended",
"intent_custom",
],
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -516,7 +529,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="addon_required")
return await self.async_step_intent_migrate()
return await self.async_step_on_supervisor({CONF_USE_ADDON: True})
return await self.async_step_installation_type()
async def async_step_manual(
self, user_input: dict[str, Any] | None = None
@ -593,6 +606,21 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="hassio_confirm")
async def async_step_intent_recommended(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Select recommended installation type."""
self._recommended_install = True
return await self.async_step_on_supervisor({CONF_USE_ADDON: True})
async def async_step_intent_custom(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Select custom installation type."""
if self._usb_discovery:
return await self.async_step_on_supervisor({CONF_USE_ADDON: True})
return await self.async_step_on_supervisor()
async def async_step_on_supervisor(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -641,31 +669,6 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
addon_info = await self._async_get_addon_info()
addon_config = addon_info.options
if user_input is not None:
self.s0_legacy_key = user_input[CONF_S0_LEGACY_KEY]
self.s2_access_control_key = user_input[CONF_S2_ACCESS_CONTROL_KEY]
self.s2_authenticated_key = user_input[CONF_S2_AUTHENTICATED_KEY]
self.s2_unauthenticated_key = user_input[CONF_S2_UNAUTHENTICATED_KEY]
self.lr_s2_access_control_key = user_input[CONF_LR_S2_ACCESS_CONTROL_KEY]
self.lr_s2_authenticated_key = user_input[CONF_LR_S2_AUTHENTICATED_KEY]
if not self._usb_discovery:
self.usb_path = user_input[CONF_USB_PATH]
addon_config_updates = {
CONF_ADDON_DEVICE: self.usb_path,
CONF_ADDON_S0_LEGACY_KEY: self.s0_legacy_key,
CONF_ADDON_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
CONF_ADDON_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
CONF_ADDON_S2_UNAUTHENTICATED_KEY: self.s2_unauthenticated_key,
CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY: self.lr_s2_access_control_key,
CONF_ADDON_LR_S2_AUTHENTICATED_KEY: self.lr_s2_authenticated_key,
}
await self._async_set_addon_config(addon_config_updates)
return await self.async_step_start_addon()
usb_path = self.usb_path or addon_config.get(CONF_ADDON_DEVICE) or ""
s0_legacy_key = addon_config.get(
CONF_ADDON_S0_LEGACY_KEY, self.s0_legacy_key or ""
)
@ -685,22 +688,67 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_ADDON_LR_S2_AUTHENTICATED_KEY, self.lr_s2_authenticated_key or ""
)
schema: VolDictType = {
vol.Optional(CONF_S0_LEGACY_KEY, default=s0_legacy_key): str,
vol.Optional(
CONF_S2_ACCESS_CONTROL_KEY, default=s2_access_control_key
): str,
vol.Optional(CONF_S2_AUTHENTICATED_KEY, default=s2_authenticated_key): str,
vol.Optional(
CONF_S2_UNAUTHENTICATED_KEY, default=s2_unauthenticated_key
): str,
vol.Optional(
CONF_LR_S2_ACCESS_CONTROL_KEY, default=lr_s2_access_control_key
): str,
vol.Optional(
CONF_LR_S2_AUTHENTICATED_KEY, default=lr_s2_authenticated_key
): str,
}
if self._recommended_install and self._usb_discovery:
# Recommended installation with USB discovery, skip asking for keys
user_input = {}
if user_input is not None:
self.s0_legacy_key = user_input.get(CONF_S0_LEGACY_KEY, s0_legacy_key)
self.s2_access_control_key = user_input.get(
CONF_S2_ACCESS_CONTROL_KEY, s2_access_control_key
)
self.s2_authenticated_key = user_input.get(
CONF_S2_AUTHENTICATED_KEY, s2_authenticated_key
)
self.s2_unauthenticated_key = user_input.get(
CONF_S2_UNAUTHENTICATED_KEY, s2_unauthenticated_key
)
self.lr_s2_access_control_key = user_input.get(
CONF_LR_S2_ACCESS_CONTROL_KEY, lr_s2_access_control_key
)
self.lr_s2_authenticated_key = user_input.get(
CONF_LR_S2_AUTHENTICATED_KEY, lr_s2_authenticated_key
)
if not self._usb_discovery:
self.usb_path = user_input[CONF_USB_PATH]
addon_config_updates = {
CONF_ADDON_DEVICE: self.usb_path,
CONF_ADDON_S0_LEGACY_KEY: self.s0_legacy_key,
CONF_ADDON_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
CONF_ADDON_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
CONF_ADDON_S2_UNAUTHENTICATED_KEY: self.s2_unauthenticated_key,
CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY: self.lr_s2_access_control_key,
CONF_ADDON_LR_S2_AUTHENTICATED_KEY: self.lr_s2_authenticated_key,
}
await self._async_set_addon_config(addon_config_updates)
return await self.async_step_start_addon()
usb_path = self.usb_path or addon_config.get(CONF_ADDON_DEVICE) or ""
schema: VolDictType = (
{}
if self._recommended_install
else {
vol.Optional(CONF_S0_LEGACY_KEY, default=s0_legacy_key): str,
vol.Optional(
CONF_S2_ACCESS_CONTROL_KEY, default=s2_access_control_key
): str,
vol.Optional(
CONF_S2_AUTHENTICATED_KEY, default=s2_authenticated_key
): str,
vol.Optional(
CONF_S2_UNAUTHENTICATED_KEY, default=s2_unauthenticated_key
): str,
vol.Optional(
CONF_LR_S2_ACCESS_CONTROL_KEY, default=lr_s2_access_control_key
): str,
vol.Optional(
CONF_LR_S2_AUTHENTICATED_KEY, default=lr_s2_authenticated_key
): str,
}
)
if not self._usb_discovery:
try:

View File

@ -131,6 +131,14 @@
"usb_path": "[%key:common::config_flow::data::usb_path%]"
},
"title": "Select your Z-Wave device"
},
"installation_type": {
"title": "Set-up Z-Wave",
"description": "Choose the installation type for your Z-Wave integration.",
"menu_options": {
"intent_recommended": "Recommended installation",
"intent_custom": "Custom installation"
}
}
}
},

View File

@ -19,7 +19,24 @@ from zwave_js_server.version import VersionInfo
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.zwave_js.config_flow import TITLE, get_usb_ports
from homeassistant.components.zwave_js.const import ADDON_SLUG, CONF_USB_PATH, DOMAIN
from homeassistant.components.zwave_js.const import (
ADDON_SLUG,
CONF_ADDON_DEVICE,
CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_LR_S2_AUTHENTICATED_KEY,
CONF_ADDON_S0_LEGACY_KEY,
CONF_ADDON_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY,
CONF_ADDON_S2_UNAUTHENTICATED_KEY,
CONF_LR_S2_ACCESS_CONTROL_KEY,
CONF_LR_S2_AUTHENTICATED_KEY,
CONF_S0_LEGACY_KEY,
CONF_S2_ACCESS_CONTROL_KEY,
CONF_S2_AUTHENTICATED_KEY,
CONF_S2_UNAUTHENTICATED_KEY,
CONF_USB_PATH,
DOMAIN,
)
from homeassistant.components.zwave_js.helpers import SERVER_VERSION_TIMEOUT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -455,6 +472,13 @@ async def test_clean_discovery_on_user_create(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -643,6 +667,14 @@ async def test_usb_discovery(
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
assert result["menu_options"] == ["intent_recommended", "intent_custom"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "install_addon"
@ -756,6 +788,13 @@ async def test_usb_discovery_addon_not_running(
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_addon_user"
@ -1511,6 +1550,13 @@ async def test_not_addon(hass: HomeAssistant, supervisor) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -1589,6 +1635,13 @@ async def test_addon_running(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -1700,6 +1753,13 @@ async def test_addon_running_failures(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -1763,6 +1823,13 @@ async def test_addon_running_already_configured(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -1810,6 +1877,13 @@ async def test_addon_installed(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -1913,6 +1987,13 @@ async def test_addon_installed_start_failure(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -1998,6 +2079,13 @@ async def test_addon_installed_failures(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -2079,6 +2167,13 @@ async def test_addon_installed_set_options_failure(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -2134,6 +2229,13 @@ async def test_addon_installed_usb_ports_failure(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -2194,6 +2296,13 @@ async def test_addon_installed_already_configured(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -2280,6 +2389,13 @@ async def test_addon_not_installed(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -2374,6 +2490,13 @@ async def test_install_addon_failure(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_custom"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
@ -2617,7 +2740,6 @@ async def test_reconfigure_not_addon_with_addon_stop_fail(
assert entry.state is config_entries.ConfigEntryState.LOADED
assert setup_entry.call_count == 1
assert unload_entry.call_count == 1
# avoid unload entry in teardown
await hass.config_entries.async_unload(entry.entry_id)
assert entry.state is config_entries.ConfigEntryState.NOT_LOADED
@ -4695,3 +4817,247 @@ async def test_get_usb_ports_sorting(hass: HomeAssistant) -> None:
"n/a - /dev/ttyUSB0, s/n: n/a",
"N/A - /dev/ttyUSB2, s/n: n/a",
]
@pytest.mark.parametrize(
"discovery_info",
[
[
Discovery(
addon="core_zwave_js",
service="zwave_js",
uuid=uuid4(),
config=ADDON_DISCOVERY_INFO,
)
]
],
)
async def test_intent_recommended_user(
hass: HomeAssistant,
supervisor,
addon_not_installed,
install_addon,
start_addon,
addon_options,
set_addon_options,
get_addon_discovery_info,
) -> None:
"""Test the intent_recommended step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_recommended"}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "install_addon"
# Make sure the flow continues when the progress task is done.
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert install_addon.call_args == call("core_zwave_js")
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_addon_user"
assert result["data_schema"].schema[CONF_USB_PATH] is not None
assert result["data_schema"].schema.get(CONF_S0_LEGACY_KEY) is None
assert result["data_schema"].schema.get(CONF_S2_ACCESS_CONTROL_KEY) is None
assert result["data_schema"].schema.get(CONF_S2_AUTHENTICATED_KEY) is None
assert result["data_schema"].schema.get(CONF_S2_UNAUTHENTICATED_KEY) is None
assert result["data_schema"].schema.get(CONF_LR_S2_ACCESS_CONTROL_KEY) is None
assert result["data_schema"].schema.get(CONF_LR_S2_AUTHENTICATED_KEY) is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USB_PATH: "/test",
},
)
assert set_addon_options.call_args == call(
"core_zwave_js",
AddonsOptions(
config={
CONF_ADDON_DEVICE: "/test",
CONF_ADDON_S0_LEGACY_KEY: "",
CONF_ADDON_S2_ACCESS_CONTROL_KEY: "",
CONF_ADDON_S2_AUTHENTICATED_KEY: "",
CONF_ADDON_S2_UNAUTHENTICATED_KEY: "",
CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY: "",
CONF_ADDON_LR_S2_AUTHENTICATED_KEY: "",
}
),
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
with (
patch(
"homeassistant.components.zwave_js.async_setup", return_value=True
) as mock_setup,
patch(
"homeassistant.components.zwave_js.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call("core_zwave_js")
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"s0_legacy_key": "",
"s2_access_control_key": "",
"s2_authenticated_key": "",
"s2_unauthenticated_key": "",
"lr_s2_access_control_key": "",
"lr_s2_authenticated_key": "",
"use_addon": True,
"integration_created_addon": True,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("usb_discovery_info", "device", "discovery_name"),
[
(
USB_DISCOVERY_INFO,
USB_DISCOVERY_INFO.device,
"zwave radio",
),
(
UsbServiceInfo(
device="/dev/zwa2",
pid="303A",
vid="4001",
serial_number="1234",
description="ZWA-2 - Nabu Casa ZWA-2",
manufacturer="Nabu Casa",
),
"/dev/zwa2",
"Home Assistant Connect ZWA-2",
),
],
)
@pytest.mark.parametrize(
"discovery_info",
[
[
Discovery(
addon="core_zwave_js",
service="zwave_js",
uuid=uuid4(),
config=ADDON_DISCOVERY_INFO,
)
]
],
)
async def test_recommended_usb_discovery(
hass: HomeAssistant,
supervisor,
addon_not_installed,
install_addon,
addon_options,
get_addon_discovery_info,
mock_usb_serial_by_id: MagicMock,
set_addon_options,
start_addon,
usb_discovery_info: UsbServiceInfo,
device: str,
discovery_name: str,
) -> None:
"""Test usb discovery success path."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USB},
data=usb_discovery_info,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "usb_confirm"
assert result["description_placeholders"] == {"name": discovery_name}
assert mock_usb_serial_by_id.call_count == 1
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "installation_type"
assert result["menu_options"] == ["intent_recommended", "intent_custom"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "intent_recommended"}
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "install_addon"
# Make sure the flow continues when the progress task is done.
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert install_addon.call_args == call("core_zwave_js")
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
assert set_addon_options.call_args == call(
"core_zwave_js",
AddonsOptions(
config={
"device": device,
"s0_legacy_key": "",
"s2_access_control_key": "",
"s2_authenticated_key": "",
"s2_unauthenticated_key": "",
"lr_s2_access_control_key": "",
"lr_s2_authenticated_key": "",
}
),
)
with (
patch(
"homeassistant.components.zwave_js.async_setup", return_value=True
) as mock_setup,
patch(
"homeassistant.components.zwave_js.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call("core_zwave_js")
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": device,
"s0_legacy_key": "",
"s2_access_control_key": "",
"s2_authenticated_key": "",
"s2_unauthenticated_key": "",
"lr_s2_access_control_key": "",
"lr_s2_authenticated_key": "",
"use_addon": True,
"integration_created_addon": True,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1