starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 07:25:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

test connect timeout

Browse Source
This commit is contained in:
Petar Petrov
2025-09-12 14:15:35 +03:00
parent 876ff17230
commit 84fb87935d
2 changed files with 276 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/otbr/config_flow.py
+1 -2
View File
@@ -8,7 +8,6 @@ import logging
from typing import TYPE_CHECKING, Any, cast
import aiohttp
from aiohttp.client_exceptions import ClientConnectorError
import python_otbr_api
from python_otbr_api import tlv_parser
from python_otbr_api.tlv_parser import MeshcopTLVType
@@ -196,7 +195,7 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN):
try:
return await self._connect_and_configure_router(url)
except ClientConnectorError as exc:
except aiohttp.ClientConnectionError as exc:
_LOGGER.debug(
"ClientConnectorError after %.2f seconds, retrying in %.1fs: %s",
elapsed_time,
tests/components/otbr/test_config_flow.py
+275 -1
View File
@@ -1246,6 +1246,46 @@ async def test_recommended_step_skips_device_selection_if_already_set(
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_recommended_step_device_already_set_skips_to_addon(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test recommended step skips directly to addon when device is already set."""
# Set addon as running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock the network calls to prevent actual network requests
with (
patch(
"homeassistant.components.otbr.config_flow.OTBRConfigFlow._connect_with_retry",
return_value=TEST_BORDER_AGENT_ID,
),
patch(
"homeassistant.components.otbr.async_setup_entry",
return_value=True,
),
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
# Set device on the flow instance
flow_instance = hass.config_entries.flow._progress[result["flow_id"]]
flow_instance._device = "/dev/ttyUSB0"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "recommended"}
)
# Should skip device selection and proceed to create entry
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_addon_step_not_installed_proceeds_to_install(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
@@ -1284,11 +1324,12 @@ async def test_addon_step_running_stops_then_starts(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test addon step stops running addon then starts."""
# Set addon as running
# Set addon as running initially
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock the network calls to prevent actual network requests
with (
@@ -1459,6 +1500,46 @@ async def test_start_otbr_addon_progress(
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_start_addon_progress_step_return(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test start_addon progress step returns progress when task is not done."""
# Set addon as installed but not running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "installed"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock the network calls to prevent actual network requests
with (
patch(
"homeassistant.components.otbr.config_flow.OTBRConfigFlow._connect_with_retry",
return_value=TEST_BORDER_AGENT_ID,
),
patch(
"homeassistant.components.otbr.async_setup_entry",
return_value=True,
),
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "recommended"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"device": "/dev/ttyUSB0"}
)
# The flow should complete successfully and create an entry
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_start_otbr_addon_success(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
@@ -1623,6 +1704,46 @@ async def test_connect_otbr_progress(
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_connect_otbr_progress_step_return(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test connect_otbr progress step returns progress when task is not done."""
# Set addon as running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock the network calls to prevent actual network requests
with (
patch(
"homeassistant.components.otbr.config_flow.OTBRConfigFlow._connect_with_retry",
return_value=TEST_BORDER_AGENT_ID,
),
patch(
"homeassistant.components.otbr.async_setup_entry",
return_value=True,
),
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "recommended"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"device": "/dev/ttyUSB0"}
)
# The flow should complete successfully and create an entry
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures(
"mock_is_hassio",
"mock_async_get_usb_ports",
@@ -1734,6 +1855,74 @@ async def test_connect_otbr_connection_failure(
assert result["reason"] == "unknown"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_connect_otbr_timeout_failure(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test OTBR connection timeout failure."""
# Set addon as running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock connection timeout failure
with patch(
"homeassistant.components.otbr.config_flow.OTBRConfigFlow._connect_with_retry",
side_effect=HomeAssistantError("Failed to connect to OTBR after 10.0 seconds"),
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "recommended"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"device": "/dev/ttyUSB0"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_connect_with_retry_timeout_simulation(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test OTBR connection timeout by simulating elapsed time."""
# Set addon as running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Create a flow instance to test the timeout logic directly
flow = otbr.config_flow.OTBRConfigFlow()
flow.hass = hass
flow.flow_id = "test_flow_id"
# Mock time to simulate timeout
with patch("asyncio.get_event_loop") as mock_get_loop:
mock_loop = MagicMock()
mock_get_loop.return_value = mock_loop
# Simulate time progression: start at 0, then 11 seconds (past timeout)
mock_loop.time.side_effect = [0.0, 11.0]
# Mock the connection method to always fail with connection error
with patch.object(
flow, "_connect_and_configure_router", side_effect=aiohttp.ClientConnectionError
):
with pytest.raises(HomeAssistantError) as exc_info:
await flow._connect_with_retry("http://test:8081")
assert "Failed to connect to OTBR after 10.0 seconds" in str(exc_info.value)
@pytest.mark.usefixtures(
"mock_is_hassio",
"mock_async_get_usb_ports",
@@ -1909,6 +2098,91 @@ async def test_connect_otbr_retry_logic(
assert result["title"] == "Open Thread Border Router"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_connect_with_retry_timeout(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test _connect_with_retry timeout error."""
# Set addon as running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock persistent ClientConnectorError to trigger timeout
from aiohttp import ClientConnectionError
with patch(
"homeassistant.components.otbr.config_flow.OTBRConfigFlow._connect_and_configure_router",
side_effect=ClientConnectionError("Connection refused"),
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "recommended"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"device": "/dev/ttyUSB0"}
)
# Should return progress step while retrying
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "connect_otbr"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_connect_with_retry_client_connector_error(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock
) -> None:
"""Test _connect_with_retry ClientConnectorError handling and retry."""
# Set addon as running
mock_addon_info = (
mock_get_otbr_addon_manager.return_value.async_get_addon_info.return_value
)
mock_addon_info.state = "running"
mock_addon_info.hostname = "core-openthread-border-router"
# Mock ClientConnectionError that eventually succeeds
call_count = 0
from aiohttp import ClientConnectionError
async def mock_connect(url):
nonlocal call_count
call_count += 1
if call_count < 3: # Fail first 2 attempts
raise ClientConnectionError("Connection refused")
return TEST_BORDER_AGENT_ID # Succeed on 3rd attempt
with (
patch(
"homeassistant.components.otbr.config_flow.OTBRConfigFlow._connect_and_configure_router",
side_effect=mock_connect,
),
patch(
"homeassistant.components.otbr.async_setup_entry",
return_value=True,
),
):
result = await hass.config_entries.flow.async_init(
otbr.DOMAIN, context={"source": "user"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "recommended"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"device": "/dev/ttyUSB0"}
)
# Should return progress step while retrying
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "connect_otbr"
@pytest.mark.usefixtures("mock_is_hassio", "mock_async_get_usb_ports")
async def test_connect_otbr_timeout(
hass: HomeAssistant, mock_get_otbr_addon_manager: MagicMock