Cleanup ping (#95168)

.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.*

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<min>\d+)ms.+(?P<max>\d+)ms.+(?P<avg>\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

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

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

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(