Fix velbus matching ignored entries in config flow ()

* Fix bug #fix78826

* start using async_abort_entries_match

* fix/rewrite tests
pull/79095/head
Maikel Punie 2022-09-23 23:11:06 +02:00 committed by Paulus Schoutsen
parent 3b8f08270e
commit eb80062b26
2 changed files with 67 additions and 67 deletions
homeassistant/components/velbus
tests/components/velbus

View File

@ -10,21 +10,12 @@ import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.components import usb 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.data_entry_flow import FlowResult from homeassistant.data_entry_flow import FlowResult
from homeassistant.util import slugify from homeassistant.util import slugify
from .const import DOMAIN from .const import DOMAIN
@callback
def velbus_entries(hass: HomeAssistant) -> set[str]:
"""Return connections for Velbus domain."""
return {
entry.data[CONF_PORT] for entry in hass.config_entries.async_entries(DOMAIN)
}
class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow.""" """Handle a config flow."""
@ -51,10 +42,6 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return False return False
return True return True
def _prt_in_configuration_exists(self, prt: str) -> bool:
"""Return True if port exists in configuration."""
return prt in velbus_entries(self.hass)
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
) -> FlowResult: ) -> FlowResult:
@ -63,11 +50,9 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None: if user_input is not None:
name = slugify(user_input[CONF_NAME]) name = slugify(user_input[CONF_NAME])
prt = user_input[CONF_PORT] prt = user_input[CONF_PORT]
if not self._prt_in_configuration_exists(prt): self._async_abort_entries_match({CONF_PORT: prt})
if await self._test_connection(prt): if await self._test_connection(prt):
return self._create_device(name, prt) return self._create_device(name, prt)
else:
self._errors[CONF_PORT] = "already_configured"
else: else:
user_input = {} user_input = {}
user_input[CONF_NAME] = "" user_input[CONF_NAME] = ""
@ -93,8 +78,7 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
usb.get_serial_by_id, discovery_info.device usb.get_serial_by_id, discovery_info.device
) )
# check if this device is not already configured # check if this device is not already configured
if self._prt_in_configuration_exists(dev_path): self._async_abort_entries_match({CONF_PORT: dev_path})
return self.async_abort(reason="already_configured")
# check if we can make a valid velbus connection # check if we can make a valid velbus connection
if not await self._test_connection(dev_path): if not await self._test_connection(dev_path):
return self.async_abort(reason="cannot_connect") return self.async_abort(reason="cannot_connect")

View File

@ -7,9 +7,8 @@ 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 import usb
from homeassistant.components.velbus import config_flow
from homeassistant.components.velbus.const import DOMAIN from homeassistant.components.velbus.const import DOMAIN
from homeassistant.config_entries import SOURCE_USB from homeassistant.config_entries import SOURCE_USB, SOURCE_USER
from homeassistant.const import CONF_NAME, CONF_PORT, CONF_SOURCE from homeassistant.const import CONF_NAME, CONF_PORT, CONF_SOURCE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -53,63 +52,76 @@ def mock_controller_connection_failed():
yield yield
def init_config_flow(hass: HomeAssistant):
"""Init a configuration flow."""
flow = config_flow.VelbusConfigFlow()
flow.hass = hass
return flow
@pytest.mark.usefixtures("controller") @pytest.mark.usefixtures("controller")
async def test_user(hass: HomeAssistant): async def test_user(hass: HomeAssistant):
"""Test user config.""" """Test user config."""
flow = init_config_flow(hass) # simple user form
result = await hass.config_entries.flow.async_init(
result = await flow.async_step_user() DOMAIN, context={"source": SOURCE_USER}
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
result = await flow.async_step_user(
{CONF_NAME: "Velbus Test Serial", CONF_PORT: PORT_SERIAL}
) )
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY assert result
assert result["title"] == "velbus_test_serial" assert result.get("flow_id")
assert result["data"][CONF_PORT] == PORT_SERIAL assert result.get("type") == data_entry_flow.FlowResultType.FORM
assert result.get("step_id") == "user"
result = await flow.async_step_user( # try with a serial port
{CONF_NAME: "Velbus Test TCP", CONF_PORT: PORT_TCP} 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["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY assert result
assert result["title"] == "velbus_test_tcp" assert result.get("type") == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"][CONF_PORT] == PORT_TCP assert result.get("title") == "velbus_test_serial"
data = result.get("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") == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result.get("title") == "velbus_test_tcp"
data = result.get("data")
assert data[CONF_PORT] == PORT_TCP
@pytest.mark.usefixtures("controller_connection_failed") @pytest.mark.usefixtures("controller_connection_failed")
async def test_user_fail(hass: HomeAssistant): async def test_user_fail(hass: HomeAssistant):
"""Test user config.""" """Test user config."""
flow = init_config_flow(hass) result = await hass.config_entries.flow.async_init(
DOMAIN,
result = await flow.async_step_user( context={"source": SOURCE_USER},
{CONF_NAME: "Velbus Test Serial", CONF_PORT: PORT_SERIAL} data={CONF_NAME: "Velbus Test Serial", CONF_PORT: PORT_SERIAL},
) )
assert result["type"] == data_entry_flow.FlowResultType.FORM assert result
assert result["errors"] == {CONF_PORT: "cannot_connect"} assert result.get("type") == data_entry_flow.FlowResultType.FORM
assert result.get("errors") == {CONF_PORT: "cannot_connect"}
result = await flow.async_step_user( result = await hass.config_entries.flow.async_init(
{CONF_NAME: "Velbus Test TCP", CONF_PORT: PORT_TCP} DOMAIN,
context={"source": SOURCE_USER},
data={CONF_NAME: "Velbus Test TCP", CONF_PORT: PORT_TCP},
) )
assert result["type"] == data_entry_flow.FlowResultType.FORM assert result
assert result["errors"] == {CONF_PORT: "cannot_connect"} assert result.get("type") == data_entry_flow.FlowResultType.FORM
assert result.get("errors") == {CONF_PORT: "cannot_connect"}
@pytest.mark.usefixtures("config_entry") @pytest.mark.usefixtures("config_entry")
async def test_abort_if_already_setup(hass: HomeAssistant): async def test_abort_if_already_setup(hass: HomeAssistant):
"""Test we abort if Velbus is already setup.""" """Test we abort if Velbus is already setup."""
flow = init_config_flow(hass) result = await hass.config_entries.flow.async_init(
DOMAIN,
result = await flow.async_step_user({CONF_PORT: PORT_TCP, CONF_NAME: "velbus test"}) context={"source": SOURCE_USER},
assert result["type"] == data_entry_flow.FlowResultType.FORM data={CONF_PORT: PORT_TCP, CONF_NAME: "velbus test"},
assert result["errors"] == {"port": "already_configured"} )
assert result
assert result.get("type") == data_entry_flow.FlowResultType.ABORT
assert result.get("reason") == "already_configured"
@pytest.mark.usefixtures("controller") @pytest.mark.usefixtures("controller")
@ -121,14 +133,16 @@ async def test_flow_usb(hass: HomeAssistant):
context={CONF_SOURCE: SOURCE_USB}, context={CONF_SOURCE: SOURCE_USB},
data=DISCOVERY_INFO, data=DISCOVERY_INFO,
) )
assert result["type"] == data_entry_flow.FlowResultType.FORM assert result
assert result["step_id"] == "discovery_confirm" assert result.get("type") == data_entry_flow.FlowResultType.FORM
assert result.get("step_id") == "discovery_confirm"
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={}, user_input={},
) )
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY assert result
assert result.get("type") == data_entry_flow.FlowResultType.CREATE_ENTRY
# test an already configured discovery # test an already configured discovery
entry = MockConfigEntry( entry = MockConfigEntry(
@ -141,8 +155,9 @@ async def test_flow_usb(hass: HomeAssistant):
context={CONF_SOURCE: SOURCE_USB}, context={CONF_SOURCE: SOURCE_USB},
data=DISCOVERY_INFO, data=DISCOVERY_INFO,
) )
assert result["type"] == data_entry_flow.FlowResultType.ABORT assert result
assert result["reason"] == "already_configured" assert result.get("type") == data_entry_flow.FlowResultType.ABORT
assert result.get("reason") == "already_configured"
@pytest.mark.usefixtures("controller_connection_failed") @pytest.mark.usefixtures("controller_connection_failed")
@ -154,5 +169,6 @@ async def test_flow_usb_failed(hass: HomeAssistant):
context={CONF_SOURCE: SOURCE_USB}, context={CONF_SOURCE: SOURCE_USB},
data=DISCOVERY_INFO, data=DISCOVERY_INFO,
) )
assert result["type"] == data_entry_flow.FlowResultType.ABORT assert result
assert result["reason"] == "cannot_connect" assert result.get("type") == data_entry_flow.FlowResultType.ABORT
assert result.get("reason") == "cannot_connect"