jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-07 21:37:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rework the velbus configflow to make it more user-friendly (#135609)

Browse Source
This commit is contained in:
Maikel Punie 2025-02-25 14:02:10 +01:00 committed by GitHub
parent befed910da
commit d7301c62e2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 297 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
homeassistant/components/velbus/__init__.py
View File

@ -135,15 +135,39 @@ async def async_migrate_entry(
hass: HomeAssistant, config_entry: VelbusConfigEntry
) -> bool:
"""Migrate old entry."""
_LOGGER.debug("Migrating from version %s", config_entry.version)
cache_path = hass.config.path(STORAGE_DIR, f"velbuscache-{config_entry.entry_id}/")
if config_entry.version == 1:
# This is the config entry migration for adding the new program selection
_LOGGER.error(
"Migrating from version %s.%s", config_entry.version, config_entry.minor_version
)
# This is the config entry migration for adding the new program selection
# migrate from 1.x to 2.1
if config_entry.version < 2:
# clean the velbusCache
cache_path = hass.config.path(
STORAGE_DIR, f"velbuscache-{config_entry.entry_id}/"
)
if os.path.isdir(cache_path):
await hass.async_add_executor_job(shutil.rmtree, cache_path)
# set the new version
hass.config_entries.async_update_entry(config_entry, version=2)
_LOGGER.debug("Migration to version %s successful", config_entry.version)
# This is the config entry migration for swapping the usb unique id to the serial number
# migrate from 2.1 to 2.2
if (
config_entry.version < 3
and config_entry.minor_version == 1
and config_entry.unique_id is not None
):
# not all velbus devices have a unique id, so handle this correctly
parts = config_entry.unique_id.split("_")
# old one should have 4 item
if len(parts) == 4:
hass.config_entries.async_update_entry(config_entry, unique_id=parts[1])
# update the config entry
hass.config_entries.async_update_entry(config_entry, version=2, minor_version=2)
_LOGGER.error(
"Migration to version %s.%s successful",
config_entry.version,
config_entry.minor_version,
)
return True

110
homeassistant/components/velbus/config_flow.py
View File

@ -4,22 +4,23 @@ from __future__ import annotations
from typing import Any
import serial.tools.list_ports
import velbusaio.controller
from velbusaio.exceptions import VelbusConnectionFailed
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_NAME, CONF_PORT
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_PORT
from homeassistant.helpers.service_info.usb import UsbServiceInfo
from homeassistant.util import slugify
from .const import DOMAIN
from .const import CONF_TLS, DOMAIN
class VelbusConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 2
MINOR_VERSION = 2
def __init__(self) -> None:
"""Initialize the velbus config flow."""
@ -27,14 +28,16 @@ class VelbusConfigFlow(ConfigFlow, domain=DOMAIN):
self._device: str = ""
self._title: str = ""
def _create_device(self, name: str, prt: str) -> ConfigFlowResult:
def _create_device(self) -> ConfigFlowResult:
"""Create an entry async."""
return self.async_create_entry(title=name, data={CONF_PORT: prt})
return self.async_create_entry(
title=self._title, data={CONF_PORT: self._device}
)
async def _test_connection(self, prt: str) -> bool:
async def _test_connection(self) -> bool:
"""Try to connect to the velbus with the port specified."""
try:
controller = velbusaio.controller.Velbus(prt)
controller = velbusaio.controller.Velbus(self._device)
await controller.connect()
await controller.stop()
except VelbusConnectionFailed:
@ -46,43 +49,86 @@ class VelbusConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Step when user initializes a integration."""
self._errors = {}
return self.async_show_menu(
step_id="user", menu_options=["network", "usbselect"]
)
async def async_step_network(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle network step."""
if user_input is not None:
name = slugify(user_input[CONF_NAME])
prt = user_input[CONF_PORT]
self._async_abort_entries_match({CONF_PORT: prt})
if await self._test_connection(prt):
return self._create_device(name, prt)
self._title = "Velbus Network"
if user_input[CONF_TLS]:
self._device = "tls://"
else:
self._device = ""
if user_input[CONF_PASSWORD] != "":
self._device += f"{user_input[CONF_PASSWORD]}@"
self._device += f"{user_input[CONF_HOST]}:{user_input[CONF_PORT]}"
self._async_abort_entries_match({CONF_PORT: self._device})
if await self._test_connection():
return self._create_device()
else:
user_input = {
CONF_TLS: True,
CONF_PORT: 27015,
}
return self.async_show_form(
step_id="network",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_TLS): bool,
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT): int,
vol.Optional(CONF_PASSWORD): str,
}
),
suggested_values=user_input,
),
errors=self._errors,
)
async def async_step_usbselect(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle usb select step."""
ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports)
list_of_ports = [
f"{p}{', s/n: ' + p.serial_number if p.serial_number else ''}"
+ (f" - {p.manufacturer}" if p.manufacturer else "")
for p in ports
]
if user_input is not None:
self._title = "Velbus USB"
self._device = ports[list_of_ports.index(user_input[CONF_PORT])].device
self._async_abort_entries_match({CONF_PORT: self._device})
if await self._test_connection():
return self._create_device()
else:
user_input = {}
user_input[CONF_NAME] = ""
user_input[CONF_PORT] = ""
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_NAME, default=user_input[CONF_NAME]): str,
vol.Required(CONF_PORT, default=user_input[CONF_PORT]): str,
}
step_id="usbselect",
data_schema=self.add_suggested_values_to_schema(
vol.Schema({vol.Required(CONF_PORT): vol.In(list_of_ports)}),
suggested_values=user_input,
),
errors=self._errors,
)
async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResult:
"""Handle USB Discovery."""
await self.async_set_unique_id(
f"{discovery_info.vid}:{discovery_info.pid}_{discovery_info.serial_number}_{discovery_info.manufacturer}_{discovery_info.description}"
)
dev_path = discovery_info.device
# check if this device is not already configured
self._async_abort_entries_match({CONF_PORT: dev_path})
# check if we can make a valid velbus connection
if not await self._test_connection(dev_path):
return self.async_abort(reason="cannot_connect")
# store the data for the config step
self._device = dev_path
await self.async_set_unique_id(discovery_info.serial_number)
self._device = discovery_info.device
self._title = "Velbus USB"
self._async_abort_entries_match({CONF_PORT: self._device})
if not await self._test_connection():
return self.async_abort(reason="cannot_connect")
# call the config step
self._set_confirm_only()
return await self.async_step_discovery_confirm()
@ -92,7 +138,7 @@ class VelbusConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Handle Discovery confirmation."""
if user_input is not None:
return self._create_device(self._title, self._device)
return self._create_device()
return self.async_show_form(
step_id="discovery_confirm",

1
homeassistant/components/velbus/const.py
View File

@ -14,6 +14,7 @@ DOMAIN: Final = "velbus"
CONF_CONFIG_ENTRY: Final = "config_entry"
CONF_INTERFACE: Final = "interface"
CONF_MEMO_TEXT: Final = "memo_text"
CONF_TLS: Final = "tls"
SERVICE_SCAN: Final = "scan"
SERVICE_SYNC: Final = "sync_clock"

5
homeassistant/components/velbus/quality_scale.yaml
View File

@ -8,10 +8,7 @@ rules:
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow:
status: todo
comment: |
Dynamically build up the port parameter based on inputs provided by the user, do not fill-in a name parameter, build it up in the config flow
config-flow: done
dependency-transparency: done
docs-actions: done
docs-high-level-description: done

26
homeassistant/components/velbus/strings.json
View File

@ -7,6 +7,32 @@
"name": "The name for this Velbus connection",
"port": "Connection string"
}
},
"network": {
"title": "TCP/IP configuration",
"data": {
"tls": "Use TLS (secure connection)",
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"tls": "Enable this if you use a secure connection to your velbus interface, like a Signum.",
"host": "The IP address or hostname of the velbus interface.",
"port": "The port number of the velbus interface.",
"password": "The password of the velbus interface, this is only needed if the interface is password protected."
},
"description": "TCP/IP configuration, in case you use a Signum, velserv, velbus-tcp or any other velbus to TCP/IP interface."
},
"usbselect": {
"title": "USB configuration",
"data": {
"port": "[%key:common::config_flow::data::port%]"
},
"data_description": {
"port": "Select the serial port for your velbus USB interface."
},
"description": "Select the serial port for your velbus USB interface."
}
},
"error": {

2
tests/components/velbus/snapshots/test_diagnostics.ambr
View File

@ -10,7 +10,7 @@
'discovery_keys': dict({
}),
'domain': 'velbus',
'minor_version': 1,
'minor_version': 2,
'options': dict({
}),
'pref_disable_new_entities': False,

203
tests/components/velbus/test_config_flow.py
View File

@ -7,14 +7,14 @@ import pytest
import serial.tools.list_ports
from velbusaio.exceptions import VelbusConnectionFailed
from homeassistant.components.velbus.const import DOMAIN
from homeassistant.components.velbus.const import CONF_TLS, DOMAIN
from homeassistant.config_entries import SOURCE_USB, SOURCE_USER
from homeassistant.const import CONF_NAME, CONF_PORT, CONF_SOURCE
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.usb import UsbServiceInfo
from .const import PORT_SERIAL, PORT_TCP
from .const import PORT_SERIAL
from tests.common import MockConfigEntry
@ -27,6 +27,8 @@ DISCOVERY_INFO = UsbServiceInfo(
manufacturer="Velleman",
)
USB_DEV = "/dev/ttyACME100 - Some serial port, s/n: 1234 - Virtual serial port"
def com_port():
"""Mock of a serial port."""
@ -38,23 +40,15 @@ def com_port():
return port
@pytest.fixture(name="controller")
def mock_controller() -> Generator[MagicMock]:
"""Mock a successful velbus controller."""
with patch(
"homeassistant.components.velbus.config_flow.velbusaio.controller.Velbus",
autospec=True,
) as controller:
yield controller
@pytest.fixture(autouse=True)
def override_async_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.velbus.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
with (
patch(
"homeassistant.components.velbus.async_setup_entry", return_value=True
) as mock,
):
yield mock
@pytest.fixture(name="controller_connection_failed")
@ -65,73 +59,126 @@ def mock_controller_connection_failed():
@pytest.mark.usefixtures("controller")
async def test_user(hass: HomeAssistant) -> None:
"""Test user config."""
# simple user form
async def test_user_network_succes(hass: HomeAssistant) -> None:
"""Test user network config."""
# inttial menu show
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result
assert result.get("flow_id")
assert result.get("type") is FlowResultType.FORM
assert result.get("type") is FlowResultType.MENU
assert result.get("step_id") == "user"
# try with a serial port
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "Velbus Test Serial", CONF_PORT: PORT_SERIAL},
assert result.get("menu_options") == ["network", "usbselect"]
# select the network option
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{"next_step_id": "network"},
)
assert result.get("type") is FlowResultType.FORM
# fill in the network form
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{
CONF_TLS: False,
CONF_HOST: "velbus",
CONF_PORT: 6000,
CONF_PASSWORD: "",
},
)
assert result
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "velbus_test_serial"
assert result.get("title") == "Velbus Network"
data = result.get("data")
assert data
assert data[CONF_PORT] == "velbus:6000"
@pytest.mark.usefixtures("controller")
async def test_user_network_succes_tls(hass: HomeAssistant) -> None:
"""Test user network config."""
# inttial menu show
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result
assert result.get("flow_id")
assert result.get("type") is FlowResultType.MENU
assert result.get("step_id") == "user"
assert result.get("menu_options") == ["network", "usbselect"]
# select the network option
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{"next_step_id": "network"},
)
assert result["type"] is FlowResultType.FORM
# fill in the network form
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{
CONF_TLS: True,
CONF_HOST: "velbus",
CONF_PORT: 6000,
CONF_PASSWORD: "password",
},
)
assert result
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "Velbus Network"
data = result.get("data")
assert data
assert data[CONF_PORT] == "tls://password@velbus:6000"
@pytest.mark.usefixtures("controller")
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
async def test_user_usb_succes(hass: HomeAssistant) -> None:
"""Test user usb step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{"next_step_id": "usbselect"},
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PORT: USB_DEV,
},
)
assert result
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "Velbus USB"
data = result.get("data")
assert data
assert data[CONF_PORT] == PORT_SERIAL
# try with a ip:port combination
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "Velbus Test TCP", CONF_PORT: PORT_TCP},
)
assert result
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "velbus_test_tcp"
data = result.get("data")
assert data
assert data[CONF_PORT] == PORT_TCP
@pytest.mark.usefixtures("controller_connection_failed")
async def test_user_fail(hass: HomeAssistant) -> None:
"""Test user config."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "Velbus Test Serial", CONF_PORT: PORT_SERIAL},
)
assert result
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_PORT: "cannot_connect"}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "Velbus Test TCP", CONF_PORT: PORT_TCP},
)
assert result
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {CONF_PORT: "cannot_connect"}
@pytest.mark.usefixtures("config_entry")
async def test_abort_if_already_setup(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("controller")
async def test_network_abort_if_already_setup(hass: HomeAssistant) -> None:
"""Test we abort if Velbus is already setup."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_PORT: "127.0.0.1:3788"},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_PORT: PORT_TCP, CONF_NAME: "velbus test"},
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{"next_step_id": "network"},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_TLS: False,
CONF_HOST: "127.0.0.1",
CONF_PORT: 3788,
CONF_PASSWORD: "",
},
)
assert result
assert result.get("type") is FlowResultType.ABORT
@ -156,7 +203,7 @@ async def test_flow_usb(hass: HomeAssistant) -> None:
user_input={},
)
assert result
assert result["result"].unique_id == "0B1B:10CF_1234_Velleman_Velbus VMB1USB"
assert result["result"].unique_id == "1234"
assert result.get("type") is FlowResultType.CREATE_ENTRY
@ -167,13 +214,23 @@ async def test_flow_usb_if_already_setup(hass: HomeAssistant) -> None:
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_PORT: PORT_SERIAL},
unique_id="0B1B:10CF_1234_Velleman_Velbus VMB1USB",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USB},
data=DISCOVERY_INFO,
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result.get("flow_id"),
{"next_step_id": "usbselect"},
)
assert result
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PORT: USB_DEV,
},
)
assert result
assert result.get("type") is FlowResultType.ABORT

32
tests/components/velbus/test_init.py
View File

@ -16,6 +16,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import init_integration
from .const import PORT_TCP
from tests.common import MockConfigEntry
@ -107,16 +108,41 @@ async def test_migrate_config_entry(
"""Test successful migration of entry data."""
legacy_config = {CONF_NAME: "fake_name", CONF_PORT: "1.2.3.4:5678"}
entry = MockConfigEntry(domain=DOMAIN, unique_id="my own id", data=legacy_config)
entry.add_to_hass(hass)
assert dict(entry.data) == legacy_config
assert entry.version == 1
assert entry.minor_version == 1
entry.add_to_hass(hass)
# test in case we do not have a cache
with patch("os.path.isdir", return_value=True), patch("shutil.rmtree"):
await hass.config_entries.async_setup(entry.entry_id)
assert dict(entry.data) == legacy_config
assert entry.version == 2
assert entry.minor_version == 2
@pytest.mark.parametrize(
("unique_id", "expected"),
[("vid:pid_serial_manufacturer_decription", "serial"), (None, None)],
)
async def test_migrate_config_entry_unique_id(
hass: HomeAssistant,
controller: AsyncMock,
unique_id: str,
expected: str,
) -> None:
"""Test the migration of unique id."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_PORT: PORT_TCP, CONF_NAME: "velbus home"},
unique_id=unique_id,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.unique_id == expected
assert entry.version == 2
assert entry.minor_version == 2
async def test_api_call(