Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow configuring Z-Wave JS to talk via ESPHome (#152590) 

Co-authored-by: Artur Pragacz <49985303+arturpragacz@users.noreply.github.com>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
Co-authored-by: J. Nick Koston <nick@koston.org>
pull/152807/head^2
Paulus Schoutsen 2025-09-23 06:58:36 -04:00 committed by GitHub
parent 90bfadda9b
commit a0f67381e5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 590 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
homeassistant/components/esphome/entry_data.py
View File

@ -49,11 +49,13 @@ from aioesphomeapi import (
from aioesphomeapi.model import ButtonInfo
from bleak_esphome.backend.device import ESPHomeBluetoothDevice
from homeassistant import config_entries
from homeassistant.components.assist_satellite import AssistSatelliteConfiguration
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import discovery_flow, entity_registry as er
from homeassistant.helpers.service_info.esphome import ESPHomeServiceInfo
from homeassistant.helpers.storage import Store
from .const import DOMAIN
@ -468,7 +470,7 @@ class RuntimeEntryData:
@callback
def async_on_connect(
self, device_info: DeviceInfo, api_version: APIVersion
self, hass: HomeAssistant, device_info: DeviceInfo, api_version: APIVersion
) -> None:
"""Call when the entry has been connected."""
self.available = True
@ -484,6 +486,29 @@ class RuntimeEntryData:
# be marked as unavailable or not.
self.expected_disconnect = True
if not device_info.zwave_proxy_feature_flags:
return
assert self.client.connected_address
discovery_flow.async_create_flow(
hass,
"zwave_js",
{"source": config_entries.SOURCE_ESPHOME},
ESPHomeServiceInfo(
name=device_info.name,
zwave_home_id=device_info.zwave_home_id or None,
ip_address=self.client.connected_address,
port=self.client.port,
noise_psk=self.client.noise_psk,
),
discovery_key=discovery_flow.DiscoveryKey(
domain=DOMAIN,
key=device_info.mac_address,
version=1,
),
)
@callback
def async_register_assist_satellite_config_updated_callback(
self,

2
homeassistant/components/esphome/manager.py
View File

@ -505,7 +505,7 @@ class ESPHomeManager:
api_version = cli.api_version
assert api_version is not None, "API version must be set"
entry_data.async_on_connect(device_info, api_version)
entry_data.async_on_connect(hass, device_info, api_version)
await self._handle_dynamic_encryption_key(device_info)

32
homeassistant/components/zwave_js/__init__.py
View File

@ -91,6 +91,7 @@ from .const import (
CONF_ADDON_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY,
CONF_ADDON_S2_UNAUTHENTICATED_KEY,
CONF_ADDON_SOCKET,
CONF_DATA_COLLECTION_OPTED_IN,
CONF_INSTALLER_MODE,
CONF_INTEGRATION_CREATED_ADDON,
@ -102,9 +103,11 @@ from .const import (
CONF_S2_ACCESS_CONTROL_KEY,
CONF_S2_AUTHENTICATED_KEY,
CONF_S2_UNAUTHENTICATED_KEY,
CONF_SOCKET_PATH,
CONF_USB_PATH,
CONF_USE_ADDON,
DOMAIN,
ESPHOME_ADDON_VERSION,
EVENT_DEVICE_ADDED_TO_REGISTRY,
EVENT_VALUE_UPDATED,
LIB_LOGGER,
@ -1174,7 +1177,16 @@ async def async_ensure_addon_running(
except AddonError as err:
raise ConfigEntryNotReady(err) from err
usb_path: str = entry.data[CONF_USB_PATH]
addon_has_lr = (
addon_info.version and AwesomeVersion(addon_info.version) >= LR_ADDON_VERSION
)
addon_has_esphome = (
addon_info.version
and AwesomeVersion(addon_info.version) >= ESPHOME_ADDON_VERSION
)
usb_path: str | None = entry.data[CONF_USB_PATH]
socket_path: str | None = entry.data.get(CONF_SOCKET_PATH)
# s0_legacy_key was saved as network_key before s2 was added.
s0_legacy_key: str = entry.data.get(CONF_S0_LEGACY_KEY, "")
if not s0_legacy_key:
@ -1186,15 +1198,18 @@ async def async_ensure_addon_running(
lr_s2_authenticated_key: str = entry.data.get(CONF_LR_S2_AUTHENTICATED_KEY, "")
addon_state = addon_info.state
addon_config = {
CONF_ADDON_DEVICE: usb_path,
CONF_ADDON_S0_LEGACY_KEY: s0_legacy_key,
CONF_ADDON_S2_ACCESS_CONTROL_KEY: s2_access_control_key,
CONF_ADDON_S2_AUTHENTICATED_KEY: s2_authenticated_key,
CONF_ADDON_S2_UNAUTHENTICATED_KEY: s2_unauthenticated_key,
}
if addon_info.version and AwesomeVersion(addon_info.version) >= LR_ADDON_VERSION:
if usb_path is not None:
addon_config[CONF_ADDON_DEVICE] = usb_path
if addon_has_lr:
addon_config[CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY] = lr_s2_access_control_key
addon_config[CONF_ADDON_LR_S2_AUTHENTICATED_KEY] = lr_s2_authenticated_key
if addon_has_esphome and socket_path is not None:
addon_config[CONF_ADDON_SOCKET] = socket_path
if addon_state == AddonState.NOT_INSTALLED:
addon_manager.async_schedule_install_setup_addon(
@ -1211,7 +1226,7 @@ async def async_ensure_addon_running(
raise ConfigEntryNotReady
addon_options = addon_info.options
addon_device = addon_options[CONF_ADDON_DEVICE]
addon_device = addon_options.get(CONF_ADDON_DEVICE)
# s0_legacy_key was saved as network_key before s2 was added.
addon_s0_legacy_key = addon_options.get(CONF_ADDON_S0_LEGACY_KEY, "")
if not addon_s0_legacy_key:
@ -1235,9 +1250,7 @@ async def async_ensure_addon_running(
if s2_unauthenticated_key != addon_s2_unauthenticated_key:
updates[CONF_S2_UNAUTHENTICATED_KEY] = addon_s2_unauthenticated_key
if addon_info.version and AwesomeVersion(addon_info.version) >= AwesomeVersion(
LR_ADDON_VERSION
):
if addon_has_lr:
addon_lr_s2_access_control_key = addon_options.get(
CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY, ""
)
@ -1249,6 +1262,11 @@ async def async_ensure_addon_running(
if lr_s2_authenticated_key != addon_lr_s2_authenticated_key:
updates[CONF_LR_S2_AUTHENTICATED_KEY] = addon_lr_s2_authenticated_key
if addon_has_esphome:
addon_socket = addon_options.get(CONF_ADDON_SOCKET)
if socket_path != addon_socket:
updates[CONF_SOCKET_PATH] = addon_socket
if updates:
hass.config_entries.async_update_entry(entry, data={**entry.data, **updates})

126
homeassistant/components/zwave_js/config_flow.py
View File

@ -26,6 +26,7 @@ from homeassistant.components.hassio import (
AddonState,
)
from homeassistant.config_entries import (
SOURCE_ESPHOME,
SOURCE_USB,
ConfigEntryState,
ConfigFlow,
@ -37,6 +38,7 @@ from homeassistant.data_entry_flow import AbortFlow
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import selector
from homeassistant.helpers.hassio import is_hassio
from homeassistant.helpers.service_info.esphome import ESPHomeServiceInfo
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
from homeassistant.helpers.service_info.usb import UsbServiceInfo
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
@ -52,6 +54,7 @@ from .const import (
CONF_ADDON_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY,
CONF_ADDON_S2_UNAUTHENTICATED_KEY,
CONF_ADDON_SOCKET,
CONF_INTEGRATION_CREATED_ADDON,
CONF_KEEP_OLD_DEVICES,
CONF_LR_S2_ACCESS_CONTROL_KEY,
@ -60,6 +63,7 @@ from .const import (
CONF_S2_ACCESS_CONTROL_KEY,
CONF_S2_AUTHENTICATED_KEY,
CONF_S2_UNAUTHENTICATED_KEY,
CONF_SOCKET_PATH,
CONF_USB_PATH,
CONF_USE_ADDON,
DOMAIN,
@ -81,6 +85,7 @@ ADDON_SETUP_TIMEOUT_ROUNDS = 40
ADDON_USER_INPUT_MAP = {
CONF_ADDON_DEVICE: CONF_USB_PATH,
CONF_ADDON_SOCKET: CONF_SOCKET_PATH,
CONF_ADDON_S0_LEGACY_KEY: CONF_S0_LEGACY_KEY,
CONF_ADDON_S2_ACCESS_CONTROL_KEY: CONF_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY: CONF_S2_AUTHENTICATED_KEY,
@ -129,7 +134,7 @@ def get_manual_schema(user_input: dict[str, Any]) -> vol.Schema:
def get_on_supervisor_schema(user_input: dict[str, Any]) -> vol.Schema:
"""Return a schema for the on Supervisor step."""
default_use_addon = user_input[CONF_USE_ADDON]
return vol.Schema({vol.Optional(CONF_USE_ADDON, default=default_use_addon): bool})
return vol.Schema({vol.Required(CONF_USE_ADDON, default=default_use_addon): bool})
async def validate_input(hass: HomeAssistant, user_input: dict) -> VersionInfo:
@ -197,6 +202,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
self.lr_s2_access_control_key: str | None = None
self.lr_s2_authenticated_key: str | None = None
self.usb_path: str | None = None
self.socket_path: str | None = None # ESPHome socket
self.ws_address: str | None = None
self.restart_addon: bool = False
# If we install the add-on we should uninstall it on entry remove.
@ -214,7 +220,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
self._addon_config_updates: dict[str, Any] = {}
self._migrating = False
self._reconfigure_config_entry: ZwaveJSConfigEntry | None = None
self._usb_discovery = False
self._adapter_discovered = False
self._recommended_install = False
self._rf_region: str | None = None
@ -370,6 +376,11 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
new_addon_config = addon_config | config_updates
if not new_addon_config[CONF_ADDON_DEVICE]:
new_addon_config.pop(CONF_ADDON_DEVICE)
if not new_addon_config[CONF_ADDON_SOCKET]:
new_addon_config.pop(CONF_ADDON_SOCKET)
if new_addon_config == addon_config:
return
@ -542,7 +553,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
title = human_name.split(" - ")[0].strip()
self.context["title_placeholders"] = {CONF_NAME: title}
self._usb_discovery = True
self._adapter_discovered = True
if current_config_entries:
return await self.async_step_confirm_usb_migration()
@ -658,7 +669,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Select custom installation type."""
if self._usb_discovery:
if self._adapter_discovered:
return await self.async_step_on_supervisor({CONF_USE_ADDON: True})
return await self.async_step_on_supervisor()
@ -706,7 +717,8 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
if addon_info.state == AddonState.RUNNING:
addon_config = addon_info.options
self.usb_path = addon_config[CONF_ADDON_DEVICE]
self.usb_path = addon_config.get(CONF_ADDON_DEVICE)
self.socket_path = addon_config.get(CONF_ADDON_SOCKET)
self.s0_legacy_key = addon_config.get(CONF_ADDON_S0_LEGACY_KEY, "")
self.s2_access_control_key = addon_config.get(
CONF_ADDON_S2_ACCESS_CONTROL_KEY, ""
@ -736,14 +748,13 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
"""Ask for config for Z-Wave JS add-on."""
if user_input is not None:
self.usb_path = user_input[CONF_USB_PATH]
self.usb_path = user_input.get(CONF_USB_PATH)
self.socket_path = user_input.get(CONF_SOCKET_PATH)
return await self.async_step_network_type()
if self._usb_discovery:
if self._adapter_discovered:
return await self.async_step_network_type()
usb_path = self.usb_path or ""
try:
ports = await async_get_usb_ports(self.hass)
except OSError as err:
@ -752,7 +763,13 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
data_schema = vol.Schema(
{
vol.Required(CONF_USB_PATH, default=usb_path): vol.In(ports),
vol.Optional(
CONF_USB_PATH, description={"suggested_value": self.usb_path}
): vol.In(ports),
vol.Optional(
CONF_SOCKET_PATH,
description={"suggested_value": self.socket_path or ""},
): str,
}
)
@ -780,6 +797,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
addon_config_updates = {
CONF_ADDON_DEVICE: self.usb_path,
CONF_ADDON_SOCKET: self.socket_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,
@ -851,6 +869,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
addon_config_updates = {
CONF_ADDON_DEVICE: self.usb_path,
CONF_ADDON_SOCKET: self.socket_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,
@ -899,7 +918,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
discovery_info = await self._async_get_addon_discovery_info()
self.ws_address = f"ws://{discovery_info['host']}:{discovery_info['port']}"
if not self.unique_id or self.source == SOURCE_USB:
if not self.unique_id or self.source in (SOURCE_USB, SOURCE_ESPHOME):
if not self.version_info:
try:
self.version_info = await async_get_version_info(
@ -916,6 +935,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
updates={
CONF_URL: self.ws_address,
CONF_USB_PATH: self.usb_path,
CONF_SOCKET_PATH: self.socket_path,
CONF_S0_LEGACY_KEY: self.s0_legacy_key,
CONF_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
CONF_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
@ -938,6 +958,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
data={
CONF_URL: self.ws_address,
CONF_USB_PATH: self.usb_path,
CONF_SOCKET_PATH: self.socket_path,
CONF_S0_LEGACY_KEY: self.s0_legacy_key,
CONF_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
CONF_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
@ -974,7 +995,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
"""Confirm the user wants to reset their current controller."""
config_entry = self._reconfigure_config_entry
assert config_entry is not None
if not self._usb_discovery and not config_entry.data.get(CONF_USE_ADDON):
if not self._adapter_discovered and not config_entry.data.get(CONF_USE_ADDON):
return self.async_abort(
reason="addon_required",
description_placeholders={
@ -1062,9 +1083,10 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
"""Instruct the user to unplug the old controller."""
if user_input is not None:
if self.usb_path:
# USB discovery was used, so the device is already known.
if self._adapter_discovered:
# Discovery was used, so the device is already known.
self._addon_config_updates[CONF_ADDON_DEVICE] = self.usb_path
self._addon_config_updates[CONF_ADDON_SOCKET] = self.socket_path
return await self.async_step_start_addon()
# Now that the old controller is gone, we can scan for serial ports again
return await self.async_step_choose_serial_port()
@ -1184,10 +1206,12 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
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]
self.usb_path = user_input[CONF_USB_PATH]
self.usb_path = user_input.get(CONF_USB_PATH)
self.socket_path = user_input.get(CONF_SOCKET_PATH)
addon_config_updates = {
CONF_ADDON_DEVICE: self.usb_path,
CONF_ADDON_SOCKET: self.socket_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,
@ -1198,6 +1222,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
addon_config_updates = self._addon_config_updates | addon_config_updates
self._addon_config_updates = {}
await self._async_set_addon_config(addon_config_updates)
if addon_info.state == AddonState.RUNNING and not self.restart_addon:
@ -1212,6 +1237,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_start_addon()
usb_path = addon_config.get(CONF_ADDON_DEVICE, self.usb_path or "")
socket_path = addon_config.get(CONF_ADDON_SOCKET, self.socket_path or "")
s0_legacy_key = addon_config.get(
CONF_ADDON_S0_LEGACY_KEY, self.s0_legacy_key or ""
)
@ -1237,24 +1263,42 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
_LOGGER.error("Failed to get USB ports: %s", err)
return self.async_abort(reason="usb_ports_failed")
# Insert empty option in ports to allow setting a socket
ports = {
"": "Use Socket",
**ports,
}
data_schema = vol.Schema(
{
vol.Required(CONF_USB_PATH, default=usb_path): vol.In(ports),
vol.Optional(CONF_S0_LEGACY_KEY, default=s0_legacy_key): str,
vol.Optional(
CONF_S2_ACCESS_CONTROL_KEY, default=s2_access_control_key
CONF_USB_PATH, description={"suggested_value": usb_path}
): vol.In(ports),
vol.Optional(
CONF_SOCKET_PATH, description={"suggested_value": socket_path}
): str,
vol.Optional(
CONF_S2_AUTHENTICATED_KEY, default=s2_authenticated_key
CONF_S0_LEGACY_KEY, description={"suggested_value": s0_legacy_key}
): str,
vol.Optional(
CONF_S2_UNAUTHENTICATED_KEY, default=s2_unauthenticated_key
CONF_S2_ACCESS_CONTROL_KEY,
description={"suggested_value": s2_access_control_key},
): str,
vol.Optional(
CONF_LR_S2_ACCESS_CONTROL_KEY, default=lr_s2_access_control_key
CONF_S2_AUTHENTICATED_KEY,
description={"suggested_value": s2_authenticated_key},
): str,
vol.Optional(
CONF_LR_S2_AUTHENTICATED_KEY, default=lr_s2_authenticated_key
CONF_S2_UNAUTHENTICATED_KEY,
description={"suggested_value": s2_unauthenticated_key},
): str,
vol.Optional(
CONF_LR_S2_ACCESS_CONTROL_KEY,
description={"suggested_value": lr_s2_access_control_key},
): str,
vol.Optional(
CONF_LR_S2_AUTHENTICATED_KEY,
description={"suggested_value": lr_s2_authenticated_key},
): str,
}
)
@ -1268,8 +1312,10 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Choose a serial port."""
if user_input is not None:
self.usb_path = user_input[CONF_USB_PATH]
self.usb_path = user_input.get(CONF_USB_PATH)
self.socket_path = user_input.get(CONF_SOCKET_PATH)
self._addon_config_updates[CONF_ADDON_DEVICE] = self.usb_path
self._addon_config_updates[CONF_ADDON_SOCKET] = self.socket_path
return await self.async_step_start_addon()
try:
@ -1286,10 +1332,16 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
await self.hass.async_add_executor_job(usb.get_serial_by_id, old_usb_path),
None,
)
# Insert empty option in ports to allow setting a socket
ports = {
"": "Use Socket",
**ports,
}
data_schema = vol.Schema(
{
vol.Required(CONF_USB_PATH): vol.In(ports),
vol.Optional(CONF_USB_PATH): vol.In(ports),
vol.Optional(CONF_SOCKET_PATH): str,
}
)
return self.async_show_form(
@ -1347,6 +1399,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
**config_entry.data,
CONF_URL: ws_address,
CONF_USB_PATH: self.usb_path,
CONF_SOCKET_PATH: self.socket_path,
CONF_S0_LEGACY_KEY: self.s0_legacy_key,
CONF_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
CONF_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
@ -1396,6 +1449,7 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
{
CONF_URL: self.ws_address,
CONF_USB_PATH: self.usb_path,
CONF_SOCKET_PATH: self.socket_path,
CONF_S0_LEGACY_KEY: self.s0_legacy_key,
CONF_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
CONF_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
@ -1409,6 +1463,30 @@ class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="reconfigure_successful")
async def async_step_esphome(
self, discovery_info: ESPHomeServiceInfo
) -> ConfigFlowResult:
"""Handle a ESPHome discovery."""
if not is_hassio(self.hass):
return self.async_abort(reason="not_hassio")
if discovery_info.zwave_home_id:
await self.async_set_unique_id(str(discovery_info.zwave_home_id))
self._abort_if_unique_id_configured(
{
CONF_USB_PATH: None,
CONF_SOCKET_PATH: discovery_info.socket_path,
}
)
self.socket_path = discovery_info.socket_path
self.context["title_placeholders"] = {
CONF_NAME: f"{discovery_info.name} via ESPHome"
}
self._adapter_discovered = True
return await self.async_step_installation_type()
async def async_revert_addon_config(self, reason: str) -> ConfigFlowResult:
"""Abort the options flow.

3
homeassistant/components/zwave_js/const.py
View File

@ -12,6 +12,7 @@ from zwave_js_server.const.command_class.window_covering import (
from homeassistant.const import APPLICATION_NAME, __version__ as HA_VERSION
LR_ADDON_VERSION = AwesomeVersion("0.5.0")
ESPHOME_ADDON_VERSION = AwesomeVersion("0.24.0")
USER_AGENT = {APPLICATION_NAME: HA_VERSION}
@ -23,6 +24,7 @@ CONF_ADDON_S2_AUTHENTICATED_KEY = "s2_authenticated_key"
CONF_ADDON_S2_UNAUTHENTICATED_KEY = "s2_unauthenticated_key"
CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY = "lr_s2_access_control_key"
CONF_ADDON_LR_S2_AUTHENTICATED_KEY = "lr_s2_authenticated_key"
CONF_ADDON_SOCKET = "socket"
CONF_INSTALLER_MODE = "installer_mode"
CONF_INTEGRATION_CREATED_ADDON = "integration_created_addon"
CONF_KEEP_OLD_DEVICES = "keep_old_devices"
@ -33,6 +35,7 @@ CONF_S2_AUTHENTICATED_KEY = "s2_authenticated_key"
CONF_S2_UNAUTHENTICATED_KEY = "s2_unauthenticated_key"
CONF_LR_S2_ACCESS_CONTROL_KEY = "lr_s2_access_control_key"
CONF_LR_S2_AUTHENTICATED_KEY = "lr_s2_authenticated_key"
CONF_SOCKET_PATH = "socket_path"
CONF_USB_PATH = "usb_path"
CONF_USE_ADDON = "use_addon"
CONF_DATA_COLLECTION_OPTED_IN = "data_collection_opted_in"

3
homeassistant/components/zwave_js/strings.json
View File

@ -21,7 +21,8 @@
"not_zwave_js_addon": "Discovered add-on is not the official Z-Wave add-on.",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"reset_failed": "Failed to reset adapter.",
"usb_ports_failed": "Failed to get USB devices."
"usb_ports_failed": "Failed to get USB devices.",
"not_hassio": "ESPHome discovery requires Home Assistant to configure the Z-Wave add-on."
},
"error": {
"addon_start_failed": "Failed to start the Z-Wave add-on. Check the configuration.",

16
homeassistant/config_entries.py
View File

@ -100,6 +100,7 @@ _LOGGER = logging.getLogger(__name__)
SOURCE_BLUETOOTH = "bluetooth"
SOURCE_DHCP = "dhcp"
SOURCE_DISCOVERY = "discovery"
SOURCE_ESPHOME = "esphome"
SOURCE_HARDWARE = "hardware"
SOURCE_HASSIO = "hassio"
SOURCE_HOMEKIT = "homekit"
@ -2336,8 +2337,9 @@ class ConfigEntries:
entry: ConfigEntry,
*,
data: Mapping[str, Any] | UndefinedType = UNDEFINED,
discovery_keys: MappingProxyType[str, tuple[DiscoveryKey, ...]]
| UndefinedType = UNDEFINED,
discovery_keys: (
MappingProxyType[str, tuple[DiscoveryKey, ...]] | UndefinedType
) = UNDEFINED,
minor_version: int | UndefinedType = UNDEFINED,
options: Mapping[str, Any] | UndefinedType = UNDEFINED,
pref_disable_new_entities: bool | UndefinedType = UNDEFINED,
@ -2373,8 +2375,9 @@ class ConfigEntries:
entry: ConfigEntry,
*,
data: Mapping[str, Any] | UndefinedType = UNDEFINED,
discovery_keys: MappingProxyType[str, tuple[DiscoveryKey, ...]]
| UndefinedType = UNDEFINED,
discovery_keys: (
MappingProxyType[str, tuple[DiscoveryKey, ...]] | UndefinedType
) = UNDEFINED,
minor_version: int | UndefinedType = UNDEFINED,
options: Mapping[str, Any] | UndefinedType = UNDEFINED,
pref_disable_new_entities: bool | UndefinedType = UNDEFINED,
@ -2728,7 +2731,10 @@ class ConfigEntries:
continue
issues.add(issue.issue_id)
for domain, unique_ids in self._entries._domain_unique_id_index.items(): # noqa: SLF001
for (
domain,
unique_ids,
) in self._entries._domain_unique_id_index.items(): # noqa: SLF001
for unique_id, entries in unique_ids.items():
# We might mutate the list of entries, so we need a copy to not mess up
# the index

26
homeassistant/helpers/service_info/esphome.py Normal file
View File

@ -0,0 +1,26 @@
"""ESPHome discovery data."""
from dataclasses import dataclass
from yarl import URL
from homeassistant.data_entry_flow import BaseServiceInfo
@dataclass(slots=True)
class ESPHomeServiceInfo(BaseServiceInfo):
"""Prepared info from ESPHome entries."""
name: str
zwave_home_id: int | None
ip_address: str
port: int
noise_psk: str | None = None
@property
def socket_path(self) -> str:
"""Return the socket path to connect to the ESPHome device."""
url = URL.build(scheme="esphome", host=self.ip_address, port=self.port)
if self.noise_psk:
url = url.with_user(self.noise_psk)
return str(url)

53
tests/components/esphome/test_entry_data.py
View File

@ -1,5 +1,7 @@
"""Test ESPHome entry data."""
from unittest.mock import Mock, patch
from aioesphomeapi import (
APIClient,
EntityCategory as ESPHomeEntityCategory,
@ -8,9 +10,11 @@ from aioesphomeapi import (
)
from homeassistant.components.esphome import DOMAIN
from homeassistant.components.esphome.entry_data import RuntimeEntryData
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import discovery_flow, entity_registry as er
from homeassistant.helpers.service_info.esphome import ESPHomeServiceInfo
from .conftest import MockGenericDeviceEntryType
@ -69,3 +73,50 @@ async def test_migrate_entity_unique_id_downgrade_upgrade(
# Note that ESPHome includes the EntityInfo type in the unique id
# as this is not a 1:1 mapping to the entity platform (ie. text_sensor)
assert entry.unique_id == "11:22:33:44:55:AA-sensor-mysensor"
async def test_discover_zwave() -> None:
"""Test ESPHome discovery of Z-Wave JS."""
hass = Mock()
entry_data = RuntimeEntryData(
"mock-id",
"mock-title",
Mock(
connected_address="mock-client-address",
port=1234,
noise_psk=None,
),
None,
)
device_info = Mock(
mac_address="mock-device-info-mac",
zwave_proxy_feature_flags=1,
zwave_home_id=1234,
)
device_info.name = "mock-device-infoname"
with patch(
"homeassistant.helpers.discovery_flow.async_create_flow"
) as mock_create_flow:
entry_data.async_on_connect(
hass,
device_info,
None,
)
mock_create_flow.assert_called_once_with(
hass,
"zwave_js",
{"source": "esphome"},
ESPHomeServiceInfo(
name="mock-device-infoname",
zwave_home_id=1234,
ip_address="mock-client-address",
port=1234,
noise_psk=None,
),
discovery_key=discovery_flow.DiscoveryKey(
domain="esphome",
key="mock-device-info-mac",
version=1,
),
)

373
tests/components/zwave_js/test_config_flow.py
View File

@ -29,6 +29,8 @@ from homeassistant.components.zwave_js.const import (
CONF_ADDON_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY,
CONF_ADDON_S2_UNAUTHENTICATED_KEY,
CONF_ADDON_SOCKET,
CONF_SOCKET_PATH,
CONF_USB_PATH,
DOMAIN,
)
@ -36,6 +38,7 @@ from homeassistant.components.zwave_js.helpers import SERVER_VERSION_TIMEOUT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.service_info.esphome import ESPHomeServiceInfo
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
from homeassistant.helpers.service_info.usb import UsbServiceInfo
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
@ -49,6 +52,13 @@ ADDON_DISCOVERY_INFO = {
}
ESPHOME_DISCOVERY_INFO = ESPHomeServiceInfo(
name="mock-name",
zwave_home_id=1234,
ip_address="192.168.1.100",
port=6053,
)
USB_DISCOVERY_INFO = UsbServiceInfo(
device="/dev/zwave",
pid="AAAA",
@ -239,6 +249,7 @@ async def test_manual(hass: HomeAssistant) -> None:
assert result2["data"] == {
"url": "ws://localhost:3000",
"usb_path": None,
"socket_path": None,
"s0_legacy_key": None,
"s2_access_control_key": None,
"s2_authenticated_key": None,
@ -433,6 +444,7 @@ async def test_supervisor_discovery(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -539,6 +551,7 @@ async def test_clean_discovery_on_user_create(
assert result["data"] == {
"url": "ws://localhost:3000",
"usb_path": None,
"socket_path": None,
"s0_legacy_key": None,
"s2_access_control_key": None,
"s2_authenticated_key": None,
@ -754,6 +767,7 @@ async def test_usb_discovery(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": device,
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -866,6 +880,7 @@ async def test_usb_discovery_addon_not_running(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": USB_DISCOVERY_INFO.device,
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -976,7 +991,12 @@ async def test_usb_discovery_migration(
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": USB_DISCOVERY_INFO.device})
"core_zwave_js",
AddonsOptions(
config={
CONF_ADDON_DEVICE: USB_DISCOVERY_INFO.device,
}
),
)
await hass.async_block_till_done()
@ -1006,6 +1026,7 @@ async def test_usb_discovery_migration(
assert result["reason"] == "migration_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == USB_DISCOVERY_INFO.device
assert entry.data["socket_path"] is None
assert entry.data["use_addon"] is True
assert "keep_old_devices" not in entry.data
assert entry.unique_id == "3245146787"
@ -1104,7 +1125,12 @@ async def test_usb_discovery_migration_restore_driver_ready_timeout(
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": USB_DISCOVERY_INFO.device})
"core_zwave_js",
AddonsOptions(
config={
"device": USB_DISCOVERY_INFO.device,
}
),
)
await hass.async_block_till_done()
@ -1135,11 +1161,135 @@ async def test_usb_discovery_migration_restore_driver_ready_timeout(
assert result["reason"] == "migration_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == USB_DISCOVERY_INFO.device
assert entry.data["socket_path"] is None
assert entry.data["use_addon"] is True
assert entry.unique_id == "1234"
assert "keep_old_devices" in entry.data
@pytest.mark.usefixtures("supervisor", "addon_not_installed", "addon_info")
async def test_esphome_discovery(
hass: HomeAssistant,
install_addon: AsyncMock,
set_addon_options: AsyncMock,
start_addon: AsyncMock,
) -> None:
"""Test ESPHome discovery success path."""
# Make sure it works only on hassio
with patch(
"homeassistant.components.zwave_js.config_flow.is_hassio", return_value=False
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ESPHOME},
data=ESPHOME_DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "not_hassio"
# Test working version
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ESPHOME},
data=ESPHOME_DISCOVERY_INFO,
)
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["step_id"] == "install_addon"
assert result["type"] is FlowResultType.SHOW_PROGRESS
# 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"] == "network_type"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"network_type": "existing",
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_security_keys"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
)
assert set_addon_options.call_args == call(
"core_zwave_js",
AddonsOptions(
config={
"socket": "esphome://192.168.1.100:6053",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
}
),
)
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["result"].unique_id == str(ESPHOME_DISCOVERY_INFO.zwave_home_id)
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": None,
"socket_path": "esphome://192.168.1.100:6053",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
"use_addon": True,
"integration_created_addon": True,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.usefixtures("supervisor", "addon_installed")
async def test_discovery_addon_not_running(
hass: HomeAssistant,
@ -1239,6 +1389,7 @@ async def test_discovery_addon_not_running(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -1358,6 +1509,7 @@ async def test_discovery_addon_not_installed(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -1552,6 +1704,7 @@ async def test_not_addon(hass: HomeAssistant) -> None:
assert result["data"] == {
"url": "ws://localhost:3000",
"usb_path": None,
"socket_path": None,
"s0_legacy_key": None,
"s2_access_control_key": None,
"s2_authenticated_key": None,
@ -1612,6 +1765,7 @@ async def test_addon_running(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -1772,6 +1926,7 @@ async def test_addon_running_already_configured(
assert result["reason"] == "already_configured"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == "/test_new"
assert entry.data["socket_path"] is None
assert entry.data["s0_legacy_key"] == "new123"
assert entry.data["s2_access_control_key"] == "new456"
assert entry.data["s2_authenticated_key"] == "new789"
@ -1879,6 +2034,7 @@ async def test_addon_installed(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -2307,6 +2463,7 @@ async def test_addon_installed_already_configured(
assert result["reason"] == "already_configured"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == "/new"
assert entry.data["socket_path"] is None
assert entry.data["s0_legacy_key"] == "new123"
assert entry.data["s2_access_control_key"] == "new456"
assert entry.data["s2_authenticated_key"] == "new789"
@ -2424,6 +2581,7 @@ async def test_addon_not_installed(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
@ -2717,6 +2875,7 @@ async def test_reconfigure_not_addon_with_addon_stop_fail(
(
"entry_data",
"old_addon_options",
"form_data",
"new_addon_options",
"disconnect_calls",
),
@ -2742,6 +2901,15 @@ async def test_reconfigure_not_addon_with_addon_stop_fail(
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
0,
),
(
@ -2765,6 +2933,15 @@ async def test_reconfigure_not_addon_with_addon_stop_fail(
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
1,
),
],
@ -2778,6 +2955,7 @@ async def test_reconfigure_addon_running(
restart_addon: AsyncMock,
entry_data: dict[str, Any],
old_addon_options: dict[str, Any],
form_data: dict[str, Any],
new_addon_options: dict[str, Any],
disconnect_calls: int,
) -> None:
@ -2812,11 +2990,9 @@ async def test_reconfigure_addon_running(
assert result["step_id"] == "configure_addon_reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
new_addon_options,
result["flow_id"], form_data
)
new_addon_options["device"] = new_addon_options.pop("usb_path")
assert set_addon_options.call_args == call(
"core_zwave_js",
AddonsOptions(config=new_addon_options),
@ -2835,7 +3011,8 @@ async def test_reconfigure_addon_running(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == new_addon_options["device"]
assert entry.data["usb_path"] == new_addon_options.get("device")
assert entry.data["socket_path"] == new_addon_options.get("socket")
assert entry.data["s0_legacy_key"] == new_addon_options["s0_legacy_key"]
assert (
entry.data["s2_access_control_key"]
@ -2864,7 +3041,7 @@ async def test_reconfigure_addon_running(
@pytest.mark.usefixtures("supervisor", "addon_running")
@pytest.mark.parametrize(
("entry_data", "old_addon_options", "new_addon_options"),
("entry_data", "old_addon_options", "form_data", "new_addon_options"),
[
(
{},
@ -2887,6 +3064,15 @@ async def test_reconfigure_addon_running(
"lr_s2_access_control_key": "old654",
"lr_s2_authenticated_key": "old321",
},
{
"device": "/test",
"s0_legacy_key": "old123",
"s2_access_control_key": "old456",
"s2_authenticated_key": "old789",
"s2_unauthenticated_key": "old987",
"lr_s2_access_control_key": "old654",
"lr_s2_authenticated_key": "old321",
},
),
],
)
@ -2899,6 +3085,7 @@ async def test_reconfigure_addon_running_no_changes(
restart_addon: AsyncMock,
entry_data: dict[str, Any],
old_addon_options: dict[str, Any],
form_data: dict[str, Any],
new_addon_options: dict[str, Any],
) -> None:
"""Test reconfigure flow without changes, and add-on already running on Supervisor."""
@ -2932,19 +3119,18 @@ async def test_reconfigure_addon_running_no_changes(
assert result["step_id"] == "configure_addon_reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
new_addon_options,
result["flow_id"], form_data
)
await hass.async_block_till_done()
new_addon_options["device"] = new_addon_options.pop("usb_path")
assert set_addon_options.call_count == 0
assert restart_addon.call_count == 0
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == new_addon_options["device"]
assert entry.data["usb_path"] == new_addon_options.get("device")
assert entry.data["socket_path"] == new_addon_options.get("socket")
assert entry.data["s0_legacy_key"] == new_addon_options["s0_legacy_key"]
assert (
entry.data["s2_access_control_key"]
@ -2987,6 +3173,7 @@ async def different_device_server_version(*args):
(
"entry_data",
"old_addon_options",
"form_data",
"new_addon_options",
"disconnect_calls",
"server_version_side_effect",
@ -3013,6 +3200,48 @@ async def different_device_server_version(*args):
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
0,
different_device_server_version,
),
(
{},
{
"device": "/test",
"network_key": "old123",
"s0_legacy_key": "old123",
"s2_access_control_key": "old456",
"s2_authenticated_key": "old789",
"s2_unauthenticated_key": "old987",
"lr_s2_access_control_key": "old654",
"lr_s2_authenticated_key": "old321",
},
{
"socket_path": "esphome://mock-host:6053",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"socket": "esphome://mock-host:6053",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
0,
different_device_server_version,
),
@ -3027,6 +3256,7 @@ async def test_reconfigure_different_device(
restart_addon: AsyncMock,
entry_data: dict[str, Any],
old_addon_options: dict[str, Any],
form_data: dict[str, Any],
new_addon_options: dict[str, Any],
disconnect_calls: int,
) -> None:
@ -3062,12 +3292,10 @@ async def test_reconfigure_different_device(
assert result["step_id"] == "configure_addon_reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
new_addon_options,
result["flow_id"], form_data
)
assert set_addon_options.call_count == 1
new_addon_options["device"] = new_addon_options.pop("usb_path")
assert set_addon_options.call_args == call(
"core_zwave_js", AddonsOptions(config=new_addon_options)
)
@ -3114,6 +3342,7 @@ async def test_reconfigure_different_device(
(
"entry_data",
"old_addon_options",
"form_data",
"new_addon_options",
"disconnect_calls",
"restart_addon_side_effect",
@ -3140,6 +3369,15 @@ async def test_reconfigure_different_device(
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
0,
[SupervisorError(), None],
),
@ -3164,6 +3402,15 @@ async def test_reconfigure_different_device(
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
0,
[
SupervisorError(),
@ -3181,6 +3428,7 @@ async def test_reconfigure_addon_restart_failed(
restart_addon: AsyncMock,
entry_data: dict[str, Any],
old_addon_options: dict[str, Any],
form_data: dict[str, Any],
new_addon_options: dict[str, Any],
disconnect_calls: int,
) -> None:
@ -3216,12 +3464,10 @@ async def test_reconfigure_addon_restart_failed(
assert result["step_id"] == "configure_addon_reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
new_addon_options,
result["flow_id"], form_data
)
assert set_addon_options.call_count == 1
new_addon_options["device"] = new_addon_options.pop("usb_path")
assert set_addon_options.call_args == call(
"core_zwave_js", AddonsOptions(config=new_addon_options)
)
@ -3319,8 +3565,7 @@ async def test_reconfigure_addon_running_server_info_failure(
assert result["step_id"] == "configure_addon_reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
new_addon_options,
result["flow_id"], new_addon_options
)
await hass.async_block_till_done()
@ -3337,6 +3582,7 @@ async def test_reconfigure_addon_running_server_info_failure(
(
"entry_data",
"old_addon_options",
"form_data",
"new_addon_options",
"disconnect_calls",
),
@ -3362,6 +3608,15 @@ async def test_reconfigure_addon_running_server_info_failure(
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
0,
),
(
@ -3385,6 +3640,15 @@ async def test_reconfigure_addon_running_server_info_failure(
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
{
"device": "/new",
"s0_legacy_key": "new123",
"s2_access_control_key": "new456",
"s2_authenticated_key": "new789",
"s2_unauthenticated_key": "new987",
"lr_s2_access_control_key": "new654",
"lr_s2_authenticated_key": "new321",
},
1,
),
],
@ -3399,6 +3663,7 @@ async def test_reconfigure_addon_not_installed(
start_addon: AsyncMock,
entry_data: dict[str, Any],
old_addon_options: dict[str, Any],
form_data: dict[str, Any],
new_addon_options: dict[str, Any],
disconnect_calls: int,
) -> None:
@ -3443,11 +3708,9 @@ async def test_reconfigure_addon_not_installed(
assert result["step_id"] == "configure_addon_reconfigure"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
new_addon_options,
result["flow_id"], form_data
)
new_addon_options["device"] = new_addon_options.pop("usb_path")
assert set_addon_options.call_args == call(
"core_zwave_js", AddonsOptions(config=new_addon_options)
)
@ -3468,7 +3731,7 @@ async def test_reconfigure_addon_not_installed(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == new_addon_options["device"]
assert entry.data["usb_path"] == new_addon_options.get("device")
assert entry.data["s0_legacy_key"] == new_addon_options["s0_legacy_key"]
assert entry.data["use_addon"] is True
assert entry.data["integration_created_addon"] is True
@ -3513,6 +3776,7 @@ async def test_zeroconf(hass: HomeAssistant) -> None:
assert result["data"] == {
"url": "ws://127.0.0.1:3000",
"usb_path": None,
"socket_path": None,
"s0_legacy_key": None,
"s2_access_control_key": None,
"s2_authenticated_key": None,
@ -3578,14 +3842,30 @@ async def test_reconfigure_migrate_low_sdk_version(
@pytest.mark.usefixtures("supervisor", "addon_running")
@pytest.mark.parametrize(
(
"form_data",
"new_addon_options",
"restore_server_version_side_effect",
"final_unique_id",
"keep_old_devices",
"device_entry_count",
),
[
(None, "3245146787", False, 2),
(aiohttp.ClientError("Boom"), "5678", True, 4),
(
{CONF_USB_PATH: "/test"},
{CONF_ADDON_DEVICE: "/test"},
None,
"3245146787",
False,
2,
),
(
{CONF_SOCKET_PATH: "esphome://1.2.3.4:1234"},
{CONF_ADDON_SOCKET: "esphome://1.2.3.4:1234"},
aiohttp.ClientError("Boom"),
"5678",
True,
4,
),
],
)
async def test_reconfigure_migrate_with_addon(
@ -3598,6 +3878,8 @@ async def test_reconfigure_migrate_with_addon(
addon_options: dict[str, Any],
set_addon_options: AsyncMock,
get_server_version: AsyncMock,
form_data: dict[str, Any],
new_addon_options: dict,
restore_server_version_side_effect: Exception | None,
final_unique_id: str,
keep_old_devices: bool,
@ -3714,26 +3996,17 @@ async def test_reconfigure_migrate_with_addon(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "choose_serial_port"
data_schema = result["data_schema"]
assert data_schema is not None
assert data_schema.schema[CONF_USB_PATH]
# Ensure the old usb path is not in the list of options
with pytest.raises(InInvalid):
data_schema.schema[CONF_USB_PATH](addon_options["device"])
version_info.home_id = 5678
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USB_PATH: "/test",
},
result["flow_id"], form_data
)
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": "/test"})
"core_zwave_js", AddonsOptions(config=new_addon_options)
)
# Simulate the new connected controller hardware labels.
@ -3751,17 +4024,19 @@ async def test_reconfigure_migrate_with_addon(
assert restart_addon.call_args == call("core_zwave_js")
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Ensure add-on running would migrate the old settings back into the config entry
with patch("homeassistant.components.zwave_js.async_ensure_addon_running"):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert entry.unique_id == "5678"
get_server_version.side_effect = restore_server_version_side_effect
version_info.home_id = 3245146787
assert entry.unique_id == "5678"
get_server_version.side_effect = restore_server_version_side_effect
version_info.home_id = 3245146787
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "restore_nvm"
assert client.connect.call_count == 2
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "restore_nvm"
assert client.connect.call_count == 2
await hass.async_block_till_done()
await hass.async_block_till_done()
assert client.connect.call_count == 4
assert entry.state is config_entries.ConfigEntryState.LOADED
assert client.driver.controller.async_restore_nvm.call_count == 1
@ -3774,7 +4049,8 @@ async def test_reconfigure_migrate_with_addon(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "migration_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == "/test"
assert entry.data[CONF_USB_PATH] == new_addon_options.get(CONF_ADDON_DEVICE)
assert entry.data[CONF_SOCKET_PATH] == new_addon_options.get(CONF_ADDON_SOCKET)
assert entry.data["use_addon"] is True
assert ("keep_old_devices" in entry.data) is keep_old_devices
assert entry.unique_id == final_unique_id
@ -3931,6 +4207,7 @@ async def test_reconfigure_migrate_restore_driver_ready_timeout(
assert result["reason"] == "migration_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == "/test"
assert entry.data["socket_path"] is None
assert entry.data["use_addon"] is True
assert "keep_old_devices" in entry.data
assert entry.unique_id == "1234"
@ -4443,8 +4720,9 @@ async def test_intent_recommended_user(
assert result["step_id"] == "configure_addon_user"
data_schema = result["data_schema"]
assert data_schema is not None
assert len(data_schema.schema) == 1
assert len(data_schema.schema) == 2
assert data_schema.schema.get(CONF_USB_PATH) is not None
assert data_schema.schema.get(CONF_SOCKET_PATH) is not None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -4491,6 +4769,7 @@ async def test_intent_recommended_user(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": "/test",
"socket_path": None,
"s0_legacy_key": "",
"s2_access_control_key": "",
"s2_authenticated_key": "",
@ -4601,6 +4880,7 @@ async def test_recommended_usb_discovery(
assert result["data"] == {
"url": "ws://host1:3001",
"usb_path": device,
"socket_path": None,
"s0_legacy_key": "",
"s2_access_control_key": "",
"s2_authenticated_key": "",
@ -4860,6 +5140,7 @@ async def test_addon_rf_region_migrate_network(
assert result["reason"] == "migration_successful"
assert entry.data["url"] == "ws://host1:3001"
assert entry.data["usb_path"] == "/test"
assert entry.data["socket_path"] is None
assert entry.data["use_addon"] is True
assert entry.unique_id == "3245146787"
assert client.driver.controller.home_id == 3245146787

14
tests/helpers/test_service_info.py
View File

@ -3,6 +3,7 @@
import pytest
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from homeassistant.helpers.service_info.esphome import ESPHomeServiceInfo
# Ensure that incorrectly formatted mac addresses are rejected, even
# on a constant outside of a test
@ -21,3 +22,16 @@ def test_invalid_macaddress() -> None:
"""Test that DhcpServiceInfo raises ValueError for unformatted macaddress."""
with pytest.raises(ValueError):
DhcpServiceInfo(ip="", hostname="", macaddress="AA:BB:CC:DD:EE:FF")
def test_esphome_socket_path() -> None:
"""Test ESPHomeServiceInfo socket_path property."""
info = ESPHomeServiceInfo(
name="Hello World",
zwave_home_id=123456789,
ip_address="192.168.1.100",
port=6053,
)
assert info.socket_path == "esphome://192.168.1.100:6053"
info.noise_psk = "my-noise-psk"
assert info.socket_path == "esphome://my-noise-psk@192.168.1.100:6053"