Add sensors for AsusWRT using http(s) library (#124337)

* Additional sensors for AsusWRT using http(s) library

* Remove temperature sensors refactor from PR

* Fix test function name

* Change translation a suggested

* Requested changes
This commit is contained in:
ollo69 2024-09-03 17:11:17 +02:00 committed by GitHub
parent 56887747a6
commit 470335e27a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 391 additions and 38 deletions

View File

@ -5,6 +5,7 @@ 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
@ -40,17 +41,23 @@ 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,
)
SENSORS_TYPE_BYTES = "sensors_bytes"
SENSORS_TYPE_COUNT = "sensors_count"
SENSORS_TYPE_CPU = "sensors_cpu"
SENSORS_TYPE_LOAD_AVG = "sensors_load_avg"
SENSORS_TYPE_MEMORY = "sensors_memory"
SENSORS_TYPE_RATES = "sensors_rates"
SENSORS_TYPE_TEMPERATURES = "sensors_temperatures"
SENSORS_TYPE_UPTIME = "sensors_uptime"
WrtDevice = namedtuple("WrtDevice", ["ip", "name", "connected_to"]) # noqa: PYI024
@ -346,6 +353,7 @@ class AsusWrtHttpBridge(AsusWrtBridge):
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 {
@ -353,20 +361,49 @@ class AsusWrtHttpBridge(AsusWrtBridge):
KEY_SENSORS: SENSORS_BYTES,
KEY_METHOD: self._get_bytes,
},
SENSORS_TYPE_CPU: {
KEY_SENSORS: sensors_cpu,
KEY_METHOD: self._get_cpu_usage,
},
SENSORS_TYPE_LOAD_AVG: {
KEY_SENSORS: sensors_loadavg,
KEY_METHOD: self._get_load_avg,
},
SENSORS_TYPE_MEMORY: {
KEY_SENSORS: SENSORS_MEMORY,
KEY_METHOD: self._get_memory_usage,
},
SENSORS_TYPE_RATES: {
KEY_SENSORS: SENSORS_RATES,
KEY_METHOD: self._get_rates,
},
SENSORS_TYPE_UPTIME: {
KEY_SENSORS: SENSORS_UPTIME,
KEY_METHOD: self._get_uptime,
},
SENSORS_TYPE_TEMPERATURES: {
KEY_SENSORS: sensors_temperatures,
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:
@ -415,3 +452,25 @@ class AsusWrtHttpBridge(AsusWrtBridge):
async def _get_temperatures(self) -> Any:
"""Fetch temperatures information from the router."""
return await self._api.async_get_temperatures()
@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()
@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()
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))

View File

@ -27,7 +27,20 @@ PROTOCOL_TELNET = "telnet"
# Sensors
SENSORS_BYTES = ["sensor_rx_bytes", "sensor_tx_bytes"]
SENSORS_CONNECTED_DEVICE = ["sensor_connected_device"]
SENSORS_CPU = [
"cpu_total_usage",
"cpu1_usage",
"cpu2_usage",
"cpu3_usage",
"cpu4_usage",
"cpu5_usage",
"cpu6_usage",
"cpu7_usage",
"cpu8_usage",
]
SENSORS_LOAD_AVG = ["sensor_load_avg1", "sensor_load_avg5", "sensor_load_avg15"]
SENSORS_MEMORY = ["mem_usage_perc", "mem_free", "mem_used"]
SENSORS_RATES = ["sensor_rx_rates", "sensor_tx_rates"]
SENSORS_TEMPERATURES_LEGACY = ["2.4GHz", "5.0GHz", "CPU"]
SENSORS_TEMPERATURES = [*SENSORS_TEMPERATURES_LEGACY, "5.0GHz_2", "6.0GHz"]
SENSORS_UPTIME = ["sensor_last_boot", "sensor_uptime"]

View File

@ -24,6 +24,21 @@
},
"load_avg_15m": {
"default": "mdi:cpu-32-bit"
},
"cpu_usage": {
"default": "mdi:cpu-32-bit"
},
"cpu_core_usage": {
"default": "mdi:cpu-32-bit"
},
"memory_usage": {
"default": "mdi:memory"
},
"memory_free": {
"default": "mdi:memory"
},
"memory_used": {
"default": "mdi:memory"
}
}
}

View File

@ -11,10 +11,12 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfDataRate,
UnitOfInformation,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -30,9 +32,12 @@ from .const import (
KEY_SENSORS,
SENSORS_BYTES,
SENSORS_CONNECTED_DEVICE,
SENSORS_CPU,
SENSORS_LOAD_AVG,
SENSORS_MEMORY,
SENSORS_RATES,
SENSORS_TEMPERATURES,
SENSORS_UPTIME,
)
from .router import AsusWrtRouter
@ -46,6 +51,19 @@ class AsusWrtSensorEntityDescription(SensorEntityDescription):
UNIT_DEVICES = "Devices"
CPU_CORE_SENSORS: tuple[AsusWrtSensorEntityDescription, ...] = tuple(
AsusWrtSensorEntityDescription(
key=sens_key,
translation_key="cpu_core_usage",
translation_placeholders={"core_id": str(core_id)},
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
suggested_display_precision=1,
)
for core_id, sens_key in enumerate(SENSORS_CPU[1:], start=1)
)
CONNECTION_SENSORS: tuple[AsusWrtSensorEntityDescription, ...] = (
AsusWrtSensorEntityDescription(
key=SENSORS_CONNECTED_DEVICE[0],
@ -167,6 +185,61 @@ CONNECTION_SENSORS: tuple[AsusWrtSensorEntityDescription, ...] = (
entity_registry_enabled_default=False,
suggested_display_precision=1,
),
AsusWrtSensorEntityDescription(
key=SENSORS_MEMORY[0],
translation_key="memory_usage",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
suggested_display_precision=1,
),
AsusWrtSensorEntityDescription(
key=SENSORS_MEMORY[1],
translation_key="memory_free",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DATA_SIZE,
native_unit_of_measurement=UnitOfInformation.MEGABYTES,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
suggested_display_precision=2,
factor=1024,
),
AsusWrtSensorEntityDescription(
key=SENSORS_MEMORY[2],
translation_key="memory_used",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.DATA_SIZE,
native_unit_of_measurement=UnitOfInformation.MEGABYTES,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
suggested_display_precision=2,
factor=1024,
),
AsusWrtSensorEntityDescription(
key=SENSORS_UPTIME[0],
translation_key="last_boot",
device_class=SensorDeviceClass.TIMESTAMP,
),
AsusWrtSensorEntityDescription(
key=SENSORS_UPTIME[1],
translation_key="uptime",
state_class=SensorStateClass.TOTAL,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.SECONDS,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
AsusWrtSensorEntityDescription(
key=SENSORS_CPU[0],
translation_key="cpu_usage",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
suggested_display_precision=1,
),
*CPU_CORE_SENSORS,
)

View File

@ -88,6 +88,27 @@
},
"6ghz_temperature": {
"name": "6GHz Temperature"
},
"cpu_usage": {
"name": "CPU usage"
},
"cpu_core_usage": {
"name": "CPU core {core_id} usage"
},
"memory_usage": {
"name": "Memory usage"
},
"memory_free": {
"name": "Memory free"
},
"memory_used": {
"name": "Memory used"
},
"last_boot": {
"name": "Last boot"
},
"uptime": {
"name": "Uptime"
}
}
},

View File

@ -16,12 +16,30 @@ ASUSWRT_LEGACY_LIB = f"{ASUSWRT_BASE}.bridge.AsusWrtLegacy"
MOCK_BYTES_TOTAL = 60000000000, 50000000000
MOCK_BYTES_TOTAL_HTTP = dict(enumerate(MOCK_BYTES_TOTAL))
MOCK_CPU_USAGE = {
"cpu1_usage": 0.1,
"cpu2_usage": 0.2,
"cpu3_usage": 0.3,
"cpu4_usage": 0.4,
"cpu5_usage": 0.5,
"cpu6_usage": 0.6,
"cpu7_usage": 0.7,
"cpu8_usage": 0.8,
"cpu_total_usage": 0.9,
}
MOCK_CURRENT_TRANSFER_RATES = 20000000, 10000000
MOCK_CURRENT_TRANSFER_RATES_HTTP = dict(enumerate(MOCK_CURRENT_TRANSFER_RATES))
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_MEMORY_USAGE = {
"mem_usage_perc": 52.4,
"mem_total": 1048576,
"mem_free": 393216,
"mem_used": 655360,
}
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}
@pytest.fixture(name="patch_setup_entry")
@ -121,6 +139,11 @@ def mock_controller_connect_http(mock_devices_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
yield service_mock
@ -133,13 +156,22 @@ def mock_controller_connect_http_sens_fail(connect_http):
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
@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_detect:
yield mock_sens_detect
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

View File

@ -2,6 +2,7 @@
from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory
from pyasuswrt.exceptions import AsusWrtError, AsusWrtNotAvailableInfoError
import pytest
@ -10,10 +11,13 @@ from homeassistant.components.asuswrt.const import (
CONF_INTERFACE,
DOMAIN,
SENSORS_BYTES,
SENSORS_CPU,
SENSORS_LOAD_AVG,
SENSORS_MEMORY,
SENSORS_RATES,
SENSORS_TEMPERATURES,
SENSORS_TEMPERATURES_LEGACY,
SENSORS_UPTIME,
)
from homeassistant.components.device_tracker import CONF_CONSIDER_HOME
from homeassistant.config_entries import ConfigEntryState
@ -26,7 +30,6 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.util import slugify
from homeassistant.util.dt import utcnow
from .common import (
CONFIG_DATA_HTTP,
@ -42,7 +45,14 @@ from tests.common import MockConfigEntry, async_fire_time_changed
SENSORS_DEFAULT = [*SENSORS_BYTES, *SENSORS_RATES]
SENSORS_ALL_LEGACY = [*SENSORS_DEFAULT, *SENSORS_LOAD_AVG, *SENSORS_TEMPERATURES_LEGACY]
SENSORS_ALL_HTTP = [*SENSORS_DEFAULT, *SENSORS_LOAD_AVG, *SENSORS_TEMPERATURES]
SENSORS_ALL_HTTP = [
*SENSORS_DEFAULT,
*SENSORS_CPU,
*SENSORS_LOAD_AVG,
*SENSORS_MEMORY,
*SENSORS_TEMPERATURES,
*SENSORS_UPTIME,
]
@pytest.fixture(name="create_device_registry_devices")
@ -95,6 +105,7 @@ def _setup_entry(hass: HomeAssistant, config, sensors, unique_id=None):
async def _test_sensors(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_devices,
config,
entry_unique_id,
@ -125,7 +136,8 @@ async def _test_sensors(
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(f"{device_tracker.DOMAIN}.test").state == STATE_HOME
@ -139,7 +151,8 @@ async def _test_sensors(
# remove first tracked device
mock_devices.pop(MOCK_MACS[0])
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# consider home option set, all devices still home but only 1 device connected
@ -160,7 +173,8 @@ async def _test_sensors(
config_entry, options={CONF_CONSIDER_HOME: 0}
)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# consider home option set to 0, device "test" not home
@ -176,13 +190,16 @@ async def _test_sensors(
)
async def test_sensors_legacy(
hass: HomeAssistant,
connect_legacy,
freezer: FrozenDateTimeFactory,
mock_devices_legacy,
create_device_registry_devices,
entry_unique_id,
connect_legacy,
create_device_registry_devices,
) -> None:
"""Test creating AsusWRT default sensors and tracker with legacy protocol."""
await _test_sensors(hass, mock_devices_legacy, CONFIG_DATA_TELNET, entry_unique_id)
await _test_sensors(
hass, freezer, mock_devices_legacy, CONFIG_DATA_TELNET, entry_unique_id
)
@pytest.mark.parametrize(
@ -191,16 +208,21 @@ async def test_sensors_legacy(
)
async def test_sensors_http(
hass: HomeAssistant,
connect_http,
freezer: FrozenDateTimeFactory,
mock_devices_http,
create_device_registry_devices,
entry_unique_id,
connect_http,
create_device_registry_devices,
) -> None:
"""Test creating AsusWRT default sensors and tracker with http protocol."""
await _test_sensors(hass, mock_devices_http, CONFIG_DATA_HTTP, entry_unique_id)
await _test_sensors(
hass, freezer, mock_devices_http, CONFIG_DATA_HTTP, entry_unique_id
)
async def _test_loadavg_sensors(hass: HomeAssistant, config) -> None:
async def _test_loadavg_sensors(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, config
) -> None:
"""Test creating an AsusWRT load average sensors."""
config_entry, sensor_prefix = _setup_entry(hass, config, SENSORS_LOAD_AVG)
config_entry.add_to_hass(hass)
@ -208,7 +230,8 @@ async def _test_loadavg_sensors(hass: HomeAssistant, config) -> None:
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# assert temperature sensor available
@ -217,18 +240,22 @@ async def _test_loadavg_sensors(hass: HomeAssistant, config) -> None:
assert hass.states.get(f"{sensor_prefix}_sensor_load_avg15").state == "1.3"
async def test_loadavg_sensors_legacy(hass: HomeAssistant, connect_legacy) -> None:
async def test_loadavg_sensors_legacy(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_legacy
) -> None:
"""Test creating an AsusWRT load average sensors."""
await _test_loadavg_sensors(hass, CONFIG_DATA_TELNET)
await _test_loadavg_sensors(hass, freezer, CONFIG_DATA_TELNET)
async def test_loadavg_sensors_http(hass: HomeAssistant, connect_http) -> None:
async def test_loadavg_sensors_http(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http
) -> None:
"""Test creating an AsusWRT load average sensors."""
await _test_loadavg_sensors(hass, CONFIG_DATA_HTTP)
await _test_loadavg_sensors(hass, freezer, CONFIG_DATA_HTTP)
async def test_loadavg_sensors_unaivalable_http(
hass: HomeAssistant, connect_http
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http
) -> None:
"""Test load average sensors no available using http."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_LOAD_AVG)
@ -241,7 +268,8 @@ async def test_loadavg_sensors_unaivalable_http(
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# assert load average sensors not available
@ -271,7 +299,9 @@ async def test_temperature_sensors_http_fail(
assert not hass.states.get(f"{sensor_prefix}_6_0ghz")
async def _test_temperature_sensors(hass: HomeAssistant, config, sensors) -> str:
async def _test_temperature_sensors(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, config, sensors
) -> str:
"""Test creating a AsusWRT temperature sensors."""
config_entry, sensor_prefix = _setup_entry(hass, config, sensors)
config_entry.add_to_hass(hass)
@ -279,16 +309,19 @@ async def _test_temperature_sensors(hass: HomeAssistant, config, sensors) -> str
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
return sensor_prefix
async def test_temperature_sensors_legacy(hass: HomeAssistant, connect_legacy) -> None:
async def test_temperature_sensors_legacy(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_legacy
) -> None:
"""Test creating a AsusWRT temperature sensors."""
sensor_prefix = await _test_temperature_sensors(
hass, CONFIG_DATA_TELNET, SENSORS_TEMPERATURES_LEGACY
hass, freezer, CONFIG_DATA_TELNET, SENSORS_TEMPERATURES_LEGACY
)
# assert temperature sensor available
assert hass.states.get(f"{sensor_prefix}_2_4ghz").state == "40.2"
@ -296,10 +329,12 @@ async def test_temperature_sensors_legacy(hass: HomeAssistant, connect_legacy) -
assert not hass.states.get(f"{sensor_prefix}_5_0ghz")
async def test_temperature_sensors_http(hass: HomeAssistant, connect_http) -> None:
async def test_temperature_sensors_http(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http
) -> None:
"""Test creating a AsusWRT temperature sensors."""
sensor_prefix = await _test_temperature_sensors(
hass, CONFIG_DATA_HTTP, SENSORS_TEMPERATURES
hass, freezer, CONFIG_DATA_HTTP, SENSORS_TEMPERATURES
)
# assert temperature sensor available
assert hass.states.get(f"{sensor_prefix}_2_4ghz").state == "40.2"
@ -309,6 +344,97 @@ async def test_temperature_sensors_http(hass: HomeAssistant, connect_http) -> No
assert not hass.states.get(f"{sensor_prefix}_5_0ghz")
async def test_cpu_sensors_http_fail(
hass: HomeAssistant, connect_http_sens_fail
) -> None:
"""Test fail creating AsusWRT cpu sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_CPU)
config_entry.add_to_hass(hass)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# assert cpu availability exception is handled correctly
assert not hass.states.get(f"{sensor_prefix}_cpu1_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu2_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu3_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu4_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu5_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu6_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu7_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu8_usage")
assert not hass.states.get(f"{sensor_prefix}_cpu_total_usage")
async def test_cpu_sensors_http(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http
) -> None:
"""Test creating AsusWRT cpu sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_CPU)
config_entry.add_to_hass(hass)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# assert cpu sensors available
assert hass.states.get(f"{sensor_prefix}_cpu1_usage").state == "0.1"
assert hass.states.get(f"{sensor_prefix}_cpu2_usage").state == "0.2"
assert hass.states.get(f"{sensor_prefix}_cpu3_usage").state == "0.3"
assert hass.states.get(f"{sensor_prefix}_cpu4_usage").state == "0.4"
assert hass.states.get(f"{sensor_prefix}_cpu5_usage").state == "0.5"
assert hass.states.get(f"{sensor_prefix}_cpu6_usage").state == "0.6"
assert hass.states.get(f"{sensor_prefix}_cpu7_usage").state == "0.7"
assert hass.states.get(f"{sensor_prefix}_cpu8_usage").state == "0.8"
assert hass.states.get(f"{sensor_prefix}_cpu_total_usage").state == "0.9"
async def test_memory_sensors_http(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http
) -> None:
"""Test creating AsusWRT memory sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_MEMORY)
config_entry.add_to_hass(hass)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# assert memory sensors available
assert hass.states.get(f"{sensor_prefix}_mem_usage_perc").state == "52.4"
assert hass.states.get(f"{sensor_prefix}_mem_free").state == "384.0"
assert hass.states.get(f"{sensor_prefix}_mem_used").state == "640.0"
async def test_uptime_sensors_http(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_http
) -> None:
"""Test creating AsusWRT uptime sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA_HTTP, SENSORS_UPTIME)
config_entry.add_to_hass(hass)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# assert uptime sensors available
assert (
hass.states.get(f"{sensor_prefix}_sensor_last_boot").state
== "2024-08-02T00:47:00+00:00"
)
assert hass.states.get(f"{sensor_prefix}_sensor_uptime").state == "1625927"
@pytest.mark.parametrize(
"side_effect",
[OSError, None],
@ -359,7 +485,9 @@ async def test_connect_fail_http(
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def _test_sensors_polling_fails(hass: HomeAssistant, config, sensors) -> None:
async def _test_sensors_polling_fails(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, config, sensors
) -> None:
"""Test AsusWRT sensors are unavailable when polling fails."""
config_entry, sensor_prefix = _setup_entry(hass, config, sensors)
config_entry.add_to_hass(hass)
@ -367,7 +495,8 @@ async def _test_sensors_polling_fails(hass: HomeAssistant, config, sensors) -> N
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
for sensor_name in sensors:
@ -380,22 +509,28 @@ async def _test_sensors_polling_fails(hass: HomeAssistant, config, sensors) -> N
async def test_sensors_polling_fails_legacy(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
connect_legacy_sens_fail,
) -> None:
"""Test AsusWRT sensors are unavailable when polling fails."""
await _test_sensors_polling_fails(hass, CONFIG_DATA_TELNET, SENSORS_ALL_LEGACY)
await _test_sensors_polling_fails(
hass, freezer, CONFIG_DATA_TELNET, SENSORS_ALL_LEGACY
)
async def test_sensors_polling_fails_http(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
connect_http_sens_fail,
connect_http_sens_detect,
) -> None:
"""Test AsusWRT sensors are unavailable when polling fails."""
await _test_sensors_polling_fails(hass, CONFIG_DATA_HTTP, SENSORS_ALL_HTTP)
await _test_sensors_polling_fails(hass, freezer, CONFIG_DATA_HTTP, SENSORS_ALL_HTTP)
async def test_options_reload(hass: HomeAssistant, connect_legacy) -> None:
async def test_options_reload(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, connect_legacy
) -> None:
"""Test AsusWRT integration is reload changing an options that require this."""
config_entry = MockConfigEntry(
domain=DOMAIN,
@ -408,7 +543,8 @@ async def test_options_reload(hass: HomeAssistant, connect_legacy) -> None:
await hass.async_block_till_done()
assert connect_legacy.return_value.connection.async_connect.call_count == 1
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# change an option that requires integration reload
@ -451,7 +587,10 @@ async def test_unique_id_migration(
async def test_decorator_errors(
hass: HomeAssistant, connect_legacy, mock_available_temps
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
connect_legacy,
mock_available_temps,
) -> None:
"""Test AsusWRT sensors are unavailable on decorator type check error."""
sensors = [*SENSORS_BYTES, *SENSORS_TEMPERATURES_LEGACY]
@ -465,7 +604,8 @@ async def test_decorator_errors(
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
for sensor_name in sensors: