Add pi_hole entity "available_updates" (#56181)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Andreas Brett 2021-11-23 23:05:27 +01:00 committed by GitHub
parent 44611d7e26
commit cee5595ba7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 207 additions and 17 deletions

View File

@ -109,6 +109,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
api_token=api_key,
)
await api.get_data()
await api.get_versions()
except HoleError as ex:
_LOGGER.warning("Failed to connect: %s", ex)
raise ConfigEntryNotReady from ex
@ -117,6 +119,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Fetch data from API endpoint."""
try:
await api.get_data()
await api.get_versions()
except HoleError as err:
raise UpdateFailed(f"Failed to communicate with API: {err}") from err
@ -150,11 +153,9 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@callback
def _async_platforms(entry: ConfigEntry) -> list[str]:
"""Return platforms to be loaded / unloaded."""
platforms = ["sensor"]
platforms = ["binary_sensor", "sensor"]
if not entry.data[CONF_STATISTICS_ONLY]:
platforms.append("switch")
else:
platforms.append("binary_sensor")
return platforms

View File

@ -1,12 +1,27 @@
"""Support for getting status from a Pi-hole system."""
from __future__ import annotations
from typing import Any
from hole import Hole
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import PiHoleEntity
from .const import DATA_KEY_API, DATA_KEY_COORDINATOR, DOMAIN as PIHOLE_DOMAIN
from .const import (
BINARY_SENSOR_TYPES,
BINARY_SENSOR_TYPES_STATISTICS_ONLY,
CONF_STATISTICS_ONLY,
DATA_KEY_API,
DATA_KEY_COORDINATOR,
DOMAIN as PIHOLE_DOMAIN,
PiHoleBinarySensorEntityDescription,
)
async def async_setup_entry(
@ -15,33 +30,63 @@ async def async_setup_entry(
"""Set up the Pi-hole binary sensor."""
name = entry.data[CONF_NAME]
hole_data = hass.data[PIHOLE_DOMAIN][entry.entry_id]
binary_sensors = [
PiHoleBinarySensor(
hole_data[DATA_KEY_API],
hole_data[DATA_KEY_COORDINATOR],
name,
entry.entry_id,
description,
)
for description in BINARY_SENSOR_TYPES
]
if entry.data[CONF_STATISTICS_ONLY]:
binary_sensors += [
PiHoleBinarySensor(
hole_data[DATA_KEY_API],
hole_data[DATA_KEY_COORDINATOR],
name,
entry.entry_id,
description,
)
for description in BINARY_SENSOR_TYPES_STATISTICS_ONLY
]
async_add_entities(binary_sensors, True)
class PiHoleBinarySensor(PiHoleEntity, BinarySensorEntity):
"""Representation of a Pi-hole binary sensor."""
_attr_icon = "mdi:pi-hole"
entity_description: PiHoleBinarySensorEntityDescription
@property
def name(self) -> str:
"""Return the name of the sensor."""
return self._name
def __init__(
self,
api: Hole,
coordinator: DataUpdateCoordinator,
name: str,
server_unique_id: str,
description: PiHoleBinarySensorEntityDescription,
) -> None:
"""Initialize a Pi-hole sensor."""
super().__init__(api, coordinator, name, server_unique_id)
self.entity_description = description
@property
def unique_id(self) -> str:
"""Return the unique id of the sensor."""
return f"{self._server_unique_id}/Status"
if description.key == "status":
self._attr_name = f"{name}"
else:
self._attr_name = f"{name} {description.name}"
self._attr_unique_id = f"{self._server_unique_id}/{description.name}"
@property
def is_on(self) -> bool:
"""Return if the service is on."""
return self.api.data.get("status") == "enabled" # type: ignore[no-any-return]
return self.entity_description.state_value(self.api)
@property
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the Pi-hole."""
return self.entity_description.extra_value(self.api)

View File

@ -1,9 +1,17 @@
"""Constants for the pi_hole integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
from typing import Any
from hole import Hole
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_UPDATE,
BinarySensorEntityDescription,
)
from homeassistant.components.sensor import SensorEntityDescription
from homeassistant.const import PERCENTAGE
@ -22,6 +30,7 @@ DEFAULT_STATISTICS_ONLY = True
SERVICE_DISABLE = "disable"
SERVICE_DISABLE_ATTR_DURATION = "duration"
ATTR_BLOCKED_DOMAINS = "domains_blocked"
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=5)
DATA_KEY_API = "api"
@ -91,3 +100,62 @@ SENSOR_TYPES: tuple[PiHoleSensorEntityDescription, ...] = (
icon="mdi:domain",
),
)
@dataclass
class RequiredPiHoleBinaryDescription:
"""Represent the required attributes of the PiHole binary description."""
state_value: Callable[[Hole], bool]
@dataclass
class PiHoleBinarySensorEntityDescription(
BinarySensorEntityDescription, RequiredPiHoleBinaryDescription
):
"""Describes PiHole binary sensor entity."""
extra_value: Callable[[Hole], dict[str, Any] | None] = lambda api: None
BINARY_SENSOR_TYPES: tuple[PiHoleBinarySensorEntityDescription, ...] = (
PiHoleBinarySensorEntityDescription(
key="core_update_available",
name="Core Update Available",
device_class=DEVICE_CLASS_UPDATE,
extra_value=lambda api: {
"current_version": api.versions["core_current"],
"latest_version": api.versions["core_latest"],
},
state_value=lambda api: bool(api.versions["core_update"]),
),
PiHoleBinarySensorEntityDescription(
key="web_update_available",
name="Web Update Available",
device_class=DEVICE_CLASS_UPDATE,
extra_value=lambda api: {
"current_version": api.versions["web_current"],
"latest_version": api.versions["web_latest"],
},
state_value=lambda api: bool(api.versions["web_update"]),
),
PiHoleBinarySensorEntityDescription(
key="ftl_update_available",
name="FTL Update Available",
device_class=DEVICE_CLASS_UPDATE,
extra_value=lambda api: {
"current_version": api.versions["FTL_current"],
"latest_version": api.versions["FTL_latest"],
},
state_value=lambda api: bool(api.versions["FTL_update"]),
),
)
BINARY_SENSOR_TYPES_STATISTICS_ONLY: tuple[PiHoleBinarySensorEntityDescription, ...] = (
PiHoleBinarySensorEntityDescription(
key="status",
name="Status",
icon="mdi:pi-hole",
state_value=lambda api: bool(api.data.get("status") == "enabled"),
),
)

View File

@ -2,7 +2,7 @@
"domain": "pi_hole",
"name": "Pi-hole",
"documentation": "https://www.home-assistant.io/integrations/pi_hole",
"requirements": ["hole==0.5.1"],
"requirements": ["hole==0.6.0"],
"codeowners": ["@fabaff", "@johnluetke", "@shenxn"],
"config_flow": true,
"iot_class": "local_polling"

View File

@ -14,6 +14,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import PiHoleEntity
from .const import (
ATTR_BLOCKED_DOMAINS,
DATA_KEY_API,
DATA_KEY_COORDINATOR,
DOMAIN as PIHOLE_DOMAIN,
@ -68,3 +69,8 @@ class PiHoleSensor(PiHoleEntity, SensorEntity):
return round(self.api.data[self.entity_description.key], 2)
except TypeError:
return self.api.data[self.entity_description.key]
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the Pi-hole."""
return {ATTR_BLOCKED_DOMAINS: self.api.data["domains_being_blocked"]}

View File

@ -813,7 +813,7 @@ hkavr==0.0.5
hlk-sw16==0.0.9
# homeassistant.components.pi_hole
hole==0.5.1
hole==0.6.0
# homeassistant.components.workday
holidays==0.11.3.1

View File

@ -509,7 +509,7 @@ herepy==2.0.0
hlk-sw16==0.0.9
# homeassistant.components.pi_hole
hole==0.5.1
hole==0.6.0
# homeassistant.components.workday
holidays==0.11.3.1

View File

@ -26,6 +26,18 @@ ZERO_DATA = {
"unique_domains": 0,
}
SAMPLE_VERSIONS = {
"core_current": "v5.5",
"core_latest": "v5.6",
"core_update": True,
"web_current": "v5.7",
"web_latest": "v5.8",
"web_update": True,
"FTL_current": "v5.10",
"FTL_latest": "v5.11",
"FTL_update": True,
}
HOST = "1.2.3.4"
PORT = 80
LOCATION = "location"
@ -75,9 +87,13 @@ def _create_mocked_hole(raise_exception=False):
type(mocked_hole).get_data = AsyncMock(
side_effect=HoleError("") if raise_exception else None
)
type(mocked_hole).get_versions = AsyncMock(
side_effect=HoleError("") if raise_exception else None
)
type(mocked_hole).enable = AsyncMock()
type(mocked_hole).disable = AsyncMock()
mocked_hole.data = ZERO_DATA
mocked_hole.versions = SAMPLE_VERSIONS
return mocked_hole

View File

@ -94,6 +94,60 @@ async def test_setup_minimal_config(hass):
assert hass.states.get("binary_sensor.pi_hole").name == "Pi-Hole"
assert hass.states.get("binary_sensor.pi_hole").state == "off"
assert (
hass.states.get("binary_sensor.pi_hole_core_update_available").name
== "Pi-Hole Core Update Available"
)
assert hass.states.get("binary_sensor.pi_hole_core_update_available").state == "on"
assert (
hass.states.get("binary_sensor.pi_hole_core_update_available").attributes[
"current_version"
]
== "v5.5"
)
assert (
hass.states.get("binary_sensor.pi_hole_core_update_available").attributes[
"latest_version"
]
== "v5.6"
)
assert (
hass.states.get("binary_sensor.pi_hole_ftl_update_available").name
== "Pi-Hole FTL Update Available"
)
assert hass.states.get("binary_sensor.pi_hole_ftl_update_available").state == "on"
assert (
hass.states.get("binary_sensor.pi_hole_ftl_update_available").attributes[
"current_version"
]
== "v5.10"
)
assert (
hass.states.get("binary_sensor.pi_hole_ftl_update_available").attributes[
"latest_version"
]
== "v5.11"
)
assert (
hass.states.get("binary_sensor.pi_hole_web_update_available").name
== "Pi-Hole Web Update Available"
)
assert hass.states.get("binary_sensor.pi_hole_web_update_available").state == "on"
assert (
hass.states.get("binary_sensor.pi_hole_web_update_available").attributes[
"current_version"
]
== "v5.7"
)
assert (
hass.states.get("binary_sensor.pi_hole_web_update_available").attributes[
"latest_version"
]
== "v5.8"
)
async def test_setup_name_config(hass):
"""Tests component setup with a custom name."""