mirror of
https://github.com/home-assistant/core.git
synced 2026-09-24 07:25:52 -05:00
Add config flow support to Orvibo legacy integration (#155115)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Erik Montnemery <erik@montnemery.com> Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
peteS-UK
co-authored by
Copilot
Erik Montnemery
Joost Lekkerkerker
Copilot
Erik Montnemery
Joost Lekkerkerker
parent
bf60d57cc2
commit
51acdeb563
@@ -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)
|
||||
|
||||
@@ -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
|
||||
)
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Constants for the orvibo integration."""
|
||||
|
||||
DOMAIN = "orvibo"
|
||||
DEFAULT_NAME = "S20"
|
||||
CONF_SWITCH_LIST = "switches"
|
||||
@@ -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"],
|
||||
|
||||
@@ -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]
|
||||
@@ -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}"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Generated
+1
@@ -515,6 +515,7 @@ FLOWS = {
|
||||
"openweathermap",
|
||||
"opower",
|
||||
"oralb",
|
||||
"orvibo",
|
||||
"osoenergy",
|
||||
"otbr",
|
||||
"otp",
|
||||
|
||||
@@ -5002,7 +5002,7 @@
|
||||
"orvibo": {
|
||||
"name": "Orvibo",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"osoenergy": {
|
||||
|
||||
Generated
+3
@@ -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
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for the Orvibo integration."""
|
||||
@@ -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
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user