mirror of
https://github.com/home-assistant/core.git
synced 2025-07-14 16:57:10 +00:00
Add velbus USB dicovery (#62596)
This commit is contained in:
parent
c65a50bd2e
commit
d7f0ad29df
@ -8,6 +8,7 @@ from velbusaio.exceptions import VelbusConnectionFailed
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
|
from homeassistant.components import usb
|
||||||
from homeassistant.const import CONF_NAME, CONF_PORT
|
from homeassistant.const import CONF_NAME, CONF_PORT
|
||||||
from homeassistant.core import HomeAssistant, callback
|
from homeassistant.core import HomeAssistant, callback
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
@ -32,6 +33,8 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
"""Initialize the velbus config flow."""
|
"""Initialize the velbus config flow."""
|
||||||
self._errors: dict[str, str] = {}
|
self._errors: dict[str, str] = {}
|
||||||
|
self._device: str = ""
|
||||||
|
self._title: str = ""
|
||||||
|
|
||||||
def _create_device(self, name: str, prt: str) -> FlowResult:
|
def _create_device(self, name: str, prt: str) -> FlowResult:
|
||||||
"""Create an entry async."""
|
"""Create an entry async."""
|
||||||
@ -50,9 +53,7 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
def _prt_in_configuration_exists(self, prt: str) -> bool:
|
def _prt_in_configuration_exists(self, prt: str) -> bool:
|
||||||
"""Return True if port exists in configuration."""
|
"""Return True if port exists in configuration."""
|
||||||
if prt in velbus_entries(self.hass):
|
return prt in velbus_entries(self.hass)
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
async def async_step_user(
|
async def async_step_user(
|
||||||
self, user_input: dict[str, Any] | None = None
|
self, user_input: dict[str, Any] | None = None
|
||||||
@ -82,3 +83,37 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
),
|
),
|
||||||
errors=self._errors,
|
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({}),
|
||||||
|
)
|
||||||
|
@ -5,5 +5,24 @@
|
|||||||
"requirements": ["velbus-aio==2021.11.7"],
|
"requirements": ["velbus-aio==2021.11.7"],
|
||||||
"config_flow": true,
|
"config_flow": true,
|
||||||
"codeowners": ["@Cereal2nd", "@brefra"],
|
"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"
|
||||||
|
}
|
||||||
|
]
|
||||||
}
|
}
|
||||||
|
@ -11,6 +11,26 @@ USB = [
|
|||||||
"vid": "0572",
|
"vid": "0572",
|
||||||
"pid": "1340"
|
"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",
|
"domain": "zha",
|
||||||
"vid": "10C4",
|
"vid": "10C4",
|
||||||
|
@ -1,16 +1,41 @@
|
|||||||
"""Tests for the Velbus config flow."""
|
"""Tests for the Velbus config flow."""
|
||||||
from unittest.mock import AsyncMock, patch
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
import serial.tools.list_ports
|
||||||
from velbusaio.exceptions import VelbusConnectionFailed
|
from velbusaio.exceptions import VelbusConnectionFailed
|
||||||
|
|
||||||
from homeassistant import data_entry_flow
|
from homeassistant import data_entry_flow
|
||||||
|
from homeassistant.components import usb
|
||||||
from homeassistant.components.velbus import config_flow
|
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 homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
from .const import PORT_SERIAL, PORT_TCP
|
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)
|
@pytest.fixture(autouse=True)
|
||||||
def override_async_setup_entry() -> AsyncMock:
|
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"})
|
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["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||||
assert result["errors"] == {"port": "already_configured"}
|
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"
|
||||||
|
Loading…
x
Reference in New Issue
Block a user