evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update rfxtrx library to handle connection retries (#111366)

Browse source 
Upgrade rfxtrx library to one that leaves reconnection to home assistant and handle loss of connection by reloading the integration.

https://github.com/Danielhiversen/pyRFXtrx/releases/tag/0.31.0
This commit is contained in:
Joakim Plate 2024-02-26 19:30:22 +01:00 • committed by GitHub
parent 979fe57f7f
commit a55c56a207
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 200 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

101
tests/components/rfxtrx/test_init.py
View file

@ -1,19 +1,18 @@
"""The tests for the Rfxtrx component."""
from __future__ import annotations
from unittest.mock import ANY, call, patch
from unittest.mock import ANY, call
import RFXtrx as rfxtrxmod
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.components.rfxtrx.const import EVENT_RFXTRX_EVENT
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from .conftest import create_rfx_test_cfg, setup_rfx_test_cfg
from .conftest import setup_rfx_test_cfg
from tests.common import MockConfigEntry
from tests.typing import WebSocketGenerator
SOME_PROTOCOLS = ["ac", "arc"]
@ -130,29 +129,89 @@ async def test_ws_device_remove(
assert mock_entry.data["devices"] == {}
async def test_connect(hass: HomeAssistant) -> None:
async def test_connect(
rfxtrx, connect_mock, transport_mock, hass: HomeAssistant
) -> None:
"""Test that we attempt to connect to the device."""
entry_data = create_rfx_test_cfg(device="/dev/ttyUSBfake")
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
config_entry = await setup_rfx_test_cfg(hass, device="/dev/ttyUSBfake")
transport_mock.assert_called_once_with("/dev/ttyUSBfake")
connect_mock.assert_called_once_with(transport_mock.return_value, ANY, modes=ANY)
rfxtrx.connect.assert_called_once_with(ANY)
with patch.object(rfxtrxmod, "Connect") as connect:
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
connect.assert_called_once_with("/dev/ttyUSBfake", ANY, modes=ANY)
assert config_entry.state is ConfigEntryState.LOADED
async def test_connect_with_protocols(hass: HomeAssistant) -> None:
async def test_connect_network(
rfxtrx, connect_mock, transport_mock, hass: HomeAssistant
) -> None:
"""Test that we attempt to connect to the device."""
config_entry = await setup_rfx_test_cfg(hass, host="localhost", port=1234)
transport_mock.assert_called_once_with(("localhost", 1234))
connect_mock.assert_called_once_with(transport_mock.return_value, ANY, modes=ANY)
rfxtrx.connect.assert_called_once_with(ANY)
assert config_entry.state is ConfigEntryState.LOADED
async def test_connect_with_protocols(
rfxtrx, connect_mock, transport_mock, hass: HomeAssistant
) -> None:
"""Test that we attempt to set protocols."""
entry_data = create_rfx_test_cfg(device="/dev/ttyUSBfake", protocols=SOME_PROTOCOLS)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
config_entry = await setup_rfx_test_cfg(
hass, device="/dev/ttyUSBfake", protocols=SOME_PROTOCOLS
)
transport_mock.assert_called_once_with("/dev/ttyUSBfake")
connect_mock.assert_called_once_with(
transport_mock.return_value, ANY, modes=SOME_PROTOCOLS
)
rfxtrx.connect.assert_called_once_with(ANY)
mock_entry.add_to_hass(hass)
assert config_entry.state is ConfigEntryState.LOADED
with patch.object(rfxtrxmod, "Connect") as connect:
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
connect.assert_called_once_with("/dev/ttyUSBfake", ANY, modes=SOME_PROTOCOLS)
async def test_connect_timeout(
rfxtrx, connect_mock, transport_mock, hass: HomeAssistant
) -> None:
"""Test that we attempt to connect to the device."""
rfxtrx.connect.side_effect = TimeoutError
config_entry = await setup_rfx_test_cfg(hass, device="/dev/ttyUSBfake")
transport_mock.assert_called_once_with("/dev/ttyUSBfake")
connect_mock.assert_called_once_with(transport_mock.return_value, ANY, modes=ANY)
rfxtrx.connect.assert_called_once_with(ANY)
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_connect_failed(
rfxtrx, connect_mock, transport_mock, hass: HomeAssistant
) -> None:
"""Test that we attempt to connect to the device."""
rfxtrx.connect.side_effect = rfxtrxmod.RFXtrxTransportError
config_entry = await setup_rfx_test_cfg(hass, device="/dev/ttyUSBfake")
transport_mock.assert_called_once_with("/dev/ttyUSBfake")
connect_mock.assert_called_once_with(transport_mock.return_value, ANY, modes=ANY)
rfxtrx.connect.assert_called_once_with(ANY)
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_reconnect(rfxtrx, hass: HomeAssistant) -> None:
"""Test that we reconnect on connection loss."""
config_entry = await setup_rfx_test_cfg(hass, device="/dev/ttyUSBfake")
assert config_entry.state is ConfigEntryState.LOADED
rfxtrx.connect.call_count = 1
await hass.async_add_executor_job(
rfxtrx.event_callback,
rfxtrxmod.ConnectionLost(),
)
assert config_entry.state is ConfigEntryState.LOADED
rfxtrx.connect.call_count = 2