Introduce base entity for ping (#104197)
parent
6271fe333d
commit
6f54aaf564
|
@ -18,11 +18,11 @@ import homeassistant.helpers.config_validation as cv
|
|||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import PingDomainData
|
||||
from .const import CONF_IMPORTED_BY, CONF_PING_COUNT, DEFAULT_PING_COUNT, DOMAIN
|
||||
from .coordinator import PingUpdateCoordinator
|
||||
from .entity import BasePingEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -84,20 +84,16 @@ async def async_setup_entry(
|
|||
async_add_entities([PingBinarySensor(entry, data.coordinators[entry.entry_id])])
|
||||
|
||||
|
||||
class PingBinarySensor(CoordinatorEntity[PingUpdateCoordinator], BinarySensorEntity):
|
||||
class PingBinarySensor(BasePingEntity, BinarySensorEntity):
|
||||
"""Representation of a Ping Binary sensor."""
|
||||
|
||||
_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
|
||||
_attr_available = False
|
||||
|
||||
def __init__(
|
||||
self, config_entry: ConfigEntry, coordinator: PingUpdateCoordinator
|
||||
) -> None:
|
||||
"""Initialize the Ping Binary sensor."""
|
||||
super().__init__(coordinator)
|
||||
|
||||
self._attr_name = config_entry.title
|
||||
self._attr_unique_id = config_entry.entry_id
|
||||
super().__init__(config_entry, coordinator)
|
||||
|
||||
# if this was imported just enable it when it was enabled before
|
||||
if CONF_IMPORTED_BY in config_entry.data:
|
||||
|
@ -112,7 +108,7 @@ class PingBinarySensor(CoordinatorEntity[PingUpdateCoordinator], BinarySensorEnt
|
|||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||
"""Return the state attributes of the ICMP checo request."""
|
||||
"""Return the state attributes of the ICMP echo request."""
|
||||
if self.coordinator.data.data is None:
|
||||
return None
|
||||
return {
|
||||
|
|
|
@ -8,21 +8,26 @@ import voluptuous as vol
|
|||
from homeassistant.components.device_tracker import (
|
||||
PLATFORM_SCHEMA as BASE_PLATFORM_SCHEMA,
|
||||
AsyncSeeCallback,
|
||||
ScannerEntity,
|
||||
SourceType,
|
||||
)
|
||||
from homeassistant.components.device_tracker.config_entry import BaseTrackerEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_HOSTS, CONF_NAME
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_HOSTS,
|
||||
CONF_NAME,
|
||||
STATE_HOME,
|
||||
STATE_NOT_HOME,
|
||||
)
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import PingDomainData
|
||||
from .const import CONF_IMPORTED_BY, CONF_PING_COUNT, DOMAIN
|
||||
from .coordinator import PingUpdateCoordinator
|
||||
from .entity import BasePingEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -84,37 +89,20 @@ async def async_setup_entry(
|
|||
async_add_entities([PingDeviceTracker(entry, data.coordinators[entry.entry_id])])
|
||||
|
||||
|
||||
class PingDeviceTracker(CoordinatorEntity[PingUpdateCoordinator], ScannerEntity):
|
||||
class PingDeviceTracker(BasePingEntity, BaseTrackerEntity):
|
||||
"""Representation of a Ping device tracker."""
|
||||
|
||||
def __init__(
|
||||
self, config_entry: ConfigEntry, coordinator: PingUpdateCoordinator
|
||||
) -> None:
|
||||
"""Initialize the Ping device tracker."""
|
||||
super().__init__(coordinator)
|
||||
|
||||
self._attr_name = config_entry.title
|
||||
self.config_entry = config_entry
|
||||
|
||||
@property
|
||||
def ip_address(self) -> str:
|
||||
"""Return the primary ip address of the device."""
|
||||
return self.coordinator.data.ip_address
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self.config_entry.entry_id
|
||||
|
||||
@property
|
||||
def source_type(self) -> SourceType:
|
||||
"""Return the source type which is router."""
|
||||
return SourceType.ROUTER
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
"""Return true if ping returns is_alive."""
|
||||
return self.coordinator.data.is_alive
|
||||
def state(self) -> str:
|
||||
"""Return the state of the device."""
|
||||
if self.coordinator.data.is_alive:
|
||||
return STATE_HOME
|
||||
return STATE_NOT_HOME
|
||||
|
||||
@property
|
||||
def entity_registry_enabled_default(self) -> bool:
|
||||
|
|
|
@ -0,0 +1,28 @@
|
|||
"""Base entity for Ping integration."""
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import PingUpdateCoordinator
|
||||
|
||||
|
||||
class BasePingEntity(CoordinatorEntity[PingUpdateCoordinator]):
|
||||
"""Representation of a Ping base entity."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_name = None
|
||||
|
||||
def __init__(
|
||||
self, config_entry: ConfigEntry, coordinator: PingUpdateCoordinator
|
||||
) -> None:
|
||||
"""Initialize the Ping Binary sensor."""
|
||||
super().__init__(coordinator)
|
||||
|
||||
self._attr_unique_id = config_entry.entry_id
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self.coordinator.data.ip_address)},
|
||||
manufacturer="Ping",
|
||||
)
|
||||
|
||||
self.config_entry = config_entry
|
|
@ -72,7 +72,7 @@
|
|||
'domain': 'binary_sensor',
|
||||
'entity_category': None,
|
||||
'entity_id': 'binary_sensor.10_10_10_10',
|
||||
'has_entity_name': False,
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
|
@ -81,7 +81,7 @@
|
|||
}),
|
||||
'original_device_class': <BinarySensorDeviceClass.CONNECTIVITY: 'connectivity'>,
|
||||
'original_icon': None,
|
||||
'original_name': '10.10.10.10',
|
||||
'original_name': None,
|
||||
'platform': 'ping',
|
||||
'previous_unique_id': None,
|
||||
'supported_features': 0,
|
||||
|
@ -90,6 +90,34 @@
|
|||
})
|
||||
# ---
|
||||
# name: test_setup_and_update.1
|
||||
DeviceRegistryEntrySnapshot({
|
||||
'area_id': None,
|
||||
'config_entries': <ANY>,
|
||||
'configuration_url': None,
|
||||
'connections': set({
|
||||
}),
|
||||
'disabled_by': None,
|
||||
'entry_type': None,
|
||||
'hw_version': None,
|
||||
'id': <ANY>,
|
||||
'identifiers': set({
|
||||
tuple(
|
||||
'ping',
|
||||
'10.10.10.10',
|
||||
),
|
||||
}),
|
||||
'is_new': False,
|
||||
'manufacturer': 'Ping',
|
||||
'model': None,
|
||||
'name': '10.10.10.10',
|
||||
'name_by_user': None,
|
||||
'serial_number': None,
|
||||
'suggested_area': None,
|
||||
'sw_version': None,
|
||||
'via_device_id': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_setup_and_update.2
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'connectivity',
|
||||
|
@ -106,7 +134,7 @@
|
|||
'state': 'on',
|
||||
})
|
||||
# ---
|
||||
# name: test_setup_and_update.2
|
||||
# name: test_setup_and_update.3
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'connectivity',
|
||||
|
|
|
@ -10,7 +10,11 @@ from syrupy.filters import props
|
|||
|
||||
from homeassistant.components.ping.const import CONF_IMPORTED_BY, DOMAIN
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er, issue_registry as ir
|
||||
from homeassistant.helpers import (
|
||||
device_registry as dr,
|
||||
entity_registry as er,
|
||||
issue_registry as ir,
|
||||
)
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
@ -20,6 +24,7 @@ from tests.common import MockConfigEntry
|
|||
async def test_setup_and_update(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
|
@ -29,6 +34,10 @@ async def test_setup_and_update(
|
|||
entry = entity_registry.async_get("binary_sensor.10_10_10_10")
|
||||
assert entry == snapshot(exclude=props("unique_id"))
|
||||
|
||||
# check the device
|
||||
device = device_registry.async_get_device({(DOMAIN, "10.10.10.10")})
|
||||
assert device == snapshot
|
||||
|
||||
state = hass.states.get("binary_sensor.10_10_10_10")
|
||||
assert state == snapshot
|
||||
|
||||
|
|
|
@ -1,5 +1,9 @@
|
|||
"""Test the binary sensor platform of ping."""
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from icmplib import Host
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.ping.const import DOMAIN
|
||||
|
@ -15,6 +19,7 @@ async def test_setup_and_update(
|
|||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
config_entry: MockConfigEntry,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test sensor setup and update."""
|
||||
|
||||
|
@ -42,6 +47,24 @@ async def test_setup_and_update(
|
|||
state = hass.states.get("device_tracker.10_10_10_10")
|
||||
assert state.state == "home"
|
||||
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# check device tracker is still "home"
|
||||
state = hass.states.get("device_tracker.10_10_10_10")
|
||||
assert state.state == "home"
|
||||
|
||||
# check if device tracker updates to "not home"
|
||||
with patch(
|
||||
"homeassistant.components.ping.helpers.async_ping",
|
||||
return_value=Host(address="10.10.10.10", packets_sent=10, rtts=[]),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("device_tracker.10_10_10_10")
|
||||
assert state.state == "not_home"
|
||||
|
||||
|
||||
async def test_import_issue_creation(
|
||||
hass: HomeAssistant,
|
||||
|
|
Loading…
Reference in New Issue