mirror of https://github.com/home-assistant/core
Add Pegel Online integration (#97028)
This commit is contained in:
Michael
GitHub
parent
b4a46b9817
commit
5caa1969c5
|
|
@ -922,6 +922,8 @@ build.json @home-assistant/supervisor
|
|||
/tests/components/panel_iframe/ @home-assistant/frontend
|
||||
/homeassistant/components/peco/ @IceBotYT
|
||||
/tests/components/peco/ @IceBotYT
|
||||
/homeassistant/components/pegel_online/ @mib1185
|
||||
/tests/components/pegel_online/ @mib1185
|
||||
/homeassistant/components/persistent_notification/ @home-assistant/core
|
||||
/tests/components/persistent_notification/ @home-assistant/core
|
||||
/homeassistant/components/philips_js/ @elupus
|
||||
|
|
|
|||
|
|
@ -0,0 +1,49 @@
|
|||
"""The PEGELONLINE component."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from aiopegelonline import PegelOnline
|
||||
|
||||
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 .const import (
|
||||
CONF_STATION,
|
||||
DOMAIN,
|
||||
)
|
||||
from .coordinator import PegelOnlineDataUpdateCoordinator
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up PEGELONLINE entry."""
|
||||
station_uuid = entry.data[CONF_STATION]
|
||||
|
||||
_LOGGER.debug("Setting up station with uuid %s", station_uuid)
|
||||
|
||||
api = PegelOnline(async_get_clientsession(hass))
|
||||
station = await api.async_get_station_details(station_uuid)
|
||||
|
||||
coordinator = PegelOnlineDataUpdateCoordinator(hass, entry.title, api, station)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][entry.entry_id] = coordinator
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload PEGELONLINE entry."""
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
return unload_ok
|
||||
|
|
@ -0,0 +1,134 @@
|
|||
"""Config flow for PEGELONLINE."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from aiopegelonline import CONNECT_ERRORS, PegelOnline
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import (
|
||||
CONF_LATITUDE,
|
||||
CONF_LOCATION,
|
||||
CONF_LONGITUDE,
|
||||
CONF_RADIUS,
|
||||
UnitOfLength,
|
||||
)
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.selector import (
|
||||
LocationSelector,
|
||||
NumberSelector,
|
||||
NumberSelectorConfig,
|
||||
SelectOptionDict,
|
||||
SelectSelector,
|
||||
SelectSelectorConfig,
|
||||
SelectSelectorMode,
|
||||
)
|
||||
|
||||
from .const import CONF_STATION, DEFAULT_RADIUS, DOMAIN
|
||||
|
||||
|
||||
class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Init the FlowHandler."""
|
||||
super().__init__()
|
||||
self._data: dict[str, Any] = {}
|
||||
self._stations: dict[str, str] = {}
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
if not user_input:
|
||||
return self._show_form_user()
|
||||
|
||||
api = PegelOnline(async_get_clientsession(self.hass))
|
||||
try:
|
||||
stations = await api.async_get_nearby_stations(
|
||||
user_input[CONF_LOCATION][CONF_LATITUDE],
|
||||
user_input[CONF_LOCATION][CONF_LONGITUDE],
|
||||
user_input[CONF_RADIUS],
|
||||
)
|
||||
except CONNECT_ERRORS:
|
||||
return self._show_form_user(user_input, errors={"base": "cannot_connect"})
|
||||
|
||||
if len(stations) == 0:
|
||||
return self._show_form_user(user_input, errors={CONF_RADIUS: "no_stations"})
|
||||
|
||||
for uuid, station in stations.items():
|
||||
self._stations[uuid] = f"{station.name} {station.water_name}"
|
||||
|
||||
self._data = user_input
|
||||
|
||||
return await self.async_step_select_station()
|
||||
|
||||
async def async_step_select_station(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the step select_station of a flow initialized by the user."""
|
||||
if not user_input:
|
||||
stations = [
|
||||
SelectOptionDict(value=k, label=v) for k, v in self._stations.items()
|
||||
]
|
||||
return self.async_show_form(
|
||||
step_id="select_station",
|
||||
description_placeholders={"stations_count": str(len(self._stations))},
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_STATION): SelectSelector(
|
||||
SelectSelectorConfig(
|
||||
options=stations, mode=SelectSelectorMode.DROPDOWN
|
||||
)
|
||||
)
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
await self.async_set_unique_id(user_input[CONF_STATION])
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=self._stations[user_input[CONF_STATION]],
|
||||
data=user_input,
|
||||
)
|
||||
|
||||
def _show_form_user(
|
||||
self,
|
||||
user_input: dict[str, Any] | None = None,
|
||||
errors: dict[str, Any] | None = None,
|
||||
) -> FlowResult:
|
||||
if user_input is None:
|
||||
user_input = {}
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_LOCATION,
|
||||
default=user_input.get(
|
||||
CONF_LOCATION,
|
||||
{
|
||||
"latitude": self.hass.config.latitude,
|
||||
"longitude": self.hass.config.longitude,
|
||||
},
|
||||
),
|
||||
): LocationSelector(),
|
||||
vol.Required(
|
||||
CONF_RADIUS, default=user_input.get(CONF_RADIUS, DEFAULT_RADIUS)
|
||||
): NumberSelector(
|
||||
NumberSelectorConfig(
|
||||
min=1,
|
||||
max=100,
|
||||
step=1,
|
||||
unit_of_measurement=UnitOfLength.KILOMETERS,
|
||||
),
|
||||
),
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
"""Constants for PEGELONLINE."""
|
||||
from datetime import timedelta
|
||||
|
||||
DOMAIN = "pegel_online"
|
||||
|
||||
DEFAULT_RADIUS = "25"
|
||||
CONF_STATION = "station"
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=5)
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
"""DataUpdateCoordinator for pegel_online."""
|
||||
import logging
|
||||
|
||||
from aiopegelonline import CONNECT_ERRORS, PegelOnline, Station
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import MIN_TIME_BETWEEN_UPDATES
|
||||
from .model import PegelOnlineData
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PegelOnlineDataUpdateCoordinator(DataUpdateCoordinator[PegelOnlineData]):
|
||||
"""DataUpdateCoordinator for the pegel_online integration."""
|
||||
|
||||
def __init__(
|
||||
self, hass: HomeAssistant, name: str, api: PegelOnline, station: Station
|
||||
) -> None:
|
||||
"""Initialize the PegelOnlineDataUpdateCoordinator."""
|
||||
self.api = api
|
||||
self.station = station
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=name,
|
||||
update_interval=MIN_TIME_BETWEEN_UPDATES,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> PegelOnlineData:
|
||||
"""Fetch data from API endpoint."""
|
||||
try:
|
||||
current_measurement = await self.api.async_get_station_measurement(
|
||||
self.station.uuid
|
||||
)
|
||||
except CONNECT_ERRORS as err:
|
||||
raise UpdateFailed(f"Failed to communicate with API: {err}") from err
|
||||
|
||||
return {"current_measurement": current_measurement}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
"""The PEGELONLINE base entity."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import PegelOnlineDataUpdateCoordinator
|
||||
|
||||
|
||||
class PegelOnlineEntity(CoordinatorEntity):
|
||||
"""Representation of a PEGELONLINE entity."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_available = True
|
||||
|
||||
def __init__(self, coordinator: PegelOnlineDataUpdateCoordinator) -> None:
|
||||
"""Initialize a PEGELONLINE entity."""
|
||||
super().__init__(coordinator)
|
||||
self.station = coordinator.station
|
||||
self._attr_extra_state_attributes = {}
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device information of the entity."""
|
||||
return DeviceInfo(
|
||||
identifiers={(DOMAIN, self.station.uuid)},
|
||||
name=f"{self.station.name} {self.station.water_name}",
|
||||
manufacturer=self.station.agency,
|
||||
configuration_url=self.station.base_data_url,
|
||||
)
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
{
|
||||
"domain": "pegel_online",
|
||||
"name": "PEGELONLINE",
|
||||
"codeowners": ["@mib1185"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/pegel_online",
|
||||
"integration_type": "service",
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["aiopegelonline"],
|
||||
"requirements": ["aiopegelonline==0.0.5"]
|
||||
}
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
"""Models for PEGELONLINE."""
|
||||
|
||||
from typing import TypedDict
|
||||
|
||||
from aiopegelonline import CurrentMeasurement
|
||||
|
||||
|
||||
class PegelOnlineData(TypedDict):
|
||||
"""TypedDict for PEGELONLINE Coordinator Data."""
|
||||
|
||||
current_measurement: CurrentMeasurement
|
||||
|
|
@ -0,0 +1,89 @@
|
|||
"""PEGELONLINE sensor entities."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import PegelOnlineDataUpdateCoordinator
|
||||
from .entity import PegelOnlineEntity
|
||||
from .model import PegelOnlineData
|
||||
|
||||
|
||||
@dataclass
|
||||
class PegelOnlineRequiredKeysMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
fn_native_unit: Callable[[PegelOnlineData], str]
|
||||
fn_native_value: Callable[[PegelOnlineData], float]
|
||||
|
||||
|
||||
@dataclass
|
||||
class PegelOnlineSensorEntityDescription(
|
||||
SensorEntityDescription, PegelOnlineRequiredKeysMixin
|
||||
):
|
||||
"""PEGELONLINE sensor entity description."""
|
||||
|
||||
|
||||
SENSORS: tuple[PegelOnlineSensorEntityDescription, ...] = (
|
||||
PegelOnlineSensorEntityDescription(
|
||||
key="current_measurement",
|
||||
translation_key="current_measurement",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
fn_native_unit=lambda data: data["current_measurement"].uom,
|
||||
fn_native_value=lambda data: data["current_measurement"].value,
|
||||
icon="mdi:waves-arrow-up",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up the PEGELONLINE sensor."""
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
async_add_entities(
|
||||
[PegelOnlineSensor(coordinator, description) for description in SENSORS]
|
||||
)
|
||||
|
||||
|
||||
class PegelOnlineSensor(PegelOnlineEntity, SensorEntity):
|
||||
"""Representation of a PEGELONLINE sensor."""
|
||||
|
||||
entity_description: PegelOnlineSensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: PegelOnlineDataUpdateCoordinator,
|
||||
description: PegelOnlineSensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize a PEGELONLINE sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
self._attr_unique_id = f"{self.station.uuid}_{description.key}"
|
||||
self._attr_native_unit_of_measurement = self.entity_description.fn_native_unit(
|
||||
coordinator.data
|
||||
)
|
||||
|
||||
if self.station.latitude and self.station.longitude:
|
||||
self._attr_extra_state_attributes.update(
|
||||
{
|
||||
ATTR_LATITUDE: self.station.latitude,
|
||||
ATTR_LONGITUDE: self.station.longitude,
|
||||
}
|
||||
)
|
||||
|
||||
@property
|
||||
def native_value(self) -> float:
|
||||
"""Return the state of the device."""
|
||||
return self.entity_description.fn_native_value(self.coordinator.data)
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"description": "Select the area, where you want to search for water measuring stations",
|
||||
"data": {
|
||||
"location": "[%key:common::config_flow::data::location%]",
|
||||
"radius": "Search radius (in km)"
|
||||
}
|
||||
},
|
||||
"select_station": {
|
||||
"title": "Select the measuring station to add",
|
||||
"description": "Found {stations_count} stations in radius",
|
||||
"data": {
|
||||
"station": "Station"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"no_stations": "Could not find any station in range."
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"current_measurement": {
|
||||
"name": "Water level"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -338,6 +338,7 @@ FLOWS = {
|
|||
"p1_monitor",
|
||||
"panasonic_viera",
|
||||
"peco",
|
||||
"pegel_online",
|
||||
"philips_js",
|
||||
"pi_hole",
|
||||
"picnic",
|
||||
|
|
|
|||
|
|
@ -4136,6 +4136,12 @@
|
|||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"pegel_online": {
|
||||
"name": "PEGELONLINE",
|
||||
"integration_type": "service",
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"pencom": {
|
||||
"name": "Pencom",
|
||||
"integration_type": "hub",
|
||||
|
|
|
|||
|
|
@ -303,6 +303,9 @@ aiooncue==0.3.5
|
|||
# homeassistant.components.openexchangerates
|
||||
aioopenexchangerates==0.4.0
|
||||
|
||||
# homeassistant.components.pegel_online
|
||||
aiopegelonline==0.0.5
|
||||
|
||||
# homeassistant.components.acmeda
|
||||
aiopulse==0.4.3
|
||||
|
||||
|
|
|
|||
|
|
@ -278,6 +278,9 @@ aiooncue==0.3.5
|
|||
# homeassistant.components.openexchangerates
|
||||
aioopenexchangerates==0.4.0
|
||||
|
||||
# homeassistant.components.pegel_online
|
||||
aiopegelonline==0.0.5
|
||||
|
||||
# homeassistant.components.acmeda
|
||||
aiopulse==0.4.3
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,40 @@
|
|||
"""Tests for Pegel Online component."""
|
||||
|
||||
|
||||
class PegelOnlineMock:
|
||||
"""Class mock of PegelOnline."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
nearby_stations=None,
|
||||
station_details=None,
|
||||
station_measurement=None,
|
||||
side_effect=None,
|
||||
) -> None:
|
||||
"""Init the mock."""
|
||||
self.nearby_stations = nearby_stations
|
||||
self.station_details = station_details
|
||||
self.station_measurement = station_measurement
|
||||
self.side_effect = side_effect
|
||||
|
||||
async def async_get_nearby_stations(self, *args):
|
||||
"""Mock async_get_nearby_stations."""
|
||||
if self.side_effect:
|
||||
raise self.side_effect
|
||||
return self.nearby_stations
|
||||
|
||||
async def async_get_station_details(self, *args):
|
||||
"""Mock async_get_station_details."""
|
||||
if self.side_effect:
|
||||
raise self.side_effect
|
||||
return self.station_details
|
||||
|
||||
async def async_get_station_measurement(self, *args):
|
||||
"""Mock async_get_station_measurement."""
|
||||
if self.side_effect:
|
||||
raise self.side_effect
|
||||
return self.station_measurement
|
||||
|
||||
def override_side_effect(self, side_effect):
|
||||
"""Override the side_effect."""
|
||||
self.side_effect = side_effect
|
||||
|
|
@ -0,0 +1,209 @@
|
|||
"""Tests for Pegel Online config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from aiohttp.client_exceptions import ClientError
|
||||
from aiopegelonline import Station
|
||||
|
||||
from homeassistant.components.pegel_online.const import (
|
||||
CONF_STATION,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import (
|
||||
CONF_LATITUDE,
|
||||
CONF_LOCATION,
|
||||
CONF_LONGITUDE,
|
||||
CONF_RADIUS,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from . import PegelOnlineMock
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_USER_DATA_STEP1 = {
|
||||
CONF_LOCATION: {CONF_LATITUDE: 51.0, CONF_LONGITUDE: 13.0},
|
||||
CONF_RADIUS: 25,
|
||||
}
|
||||
|
||||
MOCK_USER_DATA_STEP2 = {CONF_STATION: "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"}
|
||||
|
||||
MOCK_CONFIG_ENTRY_DATA = {CONF_STATION: "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"}
|
||||
|
||||
MOCK_NEARBY_STATIONS = {
|
||||
"3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8": Station(
|
||||
{
|
||||
"uuid": "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"number": "501060",
|
||||
"shortname": "DRESDEN",
|
||||
"longname": "DRESDEN",
|
||||
"km": 55.63,
|
||||
"agency": "STANDORT DRESDEN",
|
||||
"longitude": 13.738831783620384,
|
||||
"latitude": 51.054459765598125,
|
||||
"water": {"shortname": "ELBE", "longname": "ELBE"},
|
||||
}
|
||||
),
|
||||
"85d686f1-xxxx-xxxx-xxxx-3207b50901a7": Station(
|
||||
{
|
||||
"uuid": "85d686f1-xxxx-xxxx-xxxx-3207b50901a7",
|
||||
"number": "501060",
|
||||
"shortname": "MEISSEN",
|
||||
"longname": "MEISSEN",
|
||||
"km": 82.2,
|
||||
"agency": "STANDORT DRESDEN",
|
||||
"longitude": 13.475467710324812,
|
||||
"latitude": 51.16440557554545,
|
||||
"water": {"shortname": "ELBE", "longname": "ELBE"},
|
||||
}
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def test_user(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow by user."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.pegel_online.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.pegel_online.config_flow.PegelOnline",
|
||||
) as pegelonline:
|
||||
pegelonline.return_value = PegelOnlineMock(nearby_stations=MOCK_NEARBY_STATIONS)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP1
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "select_station"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP2
|
||||
)
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_STATION] == "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"
|
||||
assert result["title"] == "DRESDEN ELBE"
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_setup_entry.called
|
||||
|
||||
|
||||
async def test_user_already_configured(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow by user with an already configured statioon."""
|
||||
mock_config = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=MOCK_CONFIG_ENTRY_DATA,
|
||||
unique_id=MOCK_CONFIG_ENTRY_DATA[CONF_STATION],
|
||||
)
|
||||
mock_config.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.pegel_online.config_flow.PegelOnline",
|
||||
) as pegelonline:
|
||||
pegelonline.return_value = PegelOnlineMock(nearby_stations=MOCK_NEARBY_STATIONS)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP1
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "select_station"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP2
|
||||
)
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_connection_error(hass: HomeAssistant) -> None:
|
||||
"""Test connection error during user flow."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.pegel_online.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.pegel_online.config_flow.PegelOnline",
|
||||
) as pegelonline:
|
||||
# connection issue during setup
|
||||
pegelonline.return_value = PegelOnlineMock(side_effect=ClientError)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP1
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"]["base"] == "cannot_connect"
|
||||
|
||||
# connection issue solved
|
||||
pegelonline.return_value = PegelOnlineMock(nearby_stations=MOCK_NEARBY_STATIONS)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP1
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "select_station"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP2
|
||||
)
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_STATION] == "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"
|
||||
assert result["title"] == "DRESDEN ELBE"
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_setup_entry.called
|
||||
|
||||
|
||||
async def test_user_no_stations(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow by user which does not find any station."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.pegel_online.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.pegel_online.config_flow.PegelOnline",
|
||||
) as pegelonline:
|
||||
# no stations found
|
||||
pegelonline.return_value = PegelOnlineMock(nearby_stations={})
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP1
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"][CONF_RADIUS] == "no_stations"
|
||||
|
||||
# stations found, go ahead
|
||||
pegelonline.return_value = PegelOnlineMock(nearby_stations=MOCK_NEARBY_STATIONS)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP1
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "select_station"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA_STEP2
|
||||
)
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_STATION] == "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"
|
||||
assert result["title"] == "DRESDEN ELBE"
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_setup_entry.called
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
"""Test pegel_online component."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from aiohttp.client_exceptions import ClientError
|
||||
from aiopegelonline import CurrentMeasurement, Station
|
||||
|
||||
from homeassistant.components.pegel_online.const import (
|
||||
CONF_STATION,
|
||||
DOMAIN,
|
||||
MIN_TIME_BETWEEN_UPDATES,
|
||||
)
|
||||
from homeassistant.const import STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.util import utcnow
|
||||
|
||||
from . import PegelOnlineMock
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
MOCK_CONFIG_ENTRY_DATA = {CONF_STATION: "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"}
|
||||
|
||||
MOCK_STATION_DETAILS = Station(
|
||||
{
|
||||
"uuid": "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"number": "501060",
|
||||
"shortname": "DRESDEN",
|
||||
"longname": "DRESDEN",
|
||||
"km": 55.63,
|
||||
"agency": "STANDORT DRESDEN",
|
||||
"longitude": 13.738831783620384,
|
||||
"latitude": 51.054459765598125,
|
||||
"water": {"shortname": "ELBE", "longname": "ELBE"},
|
||||
}
|
||||
)
|
||||
MOCK_STATION_MEASUREMENT = CurrentMeasurement("cm", 56)
|
||||
|
||||
|
||||
async def test_update_error(hass: HomeAssistant) -> None:
|
||||
"""Tests error during update entity."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=MOCK_CONFIG_ENTRY_DATA,
|
||||
unique_id=MOCK_CONFIG_ENTRY_DATA[CONF_STATION],
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch("homeassistant.components.pegel_online.PegelOnline") as pegelonline:
|
||||
pegelonline.return_value = PegelOnlineMock(
|
||||
station_details=MOCK_STATION_DETAILS,
|
||||
station_measurement=MOCK_STATION_MEASUREMENT,
|
||||
)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.dresden_elbe_water_level")
|
||||
assert state
|
||||
|
||||
pegelonline().override_side_effect(ClientError)
|
||||
async_fire_time_changed(hass, utcnow() + MIN_TIME_BETWEEN_UPDATES)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.dresden_elbe_water_level")
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
"""Test pegel_online component."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from aiopegelonline import CurrentMeasurement, Station
|
||||
|
||||
from homeassistant.components.pegel_online.const import CONF_STATION, DOMAIN
|
||||
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import PegelOnlineMock
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_CONFIG_ENTRY_DATA = {CONF_STATION: "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8"}
|
||||
|
||||
MOCK_STATION_DETAILS = Station(
|
||||
{
|
||||
"uuid": "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"number": "501060",
|
||||
"shortname": "DRESDEN",
|
||||
"longname": "DRESDEN",
|
||||
"km": 55.63,
|
||||
"agency": "STANDORT DRESDEN",
|
||||
"longitude": 13.738831783620384,
|
||||
"latitude": 51.054459765598125,
|
||||
"water": {"shortname": "ELBE", "longname": "ELBE"},
|
||||
}
|
||||
)
|
||||
MOCK_STATION_MEASUREMENT = CurrentMeasurement("cm", 56)
|
||||
|
||||
|
||||
async def test_sensor(hass: HomeAssistant) -> None:
|
||||
"""Tests sensor entity."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=MOCK_CONFIG_ENTRY_DATA,
|
||||
unique_id=MOCK_CONFIG_ENTRY_DATA[CONF_STATION],
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch("homeassistant.components.pegel_online.PegelOnline") as pegelonline:
|
||||
pegelonline.return_value = PegelOnlineMock(
|
||||
station_details=MOCK_STATION_DETAILS,
|
||||
station_measurement=MOCK_STATION_MEASUREMENT,
|
||||
)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.dresden_elbe_water_level")
|
||||
assert state.name == "DRESDEN ELBE Water level"
|
||||
assert state.state == "56"
|
||||
assert state.attributes[ATTR_LATITUDE] == 51.054459765598125
|
||||
assert state.attributes[ATTR_LONGITUDE] == 13.738831783620384
|
||||
Loading…
Reference in New Issue