mirror of
https://github.com/home-assistant/core.git
synced 2025-12-26 14:08:21 +00:00
New configuration flow for ZHA integration (#35161)
* Start gateway using new zigpy init. Update config entry data import. Use new zigpy startup. Fix config entry import without zha config section. Auto form Zigbee network. * Migrate config entry. * New ZHA config entry flow. Use lightweight probe() method for ZHA config entry validation when available. Failback to old behavior of setting up Zigpy app if radio lib does not provide probing. * Clean ZHA_GW_RADIO * Don't import ZHA device settings. * Update config flow tests. * Filter out empty manufacturer. * Replace port path with an by-id device name. * Rebase cleanup * Use correct mock. * Make lint happy again * Update tests/components/zha/test_config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update tests/components/zha/test_config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update tests/components/zha/test_config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Use executor pool for IO * Address comments. Use AsyncMock from tests. * Use core interface to test config flow. * Use core interface to test config_flow. * Address comments. Use core interface. * Update ZHA dependencies. * Schema guard * Use async_update_entry for migration. * Don't allow schema extra keys. Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Alexei Chetroi
GitHub
@@ -1,125 +1,237 @@
|
||||
"""Tests for ZHA config flow."""
|
||||
from unittest import mock
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
import serial.tools.list_ports
|
||||
import zigpy.config
|
||||
|
||||
from homeassistant import setup
|
||||
from homeassistant.components.zha import config_flow
|
||||
from homeassistant.components.zha.core.const import CONTROLLER, DOMAIN, ZHA_GW_RADIO
|
||||
import homeassistant.components.zha.core.registries
|
||||
from homeassistant.components.zha.core.const import CONF_RADIO_TYPE, CONTROLLER, DOMAIN
|
||||
from homeassistant.components.zha.core.registries import RADIO_TYPES
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import CONF_SOURCE
|
||||
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
|
||||
|
||||
import tests.async_mock
|
||||
from tests.async_mock import AsyncMock, MagicMock, patch, sentinel
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_user_flow(hass):
|
||||
"""Test that config flow works."""
|
||||
flow = config_flow.ZhaFlowHandler()
|
||||
flow.hass = hass
|
||||
def com_port():
|
||||
"""Mock of a serial port."""
|
||||
port = serial.tools.list_ports_common.ListPortInfo()
|
||||
port.serial_number = "1234"
|
||||
port.manufacturer = "Virtual serial port"
|
||||
port.device = "/dev/ttyUSB1234"
|
||||
port.description = "Some serial port"
|
||||
|
||||
with tests.async_mock.patch(
|
||||
"homeassistant.components.zha.config_flow.check_zigpy_connection",
|
||||
return_value=False,
|
||||
):
|
||||
result = await flow.async_step_user(
|
||||
user_input={"usb_path": "/dev/ttyUSB1", "radio_type": "ezsp"}
|
||||
)
|
||||
return port
|
||||
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
with tests.async_mock.patch(
|
||||
"homeassistant.components.zha.config_flow.check_zigpy_connection",
|
||||
return_value=True,
|
||||
):
|
||||
result = await flow.async_step_user(
|
||||
user_input={"usb_path": "/dev/ttyUSB1", "radio_type": "ezsp"}
|
||||
)
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
@patch(
|
||||
"homeassistant.components.zha.config_flow.detect_radios",
|
||||
return_value={CONF_RADIO_TYPE: "test_radio"},
|
||||
)
|
||||
async def test_user_flow(detect_mock, hass):
|
||||
"""Test user flow -- radio detected."""
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "/dev/ttyUSB1"
|
||||
assert result["data"] == {"usb_path": "/dev/ttyUSB1", "radio_type": "ezsp"}
|
||||
port = com_port()
|
||||
port_select = f"{port}, s/n: {port.serial_number} - {port.manufacturer}"
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={zigpy.config.CONF_DEVICE_PATH: port_select},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"].startswith(port.description)
|
||||
assert result["data"] == {CONF_RADIO_TYPE: "test_radio"}
|
||||
assert detect_mock.await_count == 1
|
||||
assert detect_mock.await_args[0][0] == port.device
|
||||
|
||||
|
||||
@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
|
||||
@patch(
|
||||
"homeassistant.components.zha.config_flow.detect_radios", return_value=None,
|
||||
)
|
||||
async def test_user_flow_not_detected(detect_mock, hass):
|
||||
"""Test user flow, radio not detected."""
|
||||
|
||||
port = com_port()
|
||||
port_select = f"{port}, s/n: {port.serial_number} - {port.manufacturer}"
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={zigpy.config.CONF_DEVICE_PATH: port_select},
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "pick_radio"
|
||||
assert detect_mock.await_count == 1
|
||||
assert detect_mock.await_args[0][0] == port.device
|
||||
|
||||
|
||||
async def test_user_flow_show_form(hass):
|
||||
"""Test user step form."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_USER},
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_user_flow_manual(hass):
|
||||
"""Test user flow manual entry."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={CONF_SOURCE: SOURCE_USER},
|
||||
data={zigpy.config.CONF_DEVICE_PATH: config_flow.CONF_MANUAL_PATH},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "pick_radio"
|
||||
|
||||
|
||||
async def test_pick_radio_flow(hass):
|
||||
"""Test radio picker."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: "pick_radio"}, data={CONF_RADIO_TYPE: "ezsp"}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "port_config"
|
||||
|
||||
|
||||
async def test_user_flow_existing_config_entry(hass):
|
||||
"""Test if config entry already exists."""
|
||||
MockConfigEntry(domain=DOMAIN, data={"usb_path": "/dev/ttyUSB1"}).add_to_hass(hass)
|
||||
flow = config_flow.ZhaFlowHandler()
|
||||
flow.hass = hass
|
||||
|
||||
result = await flow.async_step_user()
|
||||
|
||||
assert result["type"] == "abort"
|
||||
|
||||
|
||||
async def test_import_flow(hass):
|
||||
"""Test import from configuration.yaml ."""
|
||||
flow = config_flow.ZhaFlowHandler()
|
||||
flow.hass = hass
|
||||
|
||||
result = await flow.async_step_import(
|
||||
{"usb_path": "/dev/ttyUSB1", "radio_type": "xbee"}
|
||||
)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "/dev/ttyUSB1"
|
||||
assert result["data"] == {"usb_path": "/dev/ttyUSB1", "radio_type": "xbee"}
|
||||
|
||||
|
||||
async def test_import_flow_existing_config_entry(hass):
|
||||
"""Test import from configuration.yaml ."""
|
||||
MockConfigEntry(domain=DOMAIN, data={"usb_path": "/dev/ttyUSB1"}).add_to_hass(hass)
|
||||
flow = config_flow.ZhaFlowHandler()
|
||||
flow.hass = hass
|
||||
|
||||
result = await flow.async_step_import(
|
||||
{"usb_path": "/dev/ttyUSB1", "radio_type": "xbee"}
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
|
||||
|
||||
async def test_check_zigpy_connection():
|
||||
"""Test config flow validator."""
|
||||
async def test_probe_radios(hass):
|
||||
"""Test detect radios."""
|
||||
app_ctrl_cls = MagicMock()
|
||||
app_ctrl_cls.SCHEMA_DEVICE = zigpy.config.SCHEMA_DEVICE
|
||||
app_ctrl_cls.probe = AsyncMock(side_effect=(True, False))
|
||||
|
||||
mock_radio = tests.async_mock.MagicMock()
|
||||
mock_radio.connect = tests.async_mock.AsyncMock()
|
||||
radio_cls = tests.async_mock.MagicMock(return_value=mock_radio)
|
||||
with patch.dict(config_flow.RADIO_TYPES, {"ezsp": {CONTROLLER: app_ctrl_cls}}):
|
||||
res = await config_flow.detect_radios("/dev/null")
|
||||
assert app_ctrl_cls.probe.await_count == 1
|
||||
assert res[CONF_RADIO_TYPE] == "ezsp"
|
||||
assert zigpy.config.CONF_DEVICE in res
|
||||
assert (
|
||||
res[zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH] == "/dev/null"
|
||||
)
|
||||
|
||||
bad_radio = tests.async_mock.MagicMock()
|
||||
bad_radio.connect = tests.async_mock.AsyncMock(side_effect=Exception)
|
||||
bad_radio_cls = tests.async_mock.MagicMock(return_value=bad_radio)
|
||||
res = await config_flow.detect_radios("/dev/null")
|
||||
assert res is None
|
||||
|
||||
mock_ctrl = tests.async_mock.MagicMock()
|
||||
mock_ctrl.startup = tests.async_mock.AsyncMock()
|
||||
mock_ctrl.shutdown = tests.async_mock.AsyncMock()
|
||||
ctrl_cls = tests.async_mock.MagicMock(return_value=mock_ctrl)
|
||||
new_radios = {
|
||||
mock.sentinel.radio: {ZHA_GW_RADIO: radio_cls, CONTROLLER: ctrl_cls},
|
||||
mock.sentinel.bad_radio: {ZHA_GW_RADIO: bad_radio_cls, CONTROLLER: ctrl_cls},
|
||||
}
|
||||
|
||||
with mock.patch.dict(
|
||||
homeassistant.components.zha.core.registries.RADIO_TYPES, new_radios, clear=True
|
||||
async def test_user_port_config_fail(hass):
|
||||
"""Test port config flow."""
|
||||
app_ctrl_cls = MagicMock()
|
||||
app_ctrl_cls.SCHEMA_DEVICE = zigpy.config.SCHEMA_DEVICE
|
||||
app_ctrl_cls.probe = AsyncMock(return_value=False)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: "pick_radio"}, data={CONF_RADIO_TYPE: "ezsp"}
|
||||
)
|
||||
|
||||
with patch.dict(config_flow.RADIO_TYPES, {"ezsp": {CONTROLLER: app_ctrl_cls}}):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB33"},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "port_config"
|
||||
assert result["errors"]["base"] == "cannot_connect"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"radio_type, orig_ctrl_cls",
|
||||
((name, r[CONTROLLER]) for name, r in RADIO_TYPES.items()),
|
||||
)
|
||||
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
|
||||
async def test_user_port_config(hass, radio_type, orig_ctrl_cls):
|
||||
"""Test port config."""
|
||||
app_ctrl_cls = MagicMock()
|
||||
app_ctrl_cls.SCHEMA_DEVICE = orig_ctrl_cls.SCHEMA_DEVICE
|
||||
app_ctrl_cls.probe = AsyncMock(return_value=True)
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={CONF_SOURCE: "pick_radio"}, data={CONF_RADIO_TYPE: radio_type}
|
||||
)
|
||||
|
||||
with patch.dict(
|
||||
config_flow.RADIO_TYPES,
|
||||
{radio_type: {CONTROLLER: app_ctrl_cls, "radio_description": "radio"}},
|
||||
):
|
||||
assert not await config_flow.check_zigpy_connection(
|
||||
mock.sentinel.usb_path, mock.sentinel.unk_radio, mock.sentinel.zigbee_db
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB33"},
|
||||
)
|
||||
assert mock_radio.connect.call_count == 0
|
||||
assert bad_radio.connect.call_count == 0
|
||||
assert mock_ctrl.startup.call_count == 0
|
||||
assert mock_ctrl.shutdown.call_count == 0
|
||||
|
||||
# unsuccessful radio connect
|
||||
assert not await config_flow.check_zigpy_connection(
|
||||
mock.sentinel.usb_path, mock.sentinel.bad_radio, mock.sentinel.zigbee_db
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"].startswith("/dev/ttyUSB33")
|
||||
assert (
|
||||
result["data"][zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH]
|
||||
== "/dev/ttyUSB33"
|
||||
)
|
||||
assert mock_radio.connect.call_count == 0
|
||||
assert bad_radio.connect.call_count == 1
|
||||
assert mock_ctrl.startup.call_count == 0
|
||||
assert mock_ctrl.shutdown.call_count == 0
|
||||
assert result["data"][CONF_RADIO_TYPE] == radio_type
|
||||
|
||||
# successful radio connect
|
||||
assert await config_flow.check_zigpy_connection(
|
||||
mock.sentinel.usb_path, mock.sentinel.radio, mock.sentinel.zigbee_db
|
||||
)
|
||||
assert mock_radio.connect.call_count == 1
|
||||
assert bad_radio.connect.call_count == 1
|
||||
assert mock_ctrl.startup.call_count == 1
|
||||
assert mock_ctrl.shutdown.call_count == 1
|
||||
|
||||
def test_get_serial_by_id_no_dir():
|
||||
"""Test serial by id conversion if there's no /dev/serial/by-id."""
|
||||
p1 = patch("os.path.isdir", MagicMock(return_value=False))
|
||||
p2 = patch("os.scandir")
|
||||
with p1 as is_dir_mock, p2 as scan_mock:
|
||||
res = config_flow.get_serial_by_id(sentinel.path)
|
||||
assert res is sentinel.path
|
||||
assert is_dir_mock.call_count == 1
|
||||
assert scan_mock.call_count == 0
|
||||
|
||||
|
||||
def test_get_serial_by_id():
|
||||
"""Test serial by id conversion."""
|
||||
p1 = patch("os.path.isdir", MagicMock(return_value=True))
|
||||
p2 = patch("os.scandir")
|
||||
|
||||
def _realpath(path):
|
||||
if path is sentinel.matched_link:
|
||||
return sentinel.path
|
||||
return sentinel.serial_link_path
|
||||
|
||||
p3 = patch("os.path.realpath", side_effect=_realpath)
|
||||
with p1 as is_dir_mock, p2 as scan_mock, p3:
|
||||
res = config_flow.get_serial_by_id(sentinel.path)
|
||||
assert res is sentinel.path
|
||||
assert is_dir_mock.call_count == 1
|
||||
assert scan_mock.call_count == 1
|
||||
|
||||
entry1 = MagicMock(spec_set=os.DirEntry)
|
||||
entry1.is_symlink.return_value = True
|
||||
entry1.path = sentinel.some_path
|
||||
|
||||
entry2 = MagicMock(spec_set=os.DirEntry)
|
||||
entry2.is_symlink.return_value = False
|
||||
entry2.path = sentinel.other_path
|
||||
|
||||
entry3 = MagicMock(spec_set=os.DirEntry)
|
||||
entry3.is_symlink.return_value = True
|
||||
entry3.path = sentinel.matched_link
|
||||
|
||||
scan_mock.return_value = [entry1, entry2, entry3]
|
||||
res = config_flow.get_serial_by_id(sentinel.path)
|
||||
assert res is sentinel.matched_link
|
||||
assert is_dir_mock.call_count == 2
|
||||
assert scan_mock.call_count == 2
|
||||
|
||||
Reference in New Issue
Block a user