mirror of
https://github.com/home-assistant/core.git
synced 2025-07-24 13:47:35 +00:00
Add NextDNS binary sensor platform (#75266)
* Add binary_sensor platform * Add tests * Add quality scale * Sort coordinators * Remove quality scale * Fix docstring
This commit is contained in:
parent
753a3c0921
commit
6eb1dbdb74
@ -15,6 +15,7 @@ from nextdns import (
|
|||||||
AnalyticsProtocols,
|
AnalyticsProtocols,
|
||||||
AnalyticsStatus,
|
AnalyticsStatus,
|
||||||
ApiError,
|
ApiError,
|
||||||
|
ConnectionStatus,
|
||||||
InvalidApiKeyError,
|
InvalidApiKeyError,
|
||||||
NextDns,
|
NextDns,
|
||||||
Settings,
|
Settings,
|
||||||
@ -31,6 +32,7 @@ from homeassistant.helpers.entity import DeviceInfo
|
|||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||||
|
|
||||||
from .const import (
|
from .const import (
|
||||||
|
ATTR_CONNECTION,
|
||||||
ATTR_DNSSEC,
|
ATTR_DNSSEC,
|
||||||
ATTR_ENCRYPTION,
|
ATTR_ENCRYPTION,
|
||||||
ATTR_IP_VERSIONS,
|
ATTR_IP_VERSIONS,
|
||||||
@ -40,6 +42,7 @@ from .const import (
|
|||||||
CONF_PROFILE_ID,
|
CONF_PROFILE_ID,
|
||||||
DOMAIN,
|
DOMAIN,
|
||||||
UPDATE_INTERVAL_ANALYTICS,
|
UPDATE_INTERVAL_ANALYTICS,
|
||||||
|
UPDATE_INTERVAL_CONNECTION,
|
||||||
UPDATE_INTERVAL_SETTINGS,
|
UPDATE_INTERVAL_SETTINGS,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -131,10 +134,19 @@ class NextDnsSettingsUpdateCoordinator(NextDnsUpdateCoordinator[Settings]):
|
|||||||
return await self.nextdns.get_settings(self.profile_id)
|
return await self.nextdns.get_settings(self.profile_id)
|
||||||
|
|
||||||
|
|
||||||
|
class NextDnsConnectionUpdateCoordinator(NextDnsUpdateCoordinator[ConnectionStatus]):
|
||||||
|
"""Class to manage fetching NextDNS connection data from API."""
|
||||||
|
|
||||||
|
async def _async_update_data_internal(self) -> ConnectionStatus:
|
||||||
|
"""Update data via library."""
|
||||||
|
return await self.nextdns.connection_status(self.profile_id)
|
||||||
|
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
PLATFORMS = [Platform.BUTTON, Platform.SENSOR, Platform.SWITCH]
|
PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.SENSOR, Platform.SWITCH]
|
||||||
COORDINATORS = [
|
COORDINATORS = [
|
||||||
|
(ATTR_CONNECTION, NextDnsConnectionUpdateCoordinator, UPDATE_INTERVAL_CONNECTION),
|
||||||
(ATTR_DNSSEC, NextDnsDnssecUpdateCoordinator, UPDATE_INTERVAL_ANALYTICS),
|
(ATTR_DNSSEC, NextDnsDnssecUpdateCoordinator, UPDATE_INTERVAL_ANALYTICS),
|
||||||
(ATTR_ENCRYPTION, NextDnsEncryptionUpdateCoordinator, UPDATE_INTERVAL_ANALYTICS),
|
(ATTR_ENCRYPTION, NextDnsEncryptionUpdateCoordinator, UPDATE_INTERVAL_ANALYTICS),
|
||||||
(ATTR_IP_VERSIONS, NextDnsIpVersionsUpdateCoordinator, UPDATE_INTERVAL_ANALYTICS),
|
(ATTR_IP_VERSIONS, NextDnsIpVersionsUpdateCoordinator, UPDATE_INTERVAL_ANALYTICS),
|
||||||
|
103
homeassistant/components/nextdns/binary_sensor.py
Normal file
103
homeassistant/components/nextdns/binary_sensor.py
Normal file
@ -0,0 +1,103 @@
|
|||||||
|
"""Support for the NextDNS service."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Generic
|
||||||
|
|
||||||
|
from nextdns import ConnectionStatus
|
||||||
|
|
||||||
|
from homeassistant.components.binary_sensor import (
|
||||||
|
BinarySensorDeviceClass,
|
||||||
|
BinarySensorEntity,
|
||||||
|
BinarySensorEntityDescription,
|
||||||
|
)
|
||||||
|
from homeassistant.config_entries import ConfigEntry
|
||||||
|
from homeassistant.core import HomeAssistant, callback
|
||||||
|
from homeassistant.helpers.entity import EntityCategory
|
||||||
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||||
|
|
||||||
|
from . import NextDnsConnectionUpdateCoordinator, TCoordinatorData
|
||||||
|
from .const import ATTR_CONNECTION, DOMAIN
|
||||||
|
|
||||||
|
PARALLEL_UPDATES = 1
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class NextDnsBinarySensorRequiredKeysMixin(Generic[TCoordinatorData]):
|
||||||
|
"""Mixin for required keys."""
|
||||||
|
|
||||||
|
state: Callable[[TCoordinatorData, str], bool]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class NextDnsBinarySensorEntityDescription(
|
||||||
|
BinarySensorEntityDescription,
|
||||||
|
NextDnsBinarySensorRequiredKeysMixin[TCoordinatorData],
|
||||||
|
):
|
||||||
|
"""NextDNS binary sensor entity description."""
|
||||||
|
|
||||||
|
|
||||||
|
SENSORS = (
|
||||||
|
NextDnsBinarySensorEntityDescription[ConnectionStatus](
|
||||||
|
key="this_device_nextdns_connection_status",
|
||||||
|
entity_category=EntityCategory.DIAGNOSTIC,
|
||||||
|
name="This device NextDNS connection status",
|
||||||
|
device_class=BinarySensorDeviceClass.CONNECTIVITY,
|
||||||
|
state=lambda data, _: data.connected,
|
||||||
|
),
|
||||||
|
NextDnsBinarySensorEntityDescription[ConnectionStatus](
|
||||||
|
key="this_device_profile_connection_status",
|
||||||
|
entity_category=EntityCategory.DIAGNOSTIC,
|
||||||
|
name="This device profile connection status",
|
||||||
|
device_class=BinarySensorDeviceClass.CONNECTIVITY,
|
||||||
|
state=lambda data, profile_id: profile_id == data.profile_id,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entry: ConfigEntry,
|
||||||
|
async_add_entities: AddEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
"""Add NextDNS entities from a config_entry."""
|
||||||
|
coordinator: NextDnsConnectionUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
||||||
|
ATTR_CONNECTION
|
||||||
|
]
|
||||||
|
|
||||||
|
sensors: list[NextDnsBinarySensor] = []
|
||||||
|
for description in SENSORS:
|
||||||
|
sensors.append(NextDnsBinarySensor(coordinator, description))
|
||||||
|
|
||||||
|
async_add_entities(sensors)
|
||||||
|
|
||||||
|
|
||||||
|
class NextDnsBinarySensor(
|
||||||
|
CoordinatorEntity[NextDnsConnectionUpdateCoordinator], BinarySensorEntity
|
||||||
|
):
|
||||||
|
"""Define an NextDNS binary sensor."""
|
||||||
|
|
||||||
|
_attr_has_entity_name = True
|
||||||
|
entity_description: NextDnsBinarySensorEntityDescription
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
coordinator: NextDnsConnectionUpdateCoordinator,
|
||||||
|
description: NextDnsBinarySensorEntityDescription,
|
||||||
|
) -> None:
|
||||||
|
"""Initialize."""
|
||||||
|
super().__init__(coordinator)
|
||||||
|
self._attr_device_info = coordinator.device_info
|
||||||
|
self._attr_unique_id = f"{coordinator.profile_id}_{description.key}"
|
||||||
|
self._attr_is_on = description.state(coordinator.data, coordinator.profile_id)
|
||||||
|
self.entity_description = description
|
||||||
|
|
||||||
|
@callback
|
||||||
|
def _handle_coordinator_update(self) -> None:
|
||||||
|
"""Handle updated data from the coordinator."""
|
||||||
|
self._attr_is_on = self.entity_description.state(
|
||||||
|
self.coordinator.data, self.coordinator.profile_id
|
||||||
|
)
|
||||||
|
self.async_write_ha_state()
|
@ -1,6 +1,7 @@
|
|||||||
"""Constants for NextDNS integration."""
|
"""Constants for NextDNS integration."""
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
|
|
||||||
|
ATTR_CONNECTION = "connection"
|
||||||
ATTR_DNSSEC = "dnssec"
|
ATTR_DNSSEC = "dnssec"
|
||||||
ATTR_ENCRYPTION = "encryption"
|
ATTR_ENCRYPTION = "encryption"
|
||||||
ATTR_IP_VERSIONS = "ip_versions"
|
ATTR_IP_VERSIONS = "ip_versions"
|
||||||
@ -11,6 +12,7 @@ ATTR_STATUS = "status"
|
|||||||
CONF_PROFILE_ID = "profile_id"
|
CONF_PROFILE_ID = "profile_id"
|
||||||
CONF_PROFILE_NAME = "profile_name"
|
CONF_PROFILE_NAME = "profile_name"
|
||||||
|
|
||||||
|
UPDATE_INTERVAL_CONNECTION = timedelta(minutes=5)
|
||||||
UPDATE_INTERVAL_ANALYTICS = timedelta(minutes=10)
|
UPDATE_INTERVAL_ANALYTICS = timedelta(minutes=10)
|
||||||
UPDATE_INTERVAL_SETTINGS = timedelta(minutes=1)
|
UPDATE_INTERVAL_SETTINGS = timedelta(minutes=1)
|
||||||
|
|
||||||
|
@ -7,6 +7,7 @@ from nextdns import (
|
|||||||
AnalyticsIpVersions,
|
AnalyticsIpVersions,
|
||||||
AnalyticsProtocols,
|
AnalyticsProtocols,
|
||||||
AnalyticsStatus,
|
AnalyticsStatus,
|
||||||
|
ConnectionStatus,
|
||||||
Settings,
|
Settings,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -16,6 +17,7 @@ from homeassistant.core import HomeAssistant
|
|||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
|
|
||||||
|
CONNECTION_STATUS = ConnectionStatus(connected=True, profile_id="abcdef")
|
||||||
PROFILES = [{"id": "xyz12", "fingerprint": "aabbccdd123", "name": "Fake Profile"}]
|
PROFILES = [{"id": "xyz12", "fingerprint": "aabbccdd123", "name": "Fake Profile"}]
|
||||||
STATUS = AnalyticsStatus(
|
STATUS = AnalyticsStatus(
|
||||||
default_queries=40, allowed_queries=30, blocked_queries=20, relayed_queries=10
|
default_queries=40, allowed_queries=30, blocked_queries=20, relayed_queries=10
|
||||||
@ -129,6 +131,9 @@ async def init_integration(hass: HomeAssistant) -> MockConfigEntry:
|
|||||||
), patch(
|
), patch(
|
||||||
"homeassistant.components.nextdns.NextDns.get_settings",
|
"homeassistant.components.nextdns.NextDns.get_settings",
|
||||||
return_value=SETTINGS,
|
return_value=SETTINGS,
|
||||||
|
), patch(
|
||||||
|
"homeassistant.components.nextdns.NextDns.connection_status",
|
||||||
|
return_value=CONNECTION_STATUS,
|
||||||
):
|
):
|
||||||
entry.add_to_hass(hass)
|
entry.add_to_hass(hass)
|
||||||
await hass.config_entries.async_setup(entry.entry_id)
|
await hass.config_entries.async_setup(entry.entry_id)
|
||||||
|
86
tests/components/nextdns/test_binary_sensor.py
Normal file
86
tests/components/nextdns/test_binary_sensor.py
Normal file
@ -0,0 +1,86 @@
|
|||||||
|
"""Test binary sensor of NextDNS integration."""
|
||||||
|
from datetime import timedelta
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from nextdns import ApiError
|
||||||
|
|
||||||
|
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers import entity_registry as er
|
||||||
|
from homeassistant.util.dt import utcnow
|
||||||
|
|
||||||
|
from . import CONNECTION_STATUS, init_integration
|
||||||
|
|
||||||
|
from tests.common import async_fire_time_changed
|
||||||
|
|
||||||
|
|
||||||
|
async def test_binary_Sensor(hass: HomeAssistant) -> None:
|
||||||
|
"""Test states of the binary sensors."""
|
||||||
|
registry = er.async_get(hass)
|
||||||
|
|
||||||
|
await init_integration(hass)
|
||||||
|
|
||||||
|
state = hass.states.get(
|
||||||
|
"binary_sensor.fake_profile_this_device_nextdns_connection_status"
|
||||||
|
)
|
||||||
|
assert state
|
||||||
|
assert state.state == STATE_ON
|
||||||
|
|
||||||
|
entry = registry.async_get(
|
||||||
|
"binary_sensor.fake_profile_this_device_nextdns_connection_status"
|
||||||
|
)
|
||||||
|
assert entry
|
||||||
|
assert entry.unique_id == "xyz12_this_device_nextdns_connection_status"
|
||||||
|
|
||||||
|
state = hass.states.get(
|
||||||
|
"binary_sensor.fake_profile_this_device_profile_connection_status"
|
||||||
|
)
|
||||||
|
assert state
|
||||||
|
assert state.state == STATE_OFF
|
||||||
|
|
||||||
|
entry = registry.async_get(
|
||||||
|
"binary_sensor.fake_profile_this_device_profile_connection_status"
|
||||||
|
)
|
||||||
|
assert entry
|
||||||
|
assert entry.unique_id == "xyz12_this_device_profile_connection_status"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_availability(hass: HomeAssistant) -> None:
|
||||||
|
"""Ensure that we mark the entities unavailable correctly when service causes an error."""
|
||||||
|
await init_integration(hass)
|
||||||
|
|
||||||
|
state = hass.states.get(
|
||||||
|
"binary_sensor.fake_profile_this_device_nextdns_connection_status"
|
||||||
|
)
|
||||||
|
assert state
|
||||||
|
assert state.state != STATE_UNAVAILABLE
|
||||||
|
assert state.state == STATE_ON
|
||||||
|
|
||||||
|
future = utcnow() + timedelta(minutes=10)
|
||||||
|
with patch(
|
||||||
|
"homeassistant.components.nextdns.NextDns.connection_status",
|
||||||
|
side_effect=ApiError("API Error"),
|
||||||
|
):
|
||||||
|
async_fire_time_changed(hass, future)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
state = hass.states.get(
|
||||||
|
"binary_sensor.fake_profile_this_device_nextdns_connection_status"
|
||||||
|
)
|
||||||
|
assert state
|
||||||
|
assert state.state == STATE_UNAVAILABLE
|
||||||
|
|
||||||
|
future = utcnow() + timedelta(minutes=20)
|
||||||
|
with patch(
|
||||||
|
"homeassistant.components.nextdns.NextDns.connection_status",
|
||||||
|
return_value=CONNECTION_STATUS,
|
||||||
|
):
|
||||||
|
async_fire_time_changed(hass, future)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
state = hass.states.get(
|
||||||
|
"binary_sensor.fake_profile_this_device_nextdns_connection_status"
|
||||||
|
)
|
||||||
|
assert state
|
||||||
|
assert state.state != STATE_UNAVAILABLE
|
||||||
|
assert state.state == STATE_ON
|
Loading…
x
Reference in New Issue
Block a user