home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-09-04 20:31:36 +02:00
Code Issues Packages Projects Releases Wiki Activity

Switch asuswrt http(s) library to asusrouter package (#150426)

Browse Source
This commit is contained in:
Yevhenii Vaskivskyi
2025-08-12 20:45:39 +02:00
committed by GitHub
parent 4d426c31f9
commit 9fdc632780
13 changed files with 414 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CODEOWNERS generated
View File

@@ -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

177
homeassistant/components/asuswrt/bridge.py
View File

@@ -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)

4
homeassistant/components/asuswrt/config_flow.py
View File

@@ -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,

56
homeassistant/components/asuswrt/helpers.py Normal file
View File

@@ -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

4
homeassistant/components/asuswrt/manifest.json
View File

@@ -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"]
}

6
homeassistant/components/asuswrt/router.py
View File

@@ -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(

6
requirements_all.txt generated
View File

@@ -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

6
requirements_test_all.txt generated
View File

@@ -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

21
tests/components/asuswrt/common.py
View File

@@ -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)

163
tests/components/asuswrt/conftest.py
View File

@@ -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

13
tests/components/asuswrt/test_config_flow.py
View File

@@ -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(

95
tests/components/asuswrt/test_helpers.py Normal file
View File

@@ -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

30
tests/components/asuswrt/test_sensor.py
View File

@@ -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)