Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add options flow to NINA (#65890) 

* Added options flow

* Resolve conflicts

* Fix lint

* Implement improvements
pull/73834/head
Maximilian 2022-06-29 09:52:21 +02:00 committed by GitHub
parent aca0fd3178
commit 20680535ec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 453 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
homeassistant/components/nina/__init__.py
View File

@ -42,6 +42,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await coordinator.async_config_entry_first_refresh()
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
@ -49,6 +51,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)
class NINADataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching NINA data API."""

209
homeassistant/components/nina/config_flow.py
View File

@ -7,9 +7,14 @@ from pynina import ApiError, Nina
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_registry import (
async_entries_for_config_entry,
async_get,
)
from .const import (
_LOGGER,
@ -22,6 +27,58 @@ from .const import (
)
def swap_key_value(dict_to_sort: dict[str, str]) -> dict[str, str]:
"""Swap keys and values in dict."""
all_region_codes_swaped: dict[str, str] = {}
for key, value in dict_to_sort.items():
if value not in all_region_codes_swaped:
all_region_codes_swaped[value] = key
else:
for i in range(len(dict_to_sort)):
tmp_value: str = f"{value}_{i}"
if tmp_value not in all_region_codes_swaped:
all_region_codes_swaped[tmp_value] = key
break
return dict(sorted(all_region_codes_swaped.items(), key=lambda ele: ele[1]))
def split_regions(
_all_region_codes_sorted: dict[str, str], regions: dict[str, dict[str, Any]]
) -> dict[str, dict[str, Any]]:
"""Split regions alphabetical."""
for index, name in _all_region_codes_sorted.items():
for region_name, grouping_letters in CONST_REGION_MAPPING.items():
if name[0] in grouping_letters:
regions[region_name][index] = name
break
return regions
def prepare_user_input(
user_input: dict[str, Any], _all_region_codes_sorted: dict[str, str]
) -> dict[str, Any]:
"""Prepare the user inputs."""
tmp: dict[str, Any] = {}
for reg in user_input[CONF_REGIONS]:
tmp[_all_region_codes_sorted[reg]] = reg.split("_", 1)[0]
compact: dict[str, Any] = {}
for key, val in tmp.items():
if val in compact:
# Abenberg, St + Abenberger Wald
compact[val] = f"{compact[val]} + {key}"
break
compact[val] = key
user_input[CONF_REGIONS] = compact
return user_input
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for NINA."""
@ -50,7 +107,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
nina: Nina = Nina(async_get_clientsession(self.hass))
try:
self._all_region_codes_sorted = self.swap_key_value(
self._all_region_codes_sorted = swap_key_value(
await nina.getAllRegionalCodes()
)
except ApiError:
@ -59,7 +116,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_LOGGER.exception("Unexpected exception: %s", err)
return self.async_abort(reason="unknown")
self.split_regions()
self.regions = split_regions(self._all_region_codes_sorted, self.regions)
if user_input is not None and not errors:
user_input[CONF_REGIONS] = []
@ -69,23 +126,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
user_input[CONF_REGIONS] += group_input
if user_input[CONF_REGIONS]:
tmp: dict[str, Any] = {}
for reg in user_input[CONF_REGIONS]:
tmp[self._all_region_codes_sorted[reg]] = reg.split("_", 1)[0]
compact: dict[str, Any] = {}
for key, val in tmp.items():
if val in compact:
# Abenberg, St + Abenberger Wald
compact[val] = f"{compact[val]} + {key}"
break
compact[val] = key
user_input[CONF_REGIONS] = compact
return self.async_create_entry(title="NINA", data=user_input)
return self.async_create_entry(
title="NINA",
data=prepare_user_input(user_input, self._all_region_codes_sorted),
)
errors["base"] = "no_selection"
@ -107,26 +152,114 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
@staticmethod
def swap_key_value(dict_to_sort: dict[str, str]) -> dict[str, str]:
"""Swap keys and values in dict."""
all_region_codes_swaped: dict[str, str] = {}
@callback
def async_get_options_flow(config_entry):
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
for key, value in dict_to_sort.items():
if value not in all_region_codes_swaped:
all_region_codes_swaped[value] = key
else:
for i in range(len(dict_to_sort)):
tmp_value: str = f"{value}_{i}"
if tmp_value not in all_region_codes_swaped:
all_region_codes_swaped[tmp_value] = key
break
return dict(sorted(all_region_codes_swaped.items(), key=lambda ele: ele[1]))
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for nut."""
def split_regions(self) -> None:
"""Split regions alphabetical."""
for index, name in self._all_region_codes_sorted.items():
for region_name, grouping_letters in CONST_REGION_MAPPING.items():
if name[0] in grouping_letters:
self.regions[region_name][index] = name
break
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
self.data = dict(self.config_entry.data)
self._all_region_codes_sorted: dict[str, str] = {}
self.regions: dict[str, dict[str, Any]] = {}
for name in CONST_REGIONS:
self.regions[name] = {}
if name not in self.data:
self.data[name] = []
async def async_step_init(self, user_input=None):
"""Handle options flow."""
errors: dict[str, Any] = {}
if not self._all_region_codes_sorted:
nina: Nina = Nina(async_get_clientsession(self.hass))
try:
self._all_region_codes_sorted = swap_key_value(
await nina.getAllRegionalCodes()
)
except ApiError:
errors["base"] = "cannot_connect"
except Exception as err: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception: %s", err)
return self.async_abort(reason="unknown")
self.regions = split_regions(self._all_region_codes_sorted, self.regions)
if user_input is not None and not errors:
user_input[CONF_REGIONS] = []
for group in CONST_REGIONS:
if group_input := user_input.get(group):
user_input[CONF_REGIONS] += group_input
if user_input[CONF_REGIONS]:
user_input = prepare_user_input(
user_input, self._all_region_codes_sorted
)
entity_registry = async_get(self.hass)
entries = async_entries_for_config_entry(
entity_registry, self.config_entry.entry_id
)
removed_entities_slots = [
f"{region}-{slot_id}"
for region in self.data[CONF_REGIONS]
for slot_id in range(0, self.data[CONF_MESSAGE_SLOTS] + 1)
if slot_id > user_input[CONF_MESSAGE_SLOTS]
]
removed_entites_area = [
f"{cfg_region}-{slot_id}"
for slot_id in range(1, self.data[CONF_MESSAGE_SLOTS] + 1)
for cfg_region in self.data[CONF_REGIONS]
if cfg_region not in user_input[CONF_REGIONS]
]
for entry in entries:
for entity_uid in list(
set(removed_entities_slots + removed_entites_area)
):
if entry.unique_id == entity_uid:
entity_registry.async_remove(entry.entity_id)
self.hass.config_entries.async_update_entry(
self.config_entry, data=user_input
)
return self.async_create_entry(title="", data=None)
errors["base"] = "no_selection"
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
**{
vol.Optional(
region, default=self.data[region]
): cv.multi_select(self.regions[region])
for region in CONST_REGIONS
},
vol.Required(
CONF_MESSAGE_SLOTS,
default=self.data[CONF_MESSAGE_SLOTS],
): vol.All(int, vol.Range(min=1, max=20)),
vol.Required(
CONF_FILTER_CORONA,
default=self.data[CONF_FILTER_CORONA],
): cv.boolean,
}
),
errors=errors,
)

22
homeassistant/components/nina/strings.json
View File

@ -23,5 +23,27 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
},
"options": {
"step": {
"init": {
"title": "Options",
"data": {
"_a_to_d": "City/county (A-D)",
"_e_to_h": "City/county (E-H)",
"_i_to_l": "City/county (I-L)",
"_m_to_q": "City/county (M-Q)",
"_r_to_u": "City/county (R-U)",
"_v_to_z": "City/county (V-Z)",
"slots": "Maximum warnings per city/county",
"corona_filter": "Remove Corona Warnings"
}
}
},
"error": {
"no_selection": "Please select at least one city/county",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
}
}

22
homeassistant/components/nina/translations/en.json
View File

@ -23,5 +23,27 @@
"title": "Select city/county"
}
}
},
"options": {
"error": {
"cannot_connect": "Failed to connect",
"no_selection": "Please select at least one city/county",
"unknown": "Unexpected error"
},
"step": {
"init": {
"data": {
"_a_to_d": "City/county (A-D)",
"_e_to_h": "City/county (E-H)",
"_i_to_l": "City/county (I-L)",
"_m_to_q": "City/county (M-Q)",
"_r_to_u": "City/county (R-U)",
"_v_to_z": "City/county (V-Z)",
"corona_filter": "Remove Corona Warnings",
"slots": "Maximum warnings per city/county"
},
"title": "Options"
}
}
}
}

12
tests/components/nina/__init__.py
View File

@ -15,9 +15,19 @@ def mocked_request_function(url: str) -> dict[str, Any]:
load_fixture("sample_warning_details.json", "nina")
)
if url == "https://warnung.bund.de/api31/dashboard/083350000000.json":
dummy_response_regions: dict[str, Any] = json.loads(
load_fixture("sample_regions.json", "nina")
)
if "https://warnung.bund.de/api31/dashboard/" in url:
return dummy_response
if (
url
== "https://www.xrepository.de/api/xrepository/urn:de:bund:destatis:bevoelkerungsstatistik:schluessel:rs_2021-07-31/download/Regionalschl_ssel_2021-07-31.json"
):
return dummy_response_regions
warning_id = url.replace("https://warnung.bund.de/api31/warnings/", "").replace(
".json", ""
)

37
tests/components/nina/fixtures/sample_regions.json
View File

@ -3,14 +3,28 @@
"kennung": "urn:de:bund:destatis:bevoelkerungsstatistik:schluessel:rs_2021-07-31",
"kennungInhalt": "urn:de:bund:destatis:bevoelkerungsstatistik:schluessel:rs",
"version": "2021-07-31",
"nameKurz": "Regionalschlüssel",
"nameLang": "Gemeinden, dargestellt durch den Amtlichen Regionalschlüssel (ARS) des Statistischen Bundesamtes",
"nameKurz": [{ "value": "Regionalschlüssel", "lang": null }],
"nameLang": [
{
"value": "Gemeinden, dargestellt durch den Amtlichen Regionalschlüssel (ARS) des Statistischen Bundesamtes",
"lang": null
}
],
"nameTechnisch": "Regionalschluessel",
"herausgebernameLang": "Statistisches Bundesamt, Wiesbaden",
"herausgebernameKurz": "Destatis",
"beschreibung": "Diese Codeliste stellt alle Gemeinden Deutschlands durch den Amtlichen Regionalschlüssel (ARS) dar, wie im Gemeindeverzeichnis des Statistischen Bundesamtes enthalten. Darüber hinaus enthält die Codeliste für die Stadtstaaten Hamburg, Bremen und Berlin Einträge für Stadt-/Ortsteile bzw. Stadtbezirke. Diese Einträge sind mit einem entsprechenden Hinweis versehen.",
"versionBeschreibung": null,
"aenderungZurVorversion": "Mehrere Aenderungen",
"herausgebernameLang": [
{ "value": "Statistisches Bundesamt, Wiesbaden", "lang": null }
],
"herausgebernameKurz": [{ "value": "Destatis", "lang": null }],
"beschreibung": [
{
"value": "Diese Codeliste stellt alle Gemeinden Deutschlands durch den Amtlichen Regionalschlüssel (ARS) dar, wie im Gemeindeverzeichnis des Statistischen Bundesamtes enthalten. Darüber hinaus enthält die Codeliste für die Stadtstaaten Hamburg, Bremen und Berlin Einträge für Stadt-/Ortsteile bzw. Stadtbezirke. Diese Einträge sind mit einem entsprechenden Hinweis versehen.",
"lang": null
}
],
"versionBeschreibung": [],
"aenderungZurVorversion": [
{ "value": "Mehrere Aenderungen", "lang": null }
],
"handbuchVersion": "1.0",
"xoevHandbuch": false,
"gueltigAb": 1627682400000,
@ -23,7 +37,8 @@
"datentyp": "string",
"codeSpalte": true,
"verwendung": { "code": "REQUIRED" },
"empfohleneCodeSpalte": true
"empfohleneCodeSpalte": true,
"sprache": null
},
{
"spaltennameLang": "Bezeichnung",
@ -31,7 +46,8 @@
"datentyp": "string",
"codeSpalte": false,
"verwendung": { "code": "REQUIRED" },
"empfohleneCodeSpalte": false
"empfohleneCodeSpalte": false,
"sprache": null
},
{
"spaltennameLang": "Hinweis",
@ -39,7 +55,8 @@
"datentyp": "string",
"codeSpalte": false,
"verwendung": { "code": "OPTIONAL" },
"empfohleneCodeSpalte": false
"empfohleneCodeSpalte": false,
"sprache": null
}
],
"daten": [

194
tests/components/nina/test_config_flow.py
View File

@ -11,6 +11,7 @@ from homeassistant import data_entry_flow
from homeassistant.components.nina.const import (
CONF_FILTER_CORONA,
CONF_MESSAGE_SLOTS,
CONF_REGIONS,
CONST_REGION_A_TO_D,
CONST_REGION_E_TO_H,
CONST_REGION_I_TO_L,
@ -21,8 +22,11 @@ from homeassistant.components.nina.const import (
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import load_fixture
from . import mocked_request_function
from tests.common import MockConfigEntry, load_fixture
DUMMY_DATA: dict[str, Any] = {
CONF_MESSAGE_SLOTS: 5,
@ -35,14 +39,19 @@ DUMMY_DATA: dict[str, Any] = {
CONF_FILTER_CORONA: True,
}
DUMMY_RESPONSE: dict[str, Any] = json.loads(load_fixture("sample_regions.json", "nina"))
DUMMY_RESPONSE_REGIONS: dict[str, Any] = json.loads(
load_fixture("sample_regions.json", "nina")
)
DUMMY_RESPONSE_WARNIGNS: dict[str, Any] = json.loads(
load_fixture("sample_warnings.json", "nina")
)
async def test_show_set_form(hass: HomeAssistant) -> None:
"""Test that the setup form is served."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
return_value=DUMMY_RESPONSE,
wraps=mocked_request_function,
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@ -86,7 +95,7 @@ async def test_step_user(hass: HomeAssistant) -> None:
"""Test starting a flow by user with valid values."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
return_value=DUMMY_RESPONSE,
wraps=mocked_request_function,
), patch(
"homeassistant.components.nina.async_setup_entry",
return_value=True,
@ -104,7 +113,7 @@ async def test_step_user_no_selection(hass: HomeAssistant) -> None:
"""Test starting a flow by user with no selection."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
return_value=DUMMY_RESPONSE,
wraps=mocked_request_function,
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@ -120,7 +129,7 @@ async def test_step_user_already_configured(hass: HomeAssistant) -> None:
"""Test starting a flow by user but it was already configured."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
return_value=DUMMY_RESPONSE,
wraps=mocked_request_function,
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=DUMMY_DATA
@ -132,3 +141,176 @@ async def test_step_user_already_configured(hass: HomeAssistant) -> None:
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "single_instance_allowed"
async def test_options_flow_init(hass: HomeAssistant) -> None:
"""Test config flow options."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data={
CONF_FILTER_CORONA: DUMMY_DATA[CONF_FILTER_CORONA],
CONF_MESSAGE_SLOTS: DUMMY_DATA[CONF_MESSAGE_SLOTS],
CONST_REGION_A_TO_D: DUMMY_DATA[CONST_REGION_A_TO_D],
CONF_REGIONS: {"095760000000": "Aach"},
},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.nina.async_setup_entry", return_value=True
), patch(
"pynina.baseApi.BaseAPI._makeRequest",
wraps=mocked_request_function,
):
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONST_REGION_A_TO_D: ["072350000000_1"],
CONST_REGION_E_TO_H: [],
CONST_REGION_I_TO_L: [],
CONST_REGION_M_TO_Q: [],
CONST_REGION_R_TO_U: [],
CONST_REGION_V_TO_Z: [],
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] is None
assert dict(config_entry.data) == {
CONF_FILTER_CORONA: DUMMY_DATA[CONF_FILTER_CORONA],
CONF_MESSAGE_SLOTS: DUMMY_DATA[CONF_MESSAGE_SLOTS],
CONST_REGION_A_TO_D: ["072350000000_1"],
CONST_REGION_E_TO_H: [],
CONST_REGION_I_TO_L: [],
CONST_REGION_M_TO_Q: [],
CONST_REGION_R_TO_U: [],
CONST_REGION_V_TO_Z: [],
CONF_REGIONS: {
"072350000000": "Damflos (Trier-Saarburg - Rheinland-Pfalz)"
},
}
async def test_options_flow_with_no_selection(hass: HomeAssistant) -> None:
"""Test config flow options with no selection."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=DUMMY_DATA,
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.nina.async_setup_entry", return_value=True
), patch(
"pynina.baseApi.BaseAPI._makeRequest",
wraps=mocked_request_function,
):
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONST_REGION_A_TO_D: [],
CONST_REGION_E_TO_H: [],
CONST_REGION_I_TO_L: [],
CONST_REGION_M_TO_Q: [],
CONST_REGION_R_TO_U: [],
CONST_REGION_V_TO_Z: [],
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] == {"base": "no_selection"}
async def test_options_flow_connection_error(hass: HomeAssistant) -> None:
"""Test config flow options but no connection."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=DUMMY_DATA,
)
config_entry.add_to_hass(hass)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
side_effect=ApiError("Could not connect to Api"),
):
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_options_flow_unexpected_exception(hass: HomeAssistant) -> None:
"""Test config flow options but with an unexpected exception."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=DUMMY_DATA,
)
config_entry.add_to_hass(hass)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
side_effect=Exception("DUMMY"),
):
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
async def test_options_flow_entity_removal(hass: HomeAssistant) -> None:
"""Test if old entities are removed."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=DUMMY_DATA,
)
config_entry.add_to_hass(hass)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
wraps=mocked_request_function,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_MESSAGE_SLOTS: 2,
CONST_REGION_A_TO_D: ["072350000000", "095760000000"],
CONST_REGION_E_TO_H: [],
CONST_REGION_I_TO_L: [],
CONST_REGION_M_TO_Q: [],
CONST_REGION_R_TO_U: [],
CONST_REGION_V_TO_Z: [],
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entity_registry: er = er.async_get(hass)
entries = er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
)
assert len(entries) == 2