From 51acdeb5635688ea6cb6a248e9a7acdd0e7cfe45 Mon Sep 17 00:00:00 2001 From: peteS-UK <64092177+peteS-UK@users.noreply.github.com> Date: Thu, 26 Feb 2026 18:59:13 +0000 Subject: [PATCH] Add config flow support to Orvibo legacy integration (#155115) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Erik Montnemery Co-authored-by: Joost Lekkerkerker --- homeassistant/components/orvibo/__init__.py | 52 ++- .../components/orvibo/config_flow.py | 205 ++++++++++ homeassistant/components/orvibo/const.py | 5 + homeassistant/components/orvibo/manifest.json | 1 + homeassistant/components/orvibo/models.py | 7 + homeassistant/components/orvibo/strings.json | 71 ++++ homeassistant/components/orvibo/switch.py | 169 ++++++--- homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 2 +- requirements_test_all.txt | 3 + tests/components/orvibo/__init__.py | 1 + tests/components/orvibo/conftest.py | 54 +++ tests/components/orvibo/test_config_flow.py | 352 ++++++++++++++++++ 13 files changed, 879 insertions(+), 44 deletions(-) create mode 100644 homeassistant/components/orvibo/config_flow.py create mode 100644 homeassistant/components/orvibo/const.py create mode 100644 homeassistant/components/orvibo/models.py create mode 100644 homeassistant/components/orvibo/strings.json create mode 100644 tests/components/orvibo/__init__.py create mode 100644 tests/components/orvibo/conftest.py create mode 100644 tests/components/orvibo/test_config_flow.py diff --git a/homeassistant/components/orvibo/__init__.py b/homeassistant/components/orvibo/__init__.py index 81cddecb6721..71c6e0609c59 100644 --- a/homeassistant/components/orvibo/__init__.py +++ b/homeassistant/components/orvibo/__init__.py @@ -1 +1,51 @@ -"""The orvibo component.""" +"""The orvibo integration.""" + +import logging + +from orvibo.s20 import S20, S20Exception + +from homeassistant import core +from homeassistant.const import CONF_HOST, CONF_MAC, Platform +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady + +from .const import DOMAIN +from .models import S20ConfigEntry + +PLATFORMS = [Platform.SWITCH] + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: core.HomeAssistant, entry: S20ConfigEntry) -> bool: + """Set up platform from a ConfigEntry.""" + + try: + s20 = await hass.async_add_executor_job( + S20, + entry.data[CONF_HOST], + entry.data[CONF_MAC], + ) + _LOGGER.debug("Initialized S20 at %s", entry.data[CONF_HOST]) + except S20Exception as err: + _LOGGER.debug("S20 at %s couldn't be initialized", entry.data[CONF_HOST]) + + raise ConfigEntryNotReady( + translation_domain=DOMAIN, + translation_key="init_error", + translation_placeholders={ + "host": entry.data[CONF_HOST], + }, + ) from err + + entry.runtime_data = s20 + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: S20ConfigEntry) -> bool: + """Unload a config entry.""" + + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) diff --git a/homeassistant/components/orvibo/config_flow.py b/homeassistant/components/orvibo/config_flow.py new file mode 100644 index 000000000000..13f914e094ec --- /dev/null +++ b/homeassistant/components/orvibo/config_flow.py @@ -0,0 +1,205 @@ +"""Config flow for the orvibo integration.""" + +import asyncio +import logging +from typing import Any + +from orvibo.s20 import S20, S20Exception, discover +import voluptuous as vol + +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult +from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.device_registry import format_mac + +from .const import CONF_SWITCH_LIST, DEFAULT_NAME, DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +FULL_EDIT_SCHEMA = vol.Schema( + { + vol.Required(CONF_HOST): cv.string, + vol.Optional(CONF_MAC): cv.string, + } +) + + +class S20ConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle the config flow for Orvibo S20 switches.""" + + VERSION = 1 + MINOR_VERSION = 1 + + def __init__(self) -> None: + """Initialize an instance of the S20 config flow.""" + self.discovery_task: asyncio.Task | None = None + self._discovered_switches: dict[str, dict[str, Any]] = {} + self.chosen_switch: dict[str, Any] = {} + + async def _async_discover(self) -> None: + def _filter_discovered_switches( + switches: dict[str, dict[str, Any]], + ) -> dict[str, dict[str, Any]]: + # Get existing unique_ids from config entries + existing_ids = {entry.unique_id for entry in self._async_current_entries()} + _LOGGER.debug("Existing unique IDs: %s", existing_ids) + # Build a new filtered dict + filtered = {} + for ip, info in switches.items(): + mac_bytes = info.get("mac") + if not mac_bytes: + continue # skip if no MAC + + unique_id = format_mac(mac_bytes.hex()).lower() + if unique_id not in existing_ids: + filtered[ip] = info + _LOGGER.debug("New switches: %s", filtered) + return filtered + + # Discover S20 devices. + _LOGGER.debug("Discovering S20 switches") + + _unfiltered_switches = await self.hass.async_add_executor_job(discover) + _LOGGER.debug("All discovered switches: %s", _unfiltered_switches) + + self._discovered_switches = _filter_discovered_switches(_unfiltered_switches) + + 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=["start_discovery", "edit"] + ) + + async def _validate_input(self, user_input: dict[str, Any]) -> str | None: + """Validate user input and discover MAC if missing.""" + + if user_input.get(CONF_MAC): + user_input[CONF_MAC] = format_mac(user_input[CONF_MAC]).lower() + if len(user_input[CONF_MAC]) != 17 or user_input[CONF_MAC].count(":") != 5: + return "invalid_mac" + + try: + device = await self.hass.async_add_executor_job( + S20, + user_input[CONF_HOST], + user_input.get(CONF_MAC), + ) + + if not user_input.get(CONF_MAC): + # Using private attribute access here since S20 class doesn't have a public method to get the MAC without repeating discovery + if not device._mac: # noqa: SLF001 + return "cannot_discover" + user_input[CONF_MAC] = format_mac(device._mac.hex()).lower() # noqa: SLF001 + + except S20Exception: + return "cannot_connect" + + return None + + async def async_step_edit( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Edit a discovered or manually configured server.""" + + errors = {} + if user_input: + error = await self._validate_input(user_input) + if not error: + await self.async_set_unique_id(user_input[CONF_MAC]) + self._abort_if_unique_id_configured() + return self.async_create_entry( + title=f"{DEFAULT_NAME} ({user_input[CONF_HOST]})", data=user_input + ) + errors["base"] = error + + return self.async_show_form( + step_id="edit", + data_schema=FULL_EDIT_SCHEMA, + errors=errors, + ) + + async def async_step_start_discovery( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle a flow initialized by the user.""" + if not self.discovery_task: + self.discovery_task = self.hass.async_create_task(self._async_discover()) + return self.async_show_progress( + step_id="start_discovery", + progress_action="start_discovery", + progress_task=self.discovery_task, + ) + if self.discovery_task.done(): + try: + self.discovery_task.result() + except (S20Exception, OSError) as err: + _LOGGER.debug("Discovery task failed: %s", err) + self.discovery_task = None + return self.async_show_progress_done( + next_step_id=( + "choose_switch" if self._discovered_switches else "discovery_failed" + ) + ) + return self.async_show_progress( + step_id="start_discovery", + progress_action="start_discovery", + progress_task=self.discovery_task, + ) + + async def async_step_choose_switch( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Choose manual or discover flow.""" + _chosen_host: str + + if user_input: + _chosen_host = user_input[CONF_SWITCH_LIST] + for host, data in self._discovered_switches.items(): + if _chosen_host == host: + self.chosen_switch[CONF_HOST] = host + self.chosen_switch[CONF_MAC] = format_mac( + data[CONF_MAC].hex() + ).lower() + await self.async_set_unique_id(self.chosen_switch[CONF_MAC]) + self._abort_if_unique_id_configured() + return self.async_create_entry( + title=f"{DEFAULT_NAME} ({host})", data=self.chosen_switch + ) + _LOGGER.debug("discovered switches: %s", self._discovered_switches) + + _options = { + host: f"{host} ({format_mac(data[CONF_MAC].hex()).lower()})" + for host, data in self._discovered_switches.items() + } + return self.async_show_form( + step_id="choose_switch", + data_schema=vol.Schema({vol.Required(CONF_SWITCH_LIST): vol.In(_options)}), + ) + + async def async_step_discovery_failed( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle a failed discovery.""" + + return self.async_show_menu( + step_id="discovery_failed", menu_options=["start_discovery", "edit"] + ) + + async def async_step_import(self, user_input: dict[str, Any]) -> ConfigFlowResult: + """Handle import from configuration.yaml.""" + _LOGGER.debug("Importing config: %s", user_input) + + error = await self._validate_input(user_input) + if error: + return self.async_abort(reason=error) + + await self.async_set_unique_id(user_input[CONF_MAC]) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=user_input.get(CONF_NAME, user_input[CONF_HOST]), data=user_input + ) diff --git a/homeassistant/components/orvibo/const.py b/homeassistant/components/orvibo/const.py new file mode 100644 index 000000000000..0286588ddbe2 --- /dev/null +++ b/homeassistant/components/orvibo/const.py @@ -0,0 +1,5 @@ +"""Constants for the orvibo integration.""" + +DOMAIN = "orvibo" +DEFAULT_NAME = "S20" +CONF_SWITCH_LIST = "switches" diff --git a/homeassistant/components/orvibo/manifest.json b/homeassistant/components/orvibo/manifest.json index e3a6676b2f2f..8ec76f83513a 100644 --- a/homeassistant/components/orvibo/manifest.json +++ b/homeassistant/components/orvibo/manifest.json @@ -2,6 +2,7 @@ "domain": "orvibo", "name": "Orvibo", "codeowners": [], + "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/orvibo", "iot_class": "local_push", "loggers": ["orvibo"], diff --git a/homeassistant/components/orvibo/models.py b/homeassistant/components/orvibo/models.py new file mode 100644 index 000000000000..d702ecef61ac --- /dev/null +++ b/homeassistant/components/orvibo/models.py @@ -0,0 +1,7 @@ +"""Data models for the Orvibo integration.""" + +from orvibo.s20 import S20 + +from homeassistant.config_entries import ConfigEntry + +type S20ConfigEntry = ConfigEntry[S20] diff --git a/homeassistant/components/orvibo/strings.json b/homeassistant/components/orvibo/strings.json new file mode 100644 index 000000000000..93ab02b755e0 --- /dev/null +++ b/homeassistant/components/orvibo/strings.json @@ -0,0 +1,71 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", + "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]", + "cannot_connect": "Unable to connect to the S20 switch", + "cannot_discover": "Unable to discover MAC address of S20 switch. Please enter the MAC address.", + "invalid_mac": "Invalid MAC address format" + }, + "error": { + "cannot_connect": "[%key:component::orvibo::config::abort::cannot_connect%]", + "cannot_discover": "[%key:component::orvibo::config::abort::cannot_discover%]", + "invalid_mac": "Invalid MAC address format" + }, + "progress": { + "start_discovery": "Attempting to discover new S20 switches\n\nThis will take about 3 seconds\n\nDiscovery may fail if the switch is asleep. If your switch does not appear, please power toggle your switch before re-running discovery.", + "title": "Orvibo S20" + }, + "step": { + "choose_switch": { + "data": { + "switches": "Choose discovered switch to configure" + }, + "title": "Discovered switches" + }, + "discovery_failed": { + "description": "No S20 switches were discovered on the network. Discovery may have failed if the switch is asleep. Please power toggle your switch before re-running discovery.", + "menu_options": { + "edit": "Enter configuration manually", + "start_discovery": "Try discovering again" + }, + "title": "Discovery failed" + }, + "edit": { + "data": { + "host": "[%key:common::config_flow::data::host%]", + "mac": "MAC address" + }, + "title": "Configure Orvibo S20 switch" + }, + "user": { + "menu_options": { + "edit": "Enter configuration manually", + "start_discovery": "Discover new S20 switches" + }, + "title": "Orvibo S20 Configuration" + } + } + }, + "exceptions": { + "init_error": { + "message": "Error while initializing S20 {host}." + }, + "turn_off_error": { + "message": "Error while turning off S20 {name}." + }, + "turn_on_error": { + "message": "Error while turning on S20 {name}." + } + }, + "issues": { + "yaml_deprecation": { + "description": "The device (MAC: {mac}, Host: {host}) is configured in `configuration.yaml`. The Orvibo integration now supports UI-based configuration and this device has been migrated to the new UI. Please remove the YAML block from `configuration.yaml` to avoid future issues.", + "title": "Legacy YAML configuration detected {host}" + }, + "yaml_deprecation_import_issue": { + "description": "Attempting to import this device (MAC: {mac}, Host: {host}) from YAML has failed for reason {reason}. 1) Remove the YAML block from `configuration.yaml`, 2) Restart Home Assistant, 3) Add the device using the UI configuration flow.", + "title": "Legacy YAML configuration import issue for {host}" + } + } +} diff --git a/homeassistant/components/orvibo/switch.py b/homeassistant/components/orvibo/switch.py index 3853c10e01c8..a7a829d7b66b 100644 --- a/homeassistant/components/orvibo/switch.py +++ b/homeassistant/components/orvibo/switch.py @@ -1,13 +1,14 @@ -"""Support for Orvibo S20 Wifi Smart Switches.""" +"""Switch platform for the Orvibo integration.""" from __future__ import annotations import logging from typing import Any -from orvibo.s20 import S20, S20Exception, discover +from orvibo.s20 import S20, S20Exception import voluptuous as vol +from homeassistant import config_entries from homeassistant.components.switch import ( PLATFORM_SCHEMA as SWITCH_PLATFORM_SCHEMA, SwitchEntity, @@ -20,14 +21,25 @@ from homeassistant.const import ( CONF_SWITCHES, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers import config_validation as cv -from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.data_entry_flow import FlowResultType +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import config_validation as cv, issue_registry as ir +from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo +from homeassistant.helpers.entity_platform import ( + AddConfigEntryEntitiesCallback, + AddEntitiesCallback, +) from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType +from .const import DEFAULT_NAME, DOMAIN +from .models import S20ConfigEntry + _LOGGER = logging.getLogger(__name__) -DEFAULT_NAME = "Orvibo S20 Switch" -DEFAULT_DISCOVERY = True +DEFAULT_DISCOVERY = False + +# Library is not thread safe and uses global variables, so we limit to 1 update at a time +PARALLEL_UPDATES = 1 PLATFORM_SCHEMA = SWITCH_PLATFORM_SCHEMA.extend( { @@ -46,65 +58,138 @@ PLATFORM_SCHEMA = SWITCH_PLATFORM_SCHEMA.extend( ) -def setup_platform( +async def async_setup_platform( hass: HomeAssistant, config: ConfigType, add_entities_callback: AddEntitiesCallback, discovery_info: DiscoveryInfoType | None = None, ) -> None: - """Set up S20 switches.""" + """Set up the integration from configuration.yaml.""" + for switch in config.get(CONF_SWITCHES, []): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=switch, + ) - switch_data = {} - switches = [] - switch_conf = config.get(CONF_SWITCHES, [config]) - - if config.get(CONF_DISCOVERY): - _LOGGER.debug("Discovering S20 switches") - switch_data.update(discover()) - - for switch in switch_conf: - switch_data[switch.get(CONF_HOST)] = switch - - for host, data in switch_data.items(): - try: - switches.append( - S20Switch(data.get(CONF_NAME), S20(host, mac=data.get(CONF_MAC))) + if ( + result.get("type") is FlowResultType.ABORT + and result.get("reason") != "already_configured" + ): + ir.async_create_issue( + hass, + DOMAIN, + f"yaml_deprecation_import_issue_{switch.get('host')}_{(switch.get('mac') or 'unknown_mac').replace(':', '').lower()}", + breaks_in_ha_version="2026.9.0", + is_fixable=False, + is_persistent=False, + issue_domain=DOMAIN, + severity=ir.IssueSeverity.WARNING, + translation_key="yaml_deprecation_import_issue", + translation_placeholders={ + "reason": str(result.get("reason")), + "host": switch.get("host"), + "mac": switch.get("mac", ""), + }, ) - _LOGGER.debug("Initialized S20 at %s", host) - except S20Exception: - _LOGGER.error("S20 at %s couldn't be initialized", host) + continue - add_entities_callback(switches) + ir.async_create_issue( + hass, + DOMAIN, + f"yaml_deprecation_{switch.get('host')}_{(switch.get('mac') or 'unknown_mac').replace(':', '').lower()}", + breaks_in_ha_version="2026.9.0", + is_fixable=False, + is_persistent=False, + severity=ir.IssueSeverity.WARNING, + translation_key="yaml_deprecation", + translation_placeholders={ + "host": switch.get("host"), + "mac": switch.get("mac") or "Unknown MAC", + }, + ) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: S20ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up orvibo from a config entry.""" + async_add_entities( + [ + S20Switch( + entry.title, + entry.data[CONF_HOST], + entry.data[CONF_MAC], + entry.runtime_data, + ) + ] + ) class S20Switch(SwitchEntity): """Representation of an S20 switch.""" - def __init__(self, name, s20): + _attr_has_entity_name = True + + def __init__(self, name: str, host: str, mac: str, s20: S20) -> None: """Initialize the S20 device.""" - self._attr_name = name - self._s20 = s20 self._attr_is_on = False - self._exc = S20Exception - - def update(self) -> None: - """Update device state.""" - try: - self._attr_is_on = self._s20.on - except self._exc: - _LOGGER.exception("Error while fetching S20 state") + self._host = host + self._mac = mac + self._s20 = s20 + self._attr_unique_id = self._mac + self._name = name + self._attr_name = None + self._attr_device_info = DeviceInfo( + identifiers={ + # MAC addresses are used as unique identifiers within this domain + (DOMAIN, self._attr_unique_id) + }, + name=name, + manufacturer="Orvibo", + model="S20", + connections={(CONNECTION_NETWORK_MAC, self._mac)}, + ) def turn_on(self, **kwargs: Any) -> None: """Turn the device on.""" try: self._s20.on = True - except self._exc: - _LOGGER.exception("Error while turning on S20") + except S20Exception as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="turn_on_error", + translation_placeholders={"name": self._name}, + ) from err def turn_off(self, **kwargs: Any) -> None: """Turn the device off.""" try: self._s20.on = False - except self._exc: - _LOGGER.exception("Error while turning off S20") + except S20Exception as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="turn_off_error", + translation_placeholders={"name": self._name}, + ) from err + + def update(self) -> None: + """Update device state.""" + try: + self._attr_is_on = self._s20.on + + # If the device was previously offline, let the user know it's back! + if not self._attr_available: + _LOGGER.info("Orvibo switch %s reconnected", self._name) + self._attr_available = True + + except S20Exception as err: + # Only log the error if this is the FIRST time it failed + if self._attr_available: + _LOGGER.info( + "Error communicating with Orvibo switch %s: %s", self._name, err + ) + self._attr_available = False diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 1086fad04be7..cbb5542d493c 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -515,6 +515,7 @@ FLOWS = { "openweathermap", "opower", "oralb", + "orvibo", "osoenergy", "otbr", "otp", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index e3890c518774..0bdb5625a1fe 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -5002,7 +5002,7 @@ "orvibo": { "name": "Orvibo", "integration_type": "hub", - "config_flow": false, + "config_flow": true, "iot_class": "local_push" }, "osoenergy": { diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 5898a5198c90..257ed8a7cac3 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1499,6 +1499,9 @@ opower==0.17.0 # homeassistant.components.oralb oralb-ble==1.0.2 +# homeassistant.components.orvibo +orvibo==1.1.2 + # homeassistant.components.ourgroceries ourgroceries==1.5.4 diff --git a/tests/components/orvibo/__init__.py b/tests/components/orvibo/__init__.py new file mode 100644 index 000000000000..d069874c9098 --- /dev/null +++ b/tests/components/orvibo/__init__.py @@ -0,0 +1 @@ +"""Tests for the Orvibo integration.""" diff --git a/tests/components/orvibo/conftest.py b/tests/components/orvibo/conftest.py new file mode 100644 index 000000000000..af20da8030a4 --- /dev/null +++ b/tests/components/orvibo/conftest.py @@ -0,0 +1,54 @@ +"""Fixtures for testing the Orvibo integration (core version).""" + +from unittest.mock import patch + +# The orvibo library executes a global UDP socket bind on import. +# We force the import here inside a patch context manager to prevent parallel +# CI test workers from crashing with 'OSError: [Errno 98] Address already in use'. +with patch("socket.socket.bind"): + import orvibo.s20 # noqa: F401 + +import pytest + +from homeassistant.components.orvibo.const import DOMAIN +from homeassistant.const import CONF_HOST, CONF_MAC + +from tests.common import MockConfigEntry + + +@pytest.fixture +def mock_s20(): + """Mock the Orvibo S20 class.""" + with patch("homeassistant.components.orvibo.config_flow.S20") as mock_class: + yield mock_class + + +@pytest.fixture +def mock_discover(): + """Mock Orvibo S20 discovery returning multiple devices.""" + with patch("homeassistant.components.orvibo.config_flow.discover") as mock_func: + mock_func.return_value = { + "192.168.1.100": {"mac": b"\xac\xcf\x23\x12\x34\x56"}, + "192.168.1.101": {"mac": b"\xac\xcf\x23\x78\x9a\xbc"}, + } + yield mock_func + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Return a mock config entry for an Orvibo S20 switch.""" + return MockConfigEntry( + domain=DOMAIN, + title="Orvibo (192.168.1.10)", + data={CONF_HOST: "192.168.1.10", CONF_MAC: "aa:bb:cc:dd:ee:ff"}, + unique_id="aa:bb:cc:dd:ee:ff", + ) + + +@pytest.fixture +def mock_setup_entry(): + """Override async_setup_entry so config flow tests don't try to setup the integration.""" + with patch( + "homeassistant.components.orvibo.async_setup_entry", return_value=True + ) as mock_setup: + yield mock_setup diff --git a/tests/components/orvibo/test_config_flow.py b/tests/components/orvibo/test_config_flow.py new file mode 100644 index 000000000000..efc42528c4ee --- /dev/null +++ b/tests/components/orvibo/test_config_flow.py @@ -0,0 +1,352 @@ +"""Tests for the Orvibo config flow in Home Assistant core.""" + +import asyncio +from typing import Any +from unittest.mock import patch + +from orvibo.s20 import S20Exception +import pytest +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.components.orvibo.const import CONF_SWITCH_LIST, DEFAULT_NAME, DOMAIN +from homeassistant.const import CONF_HOST, CONF_MAC +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from tests.common import MockConfigEntry + + +async def test_user_menu_display(hass: HomeAssistant) -> None: + """Initial step displays the user menu correctly.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "user" + assert set(result["menu_options"]) == {"start_discovery", "edit"} + + +@pytest.mark.parametrize( + ("user_input", "expected_mac", "mock_mac_bytes"), + [ + ( + {CONF_HOST: "192.168.1.2", CONF_MAC: "ac:cf:23:12:34:56"}, + "ac:cf:23:12:34:56", + None, + ), + ({CONF_HOST: "192.168.1.2"}, "aa:bb:cc:dd:ee:ff", b"\xaa\xbb\xcc\xdd\xee\xff"), + ], +) +async def test_edit_flow_success( + hass: HomeAssistant, + mock_discover, + mock_setup_entry, + mock_s20, + user_input: dict[str, Any], + expected_mac: str, + mock_mac_bytes: bytes | None, +) -> None: + """Test manual flow succeeds with provided MAC or discovered MAC.""" + mock_s20.return_value._mac = mock_mac_bytes + mock_discover.return_value = {"192.168.1.2": {"mac": b"\xaa\xbb\xcc\xdd\xee\xff"}} + + 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": "edit"} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == f"{DEFAULT_NAME} (192.168.1.2)" + assert result["data"][CONF_HOST] == "192.168.1.2" + assert result["data"][CONF_MAC] == expected_mac + assert result["result"].unique_id == expected_mac + + +@pytest.mark.parametrize( + ("user_input", "expected_error", "mock_exception", "mock_mac_bytes"), + [ + ( + {CONF_HOST: "192.168.1.2", CONF_MAC: "not_a_mac"}, + "invalid_mac", + None, + b"dummy", + ), + ({CONF_HOST: "192.168.1.99"}, "cannot_discover", None, None), + ( + {CONF_HOST: "192.168.1.3", CONF_MAC: "ac:cf:23:12:34:56"}, + "cannot_connect", + S20Exception("Connection failed"), + b"dummy", + ), + ], +) +async def test_edit_flow_errors( + hass: HomeAssistant, + mock_s20, + mock_discover, + mock_setup_entry, + user_input: dict[str, Any], + expected_error: str, + mock_exception: Exception | None, + mock_mac_bytes: bytes | None, +) -> None: + """Test various errors in the manual (edit) step and recover.""" + mock_discover.return_value = {} + mock_s20.side_effect = mock_exception + mock_s20.return_value._mac = mock_mac_bytes + + 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": "edit"} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input + ) + + assert result["type"] == FlowResultType.FORM + assert result["errors"]["base"] == expected_error + + mock_s20.side_effect = None + mock_s20.return_value._mac = b"\xac\xcf\x23\x12\x34\x56" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_HOST: "192.168.1.2", CONF_MAC: "ac:cf:23:12:34:56"}, + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == f"{DEFAULT_NAME} (192.168.1.2)" + assert result["data"][CONF_HOST] == "192.168.1.2" + assert result["data"][CONF_MAC] == "ac:cf:23:12:34:56" + + +async def test_discovery_success( + hass: HomeAssistant, mock_discover, mock_setup_entry +) -> None: + """Verify discovery finds devices and completes config entry creation.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == FlowResultType.MENU + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "start_discovery"} + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_discovery" + assert result["progress_action"] == "start_discovery" + + await hass.async_block_till_done() + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "choose_switch" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_SWITCH_LIST: "192.168.1.100"} + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == f"{DEFAULT_NAME} (192.168.1.100)" + assert result["data"][CONF_HOST] == "192.168.1.100" + assert result["data"][CONF_MAC] == "ac:cf:23:12:34:56" + assert result["result"].unique_id == "ac:cf:23:12:34:56" + + +async def test_discovery_no_devices( + hass: HomeAssistant, mock_discover, mock_s20, mock_setup_entry +) -> None: + """Discovery with no found devices should go to discovery_failed and recover via edit.""" + mock_discover.return_value = {} + + 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": "start_discovery"} + ) + await hass.async_block_till_done() + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "discovery_failed" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "edit"} + ) + + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "edit" + + mock_s20.return_value._mac = b"\xaa\xbb\xcc\xdd\xee\xff" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_HOST: "192.168.1.10", CONF_MAC: "aa:bb:cc:dd:ee:ff"}, + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == f"{DEFAULT_NAME} (192.168.1.10)" + assert result["data"][CONF_HOST] == "192.168.1.10" + assert result["data"][CONF_MAC] == "aa:bb:cc:dd:ee:ff" + + +@pytest.mark.parametrize( + ("import_data", "expected_mac", "mock_mac_bytes"), + [ + ( + {CONF_HOST: "192.168.1.5", CONF_MAC: "ac:cf:23:12:34:56"}, + "ac:cf:23:12:34:56", + None, + ), + ({CONF_HOST: "192.168.1.5"}, "11:22:33:44:55:66", b"\x11\x22\x33\x44\x55\x66"), + ], +) +async def test_import_flow_success( + hass: HomeAssistant, + mock_discover, + mock_setup_entry, + mock_s20, + import_data: dict[str, Any], + expected_mac: str, + mock_mac_bytes: bytes | None, +) -> None: + """Test importing configuration.yaml entry succeeds with provided or discovered MAC.""" + mock_s20.return_value._mac = mock_mac_bytes + mock_discover.return_value = {"192.168.1.5": {"mac": b"\x11\x22\x33\x44\x55\x66"}} + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=import_data + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == "192.168.1.5" + assert result["data"][CONF_MAC] == expected_mac + + +@pytest.mark.parametrize( + ("import_data", "expected_reason", "mock_exception", "mock_mac_bytes"), + [ + ({CONF_HOST: "192.168.1.5"}, "cannot_discover", None, None), + ( + {CONF_HOST: "192.168.1.5", CONF_MAC: "ac:cf:23:12:34:56"}, + "cannot_connect", + S20Exception("Connection failed"), + b"dummy", + ), + ], +) +async def test_import_flow_errors( + hass: HomeAssistant, + mock_s20, + mock_discover, + import_data: dict[str, Any], + expected_reason: str, + mock_exception: Exception | None, + mock_mac_bytes: bytes | None, +) -> None: + """Test various abort errors in the import flow.""" + mock_discover.return_value = {} + mock_s20.side_effect = mock_exception + mock_s20.return_value._mac = mock_mac_bytes + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=import_data + ) + + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == expected_reason + + +async def test_discover_skips_existing_and_invalid_mac( + hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_discover +) -> None: + """Test discovery ignores devices already configured and devices without MACs.""" + mock_config_entry.add_to_hass(hass) + + mock_discover.return_value = { + "192.168.1.10": {"mac": b"\xaa\xbb\xcc\xdd\xee\xff"}, + "192.168.1.11": {}, + "192.168.1.12": {"mac": b"\x11\x22\x33\x44\x55\x66"}, + } + + 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": "start_discovery"} + ) + + await hass.async_block_till_done() + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "choose_switch" + + schema = result["data_schema"].schema + dropdown_options = schema[vol.Required(CONF_SWITCH_LIST)].container + + assert "192.168.1.12" in dropdown_options + assert "192.168.1.10" not in dropdown_options + assert "192.168.1.11" not in dropdown_options + + +async def test_start_discovery_shows_progress(hass: HomeAssistant) -> None: + """Test polling the flow while discovery is still in progress.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + async def delayed_executor_job(*args, **kwargs) -> dict[str, Any]: + await asyncio.sleep(0.1) + return {} + + with patch.object(hass, "async_add_executor_job", side_effect=delayed_executor_job): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {"next_step_id": "start_discovery"} + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["progress_action"] == "start_discovery" + + await hass.async_block_till_done() + + +async def test_discovery_flow_task_exception( + hass: HomeAssistant, mock_discover +) -> None: + """Test the discovery process when the background task raises an error.""" + mock_discover.side_effect = S20Exception("Network timeout") + + 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": "start_discovery"} + ) + + await hass.async_block_till_done() + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "discovery_failed"