1
0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00

Add the dial action to the FRITZ!Box Tools integration (#151095)

Co-authored-by: Norbert Rittel <norbert@rittel.de>
Co-authored-by: Michael <35783820+mib1185@users.noreply.github.com>
This commit is contained in:
Abestanis
2025-10-18 09:53:54 +02:00
committed by GitHub
parent 62e59608b0
commit 6964829699
6 changed files with 320 additions and 2 deletions

View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from datetime import datetime, timedelta
@@ -16,6 +17,7 @@ from fritzconnection.core.exceptions import (
FritzConnectionException,
FritzSecurityError,
)
from fritzconnection.lib.fritzcall import FritzCall
from fritzconnection.lib.fritzhosts import FritzHosts
from fritzconnection.lib.fritzstatus import FritzStatus
from fritzconnection.lib.fritzwlan import FritzGuestWLAN
@@ -120,6 +122,7 @@ class FritzBoxTools(DataUpdateCoordinator[UpdateCoordinatorDataType]):
self.fritz_guest_wifi: FritzGuestWLAN = None
self.fritz_hosts: FritzHosts = None
self.fritz_status: FritzStatus = None
self.fritz_call: FritzCall = None
self.host = host
self.mesh_role = MeshRoles.NONE
self.mesh_wifi_uplink = False
@@ -183,6 +186,7 @@ class FritzBoxTools(DataUpdateCoordinator[UpdateCoordinatorDataType]):
self.fritz_hosts = FritzHosts(fc=self.connection)
self.fritz_guest_wifi = FritzGuestWLAN(fc=self.connection)
self.fritz_status = FritzStatus(fc=self.connection)
self.fritz_call = FritzCall(fc=self.connection)
info = self.fritz_status.get_device_info()
_LOGGER.debug(
@@ -617,6 +621,14 @@ class FritzBoxTools(DataUpdateCoordinator[UpdateCoordinatorDataType]):
self.fritz_guest_wifi.set_password, password, length
)
async def async_trigger_dial(self, number: str, max_ring_seconds: int) -> None:
"""Trigger service to dial a number."""
try:
await self.hass.async_add_executor_job(self.fritz_call.dial, number)
await asyncio.sleep(max_ring_seconds)
finally:
await self.hass.async_add_executor_job(self.fritz_call.hangup)
async def async_trigger_cleanup(self) -> None:
"""Trigger device trackers cleanup."""
_LOGGER.debug("Device tracker cleanup triggered")

View File

@@ -62,6 +62,9 @@
},
"set_guest_wifi_password": {
"service": "mdi:form-textbox-password"
},
"dial": {
"service": "mdi:phone-dial"
}
}
}

View File

@@ -4,6 +4,7 @@ import logging
from fritzconnection.core.exceptions import (
FritzActionError,
FritzActionFailedError,
FritzConnectionException,
FritzServiceError,
)
@@ -27,6 +28,14 @@ SERVICE_SCHEMA_SET_GUEST_WIFI_PW = vol.Schema(
vol.Optional("length"): vol.Range(min=8, max=63),
}
)
SERVICE_DIAL = "dial"
SERVICE_SCHEMA_DIAL = vol.Schema(
{
vol.Required("device_id"): str,
vol.Required("number"): str,
vol.Required("max_ring_seconds"): vol.Range(min=1, max=300),
}
)
async def _async_set_guest_wifi_password(service_call: ServiceCall) -> None:
@@ -65,6 +74,46 @@ async def _async_set_guest_wifi_password(service_call: ServiceCall) -> None:
) from ex
async def _async_dial(service_call: ServiceCall) -> None:
"""Call Fritz dial service."""
target_entry_ids = await async_extract_config_entry_ids(service_call)
target_entries: list[FritzConfigEntry] = [
loaded_entry
for loaded_entry in service_call.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 = target_entry.runtime_data
try:
await avm_wrapper.async_trigger_dial(
service_call.data["number"],
max_ring_seconds=service_call.data["max_ring_seconds"],
)
except (FritzServiceError, FritzActionError) as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_parameter_unknown"
) from ex
except FritzActionFailedError as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_dial_failed"
) from ex
except FritzConnectionException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="service_not_supported"
) from ex
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up services for Fritz integration."""
@@ -75,3 +124,4 @@ def async_setup_services(hass: HomeAssistant) -> None:
_async_set_guest_wifi_password,
SERVICE_SCHEMA_SET_GUEST_WIFI_PW,
)
hass.services.async_register(DOMAIN, SERVICE_DIAL, _async_dial, SERVICE_SCHEMA_DIAL)

View File

@@ -17,3 +17,24 @@ set_guest_wifi_password:
number:
min: 8
max: 63
dial:
fields:
device_id:
required: true
selector:
device:
integration: fritz
entity:
device_class: connectivity
number:
required: true
selector:
text:
max_ring_seconds:
default: 15
required: true
selector:
number:
min: 1
max: 300
unit_of_measurement: seconds

View File

@@ -198,12 +198,33 @@
"description": "Length of the new password. It will be auto-generated if no password is set."
}
}
},
"dial": {
"name": "Dial a phone number",
"description": "Makes the FRITZ!Box dial a phone number.",
"fields": {
"device_id": {
"name": "FRITZ!Box device",
"description": "Select the FRITZ!Box to dial from."
},
"number": {
"name": "Phone number",
"description": "The phone number to dial."
},
"max_ring_seconds": {
"name": "Maximum ring duration",
"description": "The maximum number of seconds to ring after dialing."
}
}
}
},
"exceptions": {
"config_entry_not_found": {
"message": "Failed to perform action \"{service}\". Config entry for target not found"
},
"service_dial_failed": {
"message": "Failed to dial, check if the click to dial service of the FRITZ!Box is activated"
},
"service_parameter_unknown": {
"message": "Action or parameter unknown"
},

View File

@@ -2,11 +2,19 @@
from unittest.mock import patch
from fritzconnection.core.exceptions import FritzConnectionException, FritzServiceError
from fritzconnection.core.exceptions import (
FritzActionFailedError,
FritzConnectionException,
FritzServiceError,
)
import pytest
from voluptuous import MultipleInvalid
from homeassistant.components.fritz.const import DOMAIN
from homeassistant.components.fritz.services import SERVICE_SET_GUEST_WIFI_PW
from homeassistant.components.fritz.services import (
SERVICE_DIAL,
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
@@ -24,6 +32,7 @@ async def test_setup_services(hass: HomeAssistant) -> None:
services = hass.services.async_services_for_domain(DOMAIN)
assert services
assert SERVICE_SET_GUEST_WIFI_PW in services
assert SERVICE_DIAL in services
async def test_service_set_guest_wifi_password(
@@ -132,3 +141,205 @@ async def test_service_set_guest_wifi_password_unloaded(
'ServiceValidationError: Failed to perform action "set_guest_wifi_password". Config entry for target not found'
in caplog.text
)
async def test_service_dial(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service dial."""
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_dial"
) as mock_async_trigger_dial:
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{"device_id": device.id, "number": "1234567890", "max_ring_seconds": 10},
)
assert mock_async_trigger_dial.called
assert mock_async_trigger_dial.call_args.kwargs == {"max_ring_seconds": 10}
assert mock_async_trigger_dial.call_args.args == ("1234567890",)
async def test_service_dial_unknown_parameter(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service dial with unknown parameters."""
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_dial",
side_effect=FritzServiceError("boom"),
) as mock_async_trigger_dial:
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{"device_id": device.id, "number": "1234567890", "max_ring_seconds": 10},
)
assert mock_async_trigger_dial.called
assert "HomeAssistantError: Action or parameter unknown" in caplog.text
async def test_service_dial_wrong_parameter(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service dial with unknown parameters."""
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_dial",
) as mock_async_trigger_dial:
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{
"device_id": device.id,
"number": "1234567890",
"max_ring_seconds": "",
},
)
assert not mock_async_trigger_dial.called
with patch(
"homeassistant.components.fritz.coordinator.AvmWrapper.async_trigger_dial",
) as mock_async_trigger_dial:
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{
"device_id": device.id,
"number": "1234567890",
"max_ring_seconds": 0,
},
)
assert not mock_async_trigger_dial.called
async def test_service_dial_service_not_supported(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test service dial 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_dial",
side_effect=FritzConnectionException("boom"),
) as mock_async_trigger_dial:
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{"device_id": device.id, "number": "1234567890", "max_ring_seconds": 10},
)
assert mock_async_trigger_dial.called
assert "HomeAssistantError: Action not supported" in caplog.text
async def test_service_dial_failed(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
fc_class_mock,
fh_class_mock,
) -> None:
"""Test dial service when the dial help is disabled."""
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_dial",
side_effect=FritzActionFailedError("boom"),
) as mock_async_trigger_dial:
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{"device_id": device.id, "number": "1234567890", "max_ring_seconds": 10},
)
assert mock_async_trigger_dial.called
assert (
"HomeAssistantError: Failed to dial, check if the click to dial service of the FRITZ!Box is activated"
in caplog.text
)
async def test_service_dial_unloaded(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test service dial."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
with patch(
"homeassistant.components.fritz.coordinator.AvmWrapper.async_trigger_dial"
) as mock_async_trigger_dial:
await hass.services.async_call(
DOMAIN,
SERVICE_DIAL,
{"device_id": "12345678", "number": "1234567890", "max_ring_seconds": 10},
)
assert not mock_async_trigger_dial.called
assert (
f'ServiceValidationError: Failed to perform action "{SERVICE_DIAL}". Config entry for target not found'
in caplog.text
)