diff --git a/.coveragerc b/.coveragerc index 21534333583..97f26d8adc5 100644 --- a/.coveragerc +++ b/.coveragerc @@ -717,7 +717,6 @@ omit = homeassistant/components/microsoft/tts.py homeassistant/components/miflora/sensor.py homeassistant/components/mikrotik/hub.py - homeassistant/components/mikrotik/device_tracker.py homeassistant/components/mill/climate.py homeassistant/components/mill/const.py homeassistant/components/mill/sensor.py diff --git a/homeassistant/components/mikrotik/__init__.py b/homeassistant/components/mikrotik/__init__.py index 856495dc0f2..f72c79c1559 100644 --- a/homeassistant/components/mikrotik/__init__.py +++ b/homeassistant/components/mikrotik/__init__.py @@ -1,35 +1,41 @@ """The Mikrotik component.""" from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.helpers import config_validation as cv, device_registry as dr from .const import ATTR_MANUFACTURER, DOMAIN, PLATFORMS -from .hub import MikrotikDataUpdateCoordinator +from .errors import CannotConnect, LoginError +from .hub import MikrotikDataUpdateCoordinator, get_api CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False) async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: """Set up the Mikrotik component.""" - - hub = MikrotikDataUpdateCoordinator(hass, config_entry) - if not await hub.async_setup(): + try: + api = await hass.async_add_executor_job(get_api, dict(config_entry.data)) + except CannotConnect as api_error: + raise ConfigEntryNotReady from api_error + except LoginError: return False - await hub.async_config_entry_first_refresh() + coordinator = MikrotikDataUpdateCoordinator(hass, config_entry, api) + await hass.async_add_executor_job(coordinator.api.get_hub_details) + await coordinator.async_config_entry_first_refresh() - hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = hub + hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coordinator hass.config_entries.async_setup_platforms(config_entry, PLATFORMS) device_registry = dr.async_get(hass) device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, - connections={(DOMAIN, hub.serial_num)}, + connections={(DOMAIN, coordinator.serial_num)}, manufacturer=ATTR_MANUFACTURER, - model=hub.model, - name=hub.hostname, - sw_version=hub.firmware, + model=coordinator.model, + name=coordinator.hostname, + sw_version=coordinator.firmware, ) return True @@ -37,10 +43,9 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: """Unload a config entry.""" - unload_ok = await hass.config_entries.async_unload_platforms( + if unload_ok := await hass.config_entries.async_unload_platforms( config_entry, PLATFORMS - ) - - hass.data[DOMAIN].pop(config_entry.entry_id) + ): + hass.data[DOMAIN].pop(config_entry.entry_id) return unload_ok diff --git a/homeassistant/components/mikrotik/config_flow.py b/homeassistant/components/mikrotik/config_flow.py index 36b65b6f2ba..d506c2c75e4 100644 --- a/homeassistant/components/mikrotik/config_flow.py +++ b/homeassistant/components/mikrotik/config_flow.py @@ -1,6 +1,8 @@ """Config flow for Mikrotik.""" from __future__ import annotations +from typing import Any + import voluptuous as vol from homeassistant import config_entries @@ -13,6 +15,7 @@ from homeassistant.const import ( CONF_VERIFY_SSL, ) from homeassistant.core import callback +from homeassistant.data_entry_flow import FlowResult from .const import ( CONF_ARP_PING, @@ -40,7 +43,9 @@ class MikrotikFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): """Get the options flow for this handler.""" return MikrotikOptionsFlowHandler(config_entry) - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Handle a flow initialized by the user.""" errors = {} if user_input is not None: @@ -52,7 +57,7 @@ class MikrotikFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): break try: - await self.hass.async_add_executor_job(get_api, self.hass, user_input) + await self.hass.async_add_executor_job(get_api, user_input) except CannotConnect: errors["base"] = "cannot_connect" except LoginError: @@ -86,11 +91,15 @@ class MikrotikOptionsFlowHandler(config_entries.OptionsFlow): """Initialize Mikrotik options flow.""" self.config_entry = config_entry - async def async_step_init(self, user_input=None): + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Manage the Mikrotik options.""" return await self.async_step_device_tracker() - async def async_step_device_tracker(self, user_input=None): + async def async_step_device_tracker( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Manage the device tracker options.""" if user_input is not None: return self.async_create_entry(title="", data=user_input) diff --git a/homeassistant/components/mikrotik/const.py b/homeassistant/components/mikrotik/const.py index b328c10a602..2942e6981fa 100644 --- a/homeassistant/components/mikrotik/const.py +++ b/homeassistant/components/mikrotik/const.py @@ -1,33 +1,35 @@ """Constants used in the Mikrotik components.""" +from typing import Final + from homeassistant.const import Platform -DOMAIN = "mikrotik" -DEFAULT_NAME = "Mikrotik" -DEFAULT_API_PORT = 8728 -DEFAULT_DETECTION_TIME = 300 +DOMAIN: Final = "mikrotik" +DEFAULT_NAME: Final = "Mikrotik" +DEFAULT_API_PORT: Final = 8728 +DEFAULT_DETECTION_TIME: Final = 300 -ATTR_MANUFACTURER = "Mikrotik" -ATTR_SERIAL_NUMBER = "serial-number" -ATTR_FIRMWARE = "current-firmware" -ATTR_MODEL = "model" +ATTR_MANUFACTURER: Final = "Mikrotik" +ATTR_SERIAL_NUMBER: Final = "serial-number" +ATTR_FIRMWARE: Final = "current-firmware" +ATTR_MODEL: Final = "model" -CONF_ARP_PING = "arp_ping" -CONF_FORCE_DHCP = "force_dhcp" -CONF_DETECTION_TIME = "detection_time" +CONF_ARP_PING: Final = "arp_ping" +CONF_FORCE_DHCP: Final = "force_dhcp" +CONF_DETECTION_TIME: Final = "detection_time" -NAME = "name" -INFO = "info" -IDENTITY = "identity" -ARP = "arp" +NAME: Final = "name" +INFO: Final = "info" +IDENTITY: Final = "identity" +ARP: Final = "arp" -CAPSMAN = "capsman" -DHCP = "dhcp" -WIRELESS = "wireless" -IS_WIRELESS = "is_wireless" -IS_CAPSMAN = "is_capsman" +CAPSMAN: Final = "capsman" +DHCP: Final = "dhcp" +WIRELESS: Final = "wireless" +IS_WIRELESS: Final = "is_wireless" +IS_CAPSMAN: Final = "is_capsman" -MIKROTIK_SERVICES = { +MIKROTIK_SERVICES: Final = { ARP: "/ip/arp/getall", CAPSMAN: "/caps-man/registration-table/getall", DHCP: "/ip/dhcp-server/lease/getall", @@ -38,9 +40,9 @@ MIKROTIK_SERVICES = { IS_CAPSMAN: "/caps-man/interface/print", } -PLATFORMS = [Platform.DEVICE_TRACKER] +PLATFORMS: Final = [Platform.DEVICE_TRACKER] -ATTR_DEVICE_TRACKER = [ +ATTR_DEVICE_TRACKER: Final = [ "comment", "mac-address", "ssid", diff --git a/homeassistant/components/mikrotik/device_tracker.py b/homeassistant/components/mikrotik/device_tracker.py index 9389d3bea5c..4e002cdbcfe 100644 --- a/homeassistant/components/mikrotik/device_tracker.py +++ b/homeassistant/components/mikrotik/device_tracker.py @@ -1,6 +1,8 @@ """Support for Mikrotik routers as device tracker.""" from __future__ import annotations +from typing import Any + from homeassistant.components.device_tracker.config_entry import ScannerEntity from homeassistant.components.device_tracker.const import ( DOMAIN as DEVICE_TRACKER, @@ -14,7 +16,7 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity import homeassistant.util.dt as dt_util from .const import DOMAIN -from .hub import MikrotikDataUpdateCoordinator +from .hub import Device, MikrotikDataUpdateCoordinator # These are normalized to ATTR_IP and ATTR_MAC to conform # to device_tracker @@ -27,7 +29,9 @@ async def async_setup_entry( async_add_entities: AddEntitiesCallback, ) -> None: """Set up device tracker for Mikrotik component.""" - hub = hass.data[DOMAIN][config_entry.entry_id] + coordinator: MikrotikDataUpdateCoordinator = hass.data[DOMAIN][ + config_entry.entry_id + ] tracked: dict[str, MikrotikDataUpdateCoordinatorTracker] = {} @@ -42,47 +46,53 @@ async def async_setup_entry( ): if ( - entity.unique_id in hub.api.devices - or entity.unique_id not in hub.api.all_devices + entity.unique_id in coordinator.api.devices + or entity.unique_id not in coordinator.api.all_devices ): continue - hub.api.restore_device(entity.unique_id) + coordinator.api.restore_device(entity.unique_id) @callback - def update_hub(): + def update_hub() -> None: """Update the status of the device.""" - update_items(hub, async_add_entities, tracked) + update_items(coordinator, async_add_entities, tracked) - config_entry.async_on_unload(hub.async_add_listener(update_hub)) + config_entry.async_on_unload(coordinator.async_add_listener(update_hub)) update_hub() @callback -def update_items(hub, async_add_entities, tracked): +def update_items( + coordinator: MikrotikDataUpdateCoordinator, + async_add_entities: AddEntitiesCallback, + tracked: dict[str, MikrotikDataUpdateCoordinatorTracker], +): """Update tracked device state from the hub.""" - new_tracked = [] - for mac, device in hub.api.devices.items(): + new_tracked: list[MikrotikDataUpdateCoordinatorTracker] = [] + for mac, device in coordinator.api.devices.items(): if mac not in tracked: - tracked[mac] = MikrotikDataUpdateCoordinatorTracker(device, hub) + tracked[mac] = MikrotikDataUpdateCoordinatorTracker(device, coordinator) new_tracked.append(tracked[mac]) if new_tracked: async_add_entities(new_tracked) -class MikrotikDataUpdateCoordinatorTracker(CoordinatorEntity, ScannerEntity): +class MikrotikDataUpdateCoordinatorTracker( + CoordinatorEntity[MikrotikDataUpdateCoordinator], ScannerEntity +): """Representation of network device.""" - coordinator: MikrotikDataUpdateCoordinator - - def __init__(self, device, hub): + def __init__( + self, device: Device, coordinator: MikrotikDataUpdateCoordinator + ) -> None: """Initialize the tracked device.""" - super().__init__(hub) + super().__init__(coordinator) self.device = device @property - def is_connected(self): + def is_connected(self) -> bool: """Return true if the client is connected to the network.""" if ( self.device.last_seen @@ -93,7 +103,7 @@ class MikrotikDataUpdateCoordinatorTracker(CoordinatorEntity, ScannerEntity): return False @property - def source_type(self): + def source_type(self) -> str: """Return the source type of the client.""" return SOURCE_TYPE_ROUTER @@ -124,7 +134,7 @@ class MikrotikDataUpdateCoordinatorTracker(CoordinatorEntity, ScannerEntity): return self.device.mac @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any] | None: """Return the device state attributes.""" if self.is_connected: return {k: v for k, v in self.device.attrs.items() if k not in FILTER_ATTRS} diff --git a/homeassistant/components/mikrotik/hub.py b/homeassistant/components/mikrotik/hub.py index 7f2314bd057..9219159ca74 100644 --- a/homeassistant/components/mikrotik/hub.py +++ b/homeassistant/components/mikrotik/hub.py @@ -1,14 +1,18 @@ """The Mikrotik router class.""" -from datetime import timedelta +from __future__ import annotations + +from datetime import datetime, timedelta import logging import socket import ssl +from typing import Any import librouteros from librouteros.login import plain as login_plain, token as login_token +from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL -from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.core import HomeAssistant from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.util import slugify import homeassistant.util.dt as dt_util @@ -42,36 +46,36 @@ _LOGGER = logging.getLogger(__name__) class Device: """Represents a network device.""" - def __init__(self, mac, params): + def __init__(self, mac: str, params: dict[str, Any]) -> None: """Initialize the network device.""" self._mac = mac self._params = params - self._last_seen = None - self._attrs = {} - self._wireless_params = None + self._last_seen: datetime | None = None + self._attrs: dict[str, Any] = {} + self._wireless_params: dict[str, Any] = {} @property - def name(self): + def name(self) -> str: """Return device name.""" return self._params.get("host-name", self.mac) @property - def ip_address(self): + def ip_address(self) -> str: """Return device primary ip address.""" - return self._params.get("address") + return self._params["address"] @property - def mac(self): + def mac(self) -> str: """Return device mac.""" return self._mac @property - def last_seen(self): + def last_seen(self) -> datetime | None: """Return device last seen.""" return self._last_seen @property - def attrs(self): + def attrs(self) -> dict[str, Any]: """Return device attributes.""" attr_data = self._wireless_params if self._wireless_params else self._params for attr in ATTR_DEVICE_TRACKER: @@ -80,7 +84,12 @@ class Device: self._attrs["ip_address"] = self._params.get("active-address") return self._attrs - def update(self, wireless_params=None, params=None, active=False): + def update( + self, + wireless_params: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, + active: bool = False, + ) -> None: """Update Device params.""" if wireless_params: self._wireless_params = wireless_params @@ -93,27 +102,26 @@ class Device: class MikrotikData: """Handle all communication with the Mikrotik API.""" - def __init__(self, hass, config_entry, api): + def __init__( + self, hass: HomeAssistant, config_entry: ConfigEntry, api: librouteros.Api + ) -> None: """Initialize the Mikrotik Client.""" self.hass = hass self.config_entry = config_entry self.api = api - self._host = self.config_entry.data[CONF_HOST] - self.all_devices = {} - self.devices = {} - self.available = True - self.support_capsman = False - self.support_wireless = False - self.hostname = None - self.model = None - self.firmware = None - self.serial_number = None + self._host: str = self.config_entry.data[CONF_HOST] + self.all_devices: dict[str, dict[str, Any]] = {} + self.devices: dict[str, Device] = {} + self.support_capsman: bool = False + self.support_wireless: bool = False + self.hostname: str = "" + self.model: str = "" + self.firmware: str = "" + self.serial_number: str = "" @staticmethod - def load_mac(devices=None): + def load_mac(devices: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """Load dictionary using MAC address as key.""" - if not devices: - return None mac_devices = {} for device in devices: if "mac-address" in device: @@ -122,26 +130,23 @@ class MikrotikData: return mac_devices @property - def arp_enabled(self): + def arp_enabled(self) -> bool: """Return arp_ping option setting.""" - return self.config_entry.options[CONF_ARP_PING] + return self.config_entry.options.get(CONF_ARP_PING, False) @property - def force_dhcp(self): + def force_dhcp(self) -> bool: """Return force_dhcp option setting.""" - return self.config_entry.options[CONF_FORCE_DHCP] + return self.config_entry.options.get(CONF_FORCE_DHCP, False) - def get_info(self, param): + def get_info(self, param: str) -> str: """Return device model name.""" cmd = IDENTITY if param == NAME else INFO - data = self.command(MIKROTIK_SERVICES[cmd]) - return ( - data[0].get(param) # pylint: disable=unsubscriptable-object - if data - else None - ) + if data := self.command(MIKROTIK_SERVICES[cmd]): + return str(data[0].get(param)) + return "" - def get_hub_details(self): + def get_hub_details(self) -> None: """Get Hub info.""" self.hostname = self.get_info(NAME) self.model = self.get_info(ATTR_MODEL) @@ -150,24 +155,17 @@ class MikrotikData: self.support_capsman = bool(self.command(MIKROTIK_SERVICES[IS_CAPSMAN])) self.support_wireless = bool(self.command(MIKROTIK_SERVICES[IS_WIRELESS])) - def connect_to_hub(self): - """Connect to hub.""" - try: - self.api = get_api(self.hass, self.config_entry.data) - return True - except (LoginError, CannotConnect): - return False - - def get_list_from_interface(self, interface): + def get_list_from_interface(self, interface: str) -> dict[str, dict[str, Any]]: """Get devices from interface.""" - result = self.command(MIKROTIK_SERVICES[interface]) - return self.load_mac(result) if result else {} + if result := self.command(MIKROTIK_SERVICES[interface]): + return self.load_mac(result) + return {} - def restore_device(self, mac): + def restore_device(self, mac: str) -> None: """Restore a missing device after restart.""" self.devices[mac] = Device(mac, self.all_devices[mac]) - def update_devices(self): + def update_devices(self) -> None: """Get list of devices with latest status.""" arp_devices = {} device_list = {} @@ -192,7 +190,7 @@ class MikrotikData: # get new hub firmware version if updated self.firmware = self.get_info(ATTR_FIRMWARE) - except (CannotConnect, socket.timeout, OSError) as err: + except (CannotConnect, LoginError) as err: raise UpdateFailed from err if not device_list: @@ -218,11 +216,12 @@ class MikrotikData: active = True if self.arp_enabled and mac in arp_devices: active = self.do_arp_ping( - params.get("active-address"), arp_devices[mac].get("interface") + str(params.get("active-address")), + str(arp_devices[mac].get("interface")), ) self.devices[mac].update(active=active) - def do_arp_ping(self, ip_address, interface): + def do_arp_ping(self, ip_address: str, interface: str) -> bool: """Attempt to arp ping MAC address via interface.""" _LOGGER.debug("pinging - %s", ip_address) params = { @@ -234,9 +233,9 @@ class MikrotikData: } cmd = "/ping" data = self.command(cmd, params) - if data is not None: + if data: status = 0 - for result in data: # pylint: disable=not-an-iterable + for result in data: if "status" in result: status += 1 if status == len(data): @@ -246,22 +245,25 @@ class MikrotikData: return False return True - def command(self, cmd, params=None): + def command( + self, cmd: str, params: dict[str, Any] | None = None + ) -> list[dict[str, Any]]: """Retrieve data from Mikrotik API.""" try: _LOGGER.info("Running command %s", cmd) if params: - response = list(self.api(cmd=cmd, **params)) - else: - response = list(self.api(cmd=cmd)) + return list(self.api(cmd=cmd, **params)) + return list(self.api(cmd=cmd)) except ( librouteros.exceptions.ConnectionClosed, OSError, socket.timeout, ) as api_error: _LOGGER.error("Mikrotik %s connection error %s", self._host, api_error) - if not self.connect_to_hub(): - raise CannotConnect from api_error + # try to reconnect + self.api = get_api(dict(self.config_entry.data)) + # we still have to raise CannotConnect to fail the update. + raise CannotConnect from api_error except librouteros.exceptions.ProtocolError as api_error: _LOGGER.warning( "Mikrotik %s failed to retrieve data. cmd=[%s] Error: %s", @@ -269,106 +271,71 @@ class MikrotikData: cmd, api_error, ) - return None - - return response if response else None + return [] class MikrotikDataUpdateCoordinator(DataUpdateCoordinator): """Mikrotik Hub Object.""" - def __init__(self, hass, config_entry): + def __init__( + self, hass: HomeAssistant, config_entry: ConfigEntry, api: librouteros.Api + ) -> None: """Initialize the Mikrotik Client.""" self.hass = hass - self.config_entry = config_entry - self._mk_data = None + self.config_entry: ConfigEntry = config_entry + self._mk_data = MikrotikData(self.hass, self.config_entry, api) super().__init__( self.hass, _LOGGER, name=f"{DOMAIN} - {self.host}", - update_method=self.async_update, update_interval=timedelta(seconds=10), ) @property - def host(self): + def host(self) -> str: """Return the host of this hub.""" return self.config_entry.data[CONF_HOST] @property - def hostname(self): + def hostname(self) -> str: """Return the hostname of the hub.""" return self._mk_data.hostname @property - def model(self): + def model(self) -> str: """Return the model of the hub.""" return self._mk_data.model @property - def firmware(self): + def firmware(self) -> str: """Return the firmware of the hub.""" return self._mk_data.firmware @property - def serial_num(self): + def serial_num(self) -> str: """Return the serial number of the hub.""" return self._mk_data.serial_number @property - def available(self): - """Return if the hub is connected.""" - return self._mk_data.available - - @property - def option_detection_time(self): + def option_detection_time(self) -> timedelta: """Config entry option defining number of seconds from last seen to away.""" - return timedelta(seconds=self.config_entry.options[CONF_DETECTION_TIME]) + return timedelta( + seconds=self.config_entry.options.get( + CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME + ) + ) @property - def api(self): + def api(self) -> MikrotikData: """Represent Mikrotik data object.""" return self._mk_data - async def async_add_options(self): - """Populate default options for Mikrotik.""" - if not self.config_entry.options: - data = dict(self.config_entry.data) - options = { - CONF_ARP_PING: data.pop(CONF_ARP_PING, False), - CONF_FORCE_DHCP: data.pop(CONF_FORCE_DHCP, False), - CONF_DETECTION_TIME: data.pop( - CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME - ), - } - - self.hass.config_entries.async_update_entry( - self.config_entry, data=data, options=options - ) - - async def async_update(self): + async def _async_update_data(self) -> None: """Update Mikrotik devices information.""" await self.hass.async_add_executor_job(self._mk_data.update_devices) - async def async_setup(self): - """Set up the Mikrotik hub.""" - try: - api = await self.hass.async_add_executor_job( - get_api, self.hass, self.config_entry.data - ) - except CannotConnect as api_error: - raise ConfigEntryNotReady from api_error - except LoginError: - return False - self._mk_data = MikrotikData(self.hass, self.config_entry, api) - await self.async_add_options() - await self.hass.async_add_executor_job(self._mk_data.get_hub_details) - - return True - - -def get_api(hass, entry): +def get_api(entry: dict[str, Any]) -> librouteros.Api: """Connect to Mikrotik hub.""" _LOGGER.debug("Connecting to Mikrotik hub [%s]", entry[CONF_HOST]) diff --git a/tests/components/mikrotik/__init__.py b/tests/components/mikrotik/__init__.py index 6f67eea1a0a..cebbd982350 100644 --- a/tests/components/mikrotik/__init__.py +++ b/tests/components/mikrotik/__init__.py @@ -1,4 +1,7 @@ """Tests for the Mikrotik component.""" +from unittest.mock import patch + +from homeassistant.components import mikrotik from homeassistant.components.mikrotik.const import ( CONF_ARP_PING, CONF_DETECTION_TIME, @@ -14,6 +17,8 @@ from homeassistant.const import ( CONF_VERIFY_SSL, ) +from tests.common import MockConfigEntry + MOCK_DATA = { CONF_NAME: "Mikrotik", CONF_HOST: "0.0.0.0", @@ -130,3 +135,38 @@ ARP_DATA = [ "disabled": False, }, ] + + +async def setup_mikrotik_entry(hass, **kwargs): + """Set up Mikrotik integration successfully.""" + support_wireless = kwargs.get("support_wireless", True) + dhcp_data = kwargs.get("dhcp_data", DHCP_DATA) + wireless_data = kwargs.get("wireless_data", WIRELESS_DATA) + + def mock_command(self, cmd, params=None): + if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.IS_WIRELESS]: + return support_wireless + if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.DHCP]: + return dhcp_data + if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.WIRELESS]: + return wireless_data + if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.ARP]: + return ARP_DATA + return {} + + config_entry = MockConfigEntry( + domain=mikrotik.DOMAIN, data=MOCK_DATA, options=MOCK_OPTIONS + ) + config_entry.add_to_hass(hass) + + if "force_dhcp" in kwargs: + config_entry.options = {**config_entry.options, "force_dhcp": True} + + if "arp_ping" in kwargs: + config_entry.options = {**config_entry.options, "arp_ping": True} + + with patch("librouteros.connect"), patch.object( + mikrotik.hub.MikrotikData, "command", new=mock_command + ): + await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() diff --git a/tests/components/mikrotik/test_device_tracker.py b/tests/components/mikrotik/test_device_tracker.py index fbbb016d09f..e3efe6bd39d 100644 --- a/tests/components/mikrotik/test_device_tracker.py +++ b/tests/components/mikrotik/test_device_tracker.py @@ -1,13 +1,14 @@ """The tests for the Mikrotik device tracker platform.""" from datetime import timedelta +from freezegun import freeze_time import pytest from homeassistant.components import mikrotik import homeassistant.components.device_tracker as device_tracker +from homeassistant.const import STATE_UNAVAILABLE from homeassistant.helpers import device_registry as dr, entity_registry as er -from homeassistant.setup import async_setup_component -import homeassistant.util.dt as dt_util +from homeassistant.util.dt import utcnow from . import ( DEVICE_2_WIRELESS, @@ -17,12 +18,10 @@ from . import ( MOCK_DATA, MOCK_OPTIONS, WIRELESS_DATA, + setup_mikrotik_entry, ) -from .test_hub import setup_mikrotik_entry -from tests.common import MockConfigEntry, patch - -DEFAULT_DETECTION_TIME = timedelta(seconds=300) +from tests.common import MockConfigEntry, async_fire_time_changed, patch @pytest.fixture @@ -56,24 +55,11 @@ def mock_command(self, cmd, params=None): return {} -async def test_platform_manually_configured(hass): - """Test that nothing happens when configuring mikrotik through device tracker platform.""" - assert ( - await async_setup_component( - hass, - device_tracker.DOMAIN, - {device_tracker.DOMAIN: {"platform": "mikrotik"}}, - ) - is False - ) - assert mikrotik.DOMAIN not in hass.data - - async def test_device_trackers(hass, mock_device_registry_devices): """Test device_trackers created by mikrotik.""" # test devices are added from wireless list only - hub = await setup_mikrotik_entry(hass) + await setup_mikrotik_entry(hass) device_1 = hass.states.get("device_tracker.device_1") assert device_1 is not None @@ -90,7 +76,7 @@ async def test_device_trackers(hass, mock_device_registry_devices): # test device_2 is added after connecting to wireless network WIRELESS_DATA.append(DEVICE_2_WIRELESS) - await hub.async_refresh() + async_fire_time_changed(hass, utcnow() + timedelta(seconds=10)) await hass.async_block_till_done() device_2 = hass.states.get("device_tracker.device_2") @@ -104,26 +90,72 @@ async def test_device_trackers(hass, mock_device_registry_devices): # test state remains home if last_seen consider_home_interval del WIRELESS_DATA[1] # device 2 is removed from wireless list - hub.api.devices["00:00:00:00:00:02"]._last_seen = dt_util.utcnow() - timedelta( - minutes=4 - ) - await hub.async_update() - await hass.async_block_till_done() + with freeze_time(utcnow() + timedelta(minutes=4)): + async_fire_time_changed(hass, utcnow() + timedelta(minutes=4)) + await hass.async_block_till_done() device_2 = hass.states.get("device_tracker.device_2") - assert device_2.state != "not_home" + assert device_2.state == "home" # test state changes to away if last_seen > consider_home_interval - hub.api.devices["00:00:00:00:00:02"]._last_seen = dt_util.utcnow() - timedelta( - minutes=5 - ) - await hub.async_refresh() - await hass.async_block_till_done() + with freeze_time(utcnow() + timedelta(minutes=6)): + async_fire_time_changed(hass, utcnow() + timedelta(minutes=6)) + await hass.async_block_till_done() device_2 = hass.states.get("device_tracker.device_2") assert device_2.state == "not_home" +async def test_force_dhcp(hass, mock_device_registry_devices): + """Test updating hub that supports wireless with forced dhcp method.""" + + # hub supports wireless by default, force_dhcp is enabled to override + await setup_mikrotik_entry(hass, force_dhcp=False) + device_1 = hass.states.get("device_tracker.device_1") + assert device_1 + assert device_1.state == "home" + # device_2 is not on the wireless list but it is still added from DHCP + device_2 = hass.states.get("device_tracker.device_2") + assert device_2 + assert device_2.state == "home" + + +async def test_hub_not_support_wireless(hass, mock_device_registry_devices): + """Test device_trackers created when hub doesn't support wireless.""" + + await setup_mikrotik_entry(hass, support_wireless=False) + device_1 = hass.states.get("device_tracker.device_1") + assert device_1 + assert device_1.state == "home" + # device_2 is added from DHCP + device_2 = hass.states.get("device_tracker.device_2") + assert device_2 + assert device_2.state == "home" + + +async def test_arp_ping_success(hass, mock_device_registry_devices): + """Test arp ping devices to confirm they are connected.""" + + with patch.object(mikrotik.hub.MikrotikData, "do_arp_ping", return_value=True): + await setup_mikrotik_entry(hass, arp_ping=True, force_dhcp=True) + + # test wired device_2 show as home if arp ping returns True + device_2 = hass.states.get("device_tracker.device_2") + assert device_2 + assert device_2.state == "home" + + +async def test_arp_ping_timeout(hass, mock_device_registry_devices): + """Test arp ping timeout so devices are shown away.""" + with patch.object(mikrotik.hub.MikrotikData, "do_arp_ping", return_value=False): + await setup_mikrotik_entry(hass, arp_ping=True, force_dhcp=True) + + # test wired device_2 show as not_home if arp ping times out + device_2 = hass.states.get("device_tracker.device_2") + assert device_2 + assert device_2.state == "not_home" + + async def test_device_trackers_numerical_name(hass, mock_device_registry_devices): """Test device_trackers created by mikrotik with numerical device name.""" @@ -164,6 +196,13 @@ async def test_restoring_devices(hass): suggested_object_id="device_2", config_entry=config_entry, ) + registry.async_get_or_create( + device_tracker.DOMAIN, + mikrotik.DOMAIN, + "00:00:00:00:00:03", + suggested_object_id="device_3", + config_entry=config_entry, + ) await setup_mikrotik_entry(hass) @@ -174,3 +213,22 @@ async def test_restoring_devices(hass): device_2 = hass.states.get("device_tracker.device_2") assert device_2 is not None assert device_2.state == "not_home" + # device_3 is not on the DHCP list or wireless list + # so it won't be restored. + device_3 = hass.states.get("device_tracker.device_3") + assert device_3 is None + + +async def test_update_failed(hass, mock_device_registry_devices): + """Test failing to connect during update.""" + + await setup_mikrotik_entry(hass) + + with patch.object( + mikrotik.hub.MikrotikData, "command", side_effect=mikrotik.errors.CannotConnect + ): + async_fire_time_changed(hass, utcnow() + timedelta(seconds=10)) + await hass.async_block_till_done() + + device_1 = hass.states.get("device_tracker.device_1") + assert device_1.state == STATE_UNAVAILABLE diff --git a/tests/components/mikrotik/test_hub.py b/tests/components/mikrotik/test_hub.py deleted file mode 100644 index 1e056071236..00000000000 --- a/tests/components/mikrotik/test_hub.py +++ /dev/null @@ -1,120 +0,0 @@ -"""Test Mikrotik hub.""" -from unittest.mock import patch - -from homeassistant.components import mikrotik - -from . import ARP_DATA, DHCP_DATA, MOCK_DATA, MOCK_OPTIONS, WIRELESS_DATA - -from tests.common import MockConfigEntry - - -async def setup_mikrotik_entry(hass, **kwargs): - """Set up Mikrotik integration successfully.""" - support_wireless = kwargs.get("support_wireless", True) - dhcp_data = kwargs.get("dhcp_data", DHCP_DATA) - wireless_data = kwargs.get("wireless_data", WIRELESS_DATA) - - def mock_command(self, cmd, params=None): - if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.IS_WIRELESS]: - return support_wireless - if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.DHCP]: - return dhcp_data - if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.WIRELESS]: - return wireless_data - if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.ARP]: - return ARP_DATA - return {} - - config_entry = MockConfigEntry( - domain=mikrotik.DOMAIN, data=MOCK_DATA, options=MOCK_OPTIONS - ) - config_entry.add_to_hass(hass) - - if "force_dhcp" in kwargs: - config_entry.options = {**config_entry.options, "force_dhcp": True} - - if "arp_ping" in kwargs: - config_entry.options = {**config_entry.options, "arp_ping": True} - - with patch("librouteros.connect"), patch.object( - mikrotik.hub.MikrotikData, "command", new=mock_command - ): - await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - return hass.data[mikrotik.DOMAIN][config_entry.entry_id] - - -async def test_update_failed(hass): - """Test failing to connect during update.""" - - hub = await setup_mikrotik_entry(hass) - - with patch.object( - mikrotik.hub.MikrotikData, "command", side_effect=mikrotik.errors.CannotConnect - ): - await hub.async_refresh() - - assert not hub.last_update_success - - -async def test_hub_not_support_wireless(hass): - """Test updating hub devices when hub doesn't support wireless interfaces.""" - - # test that the devices are constructed from dhcp data - - hub = await setup_mikrotik_entry(hass, support_wireless=False) - - assert hub.api.devices["00:00:00:00:00:01"]._params == DHCP_DATA[0] - assert hub.api.devices["00:00:00:00:00:01"]._wireless_params is None - assert hub.api.devices["00:00:00:00:00:02"]._params == DHCP_DATA[1] - assert hub.api.devices["00:00:00:00:00:02"]._wireless_params is None - - -async def test_hub_support_wireless(hass): - """Test updating hub devices when hub support wireless interfaces.""" - - # test that the device list is from wireless data list - - hub = await setup_mikrotik_entry(hass) - - assert hub.api.support_wireless is True - assert hub.api.devices["00:00:00:00:00:01"]._params == DHCP_DATA[0] - assert hub.api.devices["00:00:00:00:00:01"]._wireless_params == WIRELESS_DATA[0] - - # devices not in wireless list will not be added - assert "00:00:00:00:00:02" not in hub.api.devices - - -async def test_force_dhcp(hass): - """Test updating hub devices with forced dhcp method.""" - - # test that the devices are constructed from dhcp data - - hub = await setup_mikrotik_entry(hass, force_dhcp=True) - - assert hub.api.support_wireless is True - assert hub.api.devices["00:00:00:00:00:01"]._params == DHCP_DATA[0] - assert hub.api.devices["00:00:00:00:00:01"]._wireless_params == WIRELESS_DATA[0] - - # devices not in wireless list are added from dhcp - assert hub.api.devices["00:00:00:00:00:02"]._params == DHCP_DATA[1] - assert hub.api.devices["00:00:00:00:00:02"]._wireless_params is None - - -async def test_arp_ping(hass): - """Test arp ping devices to confirm they are connected.""" - - # test device show as home if arp ping returns value - with patch.object(mikrotik.hub.MikrotikData, "do_arp_ping", return_value=True): - hub = await setup_mikrotik_entry(hass, arp_ping=True, force_dhcp=True) - - assert hub.api.devices["00:00:00:00:00:01"].last_seen is not None - assert hub.api.devices["00:00:00:00:00:02"].last_seen is not None - - # test device show as away if arp ping times out - with patch.object(mikrotik.hub.MikrotikData, "do_arp_ping", return_value=False): - hub = await setup_mikrotik_entry(hass, arp_ping=True, force_dhcp=True) - - assert hub.api.devices["00:00:00:00:00:01"].last_seen is not None - # this device is not wireless so it will show as away - assert hub.api.devices["00:00:00:00:00:02"].last_seen is None diff --git a/tests/components/mikrotik/test_init.py b/tests/components/mikrotik/test_init.py index 5ac408928d8..3d7927174b5 100644 --- a/tests/components/mikrotik/test_init.py +++ b/tests/components/mikrotik/test_init.py @@ -39,7 +39,6 @@ async def test_successful_config_entry(hass): await hass.config_entries.async_setup(entry.entry_id) assert entry.state == ConfigEntryState.LOADED - assert hass.data[DOMAIN][entry.entry_id] async def test_hub_conn_error(hass, mock_api):