Register service actions in async_setup of AVM Fritz!Box tools (#136380)

* move service setup into integrations async_setup

* move back to own module

* add service test

* remove unneccessary CONFIG_SCHEMA

* remove unused constant FRITZ_SERVICES

* Revert "remove unneccessary CONFIG_SCHEMA"

This reverts commit cce1ba76a067895d62d0485479002c7bebbfb511.

* remove useless CONFIG_SCHEMA from services.py

* move logic of `service_fritzbox` into services.py

* add more service tests

* simplify logic, use ServiceValidationError
This commit is contained in:
Michael 2025-01-28 17:57:02 +01:00 committed by GitHub
parent c3db493f34
commit a8c382566c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 197 additions and 111 deletions

View File

@ -12,6 +12,8 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import ( from .const import (
DATA_FRITZ, DATA_FRITZ,
@ -22,10 +24,18 @@ from .const import (
PLATFORMS, PLATFORMS,
) )
from .coordinator import AvmWrapper, FritzData from .coordinator import AvmWrapper, FritzData
from .services import async_setup_services, async_unload_services from .services import async_setup_services
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up fritzboxtools integration."""
await async_setup_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up fritzboxtools from config entry.""" """Set up fritzboxtools from config entry."""
@ -65,8 +75,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Load the other platforms like switch # Load the other platforms like switch
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
await async_setup_services(hass)
return True return True
@ -84,8 +92,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unload_ok: if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)
await async_unload_services(hass)
return unload_ok return unload_ok

View File

@ -56,9 +56,6 @@ ERROR_CANNOT_CONNECT = "cannot_connect"
ERROR_UPNP_NOT_CONFIGURED = "upnp_not_configured" ERROR_UPNP_NOT_CONFIGURED = "upnp_not_configured"
ERROR_UNKNOWN = "unknown_error" ERROR_UNKNOWN = "unknown_error"
FRITZ_SERVICES = "fritz_services"
SERVICE_SET_GUEST_WIFI_PW = "set_guest_wifi_password"
SWITCH_TYPE_DEFLECTION = "CallDeflection" SWITCH_TYPE_DEFLECTION = "CallDeflection"
SWITCH_TYPE_PORTFORWARD = "PortForward" SWITCH_TYPE_PORTFORWARD = "PortForward"
SWITCH_TYPE_PROFILE = "Profile" SWITCH_TYPE_PROFILE = "Profile"

View File

@ -16,11 +16,10 @@ from fritzconnection.core.exceptions import (
FritzActionError, FritzActionError,
FritzConnectionException, FritzConnectionException,
FritzSecurityError, FritzSecurityError,
FritzServiceError,
) )
from fritzconnection.lib.fritzhosts import FritzHosts from fritzconnection.lib.fritzhosts import FritzHosts
from fritzconnection.lib.fritzstatus import FritzStatus from fritzconnection.lib.fritzstatus import FritzStatus
from fritzconnection.lib.fritzwlan import DEFAULT_PASSWORD_LENGTH, FritzGuestWLAN from fritzconnection.lib.fritzwlan import FritzGuestWLAN
import xmltodict import xmltodict
from homeassistant.components.device_tracker import ( from homeassistant.components.device_tracker import (
@ -29,7 +28,7 @@ from homeassistant.components.device_tracker import (
DOMAIN as DEVICE_TRACKER_DOMAIN, DOMAIN as DEVICE_TRACKER_DOMAIN,
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
@ -46,7 +45,6 @@ from .const import (
DEFAULT_USERNAME, DEFAULT_USERNAME,
DOMAIN, DOMAIN,
FRITZ_EXCEPTIONS, FRITZ_EXCEPTIONS,
SERVICE_SET_GUEST_WIFI_PW,
MeshRoles, MeshRoles,
) )
@ -693,34 +691,6 @@ class FritzBoxTools(DataUpdateCoordinator[UpdateCoordinatorDataType]):
device.id, remove_config_entry_id=config_entry.entry_id device.id, remove_config_entry_id=config_entry.entry_id
) )
async def service_fritzbox(
self, service_call: ServiceCall, config_entry: ConfigEntry
) -> None:
"""Define FRITZ!Box services."""
_LOGGER.debug("FRITZ!Box service: %s", service_call.service)
if not self.connection:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="unable_to_connect"
)
try:
if service_call.service == SERVICE_SET_GUEST_WIFI_PW:
await self.async_trigger_set_guest_password(
service_call.data.get("password"),
service_call.data.get("length", DEFAULT_PASSWORD_LENGTH),
)
return
except (FritzServiceError, FritzActionError) as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_parameter_unknown"
) from ex
except FritzConnectionException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_not_supported"
) from ex
class AvmWrapper(FritzBoxTools): class AvmWrapper(FritzBoxTools):
"""Setup AVM wrapper for API calls.""" """Setup AVM wrapper for API calls."""

View File

@ -1,8 +1,6 @@
rules: rules:
# Bronze # Bronze
action-setup: action-setup: done
status: todo
comment: still in async_setup_entry, needs to be moved to async_setup
appropriate-polling: done appropriate-polling: done
brands: done brands: done
common-modules: done common-modules: done

View File

@ -1,21 +1,25 @@
"""Services for Fritz integration.""" """Services for Fritz integration."""
from __future__ import annotations
import logging import logging
from fritzconnection.core.exceptions import (
FritzActionError,
FritzConnectionException,
FritzServiceError,
)
from fritzconnection.lib.fritzwlan import DEFAULT_PASSWORD_LENGTH
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.service import async_extract_config_entry_ids from homeassistant.helpers.service import async_extract_config_entry_ids
from .const import DOMAIN, FRITZ_SERVICES, SERVICE_SET_GUEST_WIFI_PW from .const import DOMAIN
from .coordinator import AvmWrapper from .coordinator import AvmWrapper
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
SERVICE_SET_GUEST_WIFI_PW = "set_guest_wifi_password"
SERVICE_SCHEMA_SET_GUEST_WIFI_PW = vol.Schema( SERVICE_SCHEMA_SET_GUEST_WIFI_PW = vol.Schema(
{ {
vol.Required("device_id"): str, vol.Required("device_id"): str,
@ -24,71 +28,48 @@ SERVICE_SCHEMA_SET_GUEST_WIFI_PW = vol.Schema(
} }
) )
SERVICE_LIST: list[tuple[str, vol.Schema | None]] = [
(SERVICE_SET_GUEST_WIFI_PW, SERVICE_SCHEMA_SET_GUEST_WIFI_PW), async def _async_set_guest_wifi_password(service_call: ServiceCall) -> None:
] """Call Fritz set guest wifi password service."""
hass = service_call.hass
target_entry_ids = await async_extract_config_entry_ids(hass, service_call)
target_entries = [
loaded_entry
for loaded_entry in hass.config_entries.async_loaded_entries(DOMAIN)
if loaded_entry.entry_id in target_entry_ids
]
if not target_entries:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="config_entry_not_found",
translation_placeholders={"service": service_call.service},
)
for target_entry in target_entries:
_LOGGER.debug("Executing service %s", service_call.service)
avm_wrapper: AvmWrapper = hass.data[DOMAIN][target_entry.entry_id]
try:
await avm_wrapper.async_trigger_set_guest_password(
service_call.data.get("password"),
service_call.data.get("length", DEFAULT_PASSWORD_LENGTH),
)
except (FritzServiceError, FritzActionError) as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_parameter_unknown"
) from ex
except FritzConnectionException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_not_supported"
) from ex
async def async_setup_services(hass: HomeAssistant) -> None: async def async_setup_services(hass: HomeAssistant) -> None:
"""Set up services for Fritz integration.""" """Set up services for Fritz integration."""
for service, _ in SERVICE_LIST: hass.services.async_register(
if hass.services.has_service(DOMAIN, service): DOMAIN,
return SERVICE_SET_GUEST_WIFI_PW,
_async_set_guest_wifi_password,
async def async_call_fritz_service(service_call: ServiceCall) -> None: SERVICE_SCHEMA_SET_GUEST_WIFI_PW,
"""Call correct Fritz service.""" )
if not (
fritzbox_entry_ids := await _async_get_configured_avm_device(
hass, service_call
)
):
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="config_entry_not_found",
translation_placeholders={"service": service_call.service},
)
for entry_id in fritzbox_entry_ids:
_LOGGER.debug("Executing service %s", service_call.service)
avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry_id]
if config_entry := hass.config_entries.async_get_entry(entry_id):
await avm_wrapper.service_fritzbox(service_call, config_entry)
else:
_LOGGER.error(
"Executing service %s failed, no config entry found",
service_call.service,
)
for service, schema in SERVICE_LIST:
hass.services.async_register(DOMAIN, service, async_call_fritz_service, schema)
async def _async_get_configured_avm_device(
hass: HomeAssistant, service_call: ServiceCall
) -> list:
"""Get FritzBoxTools class from config entry."""
list_entry_id: list = []
for entry_id in await async_extract_config_entry_ids(hass, service_call):
config_entry = hass.config_entries.async_get_entry(entry_id)
if (
config_entry
and config_entry.domain == DOMAIN
and config_entry.state == ConfigEntryState.LOADED
):
list_entry_id.append(entry_id)
return list_entry_id
async def async_unload_services(hass: HomeAssistant) -> None:
"""Unload services for Fritz integration."""
if not hass.data.get(FRITZ_SERVICES):
return
hass.data[FRITZ_SERVICES] = False
for service, _ in SERVICE_LIST:
hass.services.async_remove(DOMAIN, service)

View File

@ -0,0 +1,134 @@
"""Tests for Fritz!Tools services."""
from unittest.mock import patch
from fritzconnection.core.exceptions import FritzConnectionException, FritzServiceError
import pytest
from homeassistant.components.fritz.const import DOMAIN
from homeassistant.components.fritz.services import SERVICE_SET_GUEST_WIFI_PW
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from .const import MOCK_USER_DATA
from tests.common import MockConfigEntry
async def test_setup_services(hass: HomeAssistant) -> None:
"""Test setup of Fritz!Tools services."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
services = hass.services.async_services_for_domain(DOMAIN)
assert services
assert SERVICE_SET_GUEST_WIFI_PW in services
async def test_service_set_guest_wifi_password(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service set_guest_wifi_password."""
assert await async_setup_component(hass, DOMAIN, {})
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
device = device_registry.async_get_device(
identifiers={(DOMAIN, "1C:ED:6F:12:34:11")}
)
assert device
with patch(
"homeassistant.components.fritz.coordinator.AvmWrapper.async_trigger_set_guest_password"
) as mock_async_trigger_set_guest_password:
await hass.services.async_call(
DOMAIN, SERVICE_SET_GUEST_WIFI_PW, {"device_id": device.id}
)
assert mock_async_trigger_set_guest_password.called
async def test_service_set_guest_wifi_password_unknown_parameter(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service set_guest_wifi_password with unknown parameter."""
assert await async_setup_component(hass, DOMAIN, {})
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
device = device_registry.async_get_device(
identifiers={(DOMAIN, "1C:ED:6F:12:34:11")}
)
assert device
with patch(
"homeassistant.components.fritz.coordinator.AvmWrapper.async_trigger_set_guest_password",
side_effect=FritzServiceError("boom"),
) as mock_async_trigger_set_guest_password:
await hass.services.async_call(
DOMAIN, SERVICE_SET_GUEST_WIFI_PW, {"device_id": device.id}
)
assert mock_async_trigger_set_guest_password.called
assert "HomeAssistantError: Action or parameter unknown" in caplog.text
async def test_service_set_guest_wifi_password_service_not_supported(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service set_guest_wifi_password with connection error."""
assert await async_setup_component(hass, DOMAIN, {})
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
device = device_registry.async_get_device(
identifiers={(DOMAIN, "1C:ED:6F:12:34:11")}
)
assert device
with patch(
"homeassistant.components.fritz.coordinator.AvmWrapper.async_trigger_set_guest_password",
side_effect=FritzConnectionException("boom"),
) as mock_async_trigger_set_guest_password:
await hass.services.async_call(
DOMAIN, SERVICE_SET_GUEST_WIFI_PW, {"device_id": device.id}
)
assert mock_async_trigger_set_guest_password.called
assert "HomeAssistantError: Action not supported" in caplog.text
async def test_service_set_guest_wifi_password_unloaded(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test service set_guest_wifi_password."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
with patch(
"homeassistant.components.fritz.coordinator.AvmWrapper.async_trigger_set_guest_password"
) as mock_async_trigger_set_guest_password:
await hass.services.async_call(
DOMAIN, SERVICE_SET_GUEST_WIFI_PW, {"device_id": "12345678"}
)
assert not mock_async_trigger_set_guest_password.called
assert (
'ServiceValidationError: Failed to perform action "set_guest_wifi_password". Config entry for target not found'
in caplog.text
)