jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-16 01:37:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enable Pihole API v6 (#145890)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Josef Zweck <josef@zweck.dev>
This commit is contained in:
HarvsG 2025-07-05 09:21:32 +01:00 committed by GitHub
parent 1e164c94b1
commit 1b21c986e8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
22 changed files with 979 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
homeassistant/components/pi_hole/__init__.py
View File

@ -4,13 +4,15 @@ from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any, Literal
from hole import Hole
from hole.exceptions import HoleError
from hole import Hole, HoleV5, HoleV6
from hole.exceptions import HoleConnectionError, HoleError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_API_VERSION,
CONF_HOST,
CONF_LOCATION,
CONF_NAME,
@ -24,7 +26,12 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_STATISTICS_ONLY, DOMAIN, MIN_TIME_BETWEEN_UPDATES
from .const import (
CONF_STATISTICS_ONLY,
DOMAIN,
MIN_TIME_BETWEEN_UPDATES,
VERSION_6_RESPONSE_TO_5_ERROR,
)
_LOGGER = logging.getLogger(__name__)
@ -51,10 +58,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: PiHoleConfigEntry) -> bo
"""Set up Pi-hole entry."""
name = entry.data[CONF_NAME]
host = entry.data[CONF_HOST]
use_tls = entry.data[CONF_SSL]
verify_tls = entry.data[CONF_VERIFY_SSL]
location = entry.data[CONF_LOCATION]
api_key = entry.data.get(CONF_API_KEY, "")
version = entry.data.get(CONF_API_VERSION)
# remove obsolet CONF_STATISTICS_ONLY from entry.data
if CONF_STATISTICS_ONLY in entry.data:
@ -96,21 +100,42 @@ async def async_setup_entry(hass: HomeAssistant, entry: PiHoleConfigEntry) -> bo
await er.async_migrate_entries(hass, entry.entry_id, update_unique_id)
session = async_get_clientsession(hass, verify_tls)
api = Hole(
host,
session,
location=location,
tls=use_tls,
api_token=api_key,
)
if version is None:
_LOGGER.debug(
"No API version specified, determining Pi-hole API version for %s", host
)
version = await determine_api_version(hass, dict(entry.data))
_LOGGER.debug("Pi-hole API version determined: %s", version)
hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_API_VERSION: version}
)
# Once API version 5 is deprecated we should instantiate Hole directly
api = api_by_version(hass, dict(entry.data), version)
async def async_update_data() -> None:
"""Fetch data from API endpoint."""
try:
await api.get_data()
await api.get_versions()
if "error" in (response := api.data):
match response["error"]:
case {
"key": key,
"message": message,
"hint": hint,
} if (
key == VERSION_6_RESPONSE_TO_5_ERROR["key"]
and message == VERSION_6_RESPONSE_TO_5_ERROR["message"]
and hint.startswith("The API is hosted at ")
and "/admin/api" in hint
):
_LOGGER.warning(
"Pi-hole API v6 returned an error that is expected when using v5 endpoints please re-configure your authentication"
)
raise ConfigEntryAuthFailed
except HoleError as err:
if str(err) == "Authentication failed: Invalid password":
raise ConfigEntryAuthFailed from err
raise UpdateFailed(f"Failed to communicate with API: {err}") from err
if not isinstance(api.data, dict):
raise ConfigEntryAuthFailed
@ -136,3 +161,91 @@ async def async_setup_entry(hass: HomeAssistant, entry: PiHoleConfigEntry) -> bo
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Pi-hole entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
def api_by_version(
hass: HomeAssistant,
entry: dict[str, Any],
version: int,
password: str | None = None,
) -> HoleV5 | HoleV6:
"""Create a pi-hole API object by API version number. Once V5 is deprecated this function can be removed."""
if password is None:
password = entry.get(CONF_API_KEY, "")
session = async_get_clientsession(hass, entry[CONF_VERIFY_SSL])
hole_kwargs = {
"host": entry[CONF_HOST],
"session": session,
"location": entry[CONF_LOCATION],
"verify_tls": entry[CONF_VERIFY_SSL],
"version": version,
}
if version == 5:
hole_kwargs["tls"] = entry.get(CONF_SSL)
hole_kwargs["api_token"] = password
elif version == 6:
hole_kwargs["protocol"] = "https" if entry.get(CONF_SSL) else "http"
hole_kwargs["password"] = password
return Hole(**hole_kwargs)
async def determine_api_version(
hass: HomeAssistant, entry: dict[str, Any]
) -> Literal[5, 6]:
"""Determine the API version of the Pi-hole instance without requiring authentication.
Neither API v5 or v6 provides an endpoint to check the version without authentication.
Version 6 provides other enddpoints that do not require authentication, so we can use those to determine the version
version 5 returns an empty list in response to unauthenticated requests.
Because we are using endpoints that are not designed for this purpose, we should log liberally to help with debugging.
"""
holeV6 = api_by_version(hass, entry, 6, password="wrong_password")
try:
await holeV6.authenticate()
except HoleConnectionError as err:
_LOGGER.error(
"Unexpected error connecting to Pi-hole v6 API at %s: %s. Trying version 5 API",
holeV6.base_url,
err,
)
# Ideally python-hole would raise a specific exception for authentication failures
except HoleError as ex_v6:
if str(ex_v6) == "Authentication failed: Invalid password":
_LOGGER.debug(
"Success connecting to Pi-hole at %s without auth, API version is : %s",
holeV6.base_url,
6,
)
return 6
_LOGGER.debug(
"Connection to %s failed: %s, trying API version 5", holeV6.base_url, ex_v6
)
holeV5 = api_by_version(hass, entry, 5, password="wrong_token")
try:
await holeV5.get_data()
except HoleConnectionError as err:
_LOGGER.error(
"Failed to connect to Pi-hole v5 API at %s: %s", holeV5.base_url, err
)
else:
# V5 API returns [] to unauthenticated requests
if not holeV5.data:
_LOGGER.debug(
"Response '[]' from API without auth, pihole API version 5 probably detected at %s",
holeV5.base_url,
)
return 5
_LOGGER.debug(
"Unexpected response from Pi-hole API at %s: %s",
holeV5.base_url,
str(holeV5.data),
)
_LOGGER.debug(
"Could not determine pi-hole API version at: %s",
holeV6.base_url,
)
raise HoleError("Could not determine Pi-hole API version")

2
homeassistant/components/pi_hole/binary_sensor.py
View File

@ -33,7 +33,7 @@ BINARY_SENSOR_TYPES: tuple[PiHoleBinarySensorEntityDescription, ...] = (
PiHoleBinarySensorEntityDescription(
key="status",
translation_key="status",
state_value=lambda api: bool(api.data.get("status") == "enabled"),
state_value=lambda api: bool(api.status == "enabled"),
),
)

88
homeassistant/components/pi_hole/config_flow.py
View File

@ -6,13 +6,13 @@ from collections.abc import Mapping
import logging
from typing import Any
from hole import Hole
from hole.exceptions import HoleError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_API_KEY,
CONF_API_VERSION,
CONF_HOST,
CONF_LOCATION,
CONF_NAME,
@ -20,8 +20,8 @@ from homeassistant.const import (
CONF_SSL,
CONF_VERIFY_SSL,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from . import Hole, api_by_version, determine_api_version
from .const import (
DEFAULT_LOCATION,
DEFAULT_NAME,
@ -55,6 +55,7 @@ class PiHoleFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_LOCATION: user_input[CONF_LOCATION],
CONF_SSL: user_input[CONF_SSL],
CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
CONF_API_KEY: user_input[CONF_API_KEY],
}
self._async_abort_entries_match(
@ -69,9 +70,6 @@ class PiHoleFlowHandler(ConfigFlow, domain=DOMAIN):
title=user_input[CONF_NAME], data=self._config
)
if CONF_API_KEY in errors:
return await self.async_step_api_key()
user_input = user_input or {}
return self.async_show_form(
step_id="user",
@ -88,6 +86,10 @@ class PiHoleFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_LOCATION,
default=user_input.get(CONF_LOCATION, DEFAULT_LOCATION),
): str,
vol.Required(
CONF_API_KEY,
default=user_input.get(CONF_API_KEY),
): str,
vol.Required(
CONF_SSL,
default=user_input.get(CONF_SSL, DEFAULT_SSL),
@ -101,25 +103,6 @@ class PiHoleFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_api_key(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle step to setup API key."""
errors = {}
if user_input is not None:
self._config[CONF_API_KEY] = user_input[CONF_API_KEY]
if not (errors := await self._async_try_connect()):
return self.async_create_entry(
title=self._config[CONF_NAME],
data=self._config,
)
return self.async_show_form(
step_id="api_key",
data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
errors=errors,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
@ -151,19 +134,50 @@ class PiHoleFlowHandler(ConfigFlow, domain=DOMAIN):
)
async def _async_try_connect(self) -> dict[str, str]:
session = async_get_clientsession(self.hass, self._config[CONF_VERIFY_SSL])
pi_hole = Hole(
self._config[CONF_HOST],
session,
location=self._config[CONF_LOCATION],
tls=self._config[CONF_SSL],
api_token=self._config.get(CONF_API_KEY),
)
"""Try to connect to the Pi-hole API and determine the version."""
try:
await pi_hole.get_data()
except HoleError as ex:
_LOGGER.debug("Connection failed: %s", ex)
version = await determine_api_version(hass=self.hass, entry=self._config)
except HoleError:
return {"base": "cannot_connect"}
if not isinstance(pi_hole.data, dict):
return {CONF_API_KEY: "invalid_auth"}
pi_hole: Hole = api_by_version(self.hass, self._config, version)
if version == 6:
try:
await pi_hole.authenticate()
_LOGGER.debug("Success authenticating with pihole API version: %s", 6)
self._config[CONF_API_VERSION] = 6
except HoleError:
_LOGGER.debug("Failed authenticating with pihole API version: %s", 6)
return {CONF_API_KEY: "invalid_auth"}
elif version == 5:
try:
await pi_hole.get_data()
if pi_hole.data is not None and "error" in pi_hole.data:
_LOGGER.debug(
"API version %s returned an unexpected error: %s",
5,
str(pi_hole.data),
)
raise HoleError(pi_hole.data) # noqa: TRY301
except HoleError as ex_v5:
_LOGGER.error(
"Connection to API version 5 failed: %s",
ex_v5,
)
return {"base": "cannot_connect"}
else:
_LOGGER.debug(
"Success connecting to, but necessarily authenticating with, pihole, API version is: %s",
5,
)
self._config[CONF_API_VERSION] = 5
# the v5 API returns an empty list to unauthenticated requests.
if not isinstance(pi_hole.data, dict):
_LOGGER.debug(
"API version %s returned %s, '[]' is expected for unauthenticated requests",
5,
pi_hole.data,
)
return {CONF_API_KEY: "invalid_auth"}
return {}

7
homeassistant/components/pi_hole/const.py
View File

@ -17,3 +17,10 @@ SERVICE_DISABLE = "disable"
SERVICE_DISABLE_ATTR_DURATION = "duration"
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=5)
# See https://github.com/pi-hole/FTL/blob/88737f6248cd3df3202eed72aeec89b9fb572631/src/webserver/lua_web.c#L83
VERSION_6_RESPONSE_TO_5_ERROR = {
"key": "bad_request",
"message": "Bad request",
"hint": "The API is hosted at pi.hole/api, not pi.hole/admin/api",
}

5
homeassistant/components/pi_hole/entity.py
View File

@ -32,7 +32,10 @@ class PiHoleEntity(CoordinatorEntity[DataUpdateCoordinator[None]]):
@property
def device_info(self) -> DeviceInfo:
"""Return the device information of the entity."""
if self.api.tls:
if (
getattr(self.api, "tls", None) # API version 5
or getattr(self.api, "protocol", None) == "https" # API version 6
):
config_url = f"https://{self.api.host}/{self.api.location}"
else:
config_url = f"http://{self.api.host}/{self.api.location}"

9
homeassistant/components/pi_hole/icons.json
View File

@ -9,15 +9,24 @@
"ads_blocked_today": {
"default": "mdi:close-octagon-outline"
},
"ads_blocked": {
"default": "mdi:close-octagon-outline"
},
"ads_percentage_today": {
"default": "mdi:close-octagon-outline"
},
"percent_ads_blocked": {
"default": "mdi:close-octagon-outline"
},
"clients_ever_seen": {
"default": "mdi:account-outline"
},
"dns_queries_today": {
"default": "mdi:comment-question-outline"
},
"dns_queries": {
"default": "mdi:comment-question-outline"
},
"domains_being_blocked": {
"default": "mdi:block-helper"
},

2
homeassistant/components/pi_hole/manifest.json
View File

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/pi_hole",
"iot_class": "local_polling",
"loggers": ["hole"],
"requirements": ["hole==0.8.0"]
"requirements": ["hole==0.9.0"]
}

108
homeassistant/components/pi_hole/sensor.py
View File

@ -2,10 +2,13 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from hole import Hole
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.const import CONF_NAME, PERCENTAGE
from homeassistant.const import CONF_API_VERSION, CONF_NAME, PERCENTAGE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
@ -18,29 +21,98 @@ SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="ads_blocked_today",
translation_key="ads_blocked_today",
suggested_display_precision=0,
),
SensorEntityDescription(
key="ads_percentage_today",
translation_key="ads_percentage_today",
native_unit_of_measurement=PERCENTAGE,
suggested_display_precision=1,
),
SensorEntityDescription(
key="clients_ever_seen",
translation_key="clients_ever_seen",
suggested_display_precision=0,
),
SensorEntityDescription(
key="dns_queries_today", translation_key="dns_queries_today"
key="dns_queries_today",
translation_key="dns_queries_today",
suggested_display_precision=0,
),
SensorEntityDescription(
key="domains_being_blocked",
translation_key="domains_being_blocked",
suggested_display_precision=0,
),
SensorEntityDescription(key="queries_cached", translation_key="queries_cached"),
SensorEntityDescription(
key="queries_forwarded", translation_key="queries_forwarded"
key="queries_cached",
translation_key="queries_cached",
suggested_display_precision=0,
),
SensorEntityDescription(
key="queries_forwarded",
translation_key="queries_forwarded",
suggested_display_precision=0,
),
SensorEntityDescription(
key="unique_clients",
translation_key="unique_clients",
suggested_display_precision=0,
),
SensorEntityDescription(
key="unique_domains",
translation_key="unique_domains",
suggested_display_precision=0,
),
)
SENSOR_TYPES_V6: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="queries.blocked",
translation_key="ads_blocked",
suggested_display_precision=0,
),
SensorEntityDescription(
key="queries.percent_blocked",
translation_key="percent_ads_blocked",
native_unit_of_measurement=PERCENTAGE,
suggested_display_precision=2,
),
SensorEntityDescription(
key="clients.total",
translation_key="clients_ever_seen",
suggested_display_precision=0,
),
SensorEntityDescription(
key="queries.total",
translation_key="dns_queries",
suggested_display_precision=0,
),
SensorEntityDescription(
key="gravity.domains_being_blocked",
translation_key="domains_being_blocked",
suggested_display_precision=0,
),
SensorEntityDescription(
key="queries.cached",
translation_key="queries_cached",
suggested_display_precision=0,
),
SensorEntityDescription(
key="queries.forwarded",
translation_key="queries_forwarded",
suggested_display_precision=0,
),
SensorEntityDescription(
key="clients.active",
translation_key="unique_clients",
suggested_display_precision=0,
),
SensorEntityDescription(
key="queries.unique_domains",
translation_key="unique_domains",
suggested_display_precision=0,
),
SensorEntityDescription(key="unique_clients", translation_key="unique_clients"),
SensorEntityDescription(key="unique_domains", translation_key="unique_domains"),
)
@ -60,7 +132,9 @@ async def async_setup_entry(
entry.entry_id,
description,
)
for description in SENSOR_TYPES
for description in (
SENSOR_TYPES if entry.data[CONF_API_VERSION] == 5 else SENSOR_TYPES_V6
)
]
async_add_entities(sensors, True)
@ -88,7 +162,19 @@ class PiHoleSensor(PiHoleEntity, SensorEntity):
@property
def native_value(self) -> StateType:
"""Return the state of the device."""
try:
return round(self.api.data[self.entity_description.key], 2) # type: ignore[no-any-return]
except TypeError:
return self.api.data[self.entity_description.key] # type: ignore[no-any-return]
return get_nested(self.api.data, self.entity_description.key)
def get_nested(data: Mapping[str, Any], key: str) -> float | int:
"""Get a value from a nested dictionary using a dot-separated key.
Ensures type safety as it iterates into the dict.
"""
current: Any = data
for part in key.split("."):
if not isinstance(current, Mapping):
raise KeyError(f"Cannot access '{part}' in non-dict {current!r}")
current = current[part]
if not isinstance(current, (float, int)):
raise TypeError(f"Value at '{key}' is not a float or int: {current!r}")
return current

27
homeassistant/components/pi_hole/strings.json
View File

@ -8,14 +8,11 @@
"name": "[%key:common::config_flow::data::name%]",
"location": "[%key:common::config_flow::data::location%]",
"ssl": "[%key:common::config_flow::data::ssl%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
}
},
"api_key": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]",
"api_key": "App Password or API Key"
}
},
"reauth_confirm": {
"title": "Reauthenticate Pi-hole",
"description": "Please enter a new API key for Pi-hole at {host}/{location}",
@ -33,6 +30,12 @@
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"issues": {
"v5_to_v6_migration": {
"title": "Recent migration from Pi-hole API v5 to v6",
"description": "You've likely updated your Pi-hole to API v6 from v5. Some sensors changed in the new API, the daily sensors were removed, and your old API token is invalid. Provide your new app password by re-authenticating in repairs or in **Settings -> Devices & services -> Pi-hole**."
}
},
"entity": {
"binary_sensor": {
"status": {
@ -44,9 +47,17 @@
"name": "Ads blocked today",
"unit_of_measurement": "ads"
},
"ads_blocked": {
"name": "Ads blocked",
"unit_of_measurement": "ads"
},
"ads_percentage_today": {
"name": "Ads percentage blocked today"
},
"percent_ads_blocked": {
"name": "Ads percentage blocked"
},
"clients_ever_seen": {
"name": "Seen clients",
"unit_of_measurement": "clients"
@ -55,6 +66,10 @@
"name": "DNS queries today",
"unit_of_measurement": "queries"
},
"dns_queries": {
"name": "DNS queries",
"unit_of_measurement": "queries"
},
"domains_being_blocked": {
"name": "Domains blocked",
"unit_of_measurement": "domains"

2
homeassistant/components/pi_hole/switch.py
View File

@ -70,7 +70,7 @@ class PiHoleSwitch(PiHoleEntity, SwitchEntity):
@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.api.status == "enabled" # type: ignore[no-any-return]
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the service."""

30
homeassistant/components/pi_hole/update.py
View File

@ -21,9 +21,9 @@ from .entity import PiHoleEntity
class PiHoleUpdateEntityDescription(UpdateEntityDescription):
"""Describes PiHole update entity."""
installed_version: Callable[[dict], str | None] = lambda api: None
latest_version: Callable[[dict], str | None] = lambda api: None
has_update: Callable[[dict], bool | None] = lambda api: None
installed_version: Callable[[Hole], str | None] = lambda api: None
latest_version: Callable[[Hole], str | None] = lambda api: None
has_update: Callable[[Hole], bool | None] = lambda api: None
release_base_url: str | None = None
title: str | None = None
@ -34,9 +34,9 @@ UPDATE_ENTITY_TYPES: tuple[PiHoleUpdateEntityDescription, ...] = (
translation_key="core_update_available",
title="Pi-hole Core",
entity_category=EntityCategory.DIAGNOSTIC,
installed_version=lambda versions: versions.get("core_current"),
latest_version=lambda versions: versions.get("core_latest"),
has_update=lambda versions: versions.get("core_update"),
installed_version=lambda api: api.core_current,
latest_version=lambda api: api.core_latest,
has_update=lambda api: api.core_update,
release_base_url="https://github.com/pi-hole/pi-hole/releases/tag",
),
PiHoleUpdateEntityDescription(
@ -44,9 +44,9 @@ UPDATE_ENTITY_TYPES: tuple[PiHoleUpdateEntityDescription, ...] = (
translation_key="web_update_available",
title="Pi-hole Web interface",
entity_category=EntityCategory.DIAGNOSTIC,
installed_version=lambda versions: versions.get("web_current"),
latest_version=lambda versions: versions.get("web_latest"),
has_update=lambda versions: versions.get("web_update"),
installed_version=lambda api: api.web_current,
latest_version=lambda api: api.web_latest,
has_update=lambda api: api.web_update,
release_base_url="https://github.com/pi-hole/AdminLTE/releases/tag",
),
PiHoleUpdateEntityDescription(
@ -54,9 +54,9 @@ UPDATE_ENTITY_TYPES: tuple[PiHoleUpdateEntityDescription, ...] = (
translation_key="ftl_update_available",
title="Pi-hole FTL DNS",
entity_category=EntityCategory.DIAGNOSTIC,
installed_version=lambda versions: versions.get("FTL_current"),
latest_version=lambda versions: versions.get("FTL_latest"),
has_update=lambda versions: versions.get("FTL_update"),
installed_version=lambda api: api.ftl_current,
latest_version=lambda api: api.ftl_latest,
has_update=lambda api: api.ftl_update,
release_base_url="https://github.com/pi-hole/FTL/releases/tag",
),
)
@ -108,15 +108,15 @@ class PiHoleUpdateEntity(PiHoleEntity, UpdateEntity):
def installed_version(self) -> str | None:
"""Version installed and in use."""
if isinstance(self.api.versions, dict):
return self.entity_description.installed_version(self.api.versions)
return self.entity_description.installed_version(self.api)
return None
@property
def latest_version(self) -> str | None:
"""Latest version available for install."""
if isinstance(self.api.versions, dict):
if self.entity_description.has_update(self.api.versions):
return self.entity_description.latest_version(self.api.versions)
if self.entity_description.has_update(self.api):
return self.entity_description.latest_version(self.api)
return self.installed_version
return None

2
requirements_all.txt generated
View File

@ -1161,7 +1161,7 @@ hko==0.3.2
hlk-sw16==0.0.9
# homeassistant.components.pi_hole
hole==0.8.0
hole==0.9.0
# homeassistant.components.holiday
# homeassistant.components.workday

2
requirements_test_all.txt generated
View File

@ -1010,7 +1010,7 @@ hko==0.3.2
hlk-sw16==0.0.9
# homeassistant.components.pi_hole
hole==0.8.0
hole==0.9.0
# homeassistant.components.holiday
# homeassistant.components.workday

1
script/hassfest/requirements.py
View File

@ -246,7 +246,6 @@ FORBIDDEN_PACKAGE_EXCEPTIONS: dict[str, dict[str, set[str]]] = {
# opower > arrow > types-python-dateutil
"arrow": {"types-python-dateutil"}
},
"pi_hole": {"hole": {"async-timeout"}},
"pvpc_hourly_pricing": {"aiopvpc": {"async-timeout"}},
"remote_rpi_gpio": {
# https://github.com/waveform80/colorzero/issues/9

215
tests/components/pi_hole/__init__.py
View File

@ -1,8 +1,9 @@
"""Tests for the pi_hole component."""
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from hole.exceptions import HoleError
from hole.exceptions import HoleConnectionError, HoleError
from homeassistant.components.pi_hole.const import (
DEFAULT_LOCATION,
@ -12,6 +13,7 @@ from homeassistant.components.pi_hole.const import (
)
from homeassistant.const import (
CONF_API_KEY,
CONF_API_VERSION,
CONF_HOST,
CONF_LOCATION,
CONF_NAME,
@ -32,6 +34,82 @@ ZERO_DATA = {
"unique_clients": 0,
"unique_domains": 0,
}
ZERO_DATA_V6 = {
"queries": {
"total": 0,
"blocked": 0,
"percent_blocked": 0,
"unique_domains": 0,
"forwarded": 0,
"cached": 0,
"frequency": 0,
"types": {
"A": 0,
"AAAA": 0,
"ANY": 0,
"SRV": 0,
"SOA": 0,
"PTR": 0,
"TXT": 0,
"NAPTR": 0,
"MX": 0,
"DS": 0,
"RRSIG": 0,
"DNSKEY": 0,
"NS": 0,
"SVCB": 0,
"HTTPS": 0,
"OTHER": 0,
},
"status": {
"UNKNOWN": 0,
"GRAVITY": 0,
"FORWARDED": 0,
"CACHE": 0,
"REGEX": 0,
"DENYLIST": 0,
"EXTERNAL_BLOCKED_IP": 0,
"EXTERNAL_BLOCKED_NULL": 0,
"EXTERNAL_BLOCKED_NXRA": 0,
"GRAVITY_CNAME": 0,
"REGEX_CNAME": 0,
"DENYLIST_CNAME": 0,
"RETRIED": 0,
"RETRIED_DNSSEC": 0,
"IN_PROGRESS": 0,
"DBBUSY": 0,
"SPECIAL_DOMAIN": 0,
"CACHE_STALE": 0,
"EXTERNAL_BLOCKED_EDE15": 0,
},
"replies": {
"UNKNOWN": 0,
"NODATA": 0,
"NXDOMAIN": 0,
"CNAME": 0,
"IP": 0,
"DOMAIN": 0,
"RRNAME": 0,
"SERVFAIL": 0,
"REFUSED": 0,
"NOTIMP": 0,
"OTHER": 0,
"DNSSEC": 0,
"NONE": 0,
"BLOB": 0,
},
},
"clients": {"active": 0, "total": 0},
"gravity": {"domains_being_blocked": 0, "last_update": 0},
"took": 0,
}
FTL_ERROR = {
"error": {
"key": "FTLnotrunning",
"message": "FTL not running",
}
}
SAMPLE_VERSIONS_WITH_UPDATES = {
"core_current": "v5.5",
@ -62,6 +140,7 @@ PORT = 80
LOCATION = "location"
NAME = "Pi hole"
API_KEY = "apikey"
API_VERSION = 6
SSL = False
VERIFY_SSL = True
@ -72,6 +151,7 @@ CONFIG_DATA_DEFAULTS = {
CONF_SSL: DEFAULT_SSL,
CONF_VERIFY_SSL: DEFAULT_VERIFY_SSL,
CONF_API_KEY: API_KEY,
CONF_API_VERSION: API_VERSION,
}
CONFIG_DATA = {
@ -81,12 +161,14 @@ CONFIG_DATA = {
CONF_API_KEY: API_KEY,
CONF_SSL: SSL,
CONF_VERIFY_SSL: VERIFY_SSL,
CONF_API_VERSION: API_VERSION,
}
CONFIG_FLOW_USER = {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_LOCATION: LOCATION,
CONF_API_KEY: API_KEY,
CONF_NAME: NAME,
CONF_SSL: SSL,
CONF_VERIFY_SSL: VERIFY_SSL,
@ -103,6 +185,7 @@ CONFIG_ENTRY_WITH_API_KEY = {
CONF_API_KEY: API_KEY,
CONF_SSL: SSL,
CONF_VERIFY_SSL: VERIFY_SSL,
CONF_API_VERSION: API_VERSION,
}
CONFIG_ENTRY_WITHOUT_API_KEY = {
@ -111,47 +194,129 @@ CONFIG_ENTRY_WITHOUT_API_KEY = {
CONF_NAME: NAME,
CONF_SSL: SSL,
CONF_VERIFY_SSL: VERIFY_SSL,
CONF_API_VERSION: API_VERSION,
}
SWITCH_ENTITY_ID = "switch.pi_hole"
def _create_mocked_hole(
raise_exception=False, has_versions=True, has_update=True, has_data=True
):
mocked_hole = MagicMock()
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()
if has_data:
mocked_hole.data = ZERO_DATA
else:
mocked_hole.data = []
if has_versions:
if has_update:
mocked_hole.versions = SAMPLE_VERSIONS_WITH_UPDATES
raise_exception: bool = False,
has_versions: bool = True,
has_update: bool = True,
has_data: bool = True,
api_version: int = 5,
incorrect_app_password: bool = False,
wrong_host: bool = False,
ftl_error: bool = False,
) -> MagicMock:
"""Return a mocked Hole API object with side effects based on constructor args."""
instances = []
def make_mock(**kwargs: Any) -> MagicMock:
mocked_hole = MagicMock()
# Set constructor kwargs as attributes
for key, value in kwargs.items():
setattr(mocked_hole, key, value)
async def authenticate_side_effect(*_args, **_kwargs):
if wrong_host:
raise HoleConnectionError("Cannot authenticate with Pi-hole: err")
password = getattr(mocked_hole, "password", None)
if (
raise_exception
or incorrect_app_password
or (api_version == 6 and password not in ["newkey", "apikey"])
):
if api_version == 6:
raise HoleError("Authentication failed: Invalid password")
raise HoleConnectionError
async def get_data_side_effect(*_args, **_kwargs):
"""Return data based on the mocked Hole instance state."""
if wrong_host:
raise HoleConnectionError("Cannot fetch data from Pi-hole: err")
password = getattr(mocked_hole, "password", None)
api_token = getattr(mocked_hole, "api_token", None)
if (
raise_exception
or incorrect_app_password
or (api_version == 5 and (not api_token or api_token == "wrong_token"))
or (api_version == 6 and password not in ["newkey", "apikey"])
):
mocked_hole.data = [] if api_version == 5 else {}
elif password in ["newkey", "apikey"] or api_token in ["newkey", "apikey"]:
mocked_hole.data = ZERO_DATA_V6 if api_version == 6 else ZERO_DATA
async def ftl_side_effect():
mocked_hole.data = FTL_ERROR
mocked_hole.authenticate = AsyncMock(side_effect=authenticate_side_effect)
mocked_hole.get_data = AsyncMock(side_effect=get_data_side_effect)
if ftl_error:
# two unauthenticated instances are created in `determine_api_version` before aync_try_connect is called
if len(instances) > 1:
mocked_hole.get_data = AsyncMock(side_effect=ftl_side_effect)
mocked_hole.get_versions = AsyncMock(return_value=None)
mocked_hole.enable = AsyncMock()
mocked_hole.disable = AsyncMock()
# Set versions and version properties
if has_versions:
versions = (
SAMPLE_VERSIONS_WITH_UPDATES
if has_update
else SAMPLE_VERSIONS_NO_UPDATES
)
mocked_hole.versions = versions
mocked_hole.ftl_current = versions["FTL_current"]
mocked_hole.ftl_latest = versions["FTL_latest"]
mocked_hole.ftl_update = versions["FTL_update"]
mocked_hole.core_current = versions["core_current"]
mocked_hole.core_latest = versions["core_latest"]
mocked_hole.core_update = versions["core_update"]
mocked_hole.web_current = versions["web_current"]
mocked_hole.web_latest = versions["web_latest"]
mocked_hole.web_update = versions["web_update"]
else:
mocked_hole.versions = SAMPLE_VERSIONS_NO_UPDATES
else:
mocked_hole.versions = None
return mocked_hole
mocked_hole.versions = None
# Set initial data
if has_data:
mocked_hole.data = ZERO_DATA_V6 if api_version == 6 else ZERO_DATA
else:
mocked_hole.data = [] if api_version == 5 else {}
instances.append(mocked_hole)
return mocked_hole
# Return a factory function for patching
make_mock.instances = instances
return make_mock
def _patch_init_hole(mocked_hole):
return patch("homeassistant.components.pi_hole.Hole", return_value=mocked_hole)
"""Patch the Hole class in the main integration."""
def side_effect(*args, **kwargs):
return mocked_hole(**kwargs)
return patch("homeassistant.components.pi_hole.Hole", side_effect=side_effect)
def _patch_config_flow_hole(mocked_hole):
"""Patch the Hole class in the config flow."""
def side_effect(*args, **kwargs):
return mocked_hole(**kwargs)
return patch(
"homeassistant.components.pi_hole.config_flow.Hole", return_value=mocked_hole
"homeassistant.components.pi_hole.config_flow.Hole", side_effect=side_effect
)
def _patch_setup_hole():
"""Patch async_setup_entry for the integration."""
return patch(
"homeassistant.components.pi_hole.async_setup_entry", return_value=True
)

1
tests/components/pi_hole/snapshots/test_diagnostics.ambr
View File

@ -16,6 +16,7 @@
'entry': dict({
'data': dict({
'api_key': '**REDACTED**',
'api_version': 5,
'host': '1.2.3.4:80',
'location': 'admin',
'name': 'Pi-Hole',

129
tests/components/pi_hole/test_config_flow.py
View File

@ -3,16 +3,15 @@
from homeassistant.components import pi_hole
from homeassistant.components.pi_hole.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_KEY
from homeassistant.const import CONF_API_KEY, CONF_API_VERSION
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import (
CONFIG_DATA_DEFAULTS,
CONFIG_ENTRY_WITH_API_KEY,
CONFIG_ENTRY_WITHOUT_API_KEY,
CONFIG_FLOW_API_KEY,
CONFIG_FLOW_USER,
FTL_ERROR,
NAME,
ZERO_DATA,
_create_mocked_hole,
@ -24,10 +23,14 @@ from . import (
from tests.common import MockConfigEntry
async def test_flow_user_with_api_key(hass: HomeAssistant) -> None:
async def test_flow_user_with_api_key_v6(hass: HomeAssistant) -> None:
"""Test user initialized flow with api key needed."""
mocked_hole = _create_mocked_hole(has_data=False)
with _patch_config_flow_hole(mocked_hole), _patch_setup_hole() as mock_setup:
mocked_hole = _create_mocked_hole(has_data=False, api_version=6)
with (
_patch_init_hole(mocked_hole),
_patch_config_flow_hole(mocked_hole),
_patch_setup_hole() as mock_setup,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
@ -38,27 +41,19 @@ async def test_flow_user_with_api_key(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG_FLOW_USER,
user_input={**CONFIG_FLOW_USER, CONF_API_KEY: "invalid_password"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "api_key"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_API_KEY: "some_key"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "api_key"
# we have had no response from the server yet, so we expect an error
assert result["errors"] == {CONF_API_KEY: "invalid_auth"}
mocked_hole.data = ZERO_DATA
# now we have a valid passiword
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG_FLOW_API_KEY,
user_input=CONFIG_FLOW_USER,
)
# form should be complete with a valid config entry
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == NAME
assert result["data"] == CONFIG_ENTRY_WITH_API_KEY
mock_setup.assert_called_once()
@ -72,10 +67,15 @@ async def test_flow_user_with_api_key(hass: HomeAssistant) -> None:
assert result["reason"] == "already_configured"
async def test_flow_user_without_api_key(hass: HomeAssistant) -> None:
"""Test user initialized flow without api key needed."""
mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_setup_hole() as mock_setup:
async def test_flow_user_with_api_key_v5(hass: HomeAssistant) -> None:
"""Test user initialized flow with api key needed."""
mocked_hole = _create_mocked_hole(api_version=5)
with (
_patch_init_hole(mocked_hole),
_patch_config_flow_hole(mocked_hole),
_patch_setup_hole() as mock_setup,
):
# start the flow as a user initiated flow
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
@ -84,32 +84,72 @@ async def test_flow_user_without_api_key(hass: HomeAssistant) -> None:
assert result["step_id"] == "user"
assert result["errors"] == {}
# configure the flow with an invalid api key
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={**CONFIG_FLOW_USER, CONF_API_KEY: "wrong_token"},
)
# confirm an invalid authentication error
assert result["errors"] == {CONF_API_KEY: "invalid_auth"}
# configure the flow with a valid api key
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG_FLOW_USER,
)
# in API V5 we get data to confirm authentication
assert mocked_hole.instances[-1].data == ZERO_DATA
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == NAME
assert result["data"] == CONFIG_ENTRY_WITHOUT_API_KEY
assert result["data"] == {**CONFIG_ENTRY_WITH_API_KEY, CONF_API_VERSION: 5}
mock_setup.assert_called_once()
# duplicated server
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_FLOW_USER,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_flow_user_invalid(hass: HomeAssistant) -> None:
"""Test user initialized flow with invalid server."""
"""Test user initialized flow with completely invalid server."""
mocked_hole = _create_mocked_hole(raise_exception=True)
with _patch_config_flow_hole(mocked_hole):
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_FLOW_USER
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
assert result["errors"] == {"api_key": "invalid_auth"}
async def test_flow_user_invalid_v6(hass: HomeAssistant) -> None:
"""Test user initialized flow with invalid server - typically a V6 API and a incorrect app password."""
mocked_hole = _create_mocked_hole(
has_data=True, api_version=6, incorrect_app_password=True
)
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_FLOW_USER
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"api_key": "invalid_auth"}
async def test_flow_reauth(hass: HomeAssistant) -> None:
"""Test reauth flow."""
mocked_hole = _create_mocked_hole(has_data=False)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONFIG_DATA_DEFAULTS)
mocked_hole = _create_mocked_hole(has_data=False, api_version=5)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN,
data={**CONFIG_DATA_DEFAULTS, CONF_API_VERSION: 5, CONF_API_KEY: "oldkey"},
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole), _patch_config_flow_hole(mocked_hole):
assert not await hass.config_entries.async_setup(entry.entry_id)
@ -120,9 +160,7 @@ async def test_flow_reauth(hass: HomeAssistant) -> None:
assert len(flows) == 1
assert flows[0]["step_id"] == "reauth_confirm"
assert flows[0]["context"]["entry_id"] == entry.entry_id
mocked_hole.data = ZERO_DATA
mocked_hole.instances[-1].api_token = "newkey"
result = await hass.config_entries.flow.async_configure(
flows[0]["flow_id"],
user_input={CONF_API_KEY: "newkey"},
@ -131,3 +169,28 @@ async def test_flow_reauth(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_API_KEY] == "newkey"
async def test_flow_user_invalid_host(hass: HomeAssistant) -> None:
"""Test user initialized flow with invalid server host address."""
mocked_hole = _create_mocked_hole(api_version=6, wrong_host=True)
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_FLOW_USER
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_flow_error_response(hass: HomeAssistant) -> None:
"""Test user initialized flow but dataotherbase errors occur."""
mocked_hole = _create_mocked_hole(api_version=5, ftl_error=True, has_data=False)
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_FLOW_USER
)
assert mocked_hole.instances[-1].data == FTL_ERROR
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}

5
tests/components/pi_hole/test_diagnostics.py
View File

@ -19,9 +19,10 @@ async def test_diagnostics(
snapshot: SnapshotAssertion,
) -> None:
"""Tests diagnostics."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=5)
config_entry = {**CONFIG_DATA_DEFAULTS, "api_version": 5}
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data=CONFIG_DATA_DEFAULTS, entry_id="pi_hole_mock_entry"
domain=pi_hole.DOMAIN, data=config_entry, entry_id="pi_hole_mock_entry"
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):

155
tests/components/pi_hole/test_init.py
View File

@ -16,6 +16,7 @@ from homeassistant.components.pi_hole.const import (
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_API_VERSION,
CONF_HOST,
CONF_LOCATION,
CONF_NAME,
@ -27,7 +28,7 @@ from . import (
API_KEY,
CONFIG_DATA,
CONFIG_DATA_DEFAULTS,
CONFIG_ENTRY_WITHOUT_API_KEY,
DEFAULT_VERIFY_SSL,
SWITCH_ENTITY_ID,
_create_mocked_hole,
_patch_init_hole,
@ -38,32 +39,62 @@ from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("config_entry_data", "expected_api_token"),
[(CONFIG_DATA_DEFAULTS, API_KEY), (CONFIG_ENTRY_WITHOUT_API_KEY, "")],
[(CONFIG_DATA_DEFAULTS, API_KEY)],
)
async def test_setup_api(
async def test_setup_api_v6(
hass: HomeAssistant, config_entry_data: dict, expected_api_token: str
) -> None:
"""Tests the API object is created with the expected parameters."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=6)
config_entry_data = {**config_entry_data}
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=config_entry_data)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole) as patched_init_hole:
assert await hass.config_entries.async_setup(entry.entry_id)
patched_init_hole.assert_called_once_with(
host=config_entry_data[CONF_HOST],
session=ANY,
password=expected_api_token,
location=config_entry_data[CONF_LOCATION],
protocol="http",
version=6,
verify_tls=DEFAULT_VERIFY_SSL,
)
@pytest.mark.parametrize(
("config_entry_data", "expected_api_token"),
[({**CONFIG_DATA_DEFAULTS}, API_KEY)],
)
async def test_setup_api_v5(
hass: HomeAssistant, config_entry_data: dict, expected_api_token: str
) -> None:
"""Tests the API object is created with the expected parameters."""
mocked_hole = _create_mocked_hole(api_version=5)
config_entry_data = {**config_entry_data}
config_entry_data[CONF_API_VERSION] = 5
config_entry_data = {**config_entry_data, CONF_STATISTICS_ONLY: True}
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=config_entry_data)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole) as patched_init_hole:
assert await hass.config_entries.async_setup(entry.entry_id)
patched_init_hole.assert_called_once_with(
config_entry_data[CONF_HOST],
ANY,
host=config_entry_data[CONF_HOST],
session=ANY,
api_token=expected_api_token,
location=config_entry_data[CONF_LOCATION],
tls=config_entry_data[CONF_SSL],
version=5,
verify_tls=DEFAULT_VERIFY_SSL,
)
async def test_setup_with_defaults(hass: HomeAssistant) -> None:
async def test_setup_with_defaults_v5(hass: HomeAssistant) -> None:
"""Tests component setup with default config."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=5)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_STATISTICS_ONLY: True}
domain=pi_hole.DOMAIN,
data={**CONFIG_DATA_DEFAULTS, CONF_API_VERSION: 5, CONF_STATISTICS_ONLY: True},
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
@ -110,9 +141,87 @@ async def test_setup_with_defaults(hass: HomeAssistant) -> None:
assert state.state == "off"
async def test_setup_with_defaults_v6(hass: HomeAssistant) -> None:
"""Tests component setup with default config."""
mocked_hole = _create_mocked_hole(
api_version=6, has_data=True, incorrect_app_password=False
)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_STATISTICS_ONLY: True}
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
state = hass.states.get("sensor.pi_hole_ads_blocked")
assert state is not None
assert state.name == "Pi-Hole Ads blocked"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_ads_percentage_blocked")
assert state.name == "Pi-Hole Ads percentage blocked"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_dns_queries_cached")
assert state.name == "Pi-Hole DNS queries cached"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_dns_queries_forwarded")
assert state.name == "Pi-Hole DNS queries forwarded"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_dns_queries")
assert state.name == "Pi-Hole DNS queries"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_dns_unique_clients")
assert state.name == "Pi-Hole DNS unique clients"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_dns_unique_domains")
assert state.name == "Pi-Hole DNS unique domains"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_domains_blocked")
assert state.name == "Pi-Hole Domains blocked"
assert state.state == "0"
state = hass.states.get("sensor.pi_hole_seen_clients")
assert state.name == "Pi-Hole Seen clients"
assert state.state == "0"
state = hass.states.get("binary_sensor.pi_hole_status")
assert state.name == "Pi-Hole Status"
assert state.state == "off"
async def test_setup_without_api_version(hass: HomeAssistant) -> None:
"""Tests component setup without API version."""
mocked_hole = _create_mocked_hole(api_version=6)
config = {**CONFIG_DATA_DEFAULTS}
config.pop(CONF_API_VERSION)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=config)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
assert entry.data[CONF_API_VERSION] == 6
mocked_hole = _create_mocked_hole(api_version=5)
config = {**CONFIG_DATA_DEFAULTS}
config.pop(CONF_API_VERSION)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=config)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
assert entry.data[CONF_API_VERSION] == 5
async def test_setup_name_config(hass: HomeAssistant) -> None:
"""Tests component setup with a custom name."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=6)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_NAME: "Custom"}
)
@ -122,16 +231,15 @@ async def test_setup_name_config(hass: HomeAssistant) -> None:
await hass.async_block_till_done()
assert (
hass.states.get("sensor.custom_ads_blocked_today").name
== "Custom Ads blocked today"
)
assert hass.states.get("sensor.custom_ads_blocked").name == "Custom Ads blocked"
async def test_switch(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) -> None:
"""Test Pi-hole switch."""
mocked_hole = _create_mocked_hole()
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONFIG_DATA)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA, CONF_API_VERSION: 5}
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
@ -145,7 +253,7 @@ async def test_switch(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) ->
{"entity_id": SWITCH_ENTITY_ID},
blocking=True,
)
mocked_hole.enable.assert_called_once()
mocked_hole.instances[-1].enable.assert_called_once()
await hass.services.async_call(
switch.DOMAIN,
@ -153,17 +261,17 @@ async def test_switch(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) ->
{"entity_id": SWITCH_ENTITY_ID},
blocking=True,
)
mocked_hole.disable.assert_called_once_with(True)
mocked_hole.instances[-1].disable.assert_called_once_with(True)
# Failed calls
type(mocked_hole).enable = AsyncMock(side_effect=HoleError("Error1"))
mocked_hole.instances[-1].enable = AsyncMock(side_effect=HoleError("Error1"))
await hass.services.async_call(
switch.DOMAIN,
switch.SERVICE_TURN_ON,
{"entity_id": SWITCH_ENTITY_ID},
blocking=True,
)
type(mocked_hole).disable = AsyncMock(side_effect=HoleError("Error2"))
mocked_hole.instances[-1].disable = AsyncMock(side_effect=HoleError("Error2"))
await hass.services.async_call(
switch.DOMAIN,
switch.SERVICE_TURN_OFF,
@ -171,6 +279,7 @@ async def test_switch(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) ->
blocking=True,
)
errors = [x for x in caplog.records if x.levelno == logging.ERROR]
assert errors[-2].message == "Unable to enable Pi-hole: Error1"
assert errors[-1].message == "Unable to disable Pi-hole: Error2"
@ -178,7 +287,7 @@ async def test_switch(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) ->
async def test_disable_service_call(hass: HomeAssistant) -> None:
"""Test disable service call with no Pi-hole named."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=6)
with _patch_init_hole(mocked_hole):
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONFIG_DATA)
entry.add_to_hass(hass)
@ -199,7 +308,7 @@ async def test_disable_service_call(hass: HomeAssistant) -> None:
blocking=True,
)
mocked_hole.disable.assert_called_with(1)
mocked_hole.instances[-1].disable.assert_called_with(1)
async def test_unload(hass: HomeAssistant) -> None:
@ -209,7 +318,7 @@ async def test_unload(hass: HomeAssistant) -> None:
data={**CONFIG_DATA_DEFAULTS, CONF_HOST: "pi.hole"},
)
entry.add_to_hass(hass)
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=6)
with _patch_init_hole(mocked_hole):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
@ -222,7 +331,7 @@ async def test_unload(hass: HomeAssistant) -> None:
async def test_remove_obsolete(hass: HomeAssistant) -> None:
"""Test removing obsolete config entry parameters."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=6)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_STATISTICS_ONLY: True}
)

136
tests/components/pi_hole/test_repairs.py Normal file
View File

@ -0,0 +1,136 @@
"""Test pi_hole component."""
from datetime import timedelta
from unittest.mock import AsyncMock
from hole.exceptions import HoleConnectionError, HoleError
import pytest
import homeassistant
from homeassistant.components import pi_hole
from homeassistant.components.pi_hole.const import VERSION_6_RESPONSE_TO_5_ERROR
from homeassistant.const import CONF_API_VERSION, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.util import dt as dt_util
from . import CONFIG_DATA_DEFAULTS, ZERO_DATA, _create_mocked_hole, _patch_init_hole
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_change_api_5_to_6(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Tests a user with an API version 5 config entry that is updated to API version 6."""
mocked_hole = _create_mocked_hole(api_version=5)
# setu up a valid API version 5 config entry
entry = MockConfigEntry(
domain=pi_hole.DOMAIN,
data={**CONFIG_DATA_DEFAULTS, CONF_API_VERSION: 5},
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
assert mocked_hole.instances[-1].data == ZERO_DATA
# Change the mock's state after setup
mocked_hole.instances[-1].hole_version = 6
mocked_hole.instances[-1].api_token = "wrong_token"
# Patch the method on the coordinator's api reference directly
pihole_data = entry.runtime_data
assert pihole_data.api == mocked_hole.instances[-1]
pihole_data.api.get_data = AsyncMock(
side_effect=lambda: setattr(
pihole_data.api,
"data",
{"error": VERSION_6_RESPONSE_TO_5_ERROR, "took": 0.0001430511474609375},
)
)
# Now trigger the update
with pytest.raises(homeassistant.exceptions.ConfigEntryAuthFailed):
await pihole_data.coordinator.update_method()
assert pihole_data.api.data == {
"error": VERSION_6_RESPONSE_TO_5_ERROR,
"took": 0.0001430511474609375,
}
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
await hass.async_block_till_done()
# ensure a re-auth flow is created
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
assert flows[0]["step_id"] == "reauth_confirm"
assert flows[0]["context"]["entry_id"] == entry.entry_id
async def test_app_password_changing(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Tests a user with an API version 5 config entry that is updated to API version 6."""
mocked_hole = _create_mocked_hole(
api_version=6, has_data=True, incorrect_app_password=False
)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS})
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
state = hass.states.get("sensor.pi_hole_ads_blocked")
assert state is not None
assert state.name == "Pi-Hole Ads blocked"
assert state.state == "0"
# Test app password changing
async def fail_auth():
"""Set mocked data to bad_data."""
raise HoleError("Authentication failed: Invalid password")
mocked_hole.instances[-1].get_data = AsyncMock(side_effect=fail_auth)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
assert flows[0]["step_id"] == "reauth_confirm"
assert flows[0]["context"]["entry_id"] == entry.entry_id
# Test app password changing
async def fail_fetch():
"""Set mocked data to bad_data."""
raise HoleConnectionError("Cannot fetch data from Pi-hole: 200")
mocked_hole.instances[-1].get_data = AsyncMock(side_effect=fail_fetch)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
await hass.async_block_till_done()
async def test_app_failed_fetch(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Tests a user with an API version 5 config entry that is updated to API version 6."""
mocked_hole = _create_mocked_hole(
api_version=6, has_data=True, incorrect_app_password=False
)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS})
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
state = hass.states.get("sensor.pi_hole_ads_blocked")
assert state.state == "0"
# Test fetch failing changing
async def fail_fetch():
"""Set mocked data to bad_data."""
raise HoleConnectionError("Cannot fetch data from Pi-hole: 200")
mocked_hole.instances[-1].get_data = AsyncMock(side_effect=fail_fetch)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
await hass.async_block_till_done()
state = hass.states.get("sensor.pi_hole_ads_blocked")
assert state.state == STATE_UNAVAILABLE

79
tests/components/pi_hole/test_sensor.py Normal file
View File

@ -0,0 +1,79 @@
"""Test pi_hole component."""
import copy
from datetime import timedelta
from unittest.mock import AsyncMock
import pytest
from homeassistant.components import pi_hole
from homeassistant.components.pi_hole.const import CONF_STATISTICS_ONLY
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import CONFIG_DATA_DEFAULTS, ZERO_DATA_V6, _create_mocked_hole, _patch_init_hole
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_bad_data_type(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test handling of bad data. Mostly for code coverage, rather than simulating known error states."""
mocked_hole = _create_mocked_hole(
api_version=6, has_data=True, incorrect_app_password=False
)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_STATISTICS_ONLY: True}
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
bad_data = copy.deepcopy(ZERO_DATA_V6)
bad_data["queries"]["total"] = "error string"
assert bad_data != ZERO_DATA_V6
async def set_bad_data():
"""Set mocked data to bad_data."""
mocked_hole.instances[-1].data = bad_data
mocked_hole.instances[-1].get_data = AsyncMock(side_effect=set_bad_data)
# Wait a minute
future = dt_util.utcnow() + timedelta(minutes=1)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert "TypeError" in caplog.text
async def test_bad_data_key(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test handling of bad data. Mostly for code coverage, rather than simulating known error states."""
mocked_hole = _create_mocked_hole(
api_version=6, has_data=True, incorrect_app_password=False
)
entry = MockConfigEntry(
domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_STATISTICS_ONLY: True}
)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
bad_data = copy.deepcopy(ZERO_DATA_V6)
# remove a whole part of the dict tree now
bad_data["queries"] = "error string"
assert bad_data != ZERO_DATA_V6
async def set_bad_data():
"""Set mocked data to bad_data."""
mocked_hole.instances[-1].data = bad_data
mocked_hole.instances[-1].get_data = AsyncMock(side_effect=set_bad_data)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=1))
await hass.async_block_till_done()
assert mocked_hole.instances[-1].data != ZERO_DATA_V6
assert "KeyError" in caplog.text

8
tests/components/pi_hole/test_update.py
View File

@ -11,7 +11,7 @@ from tests.common import MockConfigEntry
async def test_update(hass: HomeAssistant) -> None:
"""Tests update entity."""
mocked_hole = _create_mocked_hole()
mocked_hole = _create_mocked_hole(api_version=6)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONFIG_DATA_DEFAULTS)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
@ -52,7 +52,7 @@ async def test_update(hass: HomeAssistant) -> None:
async def test_update_no_versions(hass: HomeAssistant) -> None:
"""Tests update entity when no version data available."""
mocked_hole = _create_mocked_hole(has_versions=False)
mocked_hole = _create_mocked_hole(has_versions=False, api_version=6)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONFIG_DATA_DEFAULTS)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
@ -84,7 +84,9 @@ async def test_update_no_versions(hass: HomeAssistant) -> None:
async def test_update_no_updates(hass: HomeAssistant) -> None:
"""Tests update entity when no latest data available."""
mocked_hole = _create_mocked_hole(has_versions=True, has_update=False)
mocked_hole = _create_mocked_hole(
has_versions=True, has_update=False, api_version=6
)
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONFIG_DATA_DEFAULTS)
entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):