From 9fdc63278097e4e327063fc98c542316087d1138 Mon Sep 17 00:00:00 2001 From: Yevhenii Vaskivskyi Date: Tue, 12 Aug 2025 20:45:39 +0200 Subject: [PATCH] Switch asuswrt http(s) library to asusrouter package (#150426) --- CODEOWNERS | 4 +- homeassistant/components/asuswrt/bridge.py | 177 +++++++++--------- .../components/asuswrt/config_flow.py | 4 +- homeassistant/components/asuswrt/helpers.py | 56 ++++++ .../components/asuswrt/manifest.json | 4 +- homeassistant/components/asuswrt/router.py | 6 +- requirements_all.txt | 6 +- requirements_test_all.txt | 6 +- tests/components/asuswrt/common.py | 21 ++- tests/components/asuswrt/conftest.py | 163 +++++++++++----- tests/components/asuswrt/test_config_flow.py | 13 +- tests/components/asuswrt/test_helpers.py | 95 ++++++++++ tests/components/asuswrt/test_sensor.py | 30 ++- 13 files changed, 414 insertions(+), 171 deletions(-) create mode 100644 homeassistant/components/asuswrt/helpers.py create mode 100644 tests/components/asuswrt/test_helpers.py diff --git a/CODEOWNERS b/CODEOWNERS index b9a8367ba3e..44c9d7d4547 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -156,8 +156,8 @@ build.json @home-assistant/supervisor /tests/components/assist_pipeline/ @balloob @synesthesiam /homeassistant/components/assist_satellite/ @home-assistant/core @synesthesiam /tests/components/assist_satellite/ @home-assistant/core @synesthesiam -/homeassistant/components/asuswrt/ @kennedyshead @ollo69 -/tests/components/asuswrt/ @kennedyshead @ollo69 +/homeassistant/components/asuswrt/ @kennedyshead @ollo69 @Vaskivskyi +/tests/components/asuswrt/ @kennedyshead @ollo69 @Vaskivskyi /homeassistant/components/atag/ @MatsNL /tests/components/atag/ @MatsNL /homeassistant/components/aten_pe/ @mtdcr diff --git a/homeassistant/components/asuswrt/bridge.py b/homeassistant/components/asuswrt/bridge.py index bc6f0fe6fd2..b5042d07b82 100644 --- a/homeassistant/components/asuswrt/bridge.py +++ b/homeassistant/components/asuswrt/bridge.py @@ -5,15 +5,16 @@ from __future__ import annotations from abc import ABC, abstractmethod from collections import namedtuple from collections.abc import Awaitable, Callable, Coroutine -from datetime import datetime import functools import logging from typing import Any, cast from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy from aiohttp import ClientSession -from pyasuswrt import AsusWrtError, AsusWrtHttp -from pyasuswrt.exceptions import AsusWrtNotAvailableInfoError +from asusrouter import AsusRouter, AsusRouterError +from asusrouter.modules.client import AsusClient +from asusrouter.modules.data import AsusData +from asusrouter.modules.homeassistant import convert_to_ha_data, convert_to_ha_sensors from homeassistant.const import ( CONF_HOST, @@ -41,14 +42,13 @@ from .const import ( PROTOCOL_HTTPS, PROTOCOL_TELNET, SENSORS_BYTES, - SENSORS_CPU, SENSORS_LOAD_AVG, SENSORS_MEMORY, SENSORS_RATES, - SENSORS_TEMPERATURES, SENSORS_TEMPERATURES_LEGACY, SENSORS_UPTIME, ) +from .helpers import clean_dict, translate_to_legacy SENSORS_TYPE_BYTES = "sensors_bytes" SENSORS_TYPE_COUNT = "sensors_count" @@ -310,16 +310,16 @@ class AsusWrtHttpBridge(AsusWrtBridge): def __init__(self, conf: dict[str, Any], session: ClientSession) -> None: """Initialize Bridge that use HTTP library.""" super().__init__(conf[CONF_HOST]) - self._api: AsusWrtHttp = self._get_api(conf, session) + self._api = self._get_api(conf, session) @staticmethod - def _get_api(conf: dict[str, Any], session: ClientSession) -> AsusWrtHttp: - """Get the AsusWrtHttp API.""" - return AsusWrtHttp( - conf[CONF_HOST], - conf[CONF_USERNAME], - conf.get(CONF_PASSWORD, ""), - use_https=conf[CONF_PROTOCOL] == PROTOCOL_HTTPS, + def _get_api(conf: dict[str, Any], session: ClientSession) -> AsusRouter: + """Get the AsusRouter API.""" + return AsusRouter( + hostname=conf[CONF_HOST], + username=conf[CONF_USERNAME], + password=conf.get(CONF_PASSWORD, ""), + use_ssl=conf[CONF_PROTOCOL] == PROTOCOL_HTTPS, port=conf.get(CONF_PORT), session=session, ) @@ -327,46 +327,90 @@ class AsusWrtHttpBridge(AsusWrtBridge): @property def is_connected(self) -> bool: """Get connected status.""" - return cast(bool, self._api.is_connected) + return self._api.connected async def async_connect(self) -> None: """Connect to the device.""" await self._api.async_connect() + # Collect the identity + _identity = await self._api.async_get_identity() + # get main router properties - if mac := self._api.mac: + if mac := _identity.mac: self._label_mac = format_mac(mac) - self._firmware = self._api.firmware - self._model = self._api.model + self._firmware = str(_identity.firmware) + self._model = _identity.model async def async_disconnect(self) -> None: """Disconnect to the device.""" await self._api.async_disconnect() + async def _get_data( + self, + datatype: AsusData, + force: bool = False, + ) -> dict[str, Any]: + """Get data from the device. + + This is a generic method which automatically converts to + the Home Assistant-compatible format. + """ + try: + raw = await self._api.async_get_data(datatype, force=force) + return translate_to_legacy(clean_dict(convert_to_ha_data(raw))) + except AsusRouterError as ex: + raise UpdateFailed(ex) from ex + + async def _get_sensors(self, datatype: AsusData) -> list[str]: + """Get the available sensors. + + This is a generic method which automatically converts to + the Home Assistant-compatible format. + """ + sensors = [] + try: + data = await self._api.async_get_data(datatype) + # Get the list of sensors from the raw data + # and translate in to the legacy format + sensors = translate_to_legacy(convert_to_ha_sensors(data, datatype)) + _LOGGER.debug("Available `%s` sensors: %s", datatype.value, sensors) + except AsusRouterError as ex: + _LOGGER.warning( + "Cannot get available `%s` sensors with exception: %s", + datatype.value, + ex, + ) + return sensors + async def async_get_connected_devices(self) -> dict[str, WrtDevice]: """Get list of connected devices.""" - api_devices = await self._api.async_get_connected_devices() + api_devices: dict[str, AsusClient] = await self._api.async_get_data( + AsusData.CLIENTS, force=True + ) return { - format_mac(mac): WrtDevice(dev.ip, dev.name, dev.node) + format_mac(mac): WrtDevice( + dev.connection.ip_address, dev.description.name, dev.connection.node + ) for mac, dev in api_devices.items() + if dev.connection is not None + and dev.description is not None + and dev.connection.ip_address is not None } async def async_get_available_sensors(self) -> dict[str, dict[str, Any]]: """Return a dictionary of available sensors for this bridge.""" - sensors_cpu = await self._get_available_cpu_sensors() - sensors_temperatures = await self._get_available_temperature_sensors() - sensors_loadavg = await self._get_loadavg_sensors_availability() return { SENSORS_TYPE_BYTES: { KEY_SENSORS: SENSORS_BYTES, KEY_METHOD: self._get_bytes, }, SENSORS_TYPE_CPU: { - KEY_SENSORS: sensors_cpu, + KEY_SENSORS: await self._get_sensors(AsusData.CPU), KEY_METHOD: self._get_cpu_usage, }, SENSORS_TYPE_LOAD_AVG: { - KEY_SENSORS: sensors_loadavg, + KEY_SENSORS: await self._get_sensors(AsusData.SYSINFO), KEY_METHOD: self._get_load_avg, }, SENSORS_TYPE_MEMORY: { @@ -382,95 +426,44 @@ class AsusWrtHttpBridge(AsusWrtBridge): KEY_METHOD: self._get_uptime, }, SENSORS_TYPE_TEMPERATURES: { - KEY_SENSORS: sensors_temperatures, + KEY_SENSORS: await self._get_sensors(AsusData.TEMPERATURE), KEY_METHOD: self._get_temperatures, }, } - async def _get_available_cpu_sensors(self) -> list[str]: - """Check which cpu information is available on the router.""" - try: - available_cpu = await self._api.async_get_cpu_usage() - available_sensors = [t for t in SENSORS_CPU if t in available_cpu] - except AsusWrtError as exc: - _LOGGER.warning( - ( - "Failed checking cpu sensor availability for ASUS router" - " %s. Exception: %s" - ), - self.host, - exc, - ) - return [] - return available_sensors - - async def _get_available_temperature_sensors(self) -> list[str]: - """Check which temperature information is available on the router.""" - try: - available_temps = await self._api.async_get_temperatures() - available_sensors = [ - t for t in SENSORS_TEMPERATURES if t in available_temps - ] - except AsusWrtError as exc: - _LOGGER.warning( - ( - "Failed checking temperature sensor availability for ASUS router" - " %s. Exception: %s" - ), - self.host, - exc, - ) - return [] - return available_sensors - - async def _get_loadavg_sensors_availability(self) -> list[str]: - """Check if load avg is available on the router.""" - try: - await self._api.async_get_loadavg() - except AsusWrtNotAvailableInfoError: - return [] - except AsusWrtError: - pass - return SENSORS_LOAD_AVG - - @handle_errors_and_zip(AsusWrtError, SENSORS_BYTES) async def _get_bytes(self) -> Any: """Fetch byte information from the router.""" - return await self._api.async_get_traffic_bytes() + return await self._get_data(AsusData.NETWORK) - @handle_errors_and_zip(AsusWrtError, SENSORS_RATES) async def _get_rates(self) -> Any: """Fetch rates information from the router.""" - return await self._api.async_get_traffic_rates() + data = await self._get_data(AsusData.NETWORK) + # Convert from bits/s to Bytes/s for compatibility with legacy sensors + return { + key: ( + value / 8 + if key in SENSORS_RATES and isinstance(value, (int, float)) + else value + ) + for key, value in data.items() + } - @handle_errors_and_zip(AsusWrtError, SENSORS_LOAD_AVG) async def _get_load_avg(self) -> Any: """Fetch cpu load avg information from the router.""" - return await self._api.async_get_loadavg() + return await self._get_data(AsusData.SYSINFO) - @handle_errors_and_zip(AsusWrtError, None) async def _get_temperatures(self) -> Any: """Fetch temperatures information from the router.""" - return await self._api.async_get_temperatures() + return await self._get_data(AsusData.TEMPERATURE) - @handle_errors_and_zip(AsusWrtError, None) async def _get_cpu_usage(self) -> Any: """Fetch cpu information from the router.""" - return await self._api.async_get_cpu_usage() + return await self._get_data(AsusData.CPU) - @handle_errors_and_zip(AsusWrtError, None) async def _get_memory_usage(self) -> Any: """Fetch memory information from the router.""" - return await self._api.async_get_memory_usage() + return await self._get_data(AsusData.RAM) async def _get_uptime(self) -> dict[str, Any]: """Fetch uptime from the router.""" - try: - uptimes = await self._api.async_get_uptime() - except AsusWrtError as exc: - raise UpdateFailed(exc) from exc - - last_boot = datetime.fromisoformat(uptimes["last_boot"]) - uptime = uptimes["uptime"] - - return dict(zip(SENSORS_UPTIME, [last_boot, uptime], strict=False)) + return await self._get_data(AsusData.BOOTTIME) diff --git a/homeassistant/components/asuswrt/config_flow.py b/homeassistant/components/asuswrt/config_flow.py index d58a216aaee..a86f7e08318 100644 --- a/homeassistant/components/asuswrt/config_flow.py +++ b/homeassistant/components/asuswrt/config_flow.py @@ -7,7 +7,7 @@ import os import socket from typing import Any, cast -from pyasuswrt import AsusWrtError +from asusrouter import AsusRouterError import voluptuous as vol from homeassistant.components.device_tracker import ( @@ -189,7 +189,7 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN): try: await api.async_connect() - except (AsusWrtError, OSError): + except (AsusRouterError, OSError): _LOGGER.error( "Error connecting to the AsusWrt router at %s using protocol %s", host, diff --git a/homeassistant/components/asuswrt/helpers.py b/homeassistant/components/asuswrt/helpers.py new file mode 100644 index 00000000000..0fb467e6046 --- /dev/null +++ b/homeassistant/components/asuswrt/helpers.py @@ -0,0 +1,56 @@ +"""Helpers for AsusWRT integration.""" + +from __future__ import annotations + +from typing import Any, TypeVar + +T = TypeVar("T", dict[str, Any], list[Any], None) + +TRANSLATION_MAP = { + "wan_rx": "sensor_rx_bytes", + "wan_tx": "sensor_tx_bytes", + "total_usage": "cpu_total_usage", + "usage": "mem_usage_perc", + "free": "mem_free", + "used": "mem_used", + "wan_rx_speed": "sensor_rx_rates", + "wan_tx_speed": "sensor_tx_rates", + "2ghz": "2.4GHz", + "5ghz": "5.0GHz", + "5ghz2": "5.0GHz_2", + "6ghz": "6.0GHz", + "cpu": "CPU", + "datetime": "sensor_last_boot", + "uptime": "sensor_uptime", + **{f"{num}_usage": f"cpu{num}_usage" for num in range(1, 9)}, + **{f"load_avg_{load}": f"sensor_load_avg{load}" for load in ("1", "5", "15")}, +} + + +def clean_dict(raw: dict[str, Any]) -> dict[str, Any]: + """Cleans dictionary from None values. + + The `state` key is always preserved regardless of its value. + """ + + return {k: v for k, v in raw.items() if v is not None or k.endswith("state")} + + +def translate_to_legacy(raw: T) -> T: + """Translate raw data to legacy format for dicts and lists.""" + + if raw is None: + return None + + if isinstance(raw, dict): + return {TRANSLATION_MAP.get(k, k): v for k, v in raw.items()} + + if isinstance(raw, list): + return [ + TRANSLATION_MAP[item] + if isinstance(item, str) and item in TRANSLATION_MAP + else item + for item in raw + ] + + return raw diff --git a/homeassistant/components/asuswrt/manifest.json b/homeassistant/components/asuswrt/manifest.json index f4b2e3386e9..5064642619c 100644 --- a/homeassistant/components/asuswrt/manifest.json +++ b/homeassistant/components/asuswrt/manifest.json @@ -1,11 +1,11 @@ { "domain": "asuswrt", "name": "ASUSWRT", - "codeowners": ["@kennedyshead", "@ollo69"], + "codeowners": ["@kennedyshead", "@ollo69", "@Vaskivskyi"], "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/asuswrt", "integration_type": "hub", "iot_class": "local_polling", "loggers": ["aioasuswrt", "asyncssh"], - "requirements": ["aioasuswrt==1.4.0", "pyasuswrt==0.1.21"] + "requirements": ["aioasuswrt==1.4.0", "asusrouter==1.18.1"] } diff --git a/homeassistant/components/asuswrt/router.py b/homeassistant/components/asuswrt/router.py index 3cf8d2e863d..c777535e242 100644 --- a/homeassistant/components/asuswrt/router.py +++ b/homeassistant/components/asuswrt/router.py @@ -7,7 +7,7 @@ from datetime import datetime, timedelta import logging from typing import TYPE_CHECKING, Any -from pyasuswrt import AsusWrtError +from asusrouter import AsusRouterError from homeassistant.components.device_tracker import ( CONF_CONSIDER_HOME, @@ -229,7 +229,7 @@ class AsusWrtRouter: """Set up a AsusWrt router.""" try: await self._api.async_connect() - except (AsusWrtError, OSError) as exc: + except (AsusRouterError, OSError) as exc: raise ConfigEntryNotReady from exc if not self._api.is_connected: raise ConfigEntryNotReady @@ -284,7 +284,7 @@ class AsusWrtRouter: _LOGGER.debug("Checking devices for ASUS router %s", self.host) try: wrt_devices = await self._api.async_get_connected_devices() - except (OSError, AsusWrtError) as exc: + except (OSError, AsusRouterError) as exc: if not self._connect_error: self._connect_error = True _LOGGER.error( diff --git a/requirements_all.txt b/requirements_all.txt index bbc398d77ff..a79b4450f5d 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -527,6 +527,9 @@ arris-tg2492lg==2.2.0 # homeassistant.components.ampio asmog==0.0.6 +# homeassistant.components.asuswrt +asusrouter==1.18.1 + # homeassistant.components.dlna_dmr # homeassistant.components.dlna_dms # homeassistant.components.samsungtv @@ -1839,9 +1842,6 @@ pyairvisual==2023.08.1 # homeassistant.components.aprilaire pyaprilaire==0.9.1 -# homeassistant.components.asuswrt -pyasuswrt==0.1.21 - # homeassistant.components.atag pyatag==0.3.5.3 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 0554d7c0a08..b0afbc75649 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -491,6 +491,9 @@ aranet4==2.5.1 # homeassistant.components.arcam_fmj arcam-fmj==1.8.2 +# homeassistant.components.asuswrt +asusrouter==1.18.1 + # homeassistant.components.dlna_dmr # homeassistant.components.dlna_dms # homeassistant.components.samsungtv @@ -1544,9 +1547,6 @@ pyairvisual==2023.08.1 # homeassistant.components.aprilaire pyaprilaire==0.9.1 -# homeassistant.components.asuswrt -pyasuswrt==0.1.21 - # homeassistant.components.atag pyatag==0.3.5.3 diff --git a/tests/components/asuswrt/common.py b/tests/components/asuswrt/common.py index d3953416281..541e74e5b39 100644 --- a/tests/components/asuswrt/common.py +++ b/tests/components/asuswrt/common.py @@ -1,7 +1,8 @@ """Test code shared between test files.""" +from unittest.mock import MagicMock + from aioasuswrt.asuswrt import Device as LegacyDevice -from pyasuswrt.asuswrt import Device as HttpDevice from homeassistant.components.asuswrt.const import ( CONF_SSH_KEY, @@ -59,8 +60,22 @@ MOCK_MACS = [ ] -def new_device(protocol, mac, ip, name): +def make_client(mac, ip, name, node): + """Create a modern mock client.""" + connection = MagicMock() + connection.ip_address = ip + connection.node = node + description = MagicMock() + description.name = name + description.mac = mac + client = MagicMock() + client.connection = connection + client.description = description + return client + + +def new_device(protocol, mac, ip, name, node=None): """Return a new device for specific protocol.""" if protocol in [PROTOCOL_HTTP, PROTOCOL_HTTPS]: - return HttpDevice(mac, ip, name, ROUTER_MAC_ADDR, None) + return make_client(mac, ip, name, node) return LegacyDevice(mac, ip, name) diff --git a/tests/components/asuswrt/conftest.py b/tests/components/asuswrt/conftest.py index f850a26b997..e6dd42a23fd 100644 --- a/tests/components/asuswrt/conftest.py +++ b/tests/components/asuswrt/conftest.py @@ -1,17 +1,25 @@ """Fixtures for Asuswrt component.""" +from datetime import datetime from unittest.mock import Mock, patch from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy from aioasuswrt.connection import TelnetConnection -from pyasuswrt.asuswrt import AsusWrtError, AsusWrtHttp +from asusrouter import AsusRouter, AsusRouterError +from asusrouter.modules.data import AsusData +from asusrouter.modules.identity import AsusDevice import pytest -from homeassistant.components.asuswrt.const import PROTOCOL_HTTP, PROTOCOL_SSH +from .common import ( + ASUSWRT_BASE, + MOCK_MACS, + PROTOCOL_HTTP, + PROTOCOL_SSH, + ROUTER_MAC_ADDR, + new_device, +) -from .common import ASUSWRT_BASE, MOCK_MACS, ROUTER_MAC_ADDR, new_device - -ASUSWRT_HTTP_LIB = f"{ASUSWRT_BASE}.bridge.AsusWrtHttp" +ASUSWRT_HTTP_LIB = f"{ASUSWRT_BASE}.bridge.AsusRouter" ASUSWRT_LEGACY_LIB = f"{ASUSWRT_BASE}.bridge.AsusWrtLegacy" MOCK_BYTES_TOTAL = 60000000000, 50000000000 @@ -29,8 +37,20 @@ MOCK_CPU_USAGE = { } MOCK_CURRENT_TRANSFER_RATES = 20000000, 10000000 MOCK_CURRENT_TRANSFER_RATES_HTTP = dict(enumerate(MOCK_CURRENT_TRANSFER_RATES)) +# Mock for AsusData.NETWORK return of both rates and total bytes +MOCK_CURRENT_NETWORK = { + "sensor_rx_rates": MOCK_CURRENT_TRANSFER_RATES[0] * 8, # AR works with bits + "sensor_tx_rates": MOCK_CURRENT_TRANSFER_RATES[1] * 8, # AR works with bits + "sensor_rx_bytes": MOCK_BYTES_TOTAL[0], + "sensor_tx_bytes": MOCK_BYTES_TOTAL[1], +} MOCK_LOAD_AVG_HTTP = {"load_avg_1": 1.1, "load_avg_5": 1.2, "load_avg_15": 1.3} MOCK_LOAD_AVG = list(MOCK_LOAD_AVG_HTTP.values()) +MOCK_SYSINFO = { + "sensor_load_avg1": MOCK_LOAD_AVG[0], + "sensor_load_avg5": MOCK_LOAD_AVG[1], + "sensor_load_avg15": MOCK_LOAD_AVG[2], +} MOCK_MEMORY_USAGE = { "mem_usage_perc": 52.4, "mem_total": 1048576, @@ -40,6 +60,10 @@ MOCK_MEMORY_USAGE = { MOCK_TEMPERATURES = {"2.4GHz": 40.2, "5.0GHz": 0, "CPU": 71.2} MOCK_TEMPERATURES_HTTP = {**MOCK_TEMPERATURES, "5.0GHz_2": 40.3, "6.0GHz": 40.4} MOCK_UPTIME = {"last_boot": "2024-08-02T00:47:00+00:00", "uptime": 1625927} +MOCK_BOOTTIME = { + "sensor_last_boot": datetime.fromisoformat(MOCK_UPTIME["last_boot"]), + "sensor_uptime": MOCK_UPTIME["uptime"], +} @pytest.fixture(name="patch_setup_entry") @@ -62,10 +86,14 @@ def mock_devices_legacy_fixture(): @pytest.fixture(name="mock_devices_http") def mock_devices_http_fixture(): - """Mock a list of devices.""" + """Mock a list of AsusRouter client devices for HTTP backend.""" return { - MOCK_MACS[0]: new_device(PROTOCOL_HTTP, MOCK_MACS[0], "192.168.1.2", "Test"), - MOCK_MACS[1]: new_device(PROTOCOL_HTTP, MOCK_MACS[1], "192.168.1.3", "TestTwo"), + MOCK_MACS[0]: new_device( + PROTOCOL_HTTP, MOCK_MACS[0], "192.168.1.2", "Test", "node1" + ), + MOCK_MACS[1]: new_device( + PROTOCOL_HTTP, MOCK_MACS[1], "192.168.1.3", "TestTwo", "node2" + ), } @@ -121,57 +149,90 @@ def mock_controller_connect_legacy_sens_fail(connect_legacy): @pytest.fixture(name="connect_http") def mock_controller_connect_http(mock_devices_http): """Mock a successful connection with http library.""" - with patch(ASUSWRT_HTTP_LIB, spec_set=AsusWrtHttp) as service_mock: - service_mock.return_value.is_connected = True - service_mock.return_value.mac = ROUTER_MAC_ADDR - service_mock.return_value.model = "FAKE_MODEL" - service_mock.return_value.firmware = "FAKE_FIRMWARE" - service_mock.return_value.async_get_connected_devices.return_value = ( - mock_devices_http + with patch(ASUSWRT_HTTP_LIB, spec_set=AsusRouter) as service_mock: + instance = service_mock.return_value + + # Simulate connection status + instance.connected = True + + # Identity + instance.async_get_identity.return_value = AsusDevice( + mac=ROUTER_MAC_ADDR, + model="FAKE_MODEL", + firmware="FAKE_FIRMWARE", ) - service_mock.return_value.async_get_traffic_bytes.return_value = ( - MOCK_BYTES_TOTAL_HTTP - ) - service_mock.return_value.async_get_traffic_rates.return_value = ( - MOCK_CURRENT_TRANSFER_RATES_HTTP - ) - service_mock.return_value.async_get_loadavg.return_value = MOCK_LOAD_AVG_HTTP - service_mock.return_value.async_get_temperatures.return_value = { - k: v for k, v in MOCK_TEMPERATURES_HTTP.items() if k != "5.0GHz" - } - service_mock.return_value.async_get_cpu_usage.return_value = MOCK_CPU_USAGE - service_mock.return_value.async_get_memory_usage.return_value = ( - MOCK_MEMORY_USAGE - ) - service_mock.return_value.async_get_uptime.return_value = MOCK_UPTIME + + # Data fetches via async_get_data + instance.async_get_data.side_effect = lambda datatype, *args, **kwargs: { + AsusData.CLIENTS: mock_devices_http, + AsusData.NETWORK: MOCK_CURRENT_NETWORK, + AsusData.SYSINFO: MOCK_SYSINFO, + AsusData.TEMPERATURE: { + k: v for k, v in MOCK_TEMPERATURES_HTTP.items() if k != "5.0GHz" + }, + AsusData.CPU: MOCK_CPU_USAGE, + AsusData.RAM: MOCK_MEMORY_USAGE, + AsusData.BOOTTIME: MOCK_BOOTTIME, + }[datatype] + yield service_mock +def make_async_get_data_side_effect(fail_types=None): + """Return a side effect for async_get_data that fails for specified AsusData types.""" + fail_types = set(fail_types or []) + + def side_effect(datatype, *args, **kwargs): + if datatype in fail_types: + raise AsusRouterError(f"{datatype} unavailable") + # Return valid mock data for other types + if datatype == AsusData.CLIENTS: + return {} + if datatype == AsusData.NETWORK: + return {} + if datatype == AsusData.SYSINFO: + return {} + if datatype == AsusData.TEMPERATURE: + return {} + if datatype == AsusData.CPU: + return {} + if datatype == AsusData.RAM: + return {} + if datatype == AsusData.BOOTTIME: + return {} + return {} + + return side_effect + + @pytest.fixture(name="connect_http_sens_fail") def mock_controller_connect_http_sens_fail(connect_http): - """Mock a successful connection using http library with sensors fail.""" - connect_http.return_value.mac = None - connect_http.return_value.async_get_connected_devices.side_effect = AsusWrtError - connect_http.return_value.async_get_traffic_bytes.side_effect = AsusWrtError - connect_http.return_value.async_get_traffic_rates.side_effect = AsusWrtError - connect_http.return_value.async_get_loadavg.side_effect = AsusWrtError - connect_http.return_value.async_get_temperatures.side_effect = AsusWrtError - connect_http.return_value.async_get_cpu_usage.side_effect = AsusWrtError - connect_http.return_value.async_get_memory_usage.side_effect = AsusWrtError - connect_http.return_value.async_get_uptime.side_effect = AsusWrtError + """Universal fixture to fail specified AsusData types.""" + + def _set_fail_types(fail_types): + connect_http.return_value.async_get_data.side_effect = ( + make_async_get_data_side_effect(fail_types) + ) + return connect_http + + return _set_fail_types @pytest.fixture(name="connect_http_sens_detect") def mock_controller_connect_http_sens_detect(): """Mock a successful sensor detection using http library.""" - with ( - patch( - f"{ASUSWRT_BASE}.bridge.AsusWrtHttpBridge._get_available_temperature_sensors", - return_value=[*MOCK_TEMPERATURES_HTTP], - ) as mock_sens_temp_detect, - patch( - f"{ASUSWRT_BASE}.bridge.AsusWrtHttpBridge._get_available_cpu_sensors", - return_value=[*MOCK_CPU_USAGE], - ) as mock_sens_cpu_detect, - ): - yield mock_sens_temp_detect, mock_sens_cpu_detect + + def _get_sensors_side_effect(datatype): + if datatype == AsusData.TEMPERATURE: + return list(MOCK_TEMPERATURES_HTTP) + if datatype == AsusData.CPU: + return list(MOCK_CPU_USAGE) + if datatype == AsusData.SYSINFO: + return list(MOCK_SYSINFO) + return [] + + with patch( + f"{ASUSWRT_BASE}.bridge.AsusWrtHttpBridge._get_sensors", + side_effect=_get_sensors_side_effect, + ) as mock_sens_detect: + yield mock_sens_detect diff --git a/tests/components/asuswrt/test_config_flow.py b/tests/components/asuswrt/test_config_flow.py index 83c3204d239..314bf030dbc 100644 --- a/tests/components/asuswrt/test_config_flow.py +++ b/tests/components/asuswrt/test_config_flow.py @@ -3,7 +3,8 @@ from socket import gaierror from unittest.mock import patch -from pyasuswrt import AsusWrtError +from asusrouter import AsusRouterError +from asusrouter.modules.identity import AsusDevice import pytest from homeassistant.components.asuswrt.const import ( @@ -128,7 +129,11 @@ async def test_user_http( assert flow_result["type"] is FlowResultType.FORM assert flow_result["step_id"] == "user" - connect_http.return_value.mac = unique_id + connect_http.return_value.async_get_identity.return_value = AsusDevice( + mac=unique_id, + model="FAKE_MODEL", + firmware="FAKE_FIRMWARE", + ) # test with all provided result = await hass.config_entries.flow.async_configure( @@ -297,7 +302,7 @@ async def test_on_connect_legacy_failed( @pytest.mark.parametrize( ("side_effect", "error"), [ - (AsusWrtError, "cannot_connect"), + (AsusRouterError, "cannot_connect"), (TypeError, "unknown"), (None, "cannot_connect"), ], @@ -311,7 +316,7 @@ async def test_on_connect_http_failed( context={"source": SOURCE_USER, "show_advanced_options": True}, ) - connect_http.return_value.is_connected = False + connect_http.return_value.connected = False connect_http.return_value.async_connect.side_effect = side_effect result = await hass.config_entries.flow.async_configure( diff --git a/tests/components/asuswrt/test_helpers.py b/tests/components/asuswrt/test_helpers.py new file mode 100644 index 00000000000..6573ab9361c --- /dev/null +++ b/tests/components/asuswrt/test_helpers.py @@ -0,0 +1,95 @@ +"""Tests for AsusWRT helpers.""" + +from typing import Any + +import pytest + +from homeassistant.components.asuswrt.helpers import clean_dict, translate_to_legacy + +DICT_TO_CLEAN = { + "key1": "value1", + "key2": None, + "key3_state": "value3", + "key4_state": None, + "state": None, +} + +DICT_CLEAN = { + "key1": "value1", + "key3_state": "value3", + "key4_state": None, + "state": None, +} + +TRANSLATE_0_INPUT = { + "usage": "value1", + "cpu": "value2", +} + +TRANSLATE_0_OUTPUT = { + "mem_usage_perc": "value1", + "CPU": "value2", +} + +TRANSLATE_1_INPUT = { + "wan_rx": "value1", + "wan_rrx": "value2", +} + +TRANSLATE_1_OUTPUT = { + "sensor_rx_bytes": "value1", + "wan_rrx": "value2", +} + +TRANSLATE_2_INPUT = [ + "free", + "used", +] + +TRANSLATE_2_OUTPUT = [ + "mem_free", + "mem_used", +] + +TRANSLATE_3_INPUT = [ + "2ghz", + "2ghz2", +] + +TRANSLATE_3_OUTPUT = [ + "2.4GHz", + "2ghz2", +] + + +def test_clean_dict() -> None: + """Test clean_dict method.""" + + assert clean_dict(DICT_TO_CLEAN) == DICT_CLEAN + + +@pytest.mark.parametrize( + ("input", "expected"), + [ + # Case set 0: None as input -> None on output + (None, None), + # Case set 1: Dict structure should stay intact or translated + ({"key1": "value1", "key2": None}, {"key1": "value1", "key2": None}), + (TRANSLATE_0_INPUT, TRANSLATE_0_OUTPUT), + (TRANSLATE_1_INPUT, TRANSLATE_1_OUTPUT), + ({}, {}), + # Case set 2: List structure should stay intact or translated + (["key1", "key2"], ["key1", "key2"]), + (TRANSLATE_2_INPUT, TRANSLATE_2_OUTPUT), + (TRANSLATE_3_INPUT, TRANSLATE_3_OUTPUT), + ([], []), + # Case set 3: Anything else should be simply returned + (123, 123), + ("string", "string"), + (3.1415926535, 3.1415926535), + ], +) +def test_translate(input: Any, expected: Any) -> None: + """Test translate method.""" + + assert translate_to_legacy(input) == expected diff --git a/tests/components/asuswrt/test_sensor.py b/tests/components/asuswrt/test_sensor.py index 929500f0bb7..3ce3246c1d6 100644 --- a/tests/components/asuswrt/test_sensor.py +++ b/tests/components/asuswrt/test_sensor.py @@ -2,8 +2,9 @@ from datetime import timedelta +from asusrouter import AsusRouterError +from asusrouter.modules.data import AsusData from freezegun.api import FrozenDateTimeFactory -from pyasuswrt.exceptions import AsusWrtError, AsusWrtNotAvailableInfoError import pytest from homeassistant.components import device_tracker, sensor @@ -39,6 +40,7 @@ from .common import ( ROUTER_MAC_ADDR, new_device, ) +from .conftest import make_async_get_data_side_effect from tests.common import MockConfigEntry, async_fire_time_changed @@ -260,8 +262,8 @@ async def test_loadavg_sensors_unaivalable_http( config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_LOAD_AVG) config_entry.add_to_hass(hass) - connect_http.return_value.async_get_loadavg.side_effect = ( - AsusWrtNotAvailableInfoError + connect_http.return_value.async_get_data.side_effect = ( + make_async_get_data_side_effect([AsusData.SYSINFO]) ) # initial devices setup @@ -281,6 +283,7 @@ async def test_temperature_sensors_http_fail( hass: HomeAssistant, connect_http_sens_fail ) -> None: """Test fail creating AsusWRT temperature sensors.""" + _ = connect_http_sens_fail([AsusData.TEMPERATURE]) config_entry, sensor_prefix = _setup_entry( hass, CONFIG_DATA_HTTP, SENSORS_TEMPERATURES ) @@ -347,6 +350,7 @@ async def test_cpu_sensors_http_fail( hass: HomeAssistant, connect_http_sens_fail ) -> None: """Test fail creating AsusWRT cpu sensors.""" + _ = connect_http_sens_fail([AsusData.CPU]) config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_CPU) config_entry.add_to_hass(hass) @@ -367,9 +371,13 @@ async def test_cpu_sensors_http_fail( async def test_cpu_sensors_http( - hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http + hass: HomeAssistant, + freezer: FrozenDateTimeFactory, + connect_http, + connect_http_sens_detect, ) -> None: """Test creating AsusWRT cpu sensors.""" + connect_http_sens_detect(AsusData.CPU) config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_CPU) config_entry.add_to_hass(hass) @@ -461,7 +469,7 @@ async def test_connect_fail_legacy( @pytest.mark.parametrize( "side_effect", - [AsusWrtError, None], + [AsusRouterError, None], ) async def test_connect_fail_http( hass: HomeAssistant, connect_http, side_effect @@ -476,7 +484,7 @@ async def test_connect_fail_http( config_entry.add_to_hass(hass) connect_http.return_value.async_connect.side_effect = side_effect - connect_http.return_value.is_connected = False + connect_http.return_value.connected = False # initial setup fail await hass.config_entries.async_setup(config_entry.entry_id) @@ -524,6 +532,16 @@ async def test_sensors_polling_fails_http( connect_http_sens_detect, ) -> None: """Test AsusWRT sensors are unavailable when polling fails.""" + # Fail all relevant AsusData types for HTTP sensors + fail_types = [ + AsusData.NETWORK, + AsusData.CPU, + AsusData.SYSINFO, + AsusData.RAM, + AsusData.TEMPERATURE, + AsusData.BOOTTIME, + ] + _ = connect_http_sens_fail(fail_types) await _test_sensors_polling_fails(hass, freezer, CONFIG_DATA_HTTP, SENSORS_ALL_HTTP)