home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2026-08-03 20:24:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Clean up arcam_fmj config flow (#171161)

Browse Source 
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Josh Gustafson
2026-05-20 14:58:10 -06:00
committed by GitHub
parent a3b43fc19b
commit c188fdcc8b
3 changed files with 107 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/arcam_fmj/config_flow.py
+39 -18
View File
@@ -1,9 +1,11 @@
"""Config flow to configure the Arcam FMJ component."""
import socket
from typing import Any
from urllib.parse import urlparse
from arcam.fmj.client import Client, ConnectionFailed
from arcam.fmj import ConnectionFailed
from arcam.fmj.client import Client
from arcam.fmj.utils import get_uniqueid_from_host, get_uniqueid_from_udn
import voluptuous as vol
@@ -29,26 +31,19 @@ class ArcamFmjFlowHandler(ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(uuid)
self._abort_if_unique_id_configured({CONF_HOST: host, CONF_PORT: port})
async def _async_check_and_create(self, host: str, port: int) -> ConfigFlowResult:
async def _async_try_connect(self, host: str, port: int) -> None:
"""Verify the device is reachable."""
client = Client(host, port)
try:
await client.start()
except ConnectionFailed:
return self.async_abort(reason="cannot_connect")
finally:
await client.stop()
return self.async_create_entry(
title=f"{DEFAULT_NAME} ({host})",
data={CONF_HOST: host, CONF_PORT: port},
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a discovered device."""
errors: dict[str, str] = {}
if user_input is not None:
uuid = await get_uniqueid_from_host(
async_get_clientsession(self.hass), user_input[CONF_HOST]
@@ -58,18 +53,36 @@ class ArcamFmjFlowHandler(ConfigFlow, domain=DOMAIN):
user_input[CONF_HOST], user_input[CONF_PORT], uuid
)
return await self._async_check_and_create(
user_input[CONF_HOST], user_input[CONF_PORT]
)
try:
await self._async_try_connect(
user_input[CONF_HOST], user_input[CONF_PORT]
)
except socket.gaierror:
errors["base"] = "invalid_host"
except TimeoutError:
errors["base"] = "timeout_connect"
except ConnectionRefusedError:
errors["base"] = "connection_refused"
except ConnectionFailed, OSError:
errors["base"] = "cannot_connect"
else:
return self.async_create_entry(
title=f"{DEFAULT_NAME} ({user_input[CONF_HOST]})",
data={
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
},
)
fields = {
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
}
schema = vol.Schema(fields)
if user_input is not None:
schema = self.add_suggested_values_to_schema(schema, user_input)
return self.async_show_form(
step_id="user", data_schema=vol.Schema(fields), errors=errors
)
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
@@ -79,7 +92,10 @@ class ArcamFmjFlowHandler(ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = placeholders
if user_input is not None:
return await self._async_check_and_create(self.host, self.port)
return self.async_create_entry(
title=f"{DEFAULT_NAME} ({self.host})",
data={CONF_HOST: self.host, CONF_PORT: self.port},
)
return self.async_show_form(
step_id="confirm", description_placeholders=placeholders
@@ -97,6 +113,11 @@ class ArcamFmjFlowHandler(ConfigFlow, domain=DOMAIN):
await self._async_set_unique_id_and_update(host, port, uuid)
try:
await self._async_try_connect(host, port)
except ConnectionFailed, OSError:
return self.async_abort(reason="cannot_connect")
self.host = host
self.port = DEFAULT_PORT
self.port = port
return await self.async_step_confirm()
homeassistant/components/arcam_fmj/strings.json
+6
View File
@@ -5,6 +5,12 @@
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"connection_refused": "Host refused connection",
"invalid_host": "[%key:common::config_flow::error::invalid_host%]",
"timeout_connect": "[%key:common::config_flow::error::timeout_connect%]"
},
"flow_title": "{host}",
"step": {
"confirm": {
tests/components/arcam_fmj/test_config_flow.py
+62 -7
View File
@@ -2,6 +2,7 @@
from collections.abc import Generator
from dataclasses import replace
import socket
from unittest.mock import AsyncMock, MagicMock, patch
from arcam.fmj.client import ConnectionFailed
@@ -111,21 +112,29 @@ async def test_ssdp_abort(hass: HomeAssistant) -> None:
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
"connect_exception",
[
pytest.param(ConnectionFailed, id="connection_failed"),
pytest.param(ConnectionRefusedError, id="connection_refused"),
pytest.param(OSError, id="os_error"),
pytest.param(socket.gaierror, id="gaierror"),
pytest.param(TimeoutError, id="timeout"),
],
)
async def test_ssdp_unable_to_connect(
hass: HomeAssistant, dummy_client: MagicMock
hass: HomeAssistant,
dummy_client: MagicMock,
connect_exception: type[Exception],
) -> None:
"""Test a ssdp import flow."""
dummy_client.start.side_effect = AsyncMock(side_effect=ConnectionFailed)
"""Test a ssdp import flow aborts when the device is unreachable."""
dummy_client.start.side_effect = AsyncMock(side_effect=connect_exception)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP},
data=MOCK_DISCOVER,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@@ -232,3 +241,49 @@ async def test_user_wrong(
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == f"Arcam FMJ ({MOCK_HOST})"
assert result["result"].unique_id is None
@pytest.mark.parametrize(
("connect_exception", "expected_error"),
[
pytest.param(ConnectionFailed, "cannot_connect", id="connection_failed"),
pytest.param(
ConnectionRefusedError, "connection_refused", id="connection_refused"
),
pytest.param(OSError, "cannot_connect", id="os_error"),
pytest.param(socket.gaierror, "invalid_host", id="invalid_host"),
pytest.param(TimeoutError, "timeout_connect", id="timeout_connect"),
],
)
async def test_user_unable_to_connect(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
dummy_client: MagicMock,
connect_exception: type[Exception],
expected_error: str,
) -> None:
"""Test a manual user configuration flow where the device cannot be reached."""
dummy_client.start.side_effect = AsyncMock(side_effect=connect_exception)
aioclient_mock.get(MOCK_UPNP_LOCATION, status=404)
user_input = {
CONF_HOST: MOCK_HOST,
CONF_PORT: MOCK_PORT,
}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
data=user_input,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected_error}
dummy_client.start.side_effect = AsyncMock(return_value=None)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == f"Arcam FMJ ({MOCK_HOST})"
assert result["data"] == MOCK_CONFIG_ENTRY