diff --git a/homeassistant/components/fritz/button.py b/homeassistant/components/fritz/button.py index d56350dd1d0..cfd0e09412d 100644 --- a/homeassistant/components/fritz/button.py +++ b/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) diff --git a/homeassistant/components/fritz/common.py b/homeassistant/components/fritz/common.py index 9d8bcd1ab3e..e4d5e92b742 100644 --- a/homeassistant/components/fritz/common.py +++ b/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]): diff --git a/homeassistant/components/fritz/const.py b/homeassistant/components/fritz/const.py index fb60eaef5f8..caa7d44c378 100644 --- a/homeassistant/components/fritz/const.py +++ b/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" diff --git a/tests/components/fritz/conftest.py b/tests/components/fritz/conftest.py index 2e26f67c1eb..e32ca55f65d 100644 --- a/tests/components/fritz/conftest.py +++ b/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") diff --git a/tests/components/fritz/const.py b/tests/components/fritz/const.py index 30c9f9be174..ce530e32964 100644 --- a/tests/components/fritz/const.py +++ b/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] diff --git a/tests/components/fritz/test_button.py b/tests/components/fritz/test_button.py index 106fb7f9bef..f6546296d44 100644 --- a/tests/components/fritz/test_button.py +++ b/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 diff --git a/tests/components/fritz/test_switch.py b/tests/components/fritz/test_switch.py index 722f16fa0de..91d2d42106b 100644 --- a/tests/components/fritz/test_switch.py +++ b/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) diff --git a/tests/components/fritz/test_update.py b/tests/components/fritz/test_update.py index 97c9cdec25d..991b67e6285 100644 --- a/tests/components/fritz/test_update.py +++ b/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()