Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use new pyfibaro library features (#139476)

pull/125870/merge
rappenze 2025-02-28 20:47:03 +01:00 committed by GitHub
parent 437e545116
commit 6ce48eab45
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 76 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
homeassistant/components/fibaro/__init__.py
View File

@ -7,21 +7,20 @@ from collections.abc import Callable, Mapping
import logging
from typing import Any
from pyfibaro.fibaro_client import FibaroClient
from pyfibaro.fibaro_client import (
FibaroAuthenticationFailed,
FibaroClient,
FibaroConnectFailed,
)
from pyfibaro.fibaro_device import DeviceModel
from pyfibaro.fibaro_room import RoomModel
from pyfibaro.fibaro_info import InfoModel
from pyfibaro.fibaro_scene import SceneModel
from pyfibaro.fibaro_state_resolver import FibaroEvent, FibaroStateResolver
from requests.exceptions import HTTPError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_URL, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
)
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceEntry, DeviceInfo
from homeassistant.util import slugify
@ -74,63 +73,31 @@ FIBARO_TYPEMAP = {
class FibaroController:
"""Initiate Fibaro Controller Class."""
def __init__(self, config: Mapping[str, Any]) -> None:
def __init__(
self, fibaro_client: FibaroClient, info: InfoModel, import_plugins: bool
) -> None:
"""Initialize the Fibaro controller."""
# The FibaroClient uses the correct API version automatically
self._client = FibaroClient(config[CONF_URL])
self._client.set_authentication(config[CONF_USERNAME], config[CONF_PASSWORD])
self._client = fibaro_client
self._fibaro_info = info
# Whether to import devices from plugins
self._import_plugins = config[CONF_IMPORT_PLUGINS]
self._room_map: dict[int, RoomModel] # Mapping roomId to room object
self._import_plugins = import_plugins
# Mapping roomId to room object
self._room_map = {room.fibaro_id: room for room in self._client.read_rooms()}
self._device_map: dict[int, DeviceModel] # Mapping deviceId to device object
self.fibaro_devices: dict[Platform, list[DeviceModel]] = defaultdict(
list
) # List of devices by entity platform
# All scenes
self._scenes: list[SceneModel] = []
self._scenes = self._client.read_scenes()
self._callbacks: dict[int, list[Any]] = {} # Update value callbacks by deviceId
# Event callbacks by device id
self._event_callbacks: dict[int, list[Callable[[FibaroEvent], None]]] = {}
self.hub_serial: str # Unique serial number of the hub
self.hub_name: str # The friendly name of the hub
self.hub_model: str
self.hub_software_version: str
self.hub_api_url: str = config[CONF_URL]
# Unique serial number of the hub
self.hub_serial = info.serial_number
# Device infos by fibaro device id
self._device_infos: dict[int, DeviceInfo] = {}
def connect(self) -> None:
"""Start the communication with the Fibaro controller."""
# Return value doesn't need to be checked,
# it is only relevant when connecting without credentials
self._client.connect()
info = self._client.read_info()
self.hub_serial = info.serial_number
self.hub_name = info.hc_name
self.hub_model = info.platform
self.hub_software_version = info.current_version
self._room_map = {room.fibaro_id: room for room in self._client.read_rooms()}
self._read_devices()
self._scenes = self._client.read_scenes()
def connect_with_error_handling(self) -> None:
"""Translate connect errors to easily differentiate auth and connect failures.
When there is a better error handling in the used library this can be improved.
"""
try:
self.connect()
except HTTPError as http_ex:
if http_ex.response.status_code == 403:
raise FibaroAuthFailed from http_ex
raise FibaroConnectFailed from http_ex
except Exception as ex:
raise FibaroConnectFailed from ex
def enable_state_handler(self) -> None:
"""Start StateHandler thread for monitoring updates."""
@ -310,6 +277,14 @@ class FibaroController:
"""Return list of scenes."""
return self._scenes
def read_fibaro_info(self) -> InfoModel:
"""Return the general info about the hub."""
return self._fibaro_info
def get_frontend_url(self) -> str:
"""Return the url to the Fibaro hub web UI."""
return self._client.frontend_url()
def _read_devices(self) -> None:
"""Read and process the device list."""
devices = self._client.read_devices()
@ -375,11 +350,17 @@ class FibaroController:
pass
def connect_fibaro_client(data: Mapping[str, Any]) -> tuple[InfoModel, FibaroClient]:
"""Connect to the fibaro hub and read some basic data."""
client = FibaroClient(data[CONF_URL])
info = client.connect_with_credentials(data[CONF_USERNAME], data[CONF_PASSWORD])
return (info, client)
def init_controller(data: Mapping[str, Any]) -> FibaroController:
"""Validate the user input allows us to connect to fibaro."""
controller = FibaroController(data)
controller.connect_with_error_handling()
return controller
"""Connect to the fibaro hub and init the controller."""
info, client = connect_fibaro_client(data)
return FibaroController(client, info, data[CONF_IMPORT_PLUGINS])
async def async_setup_entry(hass: HomeAssistant, entry: FibaroConfigEntry) -> bool:
@ -393,22 +374,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: FibaroConfigEntry) -> bo
raise ConfigEntryNotReady(
f"Could not connect to controller at {entry.data[CONF_URL]}"
) from connect_ex
except FibaroAuthFailed as auth_ex:
except FibaroAuthenticationFailed as auth_ex:
raise ConfigEntryAuthFailed from auth_ex
entry.runtime_data = controller
# register the hub device info separately as the hub has sometimes no entities
fibaro_info = controller.read_fibaro_info()
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, controller.hub_serial)},
serial_number=controller.hub_serial,
manufacturer="Fibaro",
name=controller.hub_name,
model=controller.hub_model,
sw_version=controller.hub_software_version,
configuration_url=controller.hub_api_url.removesuffix("/api/"),
manufacturer=fibaro_info.manufacturer_name,
name=fibaro_info.hc_name,
model=fibaro_info.model_name,
sw_version=fibaro_info.current_version,
configuration_url=controller.get_frontend_url(),
connections={(dr.CONNECTION_NETWORK_MAC, fibaro_info.mac_address)},
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -443,11 +426,3 @@ async def async_remove_config_entry_device(
return False
return True
class FibaroConnectFailed(HomeAssistantError):
"""Error to indicate we cannot connect to fibaro home center."""
class FibaroAuthFailed(HomeAssistantError):
"""Error to indicate that authentication failed on fibaro home center."""

17
homeassistant/components/fibaro/config_flow.py
View File

@ -6,6 +6,7 @@ from collections.abc import Mapping
import logging
from typing import Any
from pyfibaro.fibaro_client import FibaroAuthenticationFailed, FibaroConnectFailed
from slugify import slugify
import voluptuous as vol
@ -13,7 +14,7 @@ from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_URL, CONF_USERNAME
from homeassistant.core import HomeAssistant
from . import FibaroAuthFailed, FibaroConnectFailed, init_controller
from . import connect_fibaro_client
from .const import CONF_IMPORT_PLUGINS, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -33,16 +34,16 @@ async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
controller = await hass.async_add_executor_job(init_controller, data)
info, _ = await hass.async_add_executor_job(connect_fibaro_client, data)
_LOGGER.debug(
"Successfully connected to fibaro home center %s with name %s",
controller.hub_serial,
controller.hub_name,
info.serial_number,
info.hc_name,
)
return {
"serial_number": slugify(controller.hub_serial),
"name": controller.hub_name,
"serial_number": slugify(info.serial_number),
"name": info.hc_name,
}
@ -75,7 +76,7 @@ class FibaroConfigFlow(ConfigFlow, domain=DOMAIN):
info = await _validate_input(self.hass, user_input)
except FibaroConnectFailed:
errors["base"] = "cannot_connect"
except FibaroAuthFailed:
except FibaroAuthenticationFailed:
errors["base"] = "invalid_auth"
else:
await self.async_set_unique_id(info["serial_number"])
@ -106,7 +107,7 @@ class FibaroConfigFlow(ConfigFlow, domain=DOMAIN):
await _validate_input(self.hass, new_data)
except FibaroConnectFailed:
errors["base"] = "cannot_connect"
except FibaroAuthFailed:
except FibaroAuthenticationFailed:
errors["base"] = "invalid_auth"
else:
return self.async_update_reload_and_abort(

7
tests/components/fibaro/conftest.py
View File

@ -209,19 +209,22 @@ def mock_fibaro_client() -> Generator[Mock]:
info_mock.hc_name = TEST_NAME
info_mock.current_version = TEST_VERSION
info_mock.platform = TEST_MODEL
info_mock.manufacturer_name = "Fibaro"
info_mock.model_name = "Home Center 2"
info_mock.mac_address = "00:22:4d:b7:13:24"
with patch(
"homeassistant.components.fibaro.FibaroClient", autospec=True
) as fibaro_client_mock:
client = fibaro_client_mock.return_value
client.set_authentication.return_value = None
client.connect.return_value = True
client.connect_with_credentials.return_value = info_mock
client.read_info.return_value = info_mock
client.read_rooms.return_value = []
client.read_scenes.return_value = []
client.read_devices.return_value = []
client.register_update_handler.return_value = None
client.unregister_update_handler.return_value = None
client.frontend_url.return_value = TEST_URL.removesuffix("/api/")
yield client

58
tests/components/fibaro/test_config_flow.py
View File

@ -2,8 +2,8 @@
from unittest.mock import Mock
from pyfibaro.fibaro_client import FibaroAuthenticationFailed, FibaroConnectFailed
import pytest
from requests.exceptions import HTTPError
from homeassistant import config_entries
from homeassistant.components.fibaro import DOMAIN
@ -23,8 +23,10 @@ pytestmark = pytest.mark.usefixtures("mock_setup_entry", "mock_fibaro_client")
async def _recovery_after_failure_works(
hass: HomeAssistant, mock_fibaro_client: Mock, result: FlowResult
) -> None:
mock_fibaro_client.connect.side_effect = None
mock_fibaro_client.connect.return_value = True
mock_fibaro_client.connect_with_credentials.side_effect = None
mock_fibaro_client.connect_with_credentials.return_value = (
mock_fibaro_client.read_info()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -48,8 +50,10 @@ async def _recovery_after_failure_works(
async def _recovery_after_reauth_failure_works(
hass: HomeAssistant, mock_fibaro_client: Mock, result: FlowResult
) -> None:
mock_fibaro_client.connect.side_effect = None
mock_fibaro_client.connect.return_value = True
mock_fibaro_client.connect_with_credentials.side_effect = None
mock_fibaro_client.connect_with_credentials.return_value = (
mock_fibaro_client.read_info()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -101,7 +105,9 @@ async def test_config_flow_user_initiated_auth_failure(
assert result["step_id"] == "user"
assert result["errors"] == {}
mock_fibaro_client.connect.side_effect = HTTPError(response=Mock(status_code=403))
mock_fibaro_client.connect_with_credentials.side_effect = (
FibaroAuthenticationFailed()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -119,7 +125,7 @@ async def test_config_flow_user_initiated_auth_failure(
await _recovery_after_failure_works(hass, mock_fibaro_client, result)
async def test_config_flow_user_initiated_unknown_failure_1(
async def test_config_flow_user_initiated_connect_failure(
hass: HomeAssistant, mock_fibaro_client: Mock
) -> None:
"""Unknown failure in flow manually initialized by the user."""
@ -131,37 +137,7 @@ async def test_config_flow_user_initiated_unknown_failure_1(
assert result["step_id"] == "user"
assert result["errors"] == {}
mock_fibaro_client.connect.side_effect = HTTPError(response=Mock(status_code=500))
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_URL: TEST_URL,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
await _recovery_after_failure_works(hass, mock_fibaro_client, result)
async def test_config_flow_user_initiated_unknown_failure_2(
hass: HomeAssistant, mock_fibaro_client: Mock
) -> None:
"""Unknown failure in flow manually initialized by the user."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
mock_fibaro_client.connect.side_effect = Exception()
mock_fibaro_client.connect_with_credentials.side_effect = FibaroConnectFailed()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -208,7 +184,7 @@ async def test_reauth_connect_failure(
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
mock_fibaro_client.connect.side_effect = Exception()
mock_fibaro_client.connect_with_credentials.side_effect = FibaroConnectFailed()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -233,7 +209,9 @@ async def test_reauth_auth_failure(
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
mock_fibaro_client.connect.side_effect = HTTPError(response=Mock(status_code=403))
mock_fibaro_client.connect_with_credentials.side_effect = (
FibaroAuthenticationFailed()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],