diff --git a/homeassistant/components/velbus/config_flow.py b/homeassistant/components/velbus/config_flow.py index 3a057b482df..189dc029ded 100644 --- a/homeassistant/components/velbus/config_flow.py +++ b/homeassistant/components/velbus/config_flow.py @@ -8,6 +8,7 @@ from velbusaio.exceptions import VelbusConnectionFailed import voluptuous as vol from homeassistant import config_entries +from homeassistant.components import usb from homeassistant.const import CONF_NAME, CONF_PORT from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResult @@ -32,6 +33,8 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize the velbus config flow.""" self._errors: dict[str, str] = {} + self._device: str = "" + self._title: str = "" def _create_device(self, name: str, prt: str) -> FlowResult: """Create an entry async.""" @@ -50,9 +53,7 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): def _prt_in_configuration_exists(self, prt: str) -> bool: """Return True if port exists in configuration.""" - if prt in velbus_entries(self.hass): - return True - return False + return prt in velbus_entries(self.hass) async def async_step_user( self, user_input: dict[str, Any] | None = None @@ -82,3 +83,37 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): ), errors=self._errors, ) + + async def async_step_usb(self, discovery_info: usb.UsbServiceInfo) -> FlowResult: + """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 = await self.hass.async_add_executor_job( + usb.get_serial_by_id, discovery_info.device + ) + # check if this device is not already configured + if self._prt_in_configuration_exists(dev_path): + return self.async_abort(reason="already_configured") + # 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 + self._title = "Velbus USB" + # call the config step + self._set_confirm_only() + return await self.async_step_discovery_confirm() + + async def async_step_discovery_confirm( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle Discovery confirmation.""" + if user_input is not None: + return self._create_device(self._title, self._device) + + return self.async_show_form( + step_id="discovery_confirm", + description_placeholders={CONF_NAME: self._title}, + data_schema=vol.Schema({}), + ) diff --git a/homeassistant/components/velbus/manifest.json b/homeassistant/components/velbus/manifest.json index 63d74536378..f52ba0fd99d 100644 --- a/homeassistant/components/velbus/manifest.json +++ b/homeassistant/components/velbus/manifest.json @@ -5,5 +5,24 @@ "requirements": ["velbus-aio==2021.11.7"], "config_flow": true, "codeowners": ["@Cereal2nd", "@brefra"], - "iot_class": "local_push" + "dependencies": ["usb"], + "iot_class": "local_push", + "usb": [ + { + "vid": "10CF", + "pid": "0B1B" + }, + { + "vid": "10CF", + "pid": "0516" + }, + { + "vid": "10CF", + "pid": "0517" + }, + { + "vid": "10CF", + "pid": "0518" + } + ] } diff --git a/homeassistant/generated/usb.py b/homeassistant/generated/usb.py index d9d9da5aff0..1ba9b235f85 100644 --- a/homeassistant/generated/usb.py +++ b/homeassistant/generated/usb.py @@ -11,6 +11,26 @@ USB = [ "vid": "0572", "pid": "1340" }, + { + "domain": "velbus", + "vid": "10CF", + "pid": "0B1B" + }, + { + "domain": "velbus", + "vid": "10CF", + "pid": "0516" + }, + { + "domain": "velbus", + "vid": "10CF", + "pid": "0517" + }, + { + "domain": "velbus", + "vid": "10CF", + "pid": "0518" + }, { "domain": "zha", "vid": "10C4", diff --git a/tests/components/velbus/test_config_flow.py b/tests/components/velbus/test_config_flow.py index 01a40af1751..960eedcbd01 100644 --- a/tests/components/velbus/test_config_flow.py +++ b/tests/components/velbus/test_config_flow.py @@ -1,16 +1,41 @@ """Tests for the Velbus config flow.""" -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest +import serial.tools.list_ports from velbusaio.exceptions import VelbusConnectionFailed from homeassistant import data_entry_flow +from homeassistant.components import usb from homeassistant.components.velbus import config_flow -from homeassistant.const import CONF_NAME, CONF_PORT +from homeassistant.components.velbus.const import DOMAIN +from homeassistant.config_entries import SOURCE_USB +from homeassistant.const import CONF_NAME, CONF_PORT, CONF_SOURCE from homeassistant.core import HomeAssistant from .const import PORT_SERIAL, PORT_TCP +from tests.common import MockConfigEntry + +DISCOVERY_INFO = usb.UsbServiceInfo( + device=PORT_SERIAL, + pid="10CF", + vid="0B1B", + serial_number="1234", + description="Velbus VMB1USB", + manufacturer="Velleman", +) + + +def com_port(): + """Mock of a serial port.""" + port = serial.tools.list_ports_common.ListPortInfo(PORT_SERIAL) + port.serial_number = "1234" + port.manufacturer = "Virtual serial port" + port.device = PORT_SERIAL + port.description = "Some serial port" + return port + @pytest.fixture(autouse=True) def override_async_setup_entry() -> AsyncMock: @@ -85,3 +110,49 @@ async def test_abort_if_already_setup(hass: HomeAssistant): result = await flow.async_step_user({CONF_PORT: PORT_TCP, CONF_NAME: "velbus test"}) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["errors"] == {"port": "already_configured"} + + +@pytest.mark.usefixtures("controller") +@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) +async def test_flow_usb(hass: HomeAssistant): + """Test usb discovery flow.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_USB}, + data=DISCOVERY_INFO, + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "discovery_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY + + # test an already configured discovery + entry = MockConfigEntry( + domain=DOMAIN, + data={CONF_PORT: PORT_SERIAL}, + ) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_USB}, + data=DISCOVERY_INFO, + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT + assert result["reason"] == "already_configured" + + +@pytest.mark.usefixtures("controller_connection_failed") +@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) +async def test_flow_usb_failed(hass: HomeAssistant): + """Test usb discovery flow with a failed velbus test.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_USB}, + data=DISCOVERY_INFO, + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT + assert result["reason"] == "cannot_connect"