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

Make API key mandatory for PI-Hole (#85885)

Browse Source 
add reauth, so make api-key mandatory
This commit is contained in:
Michael 2023-01-17 03:34:42 +01:00 committed by GitHub
parent 6581bad7ce
commit 1d5ecdd4ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 252 additions and 251 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -17,7 +17,8 @@ from homeassistant.const import (
CONF_VERIFY_SSL, CONF_VERIFY_SSL,
Platform, Platform,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
@ -64,6 +65,13 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA, extra=vol.ALLOW_EXTRA,
) )
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.SENSOR,
Platform.SWITCH,
Platform.UPDATE,
]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Pi-hole integration.""" """Set up the Pi-hole integration."""
@ -103,11 +111,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
location = entry.data[CONF_LOCATION] location = entry.data[CONF_LOCATION]
api_key = entry.data.get(CONF_API_KEY) api_key = entry.data.get(CONF_API_KEY)
# For backward compatibility # remove obsolet CONF_STATISTICS_ONLY from entry.data
if CONF_STATISTICS_ONLY not in entry.data: if CONF_STATISTICS_ONLY in entry.data:
hass.config_entries.async_update_entry( entry_data = entry.data.copy()
entry, data={**entry.data, CONF_STATISTICS_ONLY: not api_key} entry_data.pop(CONF_STATISTICS_ONLY)
) hass.config_entries.async_update_entry(entry, data=entry_data)
# start reauth to force api key is present
if CONF_API_KEY not in entry.data:
raise ConfigEntryAuthFailed
_LOGGER.debug("Setting up %s integration with host %s", DOMAIN, host) _LOGGER.debug("Setting up %s integration with host %s", DOMAIN, host)
@ -125,8 +137,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
try: try:
await api.get_data() await api.get_data()
await api.get_versions() await api.get_versions()
_LOGGER.debug("async_update_data() api.data: %s", api.data)
except HoleError as err: except HoleError as err:
raise UpdateFailed(f"Failed to communicate with API: {err}") from err raise UpdateFailed(f"Failed to communicate with API: {err}") from err
if not isinstance(api.data, dict):
raise ConfigEntryAuthFailed
coordinator = DataUpdateCoordinator( coordinator = DataUpdateCoordinator(
hass, hass,
@ -142,30 +157,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await coordinator.async_config_entry_first_refresh() await coordinator.async_config_entry_first_refresh()
await hass.config_entries.async_forward_entry_setups(entry, _async_platforms(entry)) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Pi-hole entry.""" """Unload Pi-hole entry."""
unload_ok = await hass.config_entries.async_unload_platforms( unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
entry, _async_platforms(entry)
)
if unload_ok: if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok return unload_ok
@callback
def _async_platforms(entry: ConfigEntry) -> list[Platform]:
"""Return platforms to be loaded / unloaded."""
platforms = [Platform.BINARY_SENSOR, Platform.UPDATE, Platform.SENSOR]
if not entry.data[CONF_STATISTICS_ONLY]:
platforms.append(Platform.SWITCH)
return platforms
class PiHoleEntity(CoordinatorEntity): class PiHoleEntity(CoordinatorEntity):
"""Representation of a Pi-hole entity.""" """Representation of a Pi-hole entity."""

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

@ -15,8 +15,6 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import PiHoleEntity from . import PiHoleEntity
from .const import ( from .const import (
BINARY_SENSOR_TYPES, BINARY_SENSOR_TYPES,
BINARY_SENSOR_TYPES_STATISTICS_ONLY,
CONF_STATISTICS_ONLY,
DATA_KEY_API, DATA_KEY_API,
DATA_KEY_COORDINATOR, DATA_KEY_COORDINATOR,
DOMAIN as PIHOLE_DOMAIN, DOMAIN as PIHOLE_DOMAIN,
@ -42,18 +40,6 @@ async def async_setup_entry(
for description in BINARY_SENSOR_TYPES 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) async_add_entities(binary_sensors, True)

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

@ -1,6 +1,7 @@
"""Config flow to configure the Pi-hole integration.""" """Config flow to configure the Pi-hole integration."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping
import logging import logging
from typing import Any from typing import Any
@ -26,7 +27,6 @@ from .const import (
DEFAULT_LOCATION, DEFAULT_LOCATION,
DEFAULT_NAME, DEFAULT_NAME,
DEFAULT_SSL, DEFAULT_SSL,
DEFAULT_STATISTICS_ONLY,
DEFAULT_VERIFY_SSL, DEFAULT_VERIFY_SSL,
DOMAIN, DOMAIN,
) )
@ -47,65 +47,29 @@ class PiHoleFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Handle a flow initiated by the user.""" """Handle a flow initiated by the user."""
return await self.async_step_init(user_input)
async def async_step_import(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by import."""
return await self.async_step_init(user_input, is_import=True)
async def async_step_init(
self, user_input: dict[str, Any] | None, is_import: bool = False
) -> FlowResult:
"""Handle init step of a flow."""
errors = {} errors = {}
if user_input is not None: if user_input is not None:
host = (
user_input[CONF_HOST]
if is_import
else f"{user_input[CONF_HOST]}:{user_input[CONF_PORT]}"
)
name = user_input[CONF_NAME]
location = user_input[CONF_LOCATION]
tls = user_input[CONF_SSL]
verify_tls = user_input[CONF_VERIFY_SSL]
endpoint = f"{host}/{location}"
if await self._async_endpoint_existed(endpoint):
return self.async_abort(reason="already_configured")
try:
await self._async_try_connect(host, location, tls, verify_tls)
except HoleError as ex:
_LOGGER.debug("Connection failed: %s", ex)
if is_import:
_LOGGER.error("Failed to import: %s", ex)
return self.async_abort(reason="cannot_connect")
errors["base"] = "cannot_connect"
else:
self._config = { self._config = {
CONF_HOST: host, CONF_HOST: f"{user_input[CONF_HOST]}:{user_input[CONF_PORT]}",
CONF_NAME: name, CONF_NAME: user_input[CONF_NAME],
CONF_LOCATION: location, CONF_LOCATION: user_input[CONF_LOCATION],
CONF_SSL: tls, CONF_SSL: user_input[CONF_SSL],
CONF_VERIFY_SSL: verify_tls, CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
CONF_API_KEY: user_input[CONF_API_KEY],
}
self._async_abort_entries_match(
{
CONF_HOST: f"{user_input[CONF_HOST]}:{user_input[CONF_PORT]}",
CONF_LOCATION: user_input[CONF_LOCATION],
} }
if is_import:
api_key = user_input.get(CONF_API_KEY)
return self.async_create_entry(
title=name,
data={
**self._config,
CONF_STATISTICS_ONLY: api_key is None,
CONF_API_KEY: api_key,
},
) )
self._config[CONF_STATISTICS_ONLY] = user_input[CONF_STATISTICS_ONLY]
if self._config[CONF_STATISTICS_ONLY]: if not (errors := await self._async_try_connect()):
return self.async_create_entry(title=name, data=self._config) return self.async_create_entry(
return await self.async_step_api_key() title=user_input[CONF_NAME], data=self._config
)
user_input = user_input or {} user_input = user_input or {}
return self.async_show_form( return self.async_show_form(
@ -116,6 +80,7 @@ class PiHoleFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
vol.Required( vol.Required(
CONF_PORT, default=user_input.get(CONF_PORT, 80) CONF_PORT, default=user_input.get(CONF_PORT, 80)
): vol.Coerce(int), ): vol.Coerce(int),
vol.Required(CONF_API_KEY): str,
vol.Required( vol.Required(
CONF_NAME, default=user_input.get(CONF_NAME, DEFAULT_NAME) CONF_NAME, default=user_input.get(CONF_NAME, DEFAULT_NAME)
): str, ): str,
@ -123,12 +88,6 @@ class PiHoleFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_LOCATION, CONF_LOCATION,
default=user_input.get(CONF_LOCATION, DEFAULT_LOCATION), default=user_input.get(CONF_LOCATION, DEFAULT_LOCATION),
): str, ): str,
vol.Required(
CONF_STATISTICS_ONLY,
default=user_input.get(
CONF_STATISTICS_ONLY, DEFAULT_STATISTICS_ONLY
),
): bool,
vol.Required( vol.Required(
CONF_SSL, CONF_SSL,
default=user_input.get(CONF_SSL, DEFAULT_SSL), default=user_input.get(CONF_SSL, DEFAULT_SSL),
@ -142,23 +101,93 @@ class PiHoleFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors, errors=errors,
) )
async def async_step_api_key( async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
self, user_input: dict[str, Any] | None = None """Handle a flow initiated by import."""
) -> FlowResult:
"""Handle step to setup API key.""" host = user_input[CONF_HOST]
if user_input is not None: name = user_input[CONF_NAME]
location = user_input[CONF_LOCATION]
tls = user_input[CONF_SSL]
verify_tls = user_input[CONF_VERIFY_SSL]
endpoint = f"{host}/{location}"
if await self._async_endpoint_existed(endpoint):
return self.async_abort(reason="already_configured")
try:
await self._async_try_connect_legacy(host, location, tls, verify_tls)
except HoleError as ex:
_LOGGER.debug("Connection failed: %s", ex)
_LOGGER.error("Failed to import: %s", ex)
return self.async_abort(reason="cannot_connect")
self._config = {
CONF_HOST: host,
CONF_NAME: name,
CONF_LOCATION: location,
CONF_SSL: tls,
CONF_VERIFY_SSL: verify_tls,
}
api_key = user_input.get(CONF_API_KEY)
return self.async_create_entry( return self.async_create_entry(
title=self._config[CONF_NAME], title=name,
data={ data={
**self._config, **self._config,
CONF_API_KEY: user_input.get(CONF_API_KEY, ""), CONF_STATISTICS_ONLY: api_key is None,
CONF_API_KEY: api_key,
}, },
) )
return self.async_show_form( async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
step_id="api_key", """Perform reauth upon an API authentication error."""
data_schema=vol.Schema({vol.Optional(CONF_API_KEY): str}), self._config = dict(entry_data)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self,
user_input: dict[str, Any] | None = None,
) -> FlowResult:
"""Perform reauth confirm upon an API authentication error."""
errors = {}
if user_input is not None:
self._config = {**self._config, CONF_API_KEY: user_input[CONF_API_KEY]}
if not (errors := await self._async_try_connect()):
entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
) )
assert entry
self.hass.config_entries.async_update_entry(entry, data=self._config)
self.hass.async_create_task(
self.hass.config_entries.async_reload(self.context["entry_id"])
)
return self.async_abort(reason="reauth_successful")
return self.async_show_form(
step_id="reauth_confirm",
description_placeholders={
CONF_HOST: self._config[CONF_HOST],
CONF_LOCATION: self._config[CONF_LOCATION],
},
data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
errors=errors,
)
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[CONF_API_KEY],
)
try:
await pi_hole.get_data()
except HoleError as ex:
_LOGGER.debug("Connection failed: %s", ex)
return {"base": "cannot_connect"}
if not isinstance(pi_hole.data, dict):
return {CONF_API_KEY: "invalid_auth"}
return {}
async def _async_endpoint_existed(self, endpoint: str) -> bool: async def _async_endpoint_existed(self, endpoint: str) -> bool:
existing_endpoints = [ existing_endpoints = [
@ -167,7 +196,7 @@ class PiHoleFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
] ]
return endpoint in existing_endpoints return endpoint in existing_endpoints
async def _async_try_connect( async def _async_try_connect_legacy(
self, host: str, location: str, tls: bool, verify_tls: bool self, host: str, location: str, tls: bool, verify_tls: bool
) -> None: ) -> None:
session = async_get_clientsession(self.hass, verify_tls) session = async_get_clientsession(self.hass, verify_tls)

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

@ -154,9 +154,6 @@ BINARY_SENSOR_TYPES: tuple[PiHoleBinarySensorEntityDescription, ...] = (
}, },
state_value=lambda api: bool(api.versions["FTL_update"]), state_value=lambda api: bool(api.versions["FTL_update"]),
), ),
)
BINARY_SENSOR_TYPES_STATISTICS_ONLY: tuple[PiHoleBinarySensorEntityDescription, ...] = (
PiHoleBinarySensorEntityDescription( PiHoleBinarySensorEntityDescription(
key="status", key="status",
name="Status", name="Status",

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

@ -8,28 +8,25 @@
"name": "[%key:common::config_flow::data::name%]", "name": "[%key:common::config_flow::data::name%]",
"location": "[%key:common::config_flow::data::location%]", "location": "[%key:common::config_flow::data::location%]",
"api_key": "[%key:common::config_flow::data::api_key%]", "api_key": "[%key:common::config_flow::data::api_key%]",
"statistics_only": "Statistics Only",
"ssl": "[%key:common::config_flow::data::ssl%]", "ssl": "[%key:common::config_flow::data::ssl%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]" "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
} }
}, },
"api_key": { "reauth_confirm": {
"title": "PI-Hole [%key:common::config_flow::title::reauth%]",
"description": "Please enter a new api key for PI-Hole at {host}/{location}",
"data": { "data": {
"api_key": "[%key:common::config_flow::data::api_key%]" "api_key": "[%key:common::config_flow::data::api_key%]"
} }
} }
}, },
"error": { "error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
}, },
"abort": { "abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]" "already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
} "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"issues": {
"deprecated_yaml": {
"title": "The PI-Hole YAML configuration is being removed",
"description": "Configuring PI-Hole using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the PI-Hole YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
} }
} }
} }

19
homeassistant/components/pi_hole/translations/en.json
View File

@ -1,16 +1,20 @@
{ {
"config": { "config": {
"abort": { "abort": {
"already_configured": "Service is already configured" "already_configured": "Service is already configured",
"reauth_successful": "Re-authentication was successful"
}, },
"error": { "error": {
"cannot_connect": "Failed to connect" "cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication"
}, },
"step": { "step": {
"api_key": { "reauth_confirm": {
"data": { "data": {
"api_key": "API Key" "api_key": "API Key"
} },
"description": "Please enter a new api key for PI-Hole at {host}/{location}",
"title": "PI-Hole Reauthenticate Integration"
}, },
"user": { "user": {
"data": { "data": {
@ -20,16 +24,9 @@
"name": "Name", "name": "Name",
"port": "Port", "port": "Port",
"ssl": "Uses an SSL certificate", "ssl": "Uses an SSL certificate",
"statistics_only": "Statistics Only",
"verify_ssl": "Verify SSL certificate" "verify_ssl": "Verify SSL certificate"
} }
} }
} }
},
"issues": {
"deprecated_yaml": {
"description": "Configuring PI-Hole using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the PI-Hole YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue.",
"title": "The PI-Hole YAML configuration is being removed"
}
} }
} }

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

@ -3,7 +3,13 @@ from unittest.mock import AsyncMock, MagicMock, patch
from hole.exceptions import HoleError from hole.exceptions import HoleError
from homeassistant.components.pi_hole.const import CONF_STATISTICS_ONLY from homeassistant.components.pi_hole.const import (
CONF_STATISTICS_ONLY,
DEFAULT_LOCATION,
DEFAULT_NAME,
DEFAULT_SSL,
DEFAULT_VERIFY_SSL,
)
from homeassistant.const import ( from homeassistant.const import (
CONF_API_KEY, CONF_API_KEY,
CONF_HOST, CONF_HOST,
@ -47,7 +53,16 @@ API_KEY = "apikey"
SSL = False SSL = False
VERIFY_SSL = True VERIFY_SSL = True
CONF_DATA = { CONFIG_DATA_DEFAULTS = {
CONF_HOST: f"{HOST}:{PORT}",
CONF_LOCATION: DEFAULT_LOCATION,
CONF_NAME: DEFAULT_NAME,
CONF_SSL: DEFAULT_SSL,
CONF_VERIFY_SSL: DEFAULT_VERIFY_SSL,
CONF_API_KEY: API_KEY,
}
CONFIG_DATA = {
CONF_HOST: f"{HOST}:{PORT}", CONF_HOST: f"{HOST}:{PORT}",
CONF_LOCATION: LOCATION, CONF_LOCATION: LOCATION,
CONF_NAME: NAME, CONF_NAME: NAME,
@ -56,34 +71,35 @@ CONF_DATA = {
CONF_VERIFY_SSL: VERIFY_SSL, CONF_VERIFY_SSL: VERIFY_SSL,
} }
CONF_CONFIG_FLOW_USER = { CONFIG_FLOW_USER = {
CONF_HOST: HOST, CONF_HOST: HOST,
CONF_PORT: PORT, CONF_PORT: PORT,
CONF_API_KEY: API_KEY,
CONF_LOCATION: LOCATION, CONF_LOCATION: LOCATION,
CONF_NAME: NAME, CONF_NAME: NAME,
CONF_STATISTICS_ONLY: False,
CONF_SSL: SSL, CONF_SSL: SSL,
CONF_VERIFY_SSL: VERIFY_SSL, CONF_VERIFY_SSL: VERIFY_SSL,
} }
CONF_CONFIG_FLOW_API_KEY = { CONFIG_FLOW_API_KEY = {
CONF_API_KEY: API_KEY, CONF_API_KEY: API_KEY,
} }
CONF_CONFIG_ENTRY = { CONFIG_ENTRY = {
CONF_HOST: f"{HOST}:{PORT}", CONF_HOST: f"{HOST}:{PORT}",
CONF_LOCATION: LOCATION, CONF_LOCATION: LOCATION,
CONF_NAME: NAME, CONF_NAME: NAME,
CONF_STATISTICS_ONLY: False,
CONF_API_KEY: API_KEY, CONF_API_KEY: API_KEY,
CONF_SSL: SSL, CONF_SSL: SSL,
CONF_VERIFY_SSL: VERIFY_SSL, CONF_VERIFY_SSL: VERIFY_SSL,
} }
CONFIG_ENTRY_IMPORTED = {**CONFIG_ENTRY, CONF_STATISTICS_ONLY: False}
SWITCH_ENTITY_ID = "switch.pi_hole" SWITCH_ENTITY_ID = "switch.pi_hole"
def _create_mocked_hole(raise_exception=False, has_versions=True): def _create_mocked_hole(raise_exception=False, has_versions=True, has_data=True):
mocked_hole = MagicMock() mocked_hole = MagicMock()
type(mocked_hole).get_data = AsyncMock( type(mocked_hole).get_data = AsyncMock(
side_effect=HoleError("") if raise_exception else None side_effect=HoleError("") if raise_exception else None
@ -93,7 +109,10 @@ def _create_mocked_hole(raise_exception=False, has_versions=True):
) )
type(mocked_hole).enable = AsyncMock() type(mocked_hole).enable = AsyncMock()
type(mocked_hole).disable = AsyncMock() type(mocked_hole).disable = AsyncMock()
if has_data:
mocked_hole.data = ZERO_DATA mocked_hole.data = ZERO_DATA
else:
mocked_hole.data = []
if has_versions: if has_versions:
mocked_hole.versions = SAMPLE_VERSIONS mocked_hole.versions = SAMPLE_VERSIONS
else: else:

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

@ -2,28 +2,26 @@
import logging import logging
from unittest.mock import patch from unittest.mock import patch
from homeassistant.components.pi_hole.const import CONF_STATISTICS_ONLY, DOMAIN from homeassistant.components.pi_hole.const import DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_API_KEY from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType from homeassistant.data_entry_flow import FlowResultType
from . import ( from . import (
CONF_CONFIG_ENTRY, CONFIG_DATA,
CONF_CONFIG_FLOW_API_KEY, CONFIG_DATA_DEFAULTS,
CONF_CONFIG_FLOW_USER, CONFIG_ENTRY,
CONF_DATA, CONFIG_ENTRY_IMPORTED,
CONFIG_FLOW_USER,
NAME, NAME,
ZERO_DATA,
_create_mocked_hole, _create_mocked_hole,
_patch_config_flow_hole, _patch_config_flow_hole,
_patch_init_hole,
) )
from tests.common import MockConfigEntry
def _flow_next(hass, flow_id):
return next(
flow
for flow in hass.config_entries.flow.async_progress()
if flow["flow_id"] == flow_id
)
def _patch_setup(): def _patch_setup():
@ -33,41 +31,41 @@ def _patch_setup():
) )
async def test_flow_import(hass, caplog): async def test_flow_import(hass: HomeAssistant, caplog):
"""Test import flow.""" """Test import flow."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_setup(): with _patch_config_flow_hole(mocked_hole), _patch_setup():
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONF_DATA DOMAIN, context={"source": SOURCE_IMPORT}, data=CONFIG_DATA
) )
assert result["type"] == FlowResultType.CREATE_ENTRY assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == NAME assert result["title"] == NAME
assert result["data"] == CONF_CONFIG_ENTRY assert result["data"] == CONFIG_ENTRY_IMPORTED
# duplicated server # duplicated server
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONF_DATA DOMAIN, context={"source": SOURCE_IMPORT}, data=CONFIG_DATA
) )
assert result["type"] == FlowResultType.ABORT assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
async def test_flow_import_invalid(hass, caplog): async def test_flow_import_invalid(hass: HomeAssistant, caplog):
"""Test import flow with invalid server.""" """Test import flow with invalid server."""
mocked_hole = _create_mocked_hole(True) mocked_hole = _create_mocked_hole(True)
with _patch_config_flow_hole(mocked_hole), _patch_setup(): with _patch_config_flow_hole(mocked_hole), _patch_setup():
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONF_DATA DOMAIN, context={"source": SOURCE_IMPORT}, data=CONFIG_DATA
) )
assert result["type"] == FlowResultType.ABORT assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "cannot_connect" assert result["reason"] == "cannot_connect"
assert len([x for x in caplog.records if x.levelno == logging.ERROR]) == 1 assert len([x for x in caplog.records if x.levelno == logging.ERROR]) == 1
async def test_flow_user(hass): async def test_flow_user(hass: HomeAssistant):
"""Test user initialized flow.""" """Test user initialized flow."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole(has_data=False)
with _patch_config_flow_hole(mocked_hole), _patch_setup(): with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": SOURCE_USER}, context={"source": SOURCE_USER},
@ -75,69 +73,68 @@ async def test_flow_user(hass):
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {} assert result["errors"] == {}
_flow_next(hass, result["flow_id"])
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input=CONF_CONFIG_FLOW_USER, user_input=CONFIG_FLOW_USER,
) )
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "api_key" assert result["step_id"] == "user"
assert result["errors"] is None assert result["errors"] == {CONF_API_KEY: "invalid_auth"}
_flow_next(hass, result["flow_id"])
mocked_hole.data = ZERO_DATA
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input=CONF_CONFIG_FLOW_API_KEY, user_input=CONFIG_FLOW_USER,
) )
assert result["type"] == FlowResultType.CREATE_ENTRY assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == NAME assert result["title"] == NAME
assert result["data"] == CONF_CONFIG_ENTRY assert result["data"] == CONFIG_ENTRY
# duplicated server # duplicated server
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": SOURCE_USER}, context={"source": SOURCE_USER},
data=CONF_CONFIG_FLOW_USER, data=CONFIG_FLOW_USER,
) )
assert result["type"] == FlowResultType.ABORT assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
async def test_flow_statistics_only(hass):
"""Test user initialized flow with statistics only."""
mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
_flow_next(hass, result["flow_id"])
user_input = {**CONF_CONFIG_FLOW_USER}
user_input[CONF_STATISTICS_ONLY] = True
config_entry_data = {**CONF_CONFIG_ENTRY}
config_entry_data[CONF_STATISTICS_ONLY] = True
config_entry_data.pop(CONF_API_KEY)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=user_input,
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == NAME
assert result["data"] == config_entry_data
async def test_flow_user_invalid(hass): async def test_flow_user_invalid(hass):
"""Test user initialized flow with invalid server.""" """Test user initialized flow with invalid server."""
mocked_hole = _create_mocked_hole(True) mocked_hole = _create_mocked_hole(True)
with _patch_config_flow_hole(mocked_hole), _patch_setup(): with _patch_config_flow_hole(mocked_hole), _patch_setup():
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_CONFIG_FLOW_USER DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_FLOW_USER
) )
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"} assert result["errors"] == {"base": "cannot_connect"}
async def test_flow_reauth(hass: HomeAssistant):
"""Test reauth flow."""
mocked_hole = _create_mocked_hole(has_data=False)
entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA_DEFAULTS)
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)
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
mocked_hole.data = ZERO_DATA
result = await hass.config_entries.flow.async_configure(
flows[0]["flow_id"],
user_input={CONF_API_KEY: "newkey"},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_API_KEY] == "newkey"

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

@ -7,27 +7,16 @@ from hole.exceptions import HoleError
from homeassistant.components import pi_hole, switch from homeassistant.components import pi_hole, switch
from homeassistant.components.pi_hole.const import ( from homeassistant.components.pi_hole.const import (
CONF_STATISTICS_ONLY, CONF_STATISTICS_ONLY,
DEFAULT_LOCATION,
DEFAULT_NAME,
DEFAULT_SSL,
DEFAULT_VERIFY_SSL,
SERVICE_DISABLE, SERVICE_DISABLE,
SERVICE_DISABLE_ATTR_DURATION, SERVICE_DISABLE_ATTR_DURATION,
) )
from homeassistant.const import ( from homeassistant.config_entries import ConfigEntryState
ATTR_ENTITY_ID, from homeassistant.const import ATTR_ENTITY_ID, CONF_API_KEY, CONF_HOST
CONF_API_KEY, from homeassistant.core import HomeAssistant
CONF_HOST,
CONF_LOCATION,
CONF_NAME,
CONF_SSL,
CONF_VERIFY_SSL,
)
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from . import ( from . import (
CONF_CONFIG_ENTRY, CONFIG_DATA_DEFAULTS,
CONF_DATA,
SWITCH_ENTITY_ID, SWITCH_ENTITY_ID,
_create_mocked_hole, _create_mocked_hole,
_patch_config_flow_hole, _patch_config_flow_hole,
@ -37,7 +26,7 @@ from . import (
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
async def test_setup_minimal_config(hass): async def test_setup_minimal_config(hass: HomeAssistant):
"""Tests component setup with minimal config.""" """Tests component setup with minimal config."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole): with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
@ -88,7 +77,7 @@ async def test_setup_minimal_config(hass):
assert state.state == "off" assert state.state == "off"
async def test_setup_name_config(hass): async def test_setup_name_config(hass: HomeAssistant):
"""Tests component setup with a custom name.""" """Tests component setup with a custom name."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole): with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
@ -106,7 +95,7 @@ async def test_setup_name_config(hass):
) )
async def test_switch(hass, caplog): async def test_switch(hass: HomeAssistant, caplog):
"""Test Pi-hole switch.""" """Test Pi-hole switch."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole): with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
@ -154,7 +143,7 @@ async def test_switch(hass, caplog):
assert errors[-1].message == "Unable to disable Pi-hole: Error2" assert errors[-1].message == "Unable to disable Pi-hole: Error2"
async def test_disable_service_call(hass): async def test_disable_service_call(hass: HomeAssistant):
"""Test disable service call with no Pi-hole named.""" """Test disable service call with no Pi-hole named."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole): with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole):
@ -180,21 +169,14 @@ async def test_disable_service_call(hass):
await hass.async_block_till_done() await hass.async_block_till_done()
mocked_hole.disable.assert_called_once_with(1) mocked_hole.disable.assert_called_with(1)
async def test_unload(hass): async def test_unload(hass: HomeAssistant):
"""Test unload entities.""" """Test unload entities."""
entry = MockConfigEntry( entry = MockConfigEntry(
domain=pi_hole.DOMAIN, domain=pi_hole.DOMAIN,
data={ data={**CONFIG_DATA_DEFAULTS, CONF_HOST: "pi.hole"},
CONF_NAME: DEFAULT_NAME,
CONF_HOST: "pi.hole",
CONF_LOCATION: DEFAULT_LOCATION,
CONF_SSL: DEFAULT_SSL,
CONF_VERIFY_SSL: DEFAULT_VERIFY_SSL,
CONF_STATISTICS_ONLY: True,
},
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
@ -208,32 +190,25 @@ async def test_unload(hass):
assert entry.entry_id not in hass.data[pi_hole.DOMAIN] assert entry.entry_id not in hass.data[pi_hole.DOMAIN]
async def test_migrate(hass): async def test_remove_obsolete(hass: HomeAssistant):
"""Test migrate from old config entry.""" """Test removing obsolete config entry parameters."""
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=CONF_DATA)
entry.add_to_hass(hass)
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole): entry = MockConfigEntry(
await hass.config_entries.async_setup(entry.entry_id) domain=pi_hole.DOMAIN, data={**CONFIG_DATA_DEFAULTS, CONF_STATISTICS_ONLY: True}
await hass.async_block_till_done() )
assert entry.data == CONF_CONFIG_ENTRY
async def test_migrate_statistics_only(hass):
"""Test migrate from old config entry with statistics only."""
conf_data = {**CONF_DATA}
conf_data[CONF_API_KEY] = ""
entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=conf_data)
entry.add_to_hass(hass) entry.add_to_hass(hass)
with _patch_init_hole(mocked_hole):
assert await hass.config_entries.async_setup(entry.entry_id)
assert CONF_STATISTICS_ONLY not in entry.data
async def test_missing_api_key(hass: HomeAssistant):
"""Tests start reauth flow if api key is missing."""
mocked_hole = _create_mocked_hole() mocked_hole = _create_mocked_hole()
with _patch_config_flow_hole(mocked_hole), _patch_init_hole(mocked_hole): data = CONFIG_DATA_DEFAULTS.copy()
await hass.config_entries.async_setup(entry.entry_id) data.pop(CONF_API_KEY)
await hass.async_block_till_done() entry = MockConfigEntry(domain=pi_hole.DOMAIN, data=data)
entry.add_to_hass(hass)
config_entry_data = {**CONF_CONFIG_ENTRY} with _patch_init_hole(mocked_hole):
config_entry_data[CONF_STATISTICS_ONLY] = True assert not await hass.config_entries.async_setup(entry.entry_id)
config_entry_data[CONF_API_KEY] = "" assert entry.state == ConfigEntryState.SETUP_ERROR
assert entry.data == config_entry_data