Add config flow to dwd_weather_warnings (#91040)

* Add config flow to dwd_weather_warnings

* Add additional test for more coverage

* Apply code review changes

* Apply further code review changes

* Rename constant for configuration

* Apply code review changes

* Simplify config flow code
This commit is contained in:
andarotajo 2023-05-07 10:26:39 +02:00 committed by GitHub
parent b22c45ea29
commit bf6d429339
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 461 additions and 21 deletions

View file

@ -290,6 +290,7 @@ build.json @home-assistant/supervisor
/homeassistant/components/dunehd/ @bieniu
/tests/components/dunehd/ @bieniu
/homeassistant/components/dwd_weather_warnings/ @runningman84 @stephan192 @Hummel95 @andarotajo
/tests/components/dwd_weather_warnings/ @runningman84 @stephan192 @Hummel95 @andarotajo
/homeassistant/components/dynalite/ @ziv1234
/tests/components/dynalite/ @ziv1234
/homeassistant/components/eafm/ @Jc2k

View file

@ -1 +1,33 @@
"""The dwd_weather_warnings component."""
from __future__ import annotations
from dwdwfsapi import DwdWeatherWarningsAPI
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import CONF_REGION_IDENTIFIER, DOMAIN, PLATFORMS
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
region_identifier: str = entry.data[CONF_REGION_IDENTIFIER]
# Initialize the API.
api = await hass.async_add_executor_job(DwdWeatherWarningsAPI, region_identifier)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = api
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

View file

@ -0,0 +1,88 @@
"""Config flow for the dwd_weather_warnings integration."""
from __future__ import annotations
from typing import Any, Final
from dwdwfsapi import DwdWeatherWarningsAPI
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_NAME
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from .const import (
CONF_REGION_IDENTIFIER,
CONF_REGION_NAME,
DEFAULT_NAME,
DOMAIN,
LOGGER,
)
CONFIG_SCHEMA: Final = vol.Schema(
{
vol.Required(CONF_REGION_IDENTIFIER): cv.string,
}
)
class DwdWeatherWarningsConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle the config flow for the dwd_weather_warnings integration."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors: dict = {}
if user_input is not None:
region_identifier = user_input[CONF_REGION_IDENTIFIER]
# Validate region identifier using the API
if not await self.hass.async_add_executor_job(
DwdWeatherWarningsAPI, region_identifier
):
errors["base"] = "invalid_identifier"
if not errors:
# Set the unique ID for this config entry.
await self.async_set_unique_id(region_identifier)
self._abort_if_unique_id_configured()
# Set the name for this config entry.
name = f"{DEFAULT_NAME} {region_identifier}"
return self.async_create_entry(title=name, data=user_input)
return self.async_show_form(
step_id="user", errors=errors, data_schema=CONFIG_SCHEMA
)
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Import a config entry from configuration.yaml."""
LOGGER.debug(
"Starting import of sensor from configuration.yaml - %s", import_config
)
# Adjust data to new format.
region_identifier = import_config.pop(CONF_REGION_NAME)
import_config[CONF_REGION_IDENTIFIER] = region_identifier
# Set the unique ID for this imported entry.
await self.async_set_unique_id(import_config[CONF_REGION_IDENTIFIER])
self._abort_if_unique_id_configured()
# Validate region identifier using the API
if not await self.hass.async_add_executor_job(
DwdWeatherWarningsAPI, region_identifier
):
return self.async_abort(reason="invalid_identifier")
name = import_config.get(
CONF_NAME, f"{DEFAULT_NAME} {import_config[CONF_REGION_IDENTIFIER]}"
)
return self.async_create_entry(title=name, data=import_config)

View file

@ -6,9 +6,14 @@ from datetime import timedelta
import logging
from typing import Final
from homeassistant.const import Platform
LOGGER = logging.getLogger(__package__)
DOMAIN: Final = "dwd_weather_warnings"
CONF_REGION_NAME: Final = "region_name"
CONF_REGION_IDENTIFIER: Final = "region_identifier"
ATTR_REGION_NAME: Final = "region_name"
ATTR_REGION_ID: Final = "region_id"
@ -29,5 +34,7 @@ API_ATTR_WARNING_COLOR: Final = "color"
CURRENT_WARNING_SENSOR: Final = "current_warning_level"
ADVANCE_WARNING_SENSOR: Final = "advance_warning_level"
DEFAULT_NAME: Final = "DWD-Weather-Warnings"
DEFAULT_NAME: Final = "DWD Weather Warnings"
DEFAULT_SCAN_INTERVAL: Final = timedelta(minutes=15)
PLATFORMS: Final[list[Platform]] = [Platform.SENSOR]

View file

@ -2,6 +2,7 @@
"domain": "dwd_weather_warnings",
"name": "Deutscher Wetterdienst (DWD) Weather Warnings",
"codeowners": ["@runningman84", "@stephan192", "@Hummel95", "@andarotajo"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/dwd_weather_warnings",
"iot_class": "cloud_polling",
"loggers": ["dwdwfsapi"],

View file

@ -8,9 +8,11 @@ Unwetterwarnungen (Stufe 3)
Warnungen vor markantem Wetter (Stufe 2)
Wetterwarnungen (Stufe 1)
"""
from __future__ import annotations
from dwdwfsapi import DwdWeatherWarningsAPI
from typing import Final
import voluptuous as vol
from homeassistant.components.sensor import (
@ -18,10 +20,12 @@ from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_MONITORED_CONDITIONS, CONF_NAME
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import Throttle
@ -43,8 +47,8 @@ from .const import (
ATTR_WARNING_COUNT,
CONF_REGION_NAME,
CURRENT_WARNING_SENSOR,
DEFAULT_NAME,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
LOGGER,
)
@ -60,39 +64,59 @@ SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
icon="mdi:close-octagon-outline",
),
)
MONITORED_CONDITIONS: list[str] = [desc.key for desc in SENSOR_TYPES]
# Should be removed together with the old YAML configuration.
YAML_MONITORED_CONDITIONS: Final = [CURRENT_WARNING_SENSOR, ADVANCE_WARNING_SENSOR]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_REGION_NAME): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(
CONF_MONITORED_CONDITIONS, default=list(MONITORED_CONDITIONS)
): vol.All(cv.ensure_list, [vol.In(MONITORED_CONDITIONS)]),
CONF_MONITORED_CONDITIONS, default=YAML_MONITORED_CONDITIONS
): vol.All(cv.ensure_list, [vol.In(YAML_MONITORED_CONDITIONS)]),
}
)
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the DWD-Weather-Warnings sensor."""
name = config.get(CONF_NAME)
region_name = config.get(CONF_REGION_NAME)
"""Import the configurations from YAML to config flows."""
# Show issue as long as the YAML configuration exists.
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
)
api = WrappedDwDWWAPI(DwdWeatherWarningsAPI(region_name))
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
sensors = [
DwdWeatherWarningsSensor(api, name, description)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up entities from config entry."""
api = WrappedDwDWWAPI(hass.data[DOMAIN][entry.entry_id])
async_add_entities(
[
DwdWeatherWarningsSensor(api, entry.title, entry.unique_id, description)
for description in SENSOR_TYPES
if description.key in config[CONF_MONITORED_CONDITIONS]
]
add_entities(sensors, True)
],
True,
)
class DwdWeatherWarningsSensor(SensorEntity):
@ -104,23 +128,26 @@ class DwdWeatherWarningsSensor(SensorEntity):
self,
api,
name,
unique_id,
description: SensorEntityDescription,
) -> None:
"""Initialize a DWD-Weather-Warnings sensor."""
self._api = api
self.entity_description = description
self._attr_name = f"{name} {description.name}"
self._attr_unique_id = f"{unique_id}-{description.key}"
@property
def native_value(self):
"""Return the state of the device."""
"""Return the state of the sensor."""
if self.entity_description.key == CURRENT_WARNING_SENSOR:
return self._api.api.current_warning_level
return self._api.api.expected_warning_level
@property
def extra_state_attributes(self):
"""Return the state attributes of the DWD-Weather-Warnings."""
"""Return the state attributes of the sensor."""
data = {
ATTR_REGION_NAME: self._api.api.warncell_name,
ATTR_REGION_ID: self._api.api.warncell_id,

View file

@ -0,0 +1,25 @@
{
"config": {
"step": {
"user": {
"description": "To identify the desired region, the warncell ID / name is required.",
"data": {
"region_identifier": "Warncell ID or name"
}
}
},
"error": {
"invalid_identifier": "The specified region identifier is invalid."
},
"abort": {
"already_configured": "Warncell ID / name is already configured.",
"invalid_identifier": "[%key:component::dwd_weather_warnings::config::error::invalid_identifier%]"
}
},
"issues": {
"deprecated_yaml": {
"title": "The Deutscher Wetterdienst (DWD) Weather Warnings YAML configuration is being removed",
"description": "Configuring Deutscher Wetterdienst (DWD) Weather Warnings using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the Deutscher Wetterdienst (DWD) Weather Warnings YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
}
}
}

View file

@ -103,6 +103,7 @@ FLOWS = {
"dsmr",
"dsmr_reader",
"dunehd",
"dwd_weather_warnings",
"dynalite",
"eafm",
"easyenergy",

View file

@ -1200,7 +1200,7 @@
"dwd_weather_warnings": {
"name": "Deutscher Wetterdienst (DWD) Weather Warnings",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "cloud_polling"
},
"dweet": {

View file

@ -493,6 +493,9 @@ doorbirdpy==2.1.0
# homeassistant.components.dsmr
dsmr_parser==0.33
# homeassistant.components.dwd_weather_warnings
dwdwfsapi==1.0.6
# homeassistant.components.dynalite
dynalite_devices==0.1.47

View file

@ -0,0 +1 @@
"""Tests for Deutscher Wetterdienst (DWD) Weather Warnings."""

View file

@ -0,0 +1,16 @@
"""Configuration for Deutscher Wetterdienst (DWD) Weather Warnings tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.dwd_weather_warnings.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry

View file

@ -0,0 +1,200 @@
"""Tests for Deutscher Wetterdienst (DWD) Weather Warnings config flow."""
from typing import Final
from unittest.mock import patch
import pytest
from homeassistant.components.dwd_weather_warnings.const import (
ADVANCE_WARNING_SENSOR,
CONF_REGION_IDENTIFIER,
CONF_REGION_NAME,
CURRENT_WARNING_SENSOR,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_MONITORED_CONDITIONS, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
DEMO_CONFIG_ENTRY: Final = {
CONF_REGION_IDENTIFIER: "807111000",
}
DEMO_YAML_CONFIGURATION: Final = {
CONF_NAME: "Unit Test",
CONF_REGION_NAME: "807111000",
CONF_MONITORED_CONDITIONS: [CURRENT_WARNING_SENSOR, ADVANCE_WARNING_SENSOR],
}
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
async def test_create_entry(hass: HomeAssistant) -> None:
"""Test that the full config flow works."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=False,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=DEMO_CONFIG_ENTRY
)
# Test for invalid region identifier.
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "invalid_identifier"}
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=True,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=DEMO_CONFIG_ENTRY
)
# Test for successfully created entry.
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "DWD Weather Warnings 807111000"
assert result["data"] == {
CONF_REGION_IDENTIFIER: "807111000",
}
async def test_import_flow_full_data(hass: HomeAssistant) -> None:
"""Test import of a full YAML configuration with both success and failure."""
# Test abort due to invalid identifier.
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=False,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=DEMO_YAML_CONFIGURATION.copy(),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "invalid_identifier"
# Test successful import.
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=DEMO_YAML_CONFIGURATION.copy(),
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Unit Test"
assert result["data"] == {
CONF_NAME: "Unit Test",
CONF_REGION_IDENTIFIER: "807111000",
CONF_MONITORED_CONDITIONS: [CURRENT_WARNING_SENSOR, ADVANCE_WARNING_SENSOR],
}
async def test_import_flow_no_name(hass: HomeAssistant) -> None:
"""Test a successful import of a YAML configuration with no name set."""
data = DEMO_YAML_CONFIGURATION.copy()
data.pop(CONF_NAME)
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=data
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "DWD Weather Warnings 807111000"
assert result["data"] == {
CONF_REGION_IDENTIFIER: "807111000",
CONF_MONITORED_CONDITIONS: [CURRENT_WARNING_SENSOR, ADVANCE_WARNING_SENSOR],
}
async def test_import_flow_only_required(hass: HomeAssistant) -> None:
"""Test a successful import of a YAML configuration with only required properties."""
data = DEMO_YAML_CONFIGURATION.copy()
data.pop(CONF_NAME)
data.pop(CONF_MONITORED_CONDITIONS)
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=data
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "DWD Weather Warnings 807111000"
assert result["data"] == {
CONF_REGION_IDENTIFIER: "807111000",
}
async def test_import_flow_already_configured(hass: HomeAssistant) -> None:
"""Test aborting, if the warncell ID / name is already configured during the import."""
entry = MockConfigEntry(
domain=DOMAIN,
data=DEMO_CONFIG_ENTRY.copy(),
unique_id=DEMO_CONFIG_ENTRY[CONF_REGION_IDENTIFIER],
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=DEMO_YAML_CONFIGURATION.copy()
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_config_flow_already_configured(hass: HomeAssistant) -> None:
"""Test aborting, if the warncell ID / name is already configured during the config."""
entry = MockConfigEntry(
domain=DOMAIN,
data=DEMO_CONFIG_ENTRY.copy(),
unique_id=DEMO_CONFIG_ENTRY[CONF_REGION_IDENTIFIER],
)
entry.add_to_hass(hass)
# Start configuration of duplicate entry.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
with patch(
"homeassistant.components.dwd_weather_warnings.config_flow.DwdWeatherWarningsAPI",
return_value=True,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=DEMO_CONFIG_ENTRY
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"

View file

@ -0,0 +1,38 @@
"""Tests for Deutscher Wetterdienst (DWD) Weather Warnings integration."""
from typing import Final
from homeassistant.components.dwd_weather_warnings.const import (
ADVANCE_WARNING_SENSOR,
CONF_REGION_IDENTIFIER,
CURRENT_WARNING_SENSOR,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_MONITORED_CONDITIONS, CONF_NAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
DEMO_CONFIG_ENTRY: Final = {
CONF_NAME: "Unit Test",
CONF_REGION_IDENTIFIER: "807111000",
CONF_MONITORED_CONDITIONS: [CURRENT_WARNING_SENSOR, ADVANCE_WARNING_SENSOR],
}
async def test_load_unload_entry(hass: HomeAssistant) -> None:
"""Test loading and unloading the integration."""
entry = MockConfigEntry(domain=DOMAIN, data=DEMO_CONFIG_ENTRY)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
assert entry.entry_id in hass.data[DOMAIN]
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
assert entry.entry_id not in hass.data[DOMAIN]