diff --git a/homeassistant/components/wiz/__init__.py b/homeassistant/components/wiz/__init__.py index 12c71e392767dc..b2a45d0491a3a0 100644 --- a/homeassistant/components/wiz/__init__.py +++ b/homeassistant/components/wiz/__init__.py @@ -14,6 +14,7 @@ from homeassistant.helpers.typing import ConfigType from .const import ( + CONF_SIGNING_KEY, DISCOVER_SCAN_TIMEOUT, DISCOVERY_INTERVAL, DOMAIN, @@ -62,7 +63,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: WizConfigEntry) -> bool: """Set up the wiz integration from a config entry.""" ip_address = entry.data[CONF_HOST] _LOGGER.debug("Get bulb with IP: %s", ip_address) - bulb = wizlight(ip_address) + bulb = wizlight(ip_address, signing_key=entry.data.get(CONF_SIGNING_KEY)) try: scenes = await bulb.getSupportedScenes() await bulb.getMac() diff --git a/homeassistant/components/wiz/config_flow.py b/homeassistant/components/wiz/config_flow.py index b5ff7f5165ec6d..0149f5d3e02cd7 100644 --- a/homeassistant/components/wiz/config_flow.py +++ b/homeassistant/components/wiz/config_flow.py @@ -1,5 +1,6 @@ """Config flow for WiZ Platform.""" +from dataclasses import asdict import logging from typing import Any, override @@ -9,18 +10,52 @@ from pywizlight.exceptions import WizLightConnectionError, WizLightTimeOutError from homeassistant.components import onboarding -from homeassistant.config_entries import ConfigFlow, ConfigFlowResult -from homeassistant.const import CONF_DEVICE, CONF_HOST +from homeassistant.components.file_upload import process_uploaded_file +from homeassistant.config_entries import ( + SOURCE_IGNORE, + SOURCE_INTEGRATION_DISCOVERY, + ConfigEntryState, + ConfigFlow, + ConfigFlowResult, +) +from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_NAME +from homeassistant.core import callback from homeassistant.data_entry_flow import AbortFlow +from homeassistant.helpers import discovery_flow +from homeassistant.helpers.selector import ( + FileSelector, + FileSelectorConfig, + TextSelector, + TextSelectorConfig, + TextSelectorType, +) from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo +from homeassistant.util.json import json_loads_object from homeassistant.util.network import is_ip_address -from .const import DEFAULT_NAME, DISCOVER_SCAN_TIMEOUT, DOMAIN, WIZ_CONNECT_EXCEPTIONS +from .const import ( + CONF_LOCAL_INTEGRATION_FILE, + CONF_REMOVE_SIGNING_KEY, + CONF_SIGNING_KEY, + DEFAULT_NAME, + DISCOVER_SCAN_TIMEOUT, + DOMAIN, + WIZ_CONNECT_EXCEPTIONS, +) +from .coordinator import WizConfigEntry from .discovery import async_discover_devices -from .utils import _short_mac, name_from_bulb_type_and_mac +from .utils import ( + LocalIntegrationExport, + _short_mac, + name_from_bulb_type_and_mac, + normalize_signing_key, + parse_local_integration_export, +) _LOGGER = logging.getLogger(__name__) +SIGNING_KEY_SELECTOR = TextSelector(TextSelectorConfig(type=TextSelectorType.PASSWORD)) + class WizConfigFlow(ConfigFlow, domain=DOMAIN): """Handle a config flow for WiZ.""" @@ -33,6 +68,14 @@ class WizConfigFlow(ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize the config flow.""" self._discovered_devices: dict[str, DiscoveredBulb] = {} + self._signing_key: str | None = None + self._device_name: str | None = None + + def _entry_data(self, host: str) -> dict[str, str]: + """Return the config entry data for a device at host.""" + if self._signing_key is None: + return {CONF_HOST: host} + return {CONF_HOST: host, CONF_SIGNING_KEY: self._signing_key} @override async def async_step_dhcp( @@ -52,6 +95,8 @@ async def async_step_integration_discovery( self._discovered_device = DiscoveredBulb( discovery_info["ip_address"], discovery_info["mac_address"] ) + self._signing_key = discovery_info.get(CONF_SIGNING_KEY) + self._device_name = discovery_info.get(CONF_NAME) return await self._async_handle_discovery() async def _async_handle_discovery(self) -> ConfigFlowResult: @@ -61,14 +106,17 @@ async def _async_handle_discovery(self) -> ConfigFlowResult: ip_address = device.ip_address mac = device.mac_address await self.async_set_unique_id(mac) - self._abort_if_unique_id_configured(updates={CONF_HOST: ip_address}) + # Loaded entries are reloaded by their update listener + self._abort_if_unique_id_configured( + updates=self._entry_data(ip_address), reload_on_update=False + ) await self._async_connect_discovered_or_abort() return await self.async_step_discovery_confirm() async def _async_connect_discovered_or_abort(self) -> None: """Connect to the device and verify its responding.""" device = self._discovered_device - bulb = wizlight(device.ip_address) + bulb = wizlight(device.ip_address, signing_key=self._signing_key) try: bulbtype = await bulb.get_bulbtype() except WIZ_CONNECT_EXCEPTIONS as ex: @@ -81,7 +129,9 @@ async def _async_connect_discovered_or_abort(self) -> None: raise AbortFlow("cannot_connect") from ex finally: await bulb.async_close() - self._name = name_from_bulb_type_and_mac(bulbtype, device.mac_address) + self._name = self._device_name or name_from_bulb_type_and_mac( + bulbtype, device.mac_address + ) async def async_step_discovery_confirm( self, user_input: dict[str, Any] | None = None @@ -95,7 +145,7 @@ async def async_step_discovery_confirm( await self._async_connect_discovered_or_abort() return self.async_create_entry( title=self._name, - data={CONF_HOST: ip_address}, + data=self._entry_data(ip_address), ) self._set_confirm_only() @@ -113,7 +163,7 @@ async def async_step_pick_device( if user_input is not None: device = self._discovered_devices[user_input[CONF_DEVICE]] await self.async_set_unique_id(device.mac_address, raise_on_progress=False) - bulb = wizlight(device.ip_address) + bulb = wizlight(device.ip_address, signing_key=self._signing_key) try: bulbtype = await bulb.get_bulbtype() except WIZ_CONNECT_EXCEPTIONS: @@ -123,7 +173,7 @@ async def async_step_pick_device( return self.async_create_entry( title=name_from_bulb_type_and_mac(bulbtype, device.mac_address), - data={CONF_HOST: device.ip_address}, + data=self._entry_data(device.ip_address), ) current_unique_ids = self._async_current_ids(include_ignore=False) @@ -157,14 +207,31 @@ async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Handle a flow initialized by the user.""" + return self.async_show_menu( + step_id="user", menu_options=["manual", "local_integration"] + ) + + async def async_step_manual( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle setting up a device by IP address or network scan.""" errors = {} if user_input is not None: - if not (host := user_input[CONF_HOST]): + host = user_input[CONF_HOST] + raw_signing_key = user_input.get(CONF_SIGNING_KEY) + signing_key = ( + normalize_signing_key(raw_signing_key) if raw_signing_key else None + ) + if raw_signing_key and signing_key is None: + errors[CONF_SIGNING_KEY] = "invalid_signing_key" + elif not host: + self._signing_key = signing_key return await self.async_step_pick_device() - if not is_ip_address(user_input[CONF_HOST]): + elif not is_ip_address(host): errors["base"] = "no_ip" else: - bulb = wizlight(host) + self._signing_key = signing_key + bulb = wizlight(host, signing_key=signing_key) try: bulbtype = await bulb.get_bulbtype() mac = await bulb.getMac() @@ -178,22 +245,180 @@ async def async_step_user( _LOGGER.exception("Unexpected exception") errors["base"] = "unknown" else: - await self.async_set_unique_id(mac, raise_on_progress=False) - self._abort_if_unique_id_configured( - updates={CONF_HOST: user_input[CONF_HOST]} - ) + if ( + entry := await self.async_set_unique_id( + mac, raise_on_progress=False + ) + ) and entry.source != SOURCE_IGNORE: + self._async_update_entry( + entry, {**entry.data, **self._entry_data(host)} + ) + return self.async_abort(reason="already_configured") name = name_from_bulb_type_and_mac(bulbtype, mac) return self.async_create_entry( title=name, - data=user_input, + data=self._entry_data(host), ) finally: await bulb.async_close() return self.async_show_form( - step_id="user", + step_id="manual", + data_schema=probatio.Schema( + { + probatio.Optional(CONF_HOST, default=""): str, + probatio.Optional( + probatio.Secret(CONF_SIGNING_KEY) + ): SIGNING_KEY_SELECTOR, + } + ), + errors=errors, + ) + + async def async_step_reconfigure( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle reconfiguration of a WiZ device.""" + errors = {} + reconfigure_entry = self._get_reconfigure_entry() + host = reconfigure_entry.data[CONF_HOST] + if user_input is not None: + host = user_input[CONF_HOST] + raw_signing_key = user_input.get(CONF_SIGNING_KEY) + remove_signing_key = user_input.get(CONF_REMOVE_SIGNING_KEY, False) + if remove_signing_key: + signing_key = None + elif raw_signing_key: + signing_key = normalize_signing_key(raw_signing_key) + else: + signing_key = reconfigure_entry.data.get(CONF_SIGNING_KEY) + if raw_signing_key and remove_signing_key: + errors[CONF_SIGNING_KEY] = "signing_key_with_remove" + elif raw_signing_key and signing_key is None: + errors[CONF_SIGNING_KEY] = "invalid_signing_key" + elif not is_ip_address(host): + errors[CONF_HOST] = "no_ip" + else: + bulb = wizlight(host, signing_key=signing_key) + try: + mac = await bulb.getMac() + except WIZ_CONNECT_EXCEPTIONS: + errors["base"] = "cannot_connect" + else: + await self.async_set_unique_id(mac, raise_on_progress=False) + self._abort_if_unique_id_mismatch(reason="different_device") + self._signing_key = signing_key + self._async_update_entry(reconfigure_entry, self._entry_data(host)) + return self.async_abort(reason="reconfigure_successful") + finally: + await bulb.async_close() + + schema: dict[Any, Any] = { + probatio.Required(CONF_HOST, default=host): str, + probatio.Optional(probatio.Secret(CONF_SIGNING_KEY)): SIGNING_KEY_SELECTOR, + } + if CONF_SIGNING_KEY in reconfigure_entry.data: + schema[probatio.Optional(CONF_REMOVE_SIGNING_KEY, default=False)] = bool + return self.async_show_form( + step_id="reconfigure", + data_schema=probatio.Schema(schema), + errors=errors, + description_placeholders={"name": reconfigure_entry.title}, + ) + + async def async_step_local_integration( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle importing a WiZ app local integration export.""" + errors = {} + if user_input is not None: + try: + export = await self.hass.async_add_executor_job( + self._read_local_integration_file, + user_input[CONF_LOCAL_INTEGRATION_FILE], + ) + except (ValueError, probatio.Invalid) as err: + _LOGGER.debug("Invalid local integration file: %s", err) + errors["base"] = "invalid_local_integration_file" + else: + return await self._async_import_local_integration(export) + + return self.async_show_form( + step_id="local_integration", data_schema=probatio.Schema( - {probatio.Optional(CONF_HOST, default=""): str} + { + probatio.Required(CONF_LOCAL_INTEGRATION_FILE): FileSelector( + FileSelectorConfig(accept=".json,application/json") + ) + } ), errors=errors, ) + + def _read_local_integration_file(self, file_id: str) -> LocalIntegrationExport: + """Read and parse an uploaded local integration export.""" + with process_uploaded_file(self.hass, file_id) as file_path: + contents = file_path.read_text(encoding="utf-8") + return parse_local_integration_export(json_loads_object(contents)) + + async def _async_import_local_integration( + self, export: LocalIntegrationExport + ) -> ConfigFlowResult: + """Add the signing key to the devices of the exported home.""" + updated = 0 + for entry in self._async_current_entries(include_ignore=False): + if entry.unique_id not in export.devices: + continue + updated += 1 + if entry.data.get(CONF_SIGNING_KEY) != export.signing_key: + self._async_update_entry( + entry, {**entry.data, CONF_SIGNING_KEY: export.signing_key} + ) + + current_unique_ids = self._async_current_ids() + new_devices = { + device.mac_address: device + for device in await async_discover_devices(self.hass, DISCOVER_SCAN_TIMEOUT) + if device.mac_address in export.devices + and device.mac_address not in current_unique_ids + } + # Replace discovery flows that were started without the signing key, + # including those of devices the scan missed + for flow in self._async_in_progress(include_uninitialized=True): + mac = flow["context"].get("unique_id") + if mac not in export.devices or mac in current_unique_ids: + continue + if mac not in new_devices: + if not (placeholders := flow["context"].get("title_placeholders")): + continue + new_devices[mac] = DiscoveredBulb(placeholders["host"], mac) + self.hass.config_entries.flow.async_abort(flow["flow_id"]) + for device in new_devices.values(): + discovery_flow.async_create_flow( + self.hass, + DOMAIN, + context={"source": SOURCE_INTEGRATION_DISCOVERY}, + data={ + **asdict(device), + CONF_NAME: export.devices[device.mac_address], + CONF_SIGNING_KEY: export.signing_key, + }, + ) + + if not updated and not new_devices: + return self.async_abort(reason="no_devices_found") + return self.async_abort( + reason="local_integration_successful", + description_placeholders={ + "updated": str(updated), + "discovered": str(len(new_devices)), + }, + ) + + @callback + def _async_update_entry(self, entry: WizConfigEntry, data: dict[str, Any]) -> None: + """Update the data of an entry and reload it.""" + self.hass.config_entries.async_update_entry(entry, data=data) + # The update listener only reloads entries that are loaded + if entry.state is not ConfigEntryState.LOADED: + self.hass.config_entries.async_schedule_reload(entry.entry_id) diff --git a/homeassistant/components/wiz/const.py b/homeassistant/components/wiz/const.py index 59cc7788a74fcc..5a18061ce36278 100644 --- a/homeassistant/components/wiz/const.py +++ b/homeassistant/components/wiz/const.py @@ -12,6 +12,10 @@ DOMAIN = "wiz" DEFAULT_NAME = "WiZ" +CONF_LOCAL_INTEGRATION_FILE = "local_integration_file" +CONF_REMOVE_SIGNING_KEY = "remove_signing_key" +CONF_SIGNING_KEY = "signing_key" + DISCOVER_SCAN_TIMEOUT = 10 DISCOVERY_INTERVAL = timedelta(minutes=15) diff --git a/homeassistant/components/wiz/diagnostics.py b/homeassistant/components/wiz/diagnostics.py index d342a797ca4821..e499f2d9053325 100644 --- a/homeassistant/components/wiz/diagnostics.py +++ b/homeassistant/components/wiz/diagnostics.py @@ -5,9 +5,11 @@ from homeassistant.components.diagnostics import async_redact_data from homeassistant.core import HomeAssistant +from .const import CONF_SIGNING_KEY from .coordinator import WizConfigEntry TO_REDACT = {"roomId", "homeId"} +TO_REDACT_ENTRY = {CONF_SIGNING_KEY} async def async_get_config_entry_diagnostics( @@ -17,7 +19,7 @@ async def async_get_config_entry_diagnostics( return { "entry": { "title": entry.title, - "data": dict(entry.data), + "data": async_redact_data(entry.data, TO_REDACT_ENTRY), }, "data": async_redact_data(entry.runtime_data.bulb.diagnostics, TO_REDACT), } diff --git a/homeassistant/components/wiz/manifest.json b/homeassistant/components/wiz/manifest.json index f76bc745af5cde..ca7448c0a093cc 100644 --- a/homeassistant/components/wiz/manifest.json +++ b/homeassistant/components/wiz/manifest.json @@ -3,7 +3,7 @@ "name": "WiZ", "codeowners": ["@sbidy", "@arturpragacz"], "config_flow": true, - "dependencies": ["network"], + "dependencies": ["file_upload", "network"], "dhcp": [ { "registered_devices": true diff --git a/homeassistant/components/wiz/strings.json b/homeassistant/components/wiz/strings.json index b3bd6120fe0107..6a98e59e11ad4e 100644 --- a/homeassistant/components/wiz/strings.json +++ b/homeassistant/components/wiz/strings.json @@ -3,13 +3,19 @@ "abort": { "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", - "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]" + "different_device": "The IP address belongs to a different WiZ device.", + "local_integration_successful": "{updated} configured devices now use the security key from the file, and {discovered} new devices were found. Add the new devices from the discovered list now; if any device gets set up without the key, import the file again.", + "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]" }, "error": { "bulb_time_out": "Cannot connect to the bulb. Maybe the bulb is offline or a wrong IP was entered. Please turn on the light and try again!", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_local_integration_file": "The file is not a valid WiZ local integration export.", + "invalid_signing_key": "The security key must be 32 hexadecimal characters.", "no_ip": "Not a valid IP address.", "no_wiz_light": "The bulb cannot be connected via WiZ integration.", + "signing_key_with_remove": "Leave the security key empty when removing it.", "unknown": "[%key:common::config_flow::error::unknown%]" }, "flow_title": "{name} ({host})", @@ -17,16 +23,47 @@ "discovery_confirm": { "description": "Do you want to set up {name} ({host})?" }, + "local_integration": { + "data": { + "local_integration_file": "Local integration file" + }, + "data_description": { + "local_integration_file": "In the WiZ app, open the home settings, select Integrations, then Local integrations, and download the file. The download link is only valid for 15 minutes." + }, + "description": "Adds the home security key to the devices of the exported home, and finds devices that are not set up yet." + }, + "manual": { + "data": { + "host": "[%key:common::config_flow::data::ip%]", + "signing_key": "Security key" + }, + "data_description": { + "signing_key": "The home security key from the local integrations export in the WiZ app. Only needed if the home only allows verified controls." + }, + "description": "If you leave the IP address empty, discovery will be used to find devices." + }, "pick_device": { "data": { "device": "[%key:common::config_flow::data::device%]" } }, - "user": { + "reconfigure": { "data": { - "host": "[%key:common::config_flow::data::ip%]" + "host": "[%key:common::config_flow::data::ip%]", + "remove_signing_key": "Remove the security key", + "signing_key": "[%key:component::wiz::config::step::manual::data::signing_key%]" }, - "description": "If you leave the IP address empty, discovery will be used to find devices." + "data_description": { + "remove_signing_key": "Send commands unsigned. Only works if the home allows all controls in the WiZ app.", + "signing_key": "[%key:component::wiz::config::step::manual::data_description::signing_key%]" + }, + "description": "Update the IP address or security key of {name}. Leave the security key empty to keep the current one." + }, + "user": { + "menu_options": { + "local_integration": "Import the local integration file from the WiZ app", + "manual": "Search the network or enter an IP address" + } } } }, diff --git a/homeassistant/components/wiz/utils.py b/homeassistant/components/wiz/utils.py index 67ebe7bb0be112..d46bea4c3566c0 100644 --- a/homeassistant/components/wiz/utils.py +++ b/homeassistant/components/wiz/utils.py @@ -1,10 +1,38 @@ """WiZ utils.""" +from dataclasses import dataclass +import re +from typing import Any + +import probatio from pywizlight import BulbType from pywizlight.bulblibrary import BulbClass from .const import DEFAULT_NAME +SIGNING_KEY_RE = re.compile(r"[0-9a-f]{32}") +MAC_RE = re.compile(r"[0-9a-f]{12}") + + +LOCAL_INTEGRATION_EXPORT_SCHEMA = probatio.Schema( + { + probatio.Required("udp_signing_key"): probatio.All( + str, probatio.Strip, probatio.Lower, probatio.Match(r"[0-9a-f]{32}\Z") + ), + probatio.Required("devices"): [dict], + }, + extra=probatio.ALLOW_EXTRA, +) + + +@dataclass(frozen=True) +class LocalIntegrationExport: + """Home exported by the local integrations page of the WiZ app.""" + + signing_key: str + # Device names by MAC address + devices: dict[str, str | None] + def _short_mac(mac: str) -> str: """Get the short mac address from the full mac.""" @@ -21,3 +49,27 @@ def name_from_bulb_type_and_mac(bulb_type: BulbType, mac: str) -> str: else: description = bulb_type.bulb_type.value return f"{DEFAULT_NAME} {description} {_short_mac(mac)}" + + +def normalize_signing_key(signing_key: str) -> str | None: + """Return the signing key in lowercase, or None if it is not 32 hex digits.""" + signing_key = signing_key.strip().lower() + if SIGNING_KEY_RE.fullmatch(signing_key) is None: + return None + return signing_key + + +def parse_local_integration_export(data: Any) -> LocalIntegrationExport: + """Parse a home exported by the local integrations page of the WiZ app.""" + home = LOCAL_INTEGRATION_EXPORT_SCHEMA(data) + devices: dict[str, str | None] = {} + for device in home["devices"]: + mac = device.get("mac_address") + # Skip entries that cannot be matched to a device on the network + if not isinstance(mac, str) or not MAC_RE.fullmatch(mac := mac.strip().lower()): + continue + name = device.get("name") + devices[mac] = name if isinstance(name, str) and name else None + if not devices: + raise ValueError("No devices with a MAC address in the export") + return LocalIntegrationExport(signing_key=home["udp_signing_key"], devices=devices) diff --git a/tests/components/wiz/__init__.py b/tests/components/wiz/__init__.py index a11a7251998562..bcc68bd9eca0ed 100644 --- a/tests/components/wiz/__init__.py +++ b/tests/components/wiz/__init__.py @@ -41,6 +41,7 @@ ) FAKE_IP = "1.1.1.1" FAKE_MAC = "abcabcabcabc" +FAKE_SIGNING_KEY = "00112233445566778899aabbccddeeff" FAKE_BULB_CONFIG = { "method": "getSystemConfig", "env": "pro", @@ -316,12 +317,13 @@ async def async_setup_integration( device: dict[str, Any] | None = None, extended_white_range: list[int] | None = None, bulb_type: BulbType | None = None, + data: dict[str, Any] | None = None, ) -> tuple[wizlight, MockConfigEntry]: """Set up the integration with a mock device.""" entry = MockConfigEntry( domain=DOMAIN, unique_id=FAKE_MAC, - data={CONF_HOST: FAKE_IP}, + data=data or {CONF_HOST: FAKE_IP}, ) entry.add_to_hass(hass) bulb = wizlight or _mocked_wizlight(device, extended_white_range, bulb_type) diff --git a/tests/components/wiz/fixtures/local_integration.json b/tests/components/wiz/fixtures/local_integration.json new file mode 100644 index 00000000000000..12536bfe76f114 --- /dev/null +++ b/tests/components/wiz/fixtures/local_integration.json @@ -0,0 +1,77 @@ +{ + "name": "Home", + "home_id": 12345678, + "region": "eu", + "udp_signing_key": "00112233445566778899aabbccddeeff", + "version": "1.0", + "creation_date": "2026-07-01T05:02:47+0000", + "update_date": "2026-10-03T15:39:36+0000", + "rooms": [ + { + "room_id": 23456789, + "name": "Bedroom", + "type": "BEDROOM", + "creation_date": "2026-07-01T05:02:47+0000", + "update_date": "2026-10-10T13:09:08+0000" + }, + { + "room_id": 23456790, + "name": "Garage", + "type": "GARAGE", + "creation_date": "2026-10-02T14:27:31+0000", + "update_date": "2026-10-02T14:27:31+0000" + } + ], + "devices": [ + { + "device_id": 34567890, + "type": "light", + "room_id": 23456789, + "group_id": null, + "name": "Bedroom lamp", + "mac_address": "abcabcabcabc", + "fw_version": "1.38.0", + "traits": { + "is_dimmable": true, + "is_tunable_white": true, + "white_range": [2200, 6500], + "is_tunable_color": true, + "supports_light_mode": true + }, + "creation_date": "2026-07-06T15:11:46+0000", + "update_date": "2026-10-07T13:10:08+0000" + }, + { + "device_id": 34567891, + "type": "light", + "room_id": 23456789, + "group_id": null, + "name": "Desk lamp", + "mac_address": "a8bb50000001", + "fw_version": "1.38.0", + "traits": { + "is_dimmable": true, + "is_tunable_white": true, + "white_range": [2700, 6500], + "is_tunable_color": true, + "supports_light_mode": true + }, + "creation_date": "2026-10-01T04:49:47+0000", + "update_date": "2026-10-10T20:18:41+0000" + }, + { + "device_id": 34567892, + "type": "plug", + "room_id": 23456789, + "group_id": null, + "name": "Kettle plug", + "mac_address": "a8bb50000002", + "fw_version": "1.39.0", + "traits": { + "has_power_meter": true + }, + "creation_date": "2026-10-01T04:49:47+0000", + "update_date": "2026-10-10T20:18:41+0000" + } + ] +} diff --git a/tests/components/wiz/snapshots/test_diagnostics.ambr b/tests/components/wiz/snapshots/test_diagnostics.ambr index 5fe9aa883a1881..1931dba36ee97e 100644 --- a/tests/components/wiz/snapshots/test_diagnostics.ambr +++ b/tests/components/wiz/snapshots/test_diagnostics.ambr @@ -9,6 +9,7 @@ 'entry': dict({ 'data': dict({ 'host': '1.1.1.1', + 'signing_key': '**REDACTED**', }), 'title': 'Mock Title', }), diff --git a/tests/components/wiz/test_config_flow.py b/tests/components/wiz/test_config_flow.py index 30f7f4c14fce8b..5f4e7d22d369e1 100644 --- a/tests/components/wiz/test_config_flow.py +++ b/tests/components/wiz/test_config_flow.py @@ -1,13 +1,24 @@ """Test the WiZ Platform config flow.""" -from unittest.mock import AsyncMock, patch +import asyncio +import json +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch import pytest +from pywizlight import BulbType +from pywizlight.discovery import DiscoveredBulb from pywizlight.exceptions import WizLightConnectionError, WizLightTimeOutError from homeassistant import config_entries -from homeassistant.components.wiz.const import DOMAIN -from homeassistant.config_entries import ConfigEntryState +from homeassistant.components.wiz.const import ( + CONF_LOCAL_INTEGRATION_FILE, + CONF_REMOVE_SIGNING_KEY, + CONF_SIGNING_KEY, + DOMAIN, +) +from homeassistant.config_entries import ConfigEntryState, ConfigFlowResult from homeassistant.const import CONF_DEVICE, CONF_HOST from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType @@ -20,6 +31,7 @@ FAKE_MAC, FAKE_RGBW_BULB, FAKE_RGBWW_BULB, + FAKE_SIGNING_KEY, FAKE_SOCKET, TEST_CONNECTION, TEST_SYSTEM_INFO, @@ -29,7 +41,11 @@ async_setup_integration, ) -from tests.common import MockConfigEntry +from tests.common import ( + MockConfigEntry, + async_load_fixture, + async_load_json_object_fixture, +) DHCP_DISCOVERY = DhcpServiceInfo( hostname="wiz_abcabc", @@ -43,14 +59,33 @@ "mac_address": FAKE_MAC, } +NEW_IP = "1.1.1.2" +NEW_MAC = "a8bb50000001" +OTHER_IP = "1.1.1.3" +OTHER_MAC = "a8bb50000002" +LOCAL_INTEGRATION_FILE_ID = "6a5f6e2c-3f5e-4c51-8d1e-9a0f3c2b1d4e" +UNLISTED_IP = "1.1.1.4" +UNLISTED_MAC = "a8bb50000003" +LOCAL_INTEGRATION_FIXTURE = "local_integration.json" -async def test_form(hass: HomeAssistant) -> None: - """Test we get the form.""" - bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) +async def _async_init_manual_flow(hass: HomeAssistant) -> ConfigFlowResult: + """Start a user flow and pick the manual setup.""" 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"] == "user" + return await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "manual"} + ) + + +async def test_form(hass: HomeAssistant) -> None: + """Test we get the form.""" + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + + result = await _async_init_manual_flow(hass) assert result["type"] is FlowResultType.FORM assert result["errors"] == {} # Patch functions @@ -83,9 +118,7 @@ async def test_form(hass: HomeAssistant) -> None: async def test_user_flow_enters_dns_name(hass: HomeAssistant) -> None: """Test we reject dns names and want ips.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) assert result["type"] is FlowResultType.FORM assert result["errors"] == {} @@ -136,9 +169,7 @@ async def test_user_form_exceptions( hass: HomeAssistant, side_effect: type[Exception], error_base: str ) -> None: """Test all user exceptions in the flow.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) bulb.get_bulbtype = AsyncMock(side_effect=side_effect) @@ -176,10 +207,8 @@ async def test_form_updates_unique_id(hass: HomeAssistant) -> None: ) entry.add_to_hass(hass) - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) - with _patch_wizlight(): + result = await _async_init_manual_flow(hass) + with _patch_discovery(), _patch_wizlight(): result2 = await hass.config_entries.flow.async_configure( result["flow_id"], TEST_CONNECTION, @@ -189,6 +218,28 @@ async def test_form_updates_unique_id(hass: HomeAssistant) -> None: assert result2["type"] is FlowResultType.ABORT assert result2["reason"] == "already_configured" assert entry.data[CONF_HOST] == FAKE_IP + assert entry.state is ConfigEntryState.LOADED + + +async def test_manual_reloads_entry_retrying_setup(hass: HomeAssistant) -> None: + """Test entering a device that failed to set up retries it right away.""" + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + bulb.getMac = AsyncMock(side_effect=WizLightTimeOutError) + _, entry = await async_setup_integration(hass, wizlight=bulb) + assert entry.state is ConfigEntryState.SETUP_RETRY + bulb.getMac = AsyncMock(return_value=FAKE_MAC) + + result = await _async_init_manual_flow(hass) + with _patch_discovery(), _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + assert entry.data == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert entry.state is ConfigEntryState.LOADED @pytest.mark.parametrize( @@ -382,12 +433,10 @@ async def test_discovered_by_dhcp_or_integration_discovery_avoid_waiting_for_ret async def test_setup_via_discovery(hass: HomeAssistant) -> None: """Test setting up via discovery.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) await hass.async_block_till_done() assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "user" + assert result["step_id"] == "manual" assert not result["errors"] with _patch_discovery(): @@ -399,11 +448,9 @@ async def test_setup_via_discovery(hass: HomeAssistant) -> None: assert not result2["errors"] # test we can try again - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "user" + assert result["step_id"] == "manual" assert not result["errors"] with _patch_discovery(): @@ -442,11 +489,9 @@ async def test_setup_via_discovery(hass: HomeAssistant) -> None: bulb.async_close.assert_awaited_once() # ignore configured devices - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "user" + assert result["step_id"] == "manual" assert not result["errors"] with _patch_discovery(): @@ -459,12 +504,10 @@ async def test_setup_via_discovery(hass: HomeAssistant) -> None: async def test_setup_via_discovery_cannot_connect(hass: HomeAssistant) -> None: """Test setting up via discovery and we fail to connect to the discovered device.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) await hass.async_block_till_done() assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "user" + assert result["step_id"] == "manual" assert not result["errors"] with _patch_discovery(): @@ -494,12 +537,10 @@ async def test_setup_via_discovery_cannot_connect(hass: HomeAssistant) -> None: async def test_setup_via_discovery_exception_finds_nothing(hass: HomeAssistant) -> None: """Test we do not find anything if discovery throws.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) await hass.async_block_till_done() assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "user" + assert result["step_id"] == "manual" assert not result["errors"] with patch( @@ -612,9 +653,7 @@ async def test_flow_replace_ignored_device(hass: HomeAssistant) -> None: entry.add_to_hass(hass) # Patch discovery to find the same ignored device with _patch_discovery(), _patch_wizlight(): - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) + result = await _async_init_manual_flow(hass) result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) await hass.async_block_till_done() assert result["type"] is FlowResultType.FORM @@ -643,3 +682,729 @@ async def test_flow_replace_ignored_device(hass: HomeAssistant) -> None: assert result["result"].unique_id == FAKE_MAC assert len(mock_setup.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1 + + +def _patch_local_integration_file(tmp_path: Path, contents: str) -> Any: + """Patch reading the uploaded local integration file.""" + file_path = tmp_path / "wiz.json" + file_path.write_text(contents, encoding="utf-8") + ctx_mock = MagicMock() + ctx_mock.__enter__.return_value = file_path + return patch( + "homeassistant.components.wiz.config_flow.process_uploaded_file", + return_value=ctx_mock, + ) + + +def _patch_discovered_devices(devices: list[DiscoveredBulb]) -> Any: + """Patch the devices found by a network scan.""" + return patch( + "homeassistant.components.wiz.discovery.find_wizlights", return_value=devices + ) + + +async def test_manual_with_signing_key(hass: HomeAssistant) -> None: + """Test a device is set up with the signing key entered by the user.""" + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + result = await _async_init_manual_flow(hass) + + with ( + patch( + "homeassistant.components.wiz.config_flow.wizlight", return_value=bulb + ) as mock_wizlight, + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + patch("homeassistant.components.wiz.async_setup", return_value=True), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: f" {FAKE_SIGNING_KEY.upper()} "}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == FAKE_MAC + mock_wizlight.assert_called_once_with(FAKE_IP, signing_key=FAKE_SIGNING_KEY) + + +@pytest.mark.parametrize( + "signing_key", + [ + pytest.param("0011", id="too_short"), + pytest.param("00" * 17, id="too_long"), + pytest.param("zz" * 16, id="not_hex"), + ], +) +async def test_manual_invalid_signing_key( + hass: HomeAssistant, signing_key: str +) -> None: + """Test an invalid signing key is rejected and the flow recovers.""" + result = await _async_init_manual_flow(hass) + + with _patch_wizlight(): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: signing_key} + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {CONF_SIGNING_KEY: "invalid_signing_key"} + + with ( + _patch_wizlight(), + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + patch("homeassistant.components.wiz.async_setup", return_value=True), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == FAKE_MAC + + +async def test_manual_pick_device_with_signing_key(hass: HomeAssistant) -> None: + """Test a picked device is set up with the signing key entered by the user.""" + result = await _async_init_manual_flow(hass) + + with _patch_discovery(): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "pick_device" + + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + with ( + patch( + "homeassistant.components.wiz.config_flow.wizlight", return_value=bulb + ) as mock_wizlight, + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + patch("homeassistant.components.wiz.async_setup", return_value=True), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_DEVICE: FAKE_MAC} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == FAKE_MAC + mock_wizlight.assert_called_once_with(FAKE_IP, signing_key=FAKE_SIGNING_KEY) + + +async def test_manual_adds_signing_key_to_existing_entry( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test setting up a configured device again adds the signing key.""" + bulb, entry = await async_setup_integration(hass) + result = await _async_init_manual_flow(hass) + + with _patch_discovery(), _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + assert entry.data == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert entry.state is ConfigEntryState.LOADED + assert "update listener" not in caplog.text + + +async def test_integration_discovery_with_signing_key(hass: HomeAssistant) -> None: + """Test a device discovered with a signing key is set up with it.""" + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + with patch( + "homeassistant.components.wiz.config_flow.wizlight", return_value=bulb + ) as mock_wizlight: + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, + data={**INTEGRATION_DISCOVERY, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "discovery_confirm" + mock_wizlight.assert_called_once_with(FAKE_IP, signing_key=FAKE_SIGNING_KEY) + + with ( + _patch_wizlight(device=bulb), + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + patch("homeassistant.components.wiz.async_setup", return_value=True), + ): + result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == FAKE_MAC + + +async def test_integration_discovery_adds_signing_key( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test discovery with a signing key adds it to the configured device.""" + bulb, entry = await async_setup_integration(hass) + + with _patch_discovery(), _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, + data={**INTEGRATION_DISCOVERY, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + assert entry.data == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert entry.state is ConfigEntryState.LOADED + assert "update listener" not in caplog.text + + +@pytest.mark.parametrize( + ("entry_data", "user_input", "expected_data"), + [ + pytest.param( + TEST_CONNECTION, + {CONF_HOST: NEW_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY.upper()}, + {CONF_HOST: NEW_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + id="add_key", + ), + pytest.param( + {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + {CONF_HOST: NEW_IP}, + {CONF_HOST: NEW_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + id="keep_key", + ), + pytest.param( + {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + {CONF_HOST: FAKE_IP, CONF_REMOVE_SIGNING_KEY: True}, + {CONF_HOST: FAKE_IP}, + id="remove_key", + ), + ], +) +async def test_reconfigure( + hass: HomeAssistant, + entry_data: dict[str, str], + user_input: dict[str, Any], + expected_data: dict[str, str], +) -> None: + """Test reconfiguring the host and signing key of a loaded device.""" + bulb, entry = await async_setup_integration(hass, data=entry_data) + assert entry.state is ConfigEntryState.LOADED + + result = await entry.start_reconfigure_flow(hass) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + + with ( + _patch_discovery(), + patch( + "homeassistant.components.wiz.config_flow.wizlight", return_value=bulb + ) as mock_flow_wizlight, + patch("homeassistant.components.wiz.wizlight", return_value=bulb), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + assert entry.data == expected_data + assert entry.state is ConfigEntryState.LOADED + mock_flow_wizlight.assert_called_once_with( + expected_data[CONF_HOST], signing_key=expected_data.get(CONF_SIGNING_KEY) + ) + + +@pytest.mark.parametrize( + ("entry_data", "offers_remove"), + [ + pytest.param(TEST_CONNECTION, False, id="without_key"), + pytest.param( + {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + True, + id="with_key", + ), + ], +) +async def test_reconfigure_offers_removing_signing_key( + hass: HomeAssistant, entry_data: dict[str, str], offers_remove: bool +) -> None: + """Test removing the signing key is only offered when there is one.""" + entry = MockConfigEntry(domain=DOMAIN, unique_id=FAKE_MAC, data=entry_data) + entry.add_to_hass(hass) + + result = await entry.start_reconfigure_flow(hass) + + assert result["type"] is FlowResultType.FORM + assert (CONF_REMOVE_SIGNING_KEY in result["data_schema"].schema) is offers_remove + + +async def test_reconfigure_remove_signing_key_with_new_key( + hass: HomeAssistant, +) -> None: + """Test entering a key while removing it is rejected and the flow recovers.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id=FAKE_MAC, + data={CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + ) + entry.add_to_hass(hass) + + result = await entry.start_reconfigure_flow(hass) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: FAKE_IP, + CONF_SIGNING_KEY: "ff" * 16, + CONF_REMOVE_SIGNING_KEY: True, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {CONF_SIGNING_KEY: "signing_key_with_remove"} + + with _patch_discovery(), _patch_wizlight(): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_REMOVE_SIGNING_KEY: True} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + assert entry.data == {CONF_HOST: FAKE_IP} + + +async def test_reconfigure_reloads_entry_not_loaded(hass: HomeAssistant) -> None: + """Test reconfiguring a device that failed to set up reloads it.""" + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + bulb.getMac = AsyncMock(side_effect=WizLightTimeOutError) + _, entry = await async_setup_integration(hass, wizlight=bulb) + assert entry.state is ConfigEntryState.SETUP_RETRY + bulb.getMac = AsyncMock(return_value=FAKE_MAC) + + result = await entry.start_reconfigure_flow(hass) + with _patch_discovery(), _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + assert entry.data == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert entry.state is ConfigEntryState.LOADED + + +@pytest.mark.parametrize( + ("user_input", "side_effect", "errors"), + [ + pytest.param( + {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: "0011"}, + None, + {CONF_SIGNING_KEY: "invalid_signing_key"}, + id="invalid_key", + ), + pytest.param( + {CONF_HOST: "this is no IP input"}, + None, + {CONF_HOST: "no_ip"}, + id="invalid_host", + ), + pytest.param( + {CONF_HOST: FAKE_IP}, + WizLightTimeOutError, + {"base": "cannot_connect"}, + id="cannot_connect", + ), + ], +) +async def test_reconfigure_errors( + hass: HomeAssistant, + user_input: dict[str, str], + side_effect: type[Exception] | None, + errors: dict[str, str], +) -> None: + """Test reconfigure errors and recovery.""" + entry = MockConfigEntry(domain=DOMAIN, unique_id=FAKE_MAC, data=TEST_CONNECTION) + entry.add_to_hass(hass) + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + bulb.getMac = AsyncMock(side_effect=side_effect, return_value=FAKE_MAC) + + result = await entry.start_reconfigure_flow(hass) + with _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == errors + + bulb.getMac = AsyncMock(return_value=FAKE_MAC) + with _patch_discovery(), _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + assert entry.data == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + + +async def test_reconfigure_different_device(hass: HomeAssistant) -> None: + """Test reconfiguring to the address of another device aborts.""" + entry = MockConfigEntry(domain=DOMAIN, unique_id=FAKE_MAC, data=TEST_CONNECTION) + entry.add_to_hass(hass) + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + bulb.getMac = AsyncMock(return_value=NEW_MAC) + + result = await entry.start_reconfigure_flow(hass) + with _patch_wizlight(device=bulb): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: NEW_IP} + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "different_device" + assert entry.data == TEST_CONNECTION + bulb.async_close.assert_awaited_once() + + +async def test_local_integration(hass: HomeAssistant, tmp_path: Path) -> None: + """Test importing a local integration export.""" + bulb, entry = await async_setup_integration(hass) + ignored_entry = MockConfigEntry( + domain=DOMAIN, unique_id=OTHER_MAC, source=config_entries.SOURCE_IGNORE + ) + ignored_entry.add_to_hass(hass) + other_home_entry = MockConfigEntry( + domain=DOMAIN, unique_id="a8bb50000004", data={CONF_HOST: "1.1.1.5"} + ) + other_home_entry.add_to_hass(hass) + with _patch_wizlight(device=bulb): + stale_flow = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, + data={"ip_address": NEW_IP, "mac_address": NEW_MAC}, + ) + unlisted_flow = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, + data={"ip_address": UNLISTED_IP, "mac_address": UNLISTED_MAC}, + ) + assert stale_flow["step_id"] == "discovery_confirm" + assert unlisted_flow["step_id"] == "discovery_confirm" + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "local_integration"} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "local_integration" + + export = await async_load_fixture(hass, LOCAL_INTEGRATION_FIXTURE, DOMAIN) + with ( + _patch_local_integration_file(tmp_path, export), + _patch_discovered_devices( + [ + DiscoveredBulb(FAKE_IP, FAKE_MAC), + DiscoveredBulb(NEW_IP, NEW_MAC), + DiscoveredBulb(OTHER_IP, OTHER_MAC), + DiscoveredBulb(UNLISTED_IP, UNLISTED_MAC), + ] + ), + _patch_wizlight(device=bulb), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "local_integration_successful" + assert result["description_placeholders"] == {"updated": "1", "discovered": "1"} + assert entry.data == {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert entry.state is ConfigEntryState.LOADED + assert other_home_entry.data == {CONF_HOST: "1.1.1.5"} + + flows = { + flow["context"]["unique_id"]: flow + for flow in hass.config_entries.flow.async_progress_by_handler(DOMAIN) + } + assert flows.keys() == {NEW_MAC, UNLISTED_MAC} + assert flows[UNLISTED_MAC]["flow_id"] == unlisted_flow["flow_id"] + assert flows[NEW_MAC]["flow_id"] != stale_flow["flow_id"] + assert flows[NEW_MAC]["context"]["title_placeholders"] == { + "name": "Desk lamp", + "host": NEW_IP, + } + + with ( + _patch_wizlight(device=bulb), + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + ): + result = await hass.config_entries.flow.async_configure( + flows[NEW_MAC]["flow_id"], {} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "Desk lamp" + assert result["data"] == {CONF_HOST: NEW_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == NEW_MAC + + +async def test_local_integration_replaces_flow_missed_by_scan( + hass: HomeAssistant, tmp_path: Path +) -> None: + """Test a pending discovery of a listed device the scan missed gets the key.""" + with _patch_wizlight(): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_DHCP}, + data=DhcpServiceInfo( + hostname="wiz_000001", ip="10.0.0.5", macaddress=NEW_MAC + ), + ) + assert result["step_id"] == "discovery_confirm" + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "local_integration"} + ) + export = await async_load_fixture(hass, LOCAL_INTEGRATION_FIXTURE, DOMAIN) + with ( + _patch_local_integration_file(tmp_path, export), + _patch_discovered_devices([]), + _patch_wizlight(), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "local_integration_successful" + assert result["description_placeholders"] == {"updated": "0", "discovered": "1"} + flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN) + assert len(flows) == 1 + assert flows[0]["context"]["source"] == config_entries.SOURCE_INTEGRATION_DISCOVERY + + with ( + _patch_wizlight(), + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + patch("homeassistant.components.wiz.async_setup", return_value=True), + ): + result = await hass.config_entries.flow.async_configure(flows[0]["flow_id"], {}) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_HOST: "10.0.0.5", CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == NEW_MAC + + +async def test_local_integration_skips_flow_still_connecting( + hass: HomeAssistant, tmp_path: Path +) -> None: + """Test a discovery that has not reached the device yet is left alone.""" + connected = asyncio.Event() + + async def _get_bulbtype() -> BulbType: + await connected.wait() + return FAKE_DIMMABLE_BULB + + bulb = _mocked_wizlight(None, None, FAKE_DIMMABLE_BULB) + bulb.get_bulbtype = AsyncMock(side_effect=_get_bulbtype) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "local_integration"} + ) + export = await async_load_fixture(hass, LOCAL_INTEGRATION_FIXTURE, DOMAIN) + + with ( + _patch_wizlight(device=bulb), + _patch_local_integration_file(tmp_path, export), + _patch_discovered_devices([]), + ): + init_task = hass.async_create_task( + hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_DHCP}, + data=DhcpServiceInfo( + hostname="wiz_000001", ip=NEW_IP, macaddress=NEW_MAC + ), + ) + ) + await asyncio.sleep(0) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + connected.set() + discovery_result = await init_task + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "no_devices_found" + assert discovery_result["type"] is FlowResultType.FORM + assert discovery_result["step_id"] == "discovery_confirm" + + +async def test_local_integration_incomplete_devices( + hass: HomeAssistant, tmp_path: Path +) -> None: + """Test devices without a name get a generated one and without a MAC are skipped.""" + export = await async_load_json_object_fixture( + hass, LOCAL_INTEGRATION_FIXTURE, DOMAIN + ) + for device in export["devices"]: + device["name"] = None + device["mac_address"] = device["mac_address"].upper() + export["devices"].append({"device_id": 1, "type": "remote", "mac_address": None}) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "local_integration"} + ) + + with ( + _patch_local_integration_file(tmp_path, json.dumps(export)), + _patch_discovered_devices([DiscoveredBulb(NEW_IP, NEW_MAC)]), + _patch_wizlight(), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "local_integration_successful" + assert result["description_placeholders"] == {"updated": "0", "discovered": "1"} + flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN) + assert len(flows) == 1 + + with ( + _patch_wizlight(), + patch("homeassistant.components.wiz.async_setup_entry", return_value=True), + patch("homeassistant.components.wiz.async_setup", return_value=True), + ): + result = await hass.config_entries.flow.async_configure(flows[0]["flow_id"], {}) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "WiZ Dimmable White 000001" + assert result["data"] == {CONF_HOST: NEW_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + assert result["result"].unique_id == NEW_MAC + + +@pytest.mark.parametrize( + "contents", + [ + pytest.param("not json", id="not_json"), + pytest.param("[]", id="not_an_object"), + pytest.param(json.dumps({"devices": [{"mac_address": FAKE_MAC}]}), id="no_key"), + pytest.param( + json.dumps( + {"udp_signing_key": "0011", "devices": [{"mac_address": FAKE_MAC}]} + ), + id="invalid_key", + ), + pytest.param( + json.dumps({"udp_signing_key": FAKE_SIGNING_KEY}), id="no_devices" + ), + pytest.param( + json.dumps({"udp_signing_key": FAKE_SIGNING_KEY, "devices": []}), + id="empty_devices", + ), + pytest.param( + json.dumps( + { + "udp_signing_key": FAKE_SIGNING_KEY, + "devices": [{"mac_address": "ab:ca:bc:ab:ca:bc"}], + } + ), + id="invalid_mac", + ), + ], +) +async def test_local_integration_invalid_file( + hass: HomeAssistant, tmp_path: Path, contents: str +) -> None: + """Test an invalid local integration export is rejected and the flow recovers.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "local_integration"} + ) + + with _patch_local_integration_file(tmp_path, contents): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "invalid_local_integration_file"} + + export = await async_load_fixture(hass, LOCAL_INTEGRATION_FIXTURE, DOMAIN) + with ( + _patch_local_integration_file(tmp_path, export), + _patch_discovered_devices([DiscoveredBulb(NEW_IP, NEW_MAC)]), + _patch_wizlight(), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "local_integration_successful" + assert result["description_placeholders"] == {"updated": "0", "discovered": "1"} + + +async def test_local_integration_no_devices( + hass: HomeAssistant, tmp_path: Path +) -> None: + """Test importing an export without configured or discovered devices.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "local_integration"} + ) + + export = await async_load_fixture(hass, LOCAL_INTEGRATION_FIXTURE, DOMAIN) + with ( + _patch_local_integration_file(tmp_path, export), + _patch_discovered_devices([DiscoveredBulb(UNLISTED_IP, UNLISTED_MAC)]), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_LOCAL_INTEGRATION_FILE: LOCAL_INTEGRATION_FILE_ID}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "no_devices_found" + assert not hass.config_entries.flow.async_progress_by_handler(DOMAIN) diff --git a/tests/components/wiz/test_diagnostics.py b/tests/components/wiz/test_diagnostics.py index 14fbdbf916a1a5..bc3fef81e89b4e 100644 --- a/tests/components/wiz/test_diagnostics.py +++ b/tests/components/wiz/test_diagnostics.py @@ -2,9 +2,11 @@ from syrupy.assertion import SnapshotAssertion +from homeassistant.components.wiz.const import CONF_SIGNING_KEY +from homeassistant.const import CONF_HOST from homeassistant.core import HomeAssistant -from . import async_setup_integration +from . import FAKE_IP, FAKE_SIGNING_KEY, async_setup_integration from tests.components.diagnostics import get_diagnostics_for_config_entry from tests.typing import ClientSessionGenerator @@ -16,6 +18,8 @@ async def test_diagnostics( snapshot: SnapshotAssertion, ) -> None: """Test generating diagnostics for a config entry.""" - _, entry = await async_setup_integration(hass) + _, entry = await async_setup_integration( + hass, data={CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY} + ) assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == snapshot diff --git a/tests/components/wiz/test_init.py b/tests/components/wiz/test_init.py index 7fcc22e0f07dcc..44fdb90d3e0138 100644 --- a/tests/components/wiz/test_init.py +++ b/tests/components/wiz/test_init.py @@ -1,9 +1,12 @@ """Tests for wiz integration.""" import datetime +from typing import Any from unittest.mock import AsyncMock, patch -from homeassistant.components.wiz.const import DOMAIN +import pytest + +from homeassistant.components.wiz.const import CONF_SIGNING_KEY, DOMAIN from homeassistant.config_entries import ConfigEntryState from homeassistant.const import ATTR_FRIENDLY_NAME, CONF_HOST, EVENT_HOMEASSISTANT_STOP from homeassistant.core import HomeAssistant @@ -13,6 +16,7 @@ from . import ( FAKE_IP, FAKE_MAC, + FAKE_SIGNING_KEY, FAKE_SOCKET, _mocked_wizlight, _patch_discovery, @@ -97,3 +101,34 @@ async def test_reload_on_title_change(hass: HomeAssistant) -> None: hass.states.get("switch.mock_title").attributes[ATTR_FRIENDLY_NAME] == "Shop Switch" ) + + +@pytest.mark.parametrize( + ("data", "signing_key"), + [ + pytest.param({CONF_HOST: FAKE_IP}, None, id="without_key"), + pytest.param( + {CONF_HOST: FAKE_IP, CONF_SIGNING_KEY: FAKE_SIGNING_KEY}, + FAKE_SIGNING_KEY, + id="with_key", + ), + ], +) +async def test_setup_uses_signing_key( + hass: HomeAssistant, data: dict[str, Any], signing_key: str | None +) -> None: + """Test the device is set up with the signing key of the entry.""" + entry = MockConfigEntry(domain=DOMAIN, unique_id=FAKE_MAC, data=data) + entry.add_to_hass(hass) + bulb = _mocked_wizlight(None, None, None) + with ( + _patch_discovery(), + patch( + "homeassistant.components.wiz.wizlight", return_value=bulb + ) as mock_wizlight, + ): + await async_setup_component(hass, DOMAIN, {DOMAIN: {}}) + await hass.async_block_till_done() + + assert entry.state is ConfigEntryState.LOADED + mock_wizlight.assert_called_once_with(FAKE_IP, signing_key=signing_key)