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

Address late review of nina ()

pull/62215/head
Maximilian 2021-12-17 15:14:59 +00:00 committed by GitHub
parent ed1ce7d9f9
commit 703b689183
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 79 additions and 81 deletions

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

@ -1,17 +1,18 @@
"""The Nina integration."""
from __future__ import annotations
from datetime import timedelta
import datetime as dt
from typing import Any
from async_timeout import timeout
from pynina import ApiError, Nina, Warning as NinaWarning
from pynina import ApiError, Nina
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import (
_LOGGER,
@ -59,33 +60,26 @@ class NINADataUpdateCoordinator(DataUpdateCoordinator):
self.warnings: dict[str, Any] = {}
self.corona_filter: bool = corona_filter
for region in regions.keys():
for region in regions:
self._nina.addRegion(region)
update_interval: timedelta = SCAN_INTERVAL
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL)
async def _async_update_data(self) -> dict[str, Any]:
"""Update data."""
try:
async with timeout(10):
async with timeout(10):
try:
await self._nina.update()
return self._parse_data()
except ApiError as err:
raise UpdateFailed(err) from err
except ApiError as err:
raise UpdateFailed(err) from err
return self._parse_data()
def _parse_data(self) -> dict[str, Any]:
"""Parse warning data."""
return_data: dict[str, Any] = {}
for (
region_id
) in self._nina.warnings: # pylint: disable=consider-using-dict-items
raw_warnings: list[NinaWarning] = self._nina.warnings[region_id]
for region_id, raw_warnings in self._nina.warnings.items():
warnings_for_regions: list[Any] = []
for raw_warn in raw_warnings:
@ -95,12 +89,22 @@ class NINADataUpdateCoordinator(DataUpdateCoordinator):
warn_obj: dict[str, Any] = {
ATTR_ID: raw_warn.id,
ATTR_HEADLINE: raw_warn.headline,
ATTR_SENT: raw_warn.sent or "",
ATTR_START: raw_warn.start or "",
ATTR_EXPIRES: raw_warn.expires or "",
ATTR_SENT: self._to_utc(raw_warn.sent),
ATTR_START: self._to_utc(raw_warn.start),
ATTR_EXPIRES: self._to_utc(raw_warn.expires),
}
warnings_for_regions.append(warn_obj)
return_data[region_id] = warnings_for_regions
return return_data
@staticmethod
def _to_utc(input_time: str) -> str | None:
if input_time:
return (
dt.datetime.fromisoformat(input_time)
.astimezone(dt_util.UTC)
.isoformat()
)
return None

20
homeassistant/components/nina/binary_sensor.py
View File

@ -53,37 +53,33 @@ class NINAMessage(CoordinatorEntity, BinarySensorEntity):
self,
coordinator: NINADataUpdateCoordinator,
region: str,
regionName: str,
slotID: int,
region_name: str,
slot_id: int,
) -> None:
"""Initialize."""
super().__init__(coordinator)
self._region: str = region
self._region_name: str = regionName
self._slot_id: int = slotID
self._warning_index: int = slotID - 1
self._warning_index: int = slot_id - 1
self._coordinator: NINADataUpdateCoordinator = coordinator
self._attr_name: str = f"Warning: {self._region_name} {self._slot_id}"
self._attr_unique_id: str = f"{self._region}-{self._slot_id}"
self._attr_name: str = f"Warning: {region_name} {slot_id}"
self._attr_unique_id: str = f"{region}-{slot_id}"
self._attr_device_class: str = BinarySensorDeviceClass.SAFETY
@property
def is_on(self) -> bool:
"""Return the state of the sensor."""
return len(self._coordinator.data[self._region]) > self._warning_index
return len(self.coordinator.data[self._region]) > self._warning_index
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return extra attributes of the sensor."""
if (
not len(self._coordinator.data[self._region]) > self._warning_index
not len(self.coordinator.data[self._region]) > self._warning_index
) or not self.is_on:
return {}
data: dict[str, Any] = self._coordinator.data[self._region][self._warning_index]
data: dict[str, Any] = self.coordinator.data[self._region][self._warning_index]
return {
ATTR_HEADLINE: data[ATTR_HEADLINE],

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

@ -8,6 +8,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from .const import (
@ -45,53 +46,46 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
has_error: bool = False
if not self._all_region_codes_sorted:
nina: Nina = Nina(async_get_clientsession(self.hass))
if len(self._all_region_codes_sorted) == 0:
try:
nina: Nina = Nina()
self._all_region_codes_sorted = self.swap_key_value(
await nina.getAllRegionalCodes()
)
self.split_regions()
except ApiError as err:
_LOGGER.warning("NINA setup error: %s", err)
except ApiError:
errors["base"] = "cannot_connect"
has_error = True
except Exception as err: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception: %s", err)
errors["base"] = "unknown"
return self.async_abort(reason="unknown")
if user_input is not None and not has_error:
config: dict[str, Any] = user_input
self.split_regions()
config[CONF_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):
config[CONF_REGIONS] += group_input
user_input[CONF_REGIONS] += group_input
if len(config[CONF_REGIONS]) > 0:
if user_input[CONF_REGIONS]:
tmp: dict[str, Any] = {}
for reg in config[CONF_REGIONS]:
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
config[CONF_REGIONS] = compact
user_input[CONF_REGIONS] = compact
return self.async_create_entry(title="NINA", data=config)
return self.async_create_entry(title="NINA", data=user_input)
errors["base"] = "no_selection"
@ -122,7 +116,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
all_region_codes_swaped[value] = key
else:
for i in range(len(dict_to_sort)):
tmp_value: str = value + "_" + str(i)
tmp_value: str = f"{value}_{i}"
if tmp_value not in all_region_codes_swaped:
all_region_codes_swaped[tmp_value] = key
break

38
tests/components/nina/test_binary_sensor.py
View File

@ -3,8 +3,6 @@ import json
from typing import Any, Dict
from unittest.mock import patch
from pynina import ApiError
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.nina.const import (
ATTR_EXPIRES,
@ -64,9 +62,9 @@ async def test_sensors(hass: HomeAssistant) -> None:
assert state_w1.state == STATE_ON
assert state_w1.attributes.get(ATTR_HEADLINE) == "Ausfall Notruf 112"
assert state_w1.attributes.get(ATTR_ID) == "mow.DE-NW-BN-SE030-20201014-30-000"
assert state_w1.attributes.get(ATTR_SENT) == "2021-10-11T05:20:00+01:00"
assert state_w1.attributes.get(ATTR_START) == "2021-11-01T05:20:00+01:00"
assert state_w1.attributes.get(ATTR_EXPIRES) == "3021-11-22T05:19:00+01:00"
assert state_w1.attributes.get(ATTR_SENT) == "2021-10-11T04:20:00+00:00"
assert state_w1.attributes.get(ATTR_START) == "2021-11-01T04:20:00+00:00"
assert state_w1.attributes.get(ATTR_EXPIRES) == "3021-11-22T04:19:00+00:00"
assert entry_w1.unique_id == "083350000000-1"
assert state_w1.attributes.get("device_class") == BinarySensorDeviceClass.SAFETY
@ -157,9 +155,9 @@ async def test_sensors_without_corona_filter(hass: HomeAssistant) -> None:
== "Corona-Verordnung des Landes: Warnstufe durch Landesgesundheitsamt ausgerufen"
)
assert state_w1.attributes.get(ATTR_ID) == "mow.DE-BW-S-SE018-20211102-18-001"
assert state_w1.attributes.get(ATTR_SENT) == "2021-11-02T20:07:16+01:00"
assert state_w1.attributes.get(ATTR_START) == ""
assert state_w1.attributes.get(ATTR_EXPIRES) == ""
assert state_w1.attributes.get(ATTR_SENT) == "2021-11-02T19:07:16+00:00"
assert state_w1.attributes.get(ATTR_START) is None
assert state_w1.attributes.get(ATTR_EXPIRES) is None
assert entry_w1.unique_id == "083350000000-1"
assert state_w1.attributes.get("device_class") == BinarySensorDeviceClass.SAFETY
@ -170,9 +168,9 @@ async def test_sensors_without_corona_filter(hass: HomeAssistant) -> None:
assert state_w2.state == STATE_ON
assert state_w2.attributes.get(ATTR_HEADLINE) == "Ausfall Notruf 112"
assert state_w2.attributes.get(ATTR_ID) == "mow.DE-NW-BN-SE030-20201014-30-000"
assert state_w2.attributes.get(ATTR_SENT) == "2021-10-11T05:20:00+01:00"
assert state_w2.attributes.get(ATTR_START) == "2021-11-01T05:20:00+01:00"
assert state_w2.attributes.get(ATTR_EXPIRES) == "3021-11-22T05:19:00+01:00"
assert state_w2.attributes.get(ATTR_SENT) == "2021-10-11T04:20:00+00:00"
assert state_w2.attributes.get(ATTR_START) == "2021-11-01T04:20:00+00:00"
assert state_w2.attributes.get(ATTR_EXPIRES) == "3021-11-22T04:19:00+00:00"
assert entry_w2.unique_id == "083350000000-2"
assert state_w2.attributes.get("device_class") == BinarySensorDeviceClass.SAFETY
@ -215,21 +213,3 @@ async def test_sensors_without_corona_filter(hass: HomeAssistant) -> None:
assert entry_w5.unique_id == "083350000000-5"
assert state_w5.attributes.get("device_class") == BinarySensorDeviceClass.SAFETY
async def test_sensors_connection_error(hass: HomeAssistant) -> None:
"""Test the creation and values of the NINA sensors with no connected."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
side_effect=ApiError("Could not connect to Api"),
):
conf_entry: MockConfigEntry = MockConfigEntry(
domain=DOMAIN, title="NINA", data=ENTRY_DATA
)
conf_entry.add_to_hass(hass)
await hass.config_entries.async_setup(conf_entry.entry_id)
await hass.async_block_till_done()
assert conf_entry.state == ConfigEntryState.SETUP_RETRY

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

@ -87,6 +87,9 @@ async def test_step_user(hass: HomeAssistant) -> None:
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
return_value=DUMMY_RESPONSE,
), patch(
"homeassistant.components.nina.async_setup_entry",
return_value=True,
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@ -128,3 +131,4 @@ 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"

20
tests/components/nina/test_init.py
View File

@ -3,6 +3,8 @@ import json
from typing import Any, Dict
from unittest.mock import patch
from pynina import ApiError
from homeassistant.components.nina.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
@ -44,3 +46,21 @@ async def test_config_entry_not_ready(hass: HomeAssistant) -> None:
entry: MockConfigEntry = await init_integration(hass)
assert entry.state == ConfigEntryState.LOADED
async def test_sensors_connection_error(hass: HomeAssistant) -> None:
"""Test the creation and values of the NINA sensors with no connected."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
side_effect=ApiError("Could not connect to Api"),
):
conf_entry: MockConfigEntry = MockConfigEntry(
domain=DOMAIN, title="NINA", data=ENTRY_DATA
)
conf_entry.add_to_hass(hass)
await hass.config_entries.async_setup(conf_entry.entry_id)
await hass.async_block_till_done()
assert conf_entry.state == ConfigEntryState.SETUP_RETRY