diff --git a/.strict-typing b/.strict-typing index 500dd076767..465892430c1 100644 --- a/.strict-typing +++ b/.strict-typing @@ -244,6 +244,7 @@ homeassistant.components.overkiz.* homeassistant.components.peco.* homeassistant.components.persistent_notification.* homeassistant.components.pi_hole.* +homeassistant.components.ping.* homeassistant.components.powerwall.* homeassistant.components.proximity.* homeassistant.components.prusalink.* diff --git a/homeassistant/components/ping/binary_sensor.py b/homeassistant/components/ping/binary_sensor.py index c8b4ce5a204..786012d466c 100644 --- a/homeassistant/components/ping/binary_sensor.py +++ b/homeassistant/components/ping/binary_sensor.py @@ -6,14 +6,14 @@ from contextlib import suppress from datetime import timedelta import logging import re -from typing import Any +from typing import TYPE_CHECKING, Any import async_timeout from icmplib import NameLookupError, async_ping import voluptuous as vol from homeassistant.components.binary_sensor import ( - PLATFORM_SCHEMA, + PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA, BinarySensorDeviceClass, BinarySensorEntity, ) @@ -53,7 +53,7 @@ PING_MATCHER_BUSYBOX = re.compile( WIN32_PING_MATCHER = re.compile(r"(?P\d+)ms.+(?P\d+)ms.+(?P\d+)ms") -PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( +PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend( { vol.Required(CONF_HOST): cv.string, vol.Optional(CONF_NAME): cv.string, @@ -89,27 +89,14 @@ async def async_setup_platform( class PingBinarySensor(RestoreEntity, BinarySensorEntity): """Representation of a Ping Binary sensor.""" + _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY + def __init__(self, name: str, ping: PingDataSubProcess | PingDataICMPLib) -> None: """Initialize the Ping Binary sensor.""" - self._available = False - self._name = name + self._attr_available = False + self._attr_name = name self._ping = ping - @property - def name(self) -> str: - """Return the name of the device.""" - return self._name - - @property - def available(self) -> bool: - """Return if we have done the first ping.""" - return self._available - - @property - def device_class(self) -> BinarySensorDeviceClass: - """Return the class of this sensor.""" - return BinarySensorDeviceClass.CONNECTIVITY - @property def is_on(self) -> bool: """Return true if the binary sensor is on.""" @@ -130,7 +117,7 @@ class PingBinarySensor(RestoreEntity, BinarySensorEntity): async def async_update(self) -> None: """Get the latest data.""" await self._ping.async_update() - self._available = True + self._attr_available = True async def async_added_to_hass(self) -> None: """Restore previous state on restart to avoid blocking startup.""" @@ -138,7 +125,7 @@ class PingBinarySensor(RestoreEntity, BinarySensorEntity): last_state = await self.async_get_last_state() if last_state is not None: - self._available = True + self._attr_available = True if last_state is None or last_state.state != STATE_ON: self._ping.data = None @@ -221,7 +208,7 @@ class PingDataSubProcess(PingData): self._ip_address, ] - async def async_ping(self): + async def async_ping(self) -> dict[str, Any] | None: """Send ICMP echo request and return details if success.""" pinger = await asyncio.create_subprocess_exec( *self._ping_cmd, @@ -249,7 +236,7 @@ class PingDataSubProcess(PingData): out_error, ) - if pinger.returncode > 1: + if pinger.returncode and pinger.returncode > 1: # returncode of 1 means the host is unreachable _LOGGER.exception( "Error running command: `%s`, return code: %s", @@ -261,9 +248,13 @@ class PingDataSubProcess(PingData): match = PING_MATCHER_BUSYBOX.search( str(out_data).rsplit("\n", maxsplit=1)[-1] ) + if TYPE_CHECKING: + assert match is not None rtt_min, rtt_avg, rtt_max = match.groups() return {"min": rtt_min, "avg": rtt_avg, "max": rtt_max, "mdev": ""} match = PING_MATCHER.search(str(out_data).rsplit("\n", maxsplit=1)[-1]) + if TYPE_CHECKING: + assert match is not None rtt_min, rtt_avg, rtt_max, rtt_mdev = match.groups() return {"min": rtt_min, "avg": rtt_avg, "max": rtt_max, "mdev": rtt_mdev} except asyncio.TimeoutError: @@ -274,7 +265,7 @@ class PingDataSubProcess(PingData): ) if pinger: with suppress(TypeError): - await pinger.kill() + await pinger.kill() # type: ignore[func-returns-value] del pinger return None diff --git a/homeassistant/components/ping/device_tracker.py b/homeassistant/components/ping/device_tracker.py index 68111df89ea..f546bd6bacc 100644 --- a/homeassistant/components/ping/device_tracker.py +++ b/homeassistant/components/ping/device_tracker.py @@ -2,14 +2,13 @@ from __future__ import annotations import asyncio -from datetime import timedelta +from datetime import datetime, timedelta import logging import subprocess from icmplib import async_multiping import voluptuous as vol -from homeassistant import util from homeassistant.components.device_tracker import ( CONF_SCAN_INTERVAL, PLATFORM_SCHEMA as BASE_PLATFORM_SCHEMA, @@ -22,6 +21,7 @@ from homeassistant.core import HomeAssistant import homeassistant.helpers.config_validation as cv from homeassistant.helpers.event import async_track_point_in_utc_time from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType +from homeassistant.util import dt as dt_util from homeassistant.util.async_ import gather_with_concurrency from homeassistant.util.process import kill_subprocess @@ -44,7 +44,14 @@ PLATFORM_SCHEMA = BASE_PLATFORM_SCHEMA.extend( class HostSubProcess: """Host object with ping detection.""" - def __init__(self, ip_address, dev_id, hass, config, privileged): + def __init__( + self, + ip_address: str, + dev_id: str, + hass: HomeAssistant, + config: ConfigType, + privileged: bool | None, + ) -> None: """Initialize the Host pinger.""" self.hass = hass self.ip_address = ip_address @@ -52,7 +59,7 @@ class HostSubProcess: self._count = config[CONF_PING_COUNT] self._ping_cmd = ["ping", "-n", "-q", "-c1", "-W1", ip_address] - def ping(self): + def ping(self) -> bool | None: """Send an ICMP echo request and return True if success.""" with subprocess.Popen( self._ping_cmd, @@ -108,7 +115,7 @@ async def async_setup_scanner( for (dev_id, ip) in config[CONF_HOSTS].items() ] - async def async_update(now): + async def async_update(now: datetime) -> None: """Update all the hosts on every interval time.""" results = await gather_with_concurrency( CONCURRENT_PING_LIMIT, @@ -124,7 +131,7 @@ async def async_setup_scanner( else: - async def async_update(now): + async def async_update(now: datetime) -> None: """Update all the hosts on every interval time.""" responses = await async_multiping( list(ip_to_dev_id), @@ -141,14 +148,14 @@ async def async_setup_scanner( ) ) - async def _async_update_interval(now): + async def _async_update_interval(now: datetime) -> None: try: await async_update(now) finally: if not hass.is_stopping: async_track_point_in_utc_time( - hass, _async_update_interval, util.dt.utcnow() + interval + hass, _async_update_interval, now + interval ) - await _async_update_interval(None) + await _async_update_interval(dt_util.now()) return True diff --git a/mypy.ini b/mypy.ini index 68095329374..e4c67bfd909 100644 --- a/mypy.ini +++ b/mypy.ini @@ -2202,6 +2202,16 @@ disallow_untyped_defs = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.ping.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.powerwall.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/tests/components/ping/test_binary_sensor.py b/tests/components/ping/test_binary_sensor.py index eae3d881fc0..3389534483f 100644 --- a/tests/components/ping/test_binary_sensor.py +++ b/tests/components/ping/test_binary_sensor.py @@ -12,13 +12,13 @@ from tests.common import get_fixture_path @pytest.fixture -def mock_ping(): +def mock_ping() -> None: """Mock icmplib.ping.""" with patch("homeassistant.components.ping.icmp_ping"): yield -async def test_reload(hass: HomeAssistant, mock_ping) -> None: +async def test_reload(hass: HomeAssistant, mock_ping: None) -> None: """Verify we can reload trend sensors.""" await setup.async_setup_component(