diff --git a/homeassistant/components/asuswrt/bridge.py b/homeassistant/components/asuswrt/bridge.py new file mode 100644 index 00000000000..9e6da0ea8f7 --- /dev/null +++ b/homeassistant/components/asuswrt/bridge.py @@ -0,0 +1,273 @@ +"""aioasuswrt and pyasuswrt bridge classes.""" +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections import namedtuple +import logging +from typing import Any, cast + +from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy + +from homeassistant.const import ( + CONF_HOST, + CONF_MODE, + CONF_PASSWORD, + CONF_PORT, + CONF_PROTOCOL, + CONF_USERNAME, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.update_coordinator import UpdateFailed + +from .const import ( + CONF_DNSMASQ, + CONF_INTERFACE, + CONF_REQUIRE_IP, + CONF_SSH_KEY, + DEFAULT_DNSMASQ, + DEFAULT_INTERFACE, + KEY_METHOD, + KEY_SENSORS, + PROTOCOL_TELNET, + SENSORS_BYTES, + SENSORS_LOAD_AVG, + SENSORS_RATES, + SENSORS_TEMPERATURES, +) + +SENSORS_TYPE_BYTES = "sensors_bytes" +SENSORS_TYPE_COUNT = "sensors_count" +SENSORS_TYPE_LOAD_AVG = "sensors_load_avg" +SENSORS_TYPE_RATES = "sensors_rates" +SENSORS_TYPE_TEMPERATURES = "sensors_temperatures" + +WrtDevice = namedtuple("WrtDevice", ["ip", "name", "connected_to"]) + +_LOGGER = logging.getLogger(__name__) + + +def _get_dict(keys: list, values: list) -> dict[str, Any]: + """Create a dict from a list of keys and values.""" + return dict(zip(keys, values)) + + +class AsusWrtBridge(ABC): + """The Base Bridge abstract class.""" + + @staticmethod + def get_bridge( + hass: HomeAssistant, conf: dict[str, Any], options: dict[str, Any] | None = None + ) -> AsusWrtBridge: + """Get Bridge instance.""" + return AsusWrtLegacyBridge(conf, options) + + def __init__(self, host: str) -> None: + """Initialize Bridge.""" + self._host = host + self._firmware: str | None = None + self._label_mac: str | None = None + self._model: str | None = None + + @property + def host(self) -> str: + """Return hostname.""" + return self._host + + @property + def firmware(self) -> str | None: + """Return firmware information.""" + return self._firmware + + @property + def label_mac(self) -> str | None: + """Return label mac information.""" + return self._label_mac + + @property + def model(self) -> str | None: + """Return model information.""" + return self._model + + @property + @abstractmethod + def is_connected(self) -> bool: + """Get connected status.""" + + @abstractmethod + async def async_connect(self) -> None: + """Connect to the device.""" + + @abstractmethod + async def async_disconnect(self) -> None: + """Disconnect to the device.""" + + @abstractmethod + async def async_get_connected_devices(self) -> dict[str, WrtDevice]: + """Get list of connected devices.""" + + @abstractmethod + async def async_get_available_sensors(self) -> dict[str, dict[str, Any]]: + """Return a dictionary of available sensors for this bridge.""" + + +class AsusWrtLegacyBridge(AsusWrtBridge): + """The Bridge that use legacy library.""" + + def __init__( + self, conf: dict[str, Any], options: dict[str, Any] | None = None + ) -> None: + """Initialize Bridge.""" + super().__init__(conf[CONF_HOST]) + self._protocol: str = conf[CONF_PROTOCOL] + self._api: AsusWrtLegacy = self._get_api(conf, options) + + @staticmethod + def _get_api( + conf: dict[str, Any], options: dict[str, Any] | None = None + ) -> AsusWrtLegacy: + """Get the AsusWrtLegacy API.""" + opt = options or {} + + return AsusWrtLegacy( + conf[CONF_HOST], + conf.get(CONF_PORT), + conf[CONF_PROTOCOL] == PROTOCOL_TELNET, + conf[CONF_USERNAME], + conf.get(CONF_PASSWORD, ""), + conf.get(CONF_SSH_KEY, ""), + conf[CONF_MODE], + opt.get(CONF_REQUIRE_IP, True), + interface=opt.get(CONF_INTERFACE, DEFAULT_INTERFACE), + dnsmasq=opt.get(CONF_DNSMASQ, DEFAULT_DNSMASQ), + ) + + @property + def is_connected(self) -> bool: + """Get connected status.""" + return cast(bool, self._api.is_connected) + + async def async_connect(self) -> None: + """Connect to the device.""" + await self._api.connection.async_connect() + + # get main router properties + if self._label_mac is None: + await self._get_label_mac() + if self._firmware is None: + await self._get_firmware() + if self._model is None: + await self._get_model() + + async def async_disconnect(self) -> None: + """Disconnect to the device.""" + if self._api is not None and self._protocol == PROTOCOL_TELNET: + self._api.connection.disconnect() + + async def async_get_connected_devices(self) -> dict[str, WrtDevice]: + """Get list of connected devices.""" + try: + api_devices = await self._api.async_get_connected_devices() + except OSError as exc: + raise UpdateFailed(exc) from exc + return { + format_mac(mac): WrtDevice(dev.ip, dev.name, None) + for mac, dev in api_devices.items() + } + + async def _get_nvram_info(self, info_type: str) -> dict[str, Any]: + """Get AsusWrt router info from nvram.""" + info = {} + try: + info = await self._api.async_get_nvram(info_type) + except OSError as exc: + _LOGGER.warning( + "Error calling method async_get_nvram(%s): %s", info_type, exc + ) + + return info + + async def _get_label_mac(self) -> None: + """Get label mac information.""" + label_mac = await self._get_nvram_info("LABEL_MAC") + if label_mac and "label_mac" in label_mac: + self._label_mac = format_mac(label_mac["label_mac"]) + + async def _get_firmware(self) -> None: + """Get firmware information.""" + firmware = await self._get_nvram_info("FIRMWARE") + if firmware and "firmver" in firmware: + firmver: str = firmware["firmver"] + if "buildno" in firmware: + firmver += f" (build {firmware['buildno']})" + self._firmware = firmver + + async def _get_model(self) -> None: + """Get model information.""" + model = await self._get_nvram_info("MODEL") + if model and "model" in model: + self._model = model["model"] + + async def async_get_available_sensors(self) -> dict[str, dict[str, Any]]: + """Return a dictionary of available sensors for this bridge.""" + sensors_temperatures = await self._get_available_temperature_sensors() + sensors_types = { + SENSORS_TYPE_BYTES: { + KEY_SENSORS: SENSORS_BYTES, + KEY_METHOD: self._get_bytes, + }, + SENSORS_TYPE_LOAD_AVG: { + KEY_SENSORS: SENSORS_LOAD_AVG, + KEY_METHOD: self._get_load_avg, + }, + SENSORS_TYPE_RATES: { + KEY_SENSORS: SENSORS_RATES, + KEY_METHOD: self._get_rates, + }, + SENSORS_TYPE_TEMPERATURES: { + KEY_SENSORS: sensors_temperatures, + KEY_METHOD: self._get_temperatures, + }, + } + return sensors_types + + async def _get_available_temperature_sensors(self) -> list[str]: + """Check which temperature information is available on the router.""" + availability = await self._api.async_find_temperature_commands() + return [SENSORS_TEMPERATURES[i] for i in range(3) if availability[i]] + + async def _get_bytes(self) -> dict[str, Any]: + """Fetch byte information from the router.""" + try: + datas = await self._api.async_get_bytes_total() + except (IndexError, OSError, ValueError) as exc: + raise UpdateFailed(exc) from exc + + return _get_dict(SENSORS_BYTES, datas) + + async def _get_rates(self) -> dict[str, Any]: + """Fetch rates information from the router.""" + try: + rates = await self._api.async_get_current_transfer_rates() + except (IndexError, OSError, ValueError) as exc: + raise UpdateFailed(exc) from exc + + return _get_dict(SENSORS_RATES, rates) + + async def _get_load_avg(self) -> dict[str, Any]: + """Fetch load average information from the router.""" + try: + avg = await self._api.async_get_loadavg() + except (IndexError, OSError, ValueError) as exc: + raise UpdateFailed(exc) from exc + + return _get_dict(SENSORS_LOAD_AVG, avg) + + async def _get_temperatures(self) -> dict[str, Any]: + """Fetch temperatures information from the router.""" + try: + temperatures: dict[str, Any] = await self._api.async_get_temperature() + except (OSError, ValueError) as exc: + raise UpdateFailed(exc) from exc + + return temperatures diff --git a/homeassistant/components/asuswrt/config_flow.py b/homeassistant/components/asuswrt/config_flow.py index 6b0056b14fa..56569d4f23b 100644 --- a/homeassistant/components/asuswrt/config_flow.py +++ b/homeassistant/components/asuswrt/config_flow.py @@ -25,13 +25,13 @@ from homeassistant.const import ( from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult from homeassistant.helpers import config_validation as cv -from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.schema_config_entry_flow import ( SchemaCommonFlowHandler, SchemaFlowFormStep, SchemaOptionsFlowHandler, ) +from .bridge import AsusWrtBridge from .const import ( CONF_DNSMASQ, CONF_INTERFACE, @@ -47,7 +47,6 @@ from .const import ( PROTOCOL_SSH, PROTOCOL_TELNET, ) -from .router import get_api, get_nvram_info LABEL_MAC = "LABEL_MAC" @@ -143,16 +142,15 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN): errors=errors or {}, ) - @staticmethod async def _async_check_connection( - user_input: dict[str, Any] + self, user_input: dict[str, Any] ) -> tuple[str, str | None]: """Attempt to connect the AsusWrt router.""" host: str = user_input[CONF_HOST] - api = get_api(user_input) + api = AsusWrtBridge.get_bridge(self.hass, user_input) try: - await api.connection.async_connect() + await api.async_connect() except OSError: _LOGGER.error("Error connecting to the AsusWrt router at %s", host) @@ -168,14 +166,9 @@ class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN): _LOGGER.error("Error connecting to the AsusWrt router at %s", host) return RESULT_CONN_ERROR, None - label_mac = await get_nvram_info(api, LABEL_MAC) - conf_protocol = user_input[CONF_PROTOCOL] - if conf_protocol == PROTOCOL_TELNET: - api.connection.disconnect() + unique_id = api.label_mac + await api.async_disconnect() - unique_id = None - if label_mac and "label_mac" in label_mac: - unique_id = format_mac(label_mac["label_mac"]) return RESULT_SUCCESS, unique_id async def async_step_user( diff --git a/homeassistant/components/asuswrt/const.py b/homeassistant/components/asuswrt/const.py index f80643f078d..1733d4c09c3 100644 --- a/homeassistant/components/asuswrt/const.py +++ b/homeassistant/components/asuswrt/const.py @@ -13,6 +13,10 @@ DEFAULT_DNSMASQ = "/var/lib/misc" DEFAULT_INTERFACE = "eth0" DEFAULT_TRACK_UNKNOWN = False +KEY_COORDINATOR = "coordinator" +KEY_METHOD = "method" +KEY_SENSORS = "sensors" + MODE_AP = "ap" MODE_ROUTER = "router" diff --git a/homeassistant/components/asuswrt/router.py b/homeassistant/components/asuswrt/router.py index 4291c21d0ed..c782a8f0f3b 100644 --- a/homeassistant/components/asuswrt/router.py +++ b/homeassistant/components/asuswrt/router.py @@ -6,22 +6,12 @@ from datetime import datetime, timedelta import logging from typing import Any -from aioasuswrt.asuswrt import AsusWrt, Device as WrtDevice - from homeassistant.components.device_tracker import ( CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME, DOMAIN as TRACKER_DOMAIN, ) from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ( - CONF_HOST, - CONF_MODE, - CONF_PASSWORD, - CONF_PORT, - CONF_PROTOCOL, - CONF_USERNAME, -) from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.helpers import entity_registry as er @@ -32,55 +22,36 @@ from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.util import dt as dt_util +from .bridge import AsusWrtBridge, WrtDevice from .const import ( CONF_DNSMASQ, CONF_INTERFACE, CONF_REQUIRE_IP, - CONF_SSH_KEY, CONF_TRACK_UNKNOWN, DEFAULT_DNSMASQ, DEFAULT_INTERFACE, DEFAULT_TRACK_UNKNOWN, DOMAIN, - PROTOCOL_TELNET, - SENSORS_BYTES, + KEY_COORDINATOR, + KEY_METHOD, + KEY_SENSORS, SENSORS_CONNECTED_DEVICE, - SENSORS_LOAD_AVG, - SENSORS_RATES, - SENSORS_TEMPERATURES, ) CONF_REQ_RELOAD = [CONF_DNSMASQ, CONF_INTERFACE, CONF_REQUIRE_IP] DEFAULT_NAME = "Asuswrt" -KEY_COORDINATOR = "coordinator" -KEY_SENSORS = "sensors" - SCAN_INTERVAL = timedelta(seconds=30) -SENSORS_TYPE_BYTES = "sensors_bytes" SENSORS_TYPE_COUNT = "sensors_count" -SENSORS_TYPE_LOAD_AVG = "sensors_load_avg" -SENSORS_TYPE_RATES = "sensors_rates" -SENSORS_TYPE_TEMPERATURES = "sensors_temperatures" _LOGGER = logging.getLogger(__name__) -def _get_dict(keys: list, values: list) -> dict[str, Any]: - """Create a dict from a list of keys and values.""" - ret_dict: dict[str, Any] = dict.fromkeys(keys) - - for index, key in enumerate(ret_dict): - ret_dict[key] = values[index] - - return ret_dict - - class AsusWrtSensorDataHandler: """Data handler for AsusWrt sensor.""" - def __init__(self, hass: HomeAssistant, api: AsusWrt) -> None: + def __init__(self, hass: HomeAssistant, api: AsusWrtBridge) -> None: """Initialize a AsusWrt sensor data handler.""" self._hass = hass self._api = api @@ -90,42 +61,6 @@ class AsusWrtSensorDataHandler: """Return number of connected devices.""" return {SENSORS_CONNECTED_DEVICE[0]: self._connected_devices} - async def _get_bytes(self) -> dict[str, Any]: - """Fetch byte information from the router.""" - try: - datas = await self._api.async_get_bytes_total() - except (OSError, ValueError) as exc: - raise UpdateFailed(exc) from exc - - return _get_dict(SENSORS_BYTES, datas) - - async def _get_rates(self) -> dict[str, Any]: - """Fetch rates information from the router.""" - try: - rates = await self._api.async_get_current_transfer_rates() - except (OSError, ValueError) as exc: - raise UpdateFailed(exc) from exc - - return _get_dict(SENSORS_RATES, rates) - - async def _get_load_avg(self) -> dict[str, Any]: - """Fetch load average information from the router.""" - try: - avg = await self._api.async_get_loadavg() - except (OSError, ValueError) as exc: - raise UpdateFailed(exc) from exc - - return _get_dict(SENSORS_LOAD_AVG, avg) - - async def _get_temperatures(self) -> dict[str, Any]: - """Fetch temperatures information from the router.""" - try: - temperatures: dict[str, Any] = await self._api.async_get_temperature() - except (OSError, ValueError) as exc: - raise UpdateFailed(exc) from exc - - return temperatures - def update_device_count(self, conn_devices: int) -> bool: """Update connected devices attribute.""" if self._connected_devices == conn_devices: @@ -134,19 +69,17 @@ class AsusWrtSensorDataHandler: return True async def get_coordinator( - self, sensor_type: str, should_poll: bool = True + self, + sensor_type: str, + update_method: Callable[[], Any] | None = None, ) -> DataUpdateCoordinator: """Get the coordinator for a specific sensor type.""" + should_poll = True if sensor_type == SENSORS_TYPE_COUNT: + should_poll = False method = self._get_connected_devices - elif sensor_type == SENSORS_TYPE_BYTES: - method = self._get_bytes - elif sensor_type == SENSORS_TYPE_LOAD_AVG: - method = self._get_load_avg - elif sensor_type == SENSORS_TYPE_RATES: - method = self._get_rates - elif sensor_type == SENSORS_TYPE_TEMPERATURES: - method = self._get_temperatures + elif update_method is not None: + method = update_method else: raise RuntimeError(f"Invalid sensor type: {sensor_type}") @@ -226,12 +159,6 @@ class AsusWrtRouter: self.hass = hass self._entry = entry - self._api: AsusWrt = None - self._protocol: str = entry.data[CONF_PROTOCOL] - self._host: str = entry.data[CONF_HOST] - self._model: str = "Asus Router" - self._sw_v: str | None = None - self._devices: dict[str, AsusWrtDevInfo] = {} self._connected_devices: int = 0 self._connect_error: bool = False @@ -248,26 +175,19 @@ class AsusWrtRouter: } self._options.update(entry.options) + self._api: AsusWrtBridge = AsusWrtBridge.get_bridge( + self.hass, dict(self._entry.data), self._options + ) + async def setup(self) -> None: """Set up a AsusWrt router.""" - self._api = get_api(dict(self._entry.data), self._options) - try: - await self._api.connection.async_connect() - except OSError as exp: - raise ConfigEntryNotReady from exp - + await self._api.async_connect() + except OSError as exc: + raise ConfigEntryNotReady from exc if not self._api.is_connected: raise ConfigEntryNotReady - # System - model = await get_nvram_info(self._api, "MODEL") - if model and "model" in model: - self._model = model["model"] - firmware = await get_nvram_info(self._api, "FIRMWARE") - if firmware and "firmver" in firmware and "buildno" in firmware: - self._sw_v = f"{firmware['firmver']} (build {firmware['buildno']})" - # Load tracked entities from registry entity_reg = er.async_get(self.hass) track_entries = er.async_entries_for_config_entry( @@ -312,24 +232,24 @@ class AsusWrtRouter: async def update_devices(self) -> None: """Update AsusWrt devices tracker.""" new_device = False - _LOGGER.debug("Checking devices for ASUS router %s", self._host) + _LOGGER.debug("Checking devices for ASUS router %s", self.host) try: - api_devices = await self._api.async_get_connected_devices() - except OSError as exc: + wrt_devices = await self._api.async_get_connected_devices() + except UpdateFailed as exc: if not self._connect_error: self._connect_error = True _LOGGER.error( "Error connecting to ASUS router %s for device update: %s", - self._host, + self.host, exc, ) return if self._connect_error: self._connect_error = False - _LOGGER.info("Reconnected to ASUS router %s", self._host) + _LOGGER.info("Reconnected to ASUS router %s", self.host) - self._connected_devices = len(api_devices) + self._connected_devices = len(wrt_devices) consider_home: int = self._options.get( CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME.total_seconds() ) @@ -337,7 +257,6 @@ class AsusWrtRouter: CONF_TRACK_UNKNOWN, DEFAULT_TRACK_UNKNOWN ) - wrt_devices = {format_mac(mac): dev for mac, dev in api_devices.items()} for device_mac, device in self._devices.items(): dev_info = wrt_devices.pop(device_mac, None) device.update(dev_info, consider_home) @@ -363,19 +282,14 @@ class AsusWrtRouter: self._sensors_data_handler = AsusWrtSensorDataHandler(self.hass, self._api) self._sensors_data_handler.update_device_count(self._connected_devices) - sensors_types: dict[str, list[str]] = { - SENSORS_TYPE_BYTES: SENSORS_BYTES, - SENSORS_TYPE_COUNT: SENSORS_CONNECTED_DEVICE, - SENSORS_TYPE_LOAD_AVG: SENSORS_LOAD_AVG, - SENSORS_TYPE_RATES: SENSORS_RATES, - SENSORS_TYPE_TEMPERATURES: await self._get_available_temperature_sensors(), - } + sensors_types = await self._api.async_get_available_sensors() + sensors_types[SENSORS_TYPE_COUNT] = {KEY_SENSORS: SENSORS_CONNECTED_DEVICE} - for sensor_type, sensor_names in sensors_types.items(): - if not sensor_names: + for sensor_type, sensor_def in sensors_types.items(): + if not (sensor_names := sensor_def.get(KEY_SENSORS)): continue coordinator = await self._sensors_data_handler.get_coordinator( - sensor_type, sensor_type != SENSORS_TYPE_COUNT + sensor_type, update_method=sensor_def.get(KEY_METHOD) ) self._sensors_coordinator[sensor_type] = { KEY_COORDINATOR: coordinator, @@ -392,31 +306,10 @@ class AsusWrtRouter: if self._sensors_data_handler.update_device_count(self._connected_devices): await coordinator.async_refresh() - async def _get_available_temperature_sensors(self) -> list[str]: - """Check which temperature information is available on the router.""" - try: - availability = await self._api.async_find_temperature_commands() - available_sensors = [ - SENSORS_TEMPERATURES[i] for i in range(3) if availability[i] - ] - except Exception as exc: # pylint: disable=broad-except - _LOGGER.debug( - ( - "Failed checking temperature sensor availability for ASUS router" - " %s. Exception: %s" - ), - self._host, - exc, - ) - return [] - - return available_sensors - async def close(self) -> None: """Close the connection.""" - if self._api is not None and self._protocol == PROTOCOL_TELNET: - self._api.connection.disconnect() - self._api = None + if self._api is not None: + await self._api.async_disconnect() for func in self._on_close: func() @@ -443,14 +336,17 @@ class AsusWrtRouter: @property def device_info(self) -> DeviceInfo: """Return the device information.""" - return DeviceInfo( + info = DeviceInfo( identifiers={(DOMAIN, self.unique_id or "AsusWRT")}, - name=self._host, - model=self._model, + name=self.host, + model=self._api.model or "Asus Router", manufacturer="Asus", - sw_version=self._sw_v, - configuration_url=f"http://{self._host}", + configuration_url=f"http://{self.host}", ) + if self._api.firmware: + info["sw_version"] = self._api.firmware + + return info @property def signal_device_new(self) -> str: @@ -465,7 +361,7 @@ class AsusWrtRouter: @property def host(self) -> str: """Return router hostname.""" - return self._host + return self._api.host @property def unique_id(self) -> str | None: @@ -475,7 +371,7 @@ class AsusWrtRouter: @property def name(self) -> str: """Return router name.""" - return self._host if self.unique_id else DEFAULT_NAME + return self.host if self.unique_id else DEFAULT_NAME @property def devices(self) -> dict[str, AsusWrtDevInfo]: @@ -486,32 +382,3 @@ class AsusWrtRouter: def sensors_coordinator(self) -> dict[str, Any]: """Return sensors coordinators.""" return self._sensors_coordinator - - -async def get_nvram_info(api: AsusWrt, info_type: str) -> dict[str, Any]: - """Get AsusWrt router info from nvram.""" - info = {} - try: - info = await api.async_get_nvram(info_type) - except OSError as exc: - _LOGGER.warning("Error calling method async_get_nvram(%s): %s", info_type, exc) - - return info - - -def get_api(conf: dict[str, Any], options: dict[str, Any] | None = None) -> AsusWrt: - """Get the AsusWrt API.""" - opt = options or {} - - return AsusWrt( - conf[CONF_HOST], - conf.get(CONF_PORT), - conf[CONF_PROTOCOL] == PROTOCOL_TELNET, - conf[CONF_USERNAME], - conf.get(CONF_PASSWORD, ""), - conf.get(CONF_SSH_KEY, ""), - conf[CONF_MODE], - opt.get(CONF_REQUIRE_IP, True), - interface=opt.get(CONF_INTERFACE, DEFAULT_INTERFACE), - dnsmasq=opt.get(CONF_DNSMASQ, DEFAULT_DNSMASQ), - ) diff --git a/homeassistant/components/asuswrt/sensor.py b/homeassistant/components/asuswrt/sensor.py index 95724ec3bb5..accd1eba59b 100644 --- a/homeassistant/components/asuswrt/sensor.py +++ b/homeassistant/components/asuswrt/sensor.py @@ -26,13 +26,15 @@ from homeassistant.helpers.update_coordinator import ( from .const import ( DATA_ASUSWRT, DOMAIN, + KEY_COORDINATOR, + KEY_SENSORS, SENSORS_BYTES, SENSORS_CONNECTED_DEVICE, SENSORS_LOAD_AVG, SENSORS_RATES, SENSORS_TEMPERATURES, ) -from .router import KEY_COORDINATOR, KEY_SENSORS, AsusWrtRouter +from .router import AsusWrtRouter @dataclass diff --git a/tests/components/asuswrt/test_config_flow.py b/tests/components/asuswrt/test_config_flow.py index aabf5d6d46b..bdee4f82f90 100644 --- a/tests/components/asuswrt/test_config_flow.py +++ b/tests/components/asuswrt/test_config_flow.py @@ -62,7 +62,7 @@ def mock_unique_id_fixture(): @pytest.fixture(name="connect") def mock_controller_connect(mock_unique_id): """Mock a successful connection.""" - with patch("homeassistant.components.asuswrt.router.AsusWrt") as service_mock: + with patch("homeassistant.components.asuswrt.bridge.AsusWrtLegacy") as service_mock: service_mock.return_value.connection.async_connect = AsyncMock() service_mock.return_value.is_connected = True service_mock.return_value.connection.disconnect = Mock() @@ -236,11 +236,12 @@ async def test_on_connect_failed(hass: HomeAssistant, side_effect, error) -> Non ) with PATCH_GET_HOST, patch( - "homeassistant.components.asuswrt.router.AsusWrt" + "homeassistant.components.asuswrt.bridge.AsusWrtLegacy" ) as asus_wrt: asus_wrt.return_value.connection.async_connect = AsyncMock( side_effect=side_effect ) + asus_wrt.return_value.async_get_nvram = AsyncMock(return_value={}) asus_wrt.return_value.is_connected = False result = await hass.config_entries.flow.async_configure( diff --git a/tests/components/asuswrt/test_sensor.py b/tests/components/asuswrt/test_sensor.py index 553902b66fd..c28d71c1a29 100644 --- a/tests/components/asuswrt/test_sensor.py +++ b/tests/components/asuswrt/test_sensor.py @@ -32,7 +32,7 @@ from homeassistant.util.dt import utcnow from tests.common import MockConfigEntry, async_fire_time_changed -ASUSWRT_LIB = "homeassistant.components.asuswrt.router.AsusWrt" +ASUSWRT_LIB = "homeassistant.components.asuswrt.bridge.AsusWrtLegacy" HOST = "myrouter.asuswrt.com" IP_ADDRESS = "192.168.1.1" @@ -311,28 +311,6 @@ async def test_loadavg_sensors( assert hass.states.get(f"{sensor_prefix}_load_avg_15m").state == "1.3" -async def test_temperature_sensors_fail( - hass: HomeAssistant, - connect, - mock_available_temps, -) -> None: - """Test fail creating AsusWRT temperature sensors.""" - config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_TEMP) - config_entry.add_to_hass(hass) - - # Only length of 3 booleans is valid. Checking the exception handling. - mock_available_temps.pop(2) - - # initial devices setup - assert await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - - # assert temperature availability exception is handled correctly - assert not hass.states.get(f"{sensor_prefix}_2_4ghz_temperature") - assert not hass.states.get(f"{sensor_prefix}_5ghz_temperature") - assert not hass.states.get(f"{sensor_prefix}_cpu_temperature") - - async def test_temperature_sensors( hass: HomeAssistant, connect,