Fix freebox pairing in bridge mode (#106131)
This commit is contained in:
parent
d4be6632cc
commit
b9a8b992d7
5 changed files with 104 additions and 26 deletions
|
@ -11,7 +11,7 @@ from homeassistant.const import CONF_HOST, CONF_PORT
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
|
|
||||||
from .const import DOMAIN
|
from .const import DOMAIN
|
||||||
from .router import get_api
|
from .router import get_api, get_hosts_list_if_supported
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
@ -69,7 +69,7 @@ class FreeboxFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
|
|
||||||
# Check permissions
|
# Check permissions
|
||||||
await fbx.system.get_config()
|
await fbx.system.get_config()
|
||||||
await fbx.lan.get_hosts_list()
|
await get_hosts_list_if_supported(fbx)
|
||||||
|
|
||||||
# Close connection
|
# Close connection
|
||||||
await fbx.close()
|
await fbx.close()
|
||||||
|
|
|
@ -64,6 +64,33 @@ async def get_api(hass: HomeAssistant, host: str) -> Freepybox:
|
||||||
return Freepybox(APP_DESC, token_file, API_VERSION)
|
return Freepybox(APP_DESC, token_file, API_VERSION)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_hosts_list_if_supported(
|
||||||
|
fbx_api: Freepybox,
|
||||||
|
) -> tuple[bool, list[dict[str, Any]]]:
|
||||||
|
"""Hosts list is not supported when freebox is configured in bridge mode."""
|
||||||
|
supports_hosts: bool = True
|
||||||
|
fbx_devices: list[dict[str, Any]] = []
|
||||||
|
try:
|
||||||
|
fbx_devices = await fbx_api.lan.get_hosts_list() or []
|
||||||
|
except HttpRequestError as err:
|
||||||
|
if (
|
||||||
|
(matcher := re.search(r"Request failed \(APIResponse: (.+)\)", str(err)))
|
||||||
|
and is_json(json_str := matcher.group(1))
|
||||||
|
and (json_resp := json.loads(json_str)).get("error_code") == "nodev"
|
||||||
|
):
|
||||||
|
# No need to retry, Host list not available
|
||||||
|
supports_hosts = False
|
||||||
|
_LOGGER.debug(
|
||||||
|
"Host list is not available using bridge mode (%s)",
|
||||||
|
json_resp.get("msg"),
|
||||||
|
)
|
||||||
|
|
||||||
|
else:
|
||||||
|
raise err
|
||||||
|
|
||||||
|
return supports_hosts, fbx_devices
|
||||||
|
|
||||||
|
|
||||||
class FreeboxRouter:
|
class FreeboxRouter:
|
||||||
"""Representation of a Freebox router."""
|
"""Representation of a Freebox router."""
|
||||||
|
|
||||||
|
@ -111,27 +138,9 @@ class FreeboxRouter:
|
||||||
|
|
||||||
# Access to Host list not available in bridge mode, API return error_code 'nodev'
|
# Access to Host list not available in bridge mode, API return error_code 'nodev'
|
||||||
if self.supports_hosts:
|
if self.supports_hosts:
|
||||||
try:
|
self.supports_hosts, fbx_devices = await get_hosts_list_if_supported(
|
||||||
fbx_devices = await self._api.lan.get_hosts_list()
|
self._api
|
||||||
except HttpRequestError as err:
|
|
||||||
if (
|
|
||||||
(
|
|
||||||
matcher := re.search(
|
|
||||||
r"Request failed \(APIResponse: (.+)\)", str(err)
|
|
||||||
)
|
)
|
||||||
)
|
|
||||||
and is_json(json_str := matcher.group(1))
|
|
||||||
and (json_resp := json.loads(json_str)).get("error_code") == "nodev"
|
|
||||||
):
|
|
||||||
# No need to retry, Host list not available
|
|
||||||
self.supports_hosts = False
|
|
||||||
_LOGGER.debug(
|
|
||||||
"Host list is not available using bridge mode (%s)",
|
|
||||||
json_resp.get("msg"),
|
|
||||||
)
|
|
||||||
|
|
||||||
else:
|
|
||||||
raise err
|
|
||||||
|
|
||||||
# Adds the Freebox itself
|
# Adds the Freebox itself
|
||||||
fbx_devices.append(
|
fbx_devices.append(
|
||||||
|
|
|
@ -112,3 +112,14 @@ def mock_router_bridge_mode(mock_device_registry_devices, router):
|
||||||
)
|
)
|
||||||
|
|
||||||
return router
|
return router
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_router_bridge_mode_error(mock_device_registry_devices, router):
|
||||||
|
"""Mock a failed connection to Freebox Bridge mode."""
|
||||||
|
|
||||||
|
router().lan.get_hosts_list = AsyncMock(
|
||||||
|
side_effect=HttpRequestError("Request failed (APIResponse: some unknown error)")
|
||||||
|
)
|
||||||
|
|
||||||
|
return router
|
||||||
|
|
|
@ -69,8 +69,8 @@ async def test_zeroconf(hass: HomeAssistant) -> None:
|
||||||
assert result["step_id"] == "link"
|
assert result["step_id"] == "link"
|
||||||
|
|
||||||
|
|
||||||
async def test_link(hass: HomeAssistant, router: Mock) -> None:
|
async def internal_test_link(hass: HomeAssistant) -> None:
|
||||||
"""Test linking."""
|
"""Test linking internal, common to both router modes."""
|
||||||
with patch(
|
with patch(
|
||||||
"homeassistant.components.freebox.async_setup_entry",
|
"homeassistant.components.freebox.async_setup_entry",
|
||||||
return_value=True,
|
return_value=True,
|
||||||
|
@ -91,6 +91,30 @@ async def test_link(hass: HomeAssistant, router: Mock) -> None:
|
||||||
assert len(mock_setup_entry.mock_calls) == 1
|
assert len(mock_setup_entry.mock_calls) == 1
|
||||||
|
|
||||||
|
|
||||||
|
async def test_link(hass: HomeAssistant, router: Mock) -> None:
|
||||||
|
"""Test link with standard router mode."""
|
||||||
|
await internal_test_link(hass)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_link_bridge_mode(hass: HomeAssistant, router_bridge_mode: Mock) -> None:
|
||||||
|
"""Test linking for a freebox in bridge mode."""
|
||||||
|
await internal_test_link(hass)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_link_bridge_mode_error(
|
||||||
|
hass: HomeAssistant, mock_router_bridge_mode_error: Mock
|
||||||
|
) -> None:
|
||||||
|
"""Test linking for a freebox in bridge mode, unknown error received from API."""
|
||||||
|
result = await hass.config_entries.flow.async_init(
|
||||||
|
DOMAIN,
|
||||||
|
context={"source": SOURCE_USER},
|
||||||
|
data={CONF_HOST: MOCK_HOST, CONF_PORT: MOCK_PORT},
|
||||||
|
)
|
||||||
|
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||||
|
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||||
|
assert result["errors"] == {"base": "cannot_connect"}
|
||||||
|
|
||||||
|
|
||||||
async def test_abort_if_already_setup(hass: HomeAssistant) -> None:
|
async def test_abort_if_already_setup(hass: HomeAssistant) -> None:
|
||||||
"""Test we abort if component is already setup."""
|
"""Test we abort if component is already setup."""
|
||||||
MockConfigEntry(
|
MockConfigEntry(
|
||||||
|
|
|
@ -1,7 +1,11 @@
|
||||||
"""Tests for the Freebox utility methods."""
|
"""Tests for the Freebox utility methods."""
|
||||||
import json
|
import json
|
||||||
|
from unittest.mock import Mock
|
||||||
|
|
||||||
from homeassistant.components.freebox.router import is_json
|
from freebox_api.exceptions import HttpRequestError
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from homeassistant.components.freebox.router import get_hosts_list_if_supported, is_json
|
||||||
|
|
||||||
from .const import DATA_LAN_GET_HOSTS_LIST_MODE_BRIDGE, DATA_WIFI_GET_GLOBAL_CONFIG
|
from .const import DATA_LAN_GET_HOSTS_LIST_MODE_BRIDGE, DATA_WIFI_GET_GLOBAL_CONFIG
|
||||||
|
|
||||||
|
@ -20,3 +24,33 @@ async def test_is_json() -> None:
|
||||||
assert not is_json("")
|
assert not is_json("")
|
||||||
assert not is_json("XXX")
|
assert not is_json("XXX")
|
||||||
assert not is_json("{XXX}")
|
assert not is_json("{XXX}")
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_hosts_list_if_supported(
|
||||||
|
router: Mock,
|
||||||
|
) -> None:
|
||||||
|
"""In router mode, get_hosts_list is supported and list is filled."""
|
||||||
|
supports_hosts, fbx_devices = await get_hosts_list_if_supported(router())
|
||||||
|
assert supports_hosts is True
|
||||||
|
# List must not be empty; but it's content depends on how many unit tests are executed...
|
||||||
|
assert fbx_devices
|
||||||
|
assert "d633d0c8-958c-43cc-e807-d881b076924b" in str(fbx_devices)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_hosts_list_if_supported_bridge(
|
||||||
|
router_bridge_mode: Mock,
|
||||||
|
) -> None:
|
||||||
|
"""In bridge mode, get_hosts_list is NOT supported and list is empty."""
|
||||||
|
supports_hosts, fbx_devices = await get_hosts_list_if_supported(
|
||||||
|
router_bridge_mode()
|
||||||
|
)
|
||||||
|
assert supports_hosts is False
|
||||||
|
assert fbx_devices == []
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_hosts_list_if_supported_bridge_error(
|
||||||
|
mock_router_bridge_mode_error: Mock,
|
||||||
|
) -> None:
|
||||||
|
"""Other exceptions must be propagated."""
|
||||||
|
with pytest.raises(HttpRequestError):
|
||||||
|
await get_hosts_list_if_supported(mock_router_bridge_mode_error())
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue