Add wake on LAN via Fritz!Box for tracked devices (#106778)

homeassistant/components/fritz/button.py

@@ -14,12 +14,13 @@ from homeassistant.components.button import (
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import EntityCategory
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from .common import AvmWrapper
-from .const import DOMAIN
+from .common import AvmWrapper, FritzData, FritzDevice, FritzDeviceBase, _is_tracked
+from .const import BUTTON_TYPE_WOL, CONNECTION_TYPE_LAN, DATA_FRITZ, DOMAIN, MeshRoles
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -70,8 +71,28 @@ async def async_setup_entry(
     _LOGGER.debug("Setting up buttons")
     avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id]
 
-    async_add_entities(
-        [FritzButton(avm_wrapper, entry.title, button) for button in BUTTONS]
+    entities_list: list[ButtonEntity] = [
+        FritzButton(avm_wrapper, entry.title, button) for button in BUTTONS
+    ]
+
+    if avm_wrapper.mesh_role == MeshRoles.SLAVE:
+        async_add_entities(entities_list)
+        return
+
+    data_fritz: FritzData = hass.data[DATA_FRITZ]
+    entities_list += _async_wol_buttons_list(avm_wrapper, data_fritz)
+
+    async_add_entities(entities_list)
+
+    @callback
+    def async_update_avm_device() -> None:
+        """Update the values of the AVM device."""
+        async_add_entities(_async_wol_buttons_list(avm_wrapper, data_fritz))
+
+    entry.async_on_unload(
+        async_dispatcher_connect(
+            hass, avm_wrapper.signal_device_new, async_update_avm_device
+        )
     )
 
 
@@ -101,3 +122,64 @@ class FritzButton(ButtonEntity):
     async def async_press(self) -> None:
         """Triggers Fritz!Box service."""
         await self.entity_description.press_action(self.avm_wrapper)
+
+
+@callback
+def _async_wol_buttons_list(
+    avm_wrapper: AvmWrapper,
+    data_fritz: FritzData,
+) -> list[FritzBoxWOLButton]:
+    """Add new WOL button entities from the AVM device."""
+    _LOGGER.debug("Setting up %s buttons", BUTTON_TYPE_WOL)
+
+    new_wols: list[FritzBoxWOLButton] = []
+
+    if avm_wrapper.unique_id not in data_fritz.wol_buttons:
+        data_fritz.wol_buttons[avm_wrapper.unique_id] = set()
+
+    for mac, device in avm_wrapper.devices.items():
+        if _is_tracked(mac, data_fritz.wol_buttons.values()):
+            _LOGGER.debug("Skipping wol button creation for device %s", device.hostname)
+            continue
+
+        if device.connection_type != CONNECTION_TYPE_LAN:
+            _LOGGER.debug(
+                "Skipping wol button creation for device %s, not connected via LAN",
+                device.hostname,
+            )
+            continue
+
+        new_wols.append(FritzBoxWOLButton(avm_wrapper, device))
+        data_fritz.wol_buttons[avm_wrapper.unique_id].add(mac)
+
+    _LOGGER.debug("Creating %s wol buttons", len(new_wols))
+    return new_wols
+
+
+class FritzBoxWOLButton(FritzDeviceBase, ButtonEntity):
+    """Defines a FRITZ!Box Tools Wake On LAN button."""
+
+    _attr_icon = "mdi:lan-pending"
+    _attr_entity_registry_enabled_default = False
+
+    def __init__(self, avm_wrapper: AvmWrapper, device: FritzDevice) -> None:
+        """Initialize Fritz!Box WOL button."""
+        super().__init__(avm_wrapper, device)
+        self._name = f"{self.hostname} Wake on LAN"
+        self._attr_unique_id = f"{self._mac}_wake_on_lan"
+        self._is_available = True
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_NETWORK_MAC, self._mac)},
+            default_manufacturer="AVM",
+            default_model="FRITZ!Box Tracked device",
+            default_name=device.hostname,
+            via_device=(
+                DOMAIN,
+                avm_wrapper.unique_id,
+            ),
+        )
+
+    async def async_press(self) -> None:
+        """Press the button."""
+        if self.mac_address:
+            await self._avm_wrapper.async_wake_on_lan(self.mac_address)

homeassistant/components/fritz/common.py

@@ -928,6 +928,16 @@ class AvmWrapper(FritzBoxTools):  # pylint: disable=hass-enforce-coordinator-mod
             NewDisallow="0" if turn_on else "1",
         )
 
+    async def async_wake_on_lan(self, mac_address: str) -> dict[str, Any]:
+        """Call X_AVM-DE_WakeOnLANByMACAddress service."""
+
+        return await self._async_service_call(
+            "Hosts",
+            "1",
+            "X_AVM-DE_WakeOnLANByMACAddress",
+            NewMACAddress=mac_address,
+        )
+
 
 @dataclass
 class FritzData:
@@ -935,6 +945,7 @@ class FritzData:
 
     tracked: dict = field(default_factory=dict)
     profile_switches: dict = field(default_factory=dict)
+    wol_buttons: dict = field(default_factory=dict)
 
 
 class FritzDeviceBase(update_coordinator.CoordinatorEntity[AvmWrapper]):

homeassistant/components/fritz/const.py

@@ -65,6 +65,8 @@ SWITCH_TYPE_PORTFORWARD = "PortForward"
 SWITCH_TYPE_PROFILE = "Profile"
 SWITCH_TYPE_WIFINETWORK = "WiFiNetwork"
 
+BUTTON_TYPE_WOL = "WakeOnLan"
+
 UPTIME_DEVIATION = 5
 
 FRITZ_EXCEPTIONS = (
@@ -79,3 +81,5 @@ FRITZ_EXCEPTIONS = (
 FRITZ_AUTH_EXCEPTIONS = (FritzAuthorizationError, FritzSecurityError)
 
 WIFI_STANDARD = {1: "2.4Ghz", 2: "5Ghz", 3: "5Ghz", 4: "Guest"}
+
+CONNECTION_TYPE_LAN = "LAN"

tests/components/fritz/conftest.py

@@ -77,13 +77,11 @@ class FritzConnectionMock:
 class FritzHostMock(FritzHosts):
     """FritzHosts mocking."""
 
-    def get_mesh_topology(self, raw=False):
-        """Retrurn mocked mesh data."""
-        return MOCK_MESH_DATA
+    get_mesh_topology = MagicMock()
+    get_mesh_topology.return_value = MOCK_MESH_DATA
 
-    def get_hosts_attributes(self):
-        """Retrurn mocked host attributes data."""
-        return MOCK_HOST_ATTRIBUTES_DATA
+    get_hosts_attributes = MagicMock()
+    get_hosts_attributes.return_value = MOCK_HOST_ATTRIBUTES_DATA
 
 
 @pytest.fixture(name="fc_data")

tests/components/fritz/const.py

@@ -27,7 +27,11 @@ MOCK_CONFIG = {
     }
 }
 MOCK_HOST = "fake_host"
-MOCK_IPS = {"fritz.box": "192.168.178.1", "printer": "192.168.178.2"}
+MOCK_IPS = {
+    "fritz.box": "192.168.178.1",
+    "printer": "192.168.178.2",
+    "server": "192.168.178.3",
+}
 MOCK_MODELNAME = "FRITZ!Box 7530 AX"
 MOCK_FIRMWARE = "256.07.29"
 MOCK_FIRMWARE_AVAILABLE = "7.50"
@@ -780,6 +784,45 @@ MOCK_MESH_DATA = {
     ],
 }
 
+MOCK_NEW_DEVICE_NODE = {
+    "uid": "n-900",
+    "device_name": "server",
+    "device_model": "",
+    "device_manufacturer": "",
+    "device_firmware_version": "",
+    "device_mac_address": "AA:BB:CC:33:44:55",
+    "is_meshed": False,
+    "mesh_role": "unknown",
+    "meshd_version": "0.0",
+    "node_interfaces": [
+        {
+            "uid": "ni-901",
+            "name": "eth0",
+            "type": "LAN",
+            "mac_address": "AA:BB:CC:33:44:55",
+            "blocking_state": "UNKNOWN",
+            "node_links": [
+                {
+                    "uid": "nl-902",
+                    "type": "LAN",
+                    "state": "CONNECTED",
+                    "last_connected": 1642872967,
+                    "node_1_uid": "n-1",
+                    "node_2_uid": "n-900",
+                    "node_interface_1_uid": "ni-31",
+                    "node_interface_2_uid": "ni-901",
+                    "max_data_rate_rx": 1000000,
+                    "max_data_rate_tx": 1000000,
+                    "cur_data_rate_rx": 0,
+                    "cur_data_rate_tx": 0,
+                    "cur_availability_rx": 99,
+                    "cur_availability_tx": 99,
+                }
+            ],
+        }
+    ],
+}
+
 MOCK_HOST_ATTRIBUTES_DATA = [
     {
         "Index": 1,
@@ -831,6 +874,31 @@ MOCK_HOST_ATTRIBUTES_DATA = [
         "X_AVM-DE_FriendlyName": "fritz.box",
         "X_AVM-DE_FriendlyNameIsWriteable": "0",
     },
+    {
+        "Index": 3,
+        "IPAddress": MOCK_IPS["server"],
+        "MACAddress": "AA:BB:CC:33:44:55",
+        "Active": True,
+        "HostName": "server",
+        "InterfaceType": "Ethernet",
+        "X_AVM-DE_Port": 1,
+        "X_AVM-DE_Speed": 1000,
+        "X_AVM-DE_UpdateAvailable": False,
+        "X_AVM-DE_UpdateSuccessful": "unknown",
+        "X_AVM-DE_InfoURL": None,
+        "X_AVM-DE_MACAddressList": None,
+        "X_AVM-DE_Model": None,
+        "X_AVM-DE_URL": f"http://{MOCK_IPS['server']}",
+        "X_AVM-DE_Guest": False,
+        "X_AVM-DE_RequestClient": "0",
+        "X_AVM-DE_VPN": False,
+        "X_AVM-DE_WANAccess": "granted",
+        "X_AVM-DE_Disallow": False,
+        "X_AVM-DE_IsMeshable": "0",
+        "X_AVM-DE_Priority": "0",
+        "X_AVM-DE_FriendlyName": "server",
+        "X_AVM-DE_FriendlyNameIsWriteable": "1",
+    },
 ]
 
 MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]

tests/components/fritz/test_button.py

@@ -1,18 +1,21 @@
 """Tests for Fritz!Tools button platform."""
 
+import copy
+from datetime import timedelta
 from unittest.mock import patch
 
 import pytest
 
 from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
-from homeassistant.components.fritz.const import DOMAIN
+from homeassistant.components.fritz.const import DOMAIN, MeshRoles
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
 from homeassistant.core import HomeAssistant
+from homeassistant.util.dt import utcnow
 
-from .const import MOCK_USER_DATA
+from .const import MOCK_MESH_DATA, MOCK_NEW_DEVICE_NODE, MOCK_USER_DATA
 
-from tests.common import MockConfigEntry
+from tests.common import MockConfigEntry, async_fire_time_changed
 
 
 async def test_button_setup(hass: HomeAssistant, fc_class_mock, fh_class_mock) -> None:
@@ -73,3 +76,113 @@ async def test_buttons(
 
         button = hass.states.get(entity_id)
         assert button.state != STATE_UNKNOWN
+
+
+async def test_wol_button(
+    hass: HomeAssistant,
+    entity_registry_enabled_by_default: None,
+    fc_class_mock,
+    fh_class_mock,
+) -> None:
+    """Test Fritz!Tools wake on LAN button."""
+    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()
+
+    assert entry.state == ConfigEntryState.LOADED
+
+    button = hass.states.get("button.printer_wake_on_lan")
+    assert button
+    assert button.state == STATE_UNKNOWN
+    with patch(
+        "homeassistant.components.fritz.common.AvmWrapper.async_wake_on_lan"
+    ) as mock_press_action:
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: "button.printer_wake_on_lan"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        mock_press_action.assert_called_once_with("AA:BB:CC:00:11:22")
+
+        button = hass.states.get("button.printer_wake_on_lan")
+        assert button.state != STATE_UNKNOWN
+
+
+async def test_wol_button_new_device(
+    hass: HomeAssistant,
+    entity_registry_enabled_by_default: None,
+    fc_class_mock,
+    fh_class_mock,
+) -> None:
+    """Test WoL button is created for new device at runtime."""
+    entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
+    entry.add_to_hass(hass)
+
+    mesh_data = copy.deepcopy(MOCK_MESH_DATA)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    assert entry.state == ConfigEntryState.LOADED
+
+    assert hass.states.get("button.printer_wake_on_lan")
+    assert not hass.states.get("button.server_wake_on_lan")
+
+    mesh_data["nodes"].append(MOCK_NEW_DEVICE_NODE)
+    fh_class_mock.get_mesh_topology.return_value = mesh_data
+
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
+    await hass.async_block_till_done(wait_background_tasks=True)
+
+    assert hass.states.get("button.printer_wake_on_lan")
+    assert hass.states.get("button.server_wake_on_lan")
+
+
+async def test_wol_button_absent_for_mesh_slave(
+    hass: HomeAssistant,
+    entity_registry_enabled_by_default: None,
+    fc_class_mock,
+    fh_class_mock,
+) -> None:
+    """Test WoL button not created if interviewed box is in slave mode."""
+    entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
+    entry.add_to_hass(hass)
+
+    slave_mesh_data = copy.deepcopy(MOCK_MESH_DATA)
+    slave_mesh_data["nodes"][0]["mesh_role"] = MeshRoles.SLAVE
+    fh_class_mock.get_mesh_topology.return_value = slave_mesh_data
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    assert entry.state == ConfigEntryState.LOADED
+
+    button = hass.states.get("button.printer_wake_on_lan")
+    assert button is None
+
+
+async def test_wol_button_absent_for_non_lan_device(
+    hass: HomeAssistant,
+    entity_registry_enabled_by_default: None,
+    fc_class_mock,
+    fh_class_mock,
+) -> None:
+    """Test WoL button not created if interviewed device is not connected via LAN."""
+    entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
+    entry.add_to_hass(hass)
+
+    printer_wifi_data = copy.deepcopy(MOCK_MESH_DATA)
+    # initialization logic uses the connection type of the `node_interface_1_uid` pair of the printer
+    # ni-230 is wifi interface of fritzbox
+    printer_node_interface = printer_wifi_data["nodes"][1]["node_interfaces"][0]
+    printer_node_interface["type"] = "WLAN"
+    printer_node_interface["node_links"][0]["node_interface_1_uid"] = "ni-230"
+    fh_class_mock.get_mesh_topology.return_value = printer_wifi_data
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    assert entry.state == ConfigEntryState.LOADED
+
+    button = hass.states.get("button.printer_wake_on_lan")
+    assert button is None

tests/components/fritz/test_switch.py

@@ -172,7 +172,7 @@ async def test_switch_setup(
     expected_wifi_names: list[str],
     fc_class_mock,
     fh_class_mock,
-):
+) -> None:
     """Test setup of Fritz!Tools switches."""
 
     entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)

tests/components/fritz/test_update.py

@@ -124,4 +124,4 @@ async def test_available_update_can_be_installed(
             {"entity_id": "update.mock_title_fritz_os"},
             blocking=True,
         )
-        assert mocked_update_call.assert_called_once
+        mocked_update_call.assert_called_once()