home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add Rehlko (formerly Kohler Energy Management) Integration (#143602)

Browse Source 
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Pete Sage
2025-04-30 06:44:16 -04:00
committed by GitHub
parent 73a1dbffeb
commit 6c633668f6
23 changed files with 2370 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS generated
View File

@ -1260,6 +1260,8 @@ build.json @home-assistant/supervisor
/tests/components/recovery_mode/ @home-assistant/core
/homeassistant/components/refoss/ @ashionky
/tests/components/refoss/ @ashionky
/homeassistant/components/rehlko/ @bdraco @peterager
/tests/components/rehlko/ @bdraco @peterager
/homeassistant/components/remote/ @home-assistant/core
/tests/components/remote/ @home-assistant/core
/homeassistant/components/remote_calendar/ @Thomas55555

95
homeassistant/components/rehlko/__init__.py Normal file
View File

@ -0,0 +1,95 @@
"""The Rehlko integration."""
from __future__ import annotations
import logging
from aiokem import AioKem, AuthenticationError
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import (
CONF_REFRESH_TOKEN,
CONNECTION_EXCEPTIONS,
DEVICE_DATA_DEVICES,
DEVICE_DATA_DISPLAY_NAME,
DEVICE_DATA_ID,
DOMAIN,
)
from .coordinator import RehlkoConfigEntry, RehlkoRuntimeData, RehlkoUpdateCoordinator
PLATFORMS = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: RehlkoConfigEntry) -> bool:
"""Set up Rehlko from a config entry."""
websession = async_get_clientsession(hass)
rehlko = AioKem(session=websession)
async def async_refresh_token_update(refresh_token: str) -> None:
"""Handle refresh token update."""
_LOGGER.debug("Saving refresh token")
# Update the config entry with the new refresh token
hass.config_entries.async_update_entry(
entry,
data={**entry.data, CONF_REFRESH_TOKEN: refresh_token},
)
rehlko.set_refresh_token_callback(async_refresh_token_update)
rehlko.set_retry_policy(retry_count=3, retry_delays=[5, 10, 20])
try:
await rehlko.authenticate(
entry.data[CONF_EMAIL],
entry.data[CONF_PASSWORD],
entry.data.get(CONF_REFRESH_TOKEN),
)
except AuthenticationError as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={CONF_EMAIL: entry.data[CONF_EMAIL]},
) from ex
except CONNECTION_EXCEPTIONS as ex:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
) from ex
coordinators: dict[int, RehlkoUpdateCoordinator] = {}
homes = await rehlko.get_homes()
entry.runtime_data = RehlkoRuntimeData(
coordinators=coordinators,
rehlko=rehlko,
homes=homes,
)
for home_data in homes:
for device_data in home_data[DEVICE_DATA_DEVICES]:
device_id = device_data[DEVICE_DATA_ID]
coordinator = RehlkoUpdateCoordinator(
hass=hass,
logger=_LOGGER,
config_entry=entry,
home_data=home_data,
device_id=device_id,
device_data=device_data,
rehlko=rehlko,
name=f"{DOMAIN} {device_data[DEVICE_DATA_DISPLAY_NAME]}",
)
# Intentionally done in series to avoid overloading
# the Rehlko API with requests
await coordinator.async_config_entry_first_refresh()
coordinators[device_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: RehlkoConfigEntry) -> bool:
"""Unload a config entry."""
await entry.runtime_data.rehlko.close()
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

103
homeassistant/components/rehlko/config_flow.py Normal file
View File

@ -0,0 +1,103 @@
"""Config flow for Rehlko integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from aiokem import AioKem, AuthenticationError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONNECTION_EXCEPTIONS, DOMAIN
_LOGGER = logging.getLogger(__name__)
class RehlkoConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Rehlko."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
errors, token_subject = await self._async_validate_or_error(user_input)
if not errors:
await self.async_set_unique_id(token_subject)
self._abort_if_unique_id_configured()
email: str = user_input[CONF_EMAIL]
normalized_email = email.lower()
return self.async_create_entry(title=normalized_email, data=user_input)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_EMAIL): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
)
async def _async_validate_or_error(
self, config: dict[str, Any]
) -> tuple[dict[str, str], str | None]:
"""Validate the user input."""
errors: dict[str, str] = {}
token_subject = None
rehlko = AioKem(session=async_get_clientsession(self.hass))
try:
await rehlko.authenticate(config[CONF_EMAIL], config[CONF_PASSWORD])
except CONNECTION_EXCEPTIONS:
errors["base"] = "cannot_connect"
except AuthenticationError:
errors[CONF_PASSWORD] = "invalid_auth"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
token_subject = rehlko.get_token_subject()
return errors, token_subject
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle reauth."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauth input."""
errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
existing_data = reauth_entry.data
description_placeholders: dict[str, str] = {
CONF_EMAIL: existing_data[CONF_EMAIL]
}
if user_input is not None:
errors, _ = await self._async_validate_or_error(
{**existing_data, **user_input}
)
if not errors:
return self.async_update_reload_and_abort(
reauth_entry,
data_updates=user_input,
)
return self.async_show_form(
description_placeholders=description_placeholders,
step_id="reauth_confirm",
data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
errors=errors,
)

25
homeassistant/components/rehlko/const.py Normal file
View File

@ -0,0 +1,25 @@
"""Constants for the Rehlko integration."""
from aiokem import CommunicationError
DOMAIN = "rehlko"
CONF_REFRESH_TOKEN = "refresh_token"
DEVICE_DATA_DEVICES = "devices"
DEVICE_DATA_PRODUCT = "product"
DEVICE_DATA_FIRMWARE_VERSION = "firmwareVersion"
DEVICE_DATA_MODEL_NAME = "modelDisplayName"
DEVICE_DATA_ID = "id"
DEVICE_DATA_DISPLAY_NAME = "displayName"
DEVICE_DATA_MAC_ADDRESS = "macAddress"
DEVICE_DATA_IS_CONNECTED = "isConnected"
KOHLER = "Kohler"
GENERATOR_DATA_DEVICE = "device"
CONNECTION_EXCEPTIONS = (
TimeoutError,
CommunicationError,
)

78
homeassistant/components/rehlko/coordinator.py Normal file
View File

@ -0,0 +1,78 @@
"""The Rehlko coordinator."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
from aiokem import AioKem, CommunicationError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
type RehlkoConfigEntry = ConfigEntry[RehlkoRuntimeData]
SCAN_INTERVAL_MINUTES = timedelta(minutes=10)
@dataclass
class RehlkoRuntimeData:
"""Dataclass to hold runtime data for the Rehlko integration."""
coordinators: dict[int, RehlkoUpdateCoordinator]
rehlko: AioKem
homes: list[dict[str, Any]]
class RehlkoUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Class to manage fetching Rehlko data API."""
config_entry: RehlkoConfigEntry
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
config_entry: RehlkoConfigEntry,
rehlko: AioKem,
home_data: dict[str, Any],
device_data: dict[str, Any],
device_id: int,
name: str,
) -> None:
"""Initialize."""
self.rehlko = rehlko
self.device_data = device_data
self.device_id = device_id
self.home_data = home_data
super().__init__(
hass=hass,
logger=logger,
config_entry=config_entry,
name=name,
update_interval=SCAN_INTERVAL_MINUTES,
)
async def _async_update_data(self) -> dict[str, Any]:
"""Update data via library."""
try:
result = await self.rehlko.get_generator_data(self.device_id)
except CommunicationError as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_failed",
) from error
return result
@property
def entry_unique_id(self) -> str:
"""Get the unique ID for the entry."""
assert self.config_entry.unique_id
return self.config_entry.unique_id

81
homeassistant/components/rehlko/entity.py Normal file
View File

@ -0,0 +1,81 @@
"""Base class for Rehlko entities."""
from __future__ import annotations
from typing import Any
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
DEVICE_DATA_DISPLAY_NAME,
DEVICE_DATA_FIRMWARE_VERSION,
DEVICE_DATA_IS_CONNECTED,
DEVICE_DATA_MAC_ADDRESS,
DEVICE_DATA_MODEL_NAME,
DEVICE_DATA_PRODUCT,
DOMAIN,
GENERATOR_DATA_DEVICE,
KOHLER,
)
from .coordinator import RehlkoUpdateCoordinator
def _get_device_connections(mac_address: str) -> set[tuple[str, str]]:
"""Get device connections."""
try:
mac_address_hex = mac_address.replace(":", "")
except ValueError: # MacAddress may be invalid if the gateway is offline
return set()
return {(dr.CONNECTION_NETWORK_MAC, mac_address_hex)}
class RehlkoEntity(CoordinatorEntity[RehlkoUpdateCoordinator]):
"""Representation of a Rehlko entity."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: RehlkoUpdateCoordinator,
device_id: int,
device_data: dict,
description: EntityDescription,
use_device_key: bool = False,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
self._device_id = device_id
self._attr_unique_id = (
f"{coordinator.entry_unique_id}_{device_id}_{description.key}"
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{coordinator.entry_unique_id}_{device_id}")},
name=device_data[DEVICE_DATA_DISPLAY_NAME],
hw_version=device_data[DEVICE_DATA_PRODUCT],
sw_version=device_data[DEVICE_DATA_FIRMWARE_VERSION],
model=device_data[DEVICE_DATA_MODEL_NAME],
manufacturer=KOHLER,
connections=_get_device_connections(device_data[DEVICE_DATA_MAC_ADDRESS]),
)
self._use_device_key = use_device_key
@property
def _device_data(self) -> dict[str, Any]:
"""Return the device data."""
return self.coordinator.data[GENERATOR_DATA_DEVICE]
@property
def _rehlko_value(self) -> str:
"""Return the sensor value."""
if self._use_device_key:
return self._device_data[self.entity_description.key]
return self.coordinator.data[self.entity_description.key]
@property
def available(self) -> bool:
"""Return if entity is available."""
return super().available and self._device_data[DEVICE_DATA_IS_CONNECTED]

18
homeassistant/components/rehlko/icons.json Normal file
View File

@ -0,0 +1,18 @@
{
"entity": {
"sensor": {
"engine_speed": {
"default": "mdi:speedometer"
},
"engine_state": {
"default": "mdi:engine"
},
"device_ip_address": {
"default": "mdi:ip-network"
},
"server_ip_address": {
"default": "mdi:server-network"
}
}
}
}

17
homeassistant/components/rehlko/manifest.json Normal file
View File

@ -0,0 +1,17 @@
{
"domain": "rehlko",
"name": "Rehlko",
"codeowners": ["@bdraco", "@peterager"],
"config_flow": true,
"dhcp": [
{
"hostname": "kohlergen*",
"macaddress": "00146F*"
}
],
"documentation": "https://www.home-assistant.io/integrations/rehlko",
"iot_class": "cloud_polling",
"loggers": ["aiokem"],
"quality_scale": "silver",
"requirements": ["aiokem==0.5.6"]
}

78
homeassistant/components/rehlko/quality_scale.yaml Normal file
View File

@ -0,0 +1,78 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
No custom actions are defined.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
No custom actions are defined.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: |
No explicit event subscriptions.
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: |
No custom actions are defined.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: |
No configuration parameters.
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Network information not useful as it is a cloud integration.
discovery: done
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: |
Device type integration.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: todo
strict-typing: todo

203
homeassistant/components/rehlko/sensor.py Normal file
View File

@ -0,0 +1,203 @@
"""Support for Rehlko sensors."""
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
REVOLUTIONS_PER_MINUTE,
EntityCategory,
UnitOfElectricPotential,
UnitOfFrequency,
UnitOfPower,
UnitOfPressure,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import DEVICE_DATA_DEVICES, DEVICE_DATA_ID
from .coordinator import RehlkoConfigEntry
from .entity import RehlkoEntity
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class RehlkoSensorEntityDescription(SensorEntityDescription):
"""Class describing Rehlko sensor entities."""
use_device_key: bool = False
SENSORS: tuple[RehlkoSensorEntityDescription, ...] = (
RehlkoSensorEntityDescription(
key="engineSpeedRpm",
translation_key="engine_speed",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=REVOLUTIONS_PER_MINUTE,
),
RehlkoSensorEntityDescription(
key="engineOilPressurePsi",
translation_key="engine_oil_pressure",
native_unit_of_measurement=UnitOfPressure.PSI,
device_class=SensorDeviceClass.PRESSURE,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="engineCoolantTempF",
translation_key="engine_coolant_temperature",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="batteryVoltageV",
translation_key="battery_voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="lubeOilTempF",
translation_key="lube_oil_temperature",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
RehlkoSensorEntityDescription(
key="controllerTempF",
translation_key="controller_temperature",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="engineCompartmentTempF",
translation_key="engine_compartment_temperature",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
RehlkoSensorEntityDescription(
key="engineFrequencyHz",
translation_key="engine_frequency",
native_unit_of_measurement=UnitOfFrequency.HERTZ,
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="totalOperationHours",
translation_key="total_operation",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
RehlkoSensorEntityDescription(
key="totalRuntimeHours",
translation_key="total_runtime",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
use_device_key=True,
),
RehlkoSensorEntityDescription(
key="runtimeSinceLastMaintenanceHours",
translation_key="runtime_since_last_maintenance",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
RehlkoSensorEntityDescription(
key="deviceIpAddress",
translation_key="device_ip_address",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
use_device_key=True,
),
RehlkoSensorEntityDescription(
key="serverIpAddress",
translation_key="server_ip_address",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
RehlkoSensorEntityDescription(
key="utilityVoltageV",
translation_key="utility_voltage",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="generatorVoltageAvgV",
translation_key="generator_voltage_avg",
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="generatorLoadW",
translation_key="generator_load",
native_unit_of_measurement=UnitOfPower.WATT,
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
),
RehlkoSensorEntityDescription(
key="generatorLoadPercent",
translation_key="generator_load_percent",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: RehlkoConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up sensors."""
homes = config_entry.runtime_data.homes
coordinators = config_entry.runtime_data.coordinators
async_add_entities(
RehlkoSensorEntity(
coordinators[device_data[DEVICE_DATA_ID]],
device_data[DEVICE_DATA_ID],
device_data,
sensor_description,
sensor_description.use_device_key,
)
for home_data in homes
for device_data in home_data[DEVICE_DATA_DEVICES]
for sensor_description in SENSORS
)
class RehlkoSensorEntity(RehlkoEntity, SensorEntity):
"""Representation of a Rehlko sensor."""
@property
def native_value(self) -> StateType:
"""Return the sensor state."""
return self._rehlko_value

99
homeassistant/components/rehlko/strings.json Normal file
View File

@ -0,0 +1,99 @@
{
"config": {
"step": {
"user": {
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"email": "The email used to log in to the Rehlko application.",
"password": "The password used to log in to the Rehlko application."
}
},
"reauth_confirm": {
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"password": "[%key:component::rehlko::config::step::user::data_description::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"entity": {
"sensor": {
"engine_speed": {
"name": "Engine speed"
},
"engine_oil_pressure": {
"name": "Engine oil pressure"
},
"engine_coolant_temperature": {
"name": "Engine coolant temperature"
},
"battery_voltage": {
"name": "Battery voltage"
},
"lube_oil_temperature": {
"name": "Lube oil temperature"
},
"controller_temperature": {
"name": "Controller temperature"
},
"engine_compartment_temperature": {
"name": "Engine compartment temperature"
},
"engine_frequency": {
"name": "Engine frequency"
},
"total_operation": {
"name": "Total operation"
},
"total_runtime": {
"name": "Total runtime"
},
"runtime_since_last_maintenance": {
"name": "Runtime since last maintenance"
},
"device_ip_address": {
"name": "Device IP address"
},
"server_ip_address": {
"name": "Server IP address"
},
"utility_voltage": {
"name": "Utility voltage"
},
"generator_voltage_average": {
"name": "Average generator voltage"
},
"generator_load": {
"name": "Generator load"
},
"generator_load_percent": {
"name": "Generator load percentage"
}
}
},
"exceptions": {
"update_failed": {
"message": "Updating data failed after retries."
},
"invalid_auth": {
"message": "Authentication failed for email {email}."
},
"cannot_connect": {
"message": "Can not connect to Rehlko servers."
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@ -518,6 +518,7 @@ FLOWS = {
"rdw",
"recollect_waste",
"refoss",
"rehlko",
"remote_calendar",
"renault",
"renson",

5
homeassistant/generated/dhcp.py generated
View File

@ -471,6 +471,11 @@ DHCP: Final[list[dict[str, str | bool]]] = [
"domain": "rainforest_eagle",
"macaddress": "D8D5B9*",
},
{
"domain": "rehlko",
"hostname": "kohlergen*",
"macaddress": "00146F*",
},
{
"domain": "reolink",
"hostname": "reolink*",

6
homeassistant/generated/integrations.json
View File

@ -5375,6 +5375,12 @@
"iot_class": "local_polling",
"single_config_entry": true
},
"rehlko": {
"name": "Rehlko",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"rejseplanen": {
"name": "Rejseplanen",
"integration_type": "hub",

3
requirements_all.txt generated
View File

@ -285,6 +285,9 @@ aiokafka==0.10.0
# homeassistant.components.kef
aiokef==0.2.16
# homeassistant.components.rehlko
aiokem==0.5.6
# homeassistant.components.lifx
aiolifx-effects==0.3.2

3
requirements_test_all.txt generated
View File

@ -267,6 +267,9 @@ aioimaplib==2.0.1
# homeassistant.components.apache_kafka
aiokafka==0.10.0
# homeassistant.components.rehlko
aiokem==0.5.6
# homeassistant.components.lifx
aiolifx-effects==0.3.2

1
tests/components/rehlko/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Rehlko Tests Package."""

100
tests/components/rehlko/conftest.py Normal file
View File

@ -0,0 +1,100 @@
"""Module for testing the Rehlko integration in Home Assistant."""
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock, Mock, patch
import pytest
from homeassistant.components.rehlko import CONF_REFRESH_TOKEN, DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_json_value_fixture
TEST_EMAIL = "MyEmail@email.com"
TEST_PASSWORD = "password"
TEST_SUBJECT = TEST_EMAIL.lower()
TEST_REFRESH_TOKEN = "my_refresh_token"
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.rehlko.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture(name="homes")
def rehlko_homes_fixture() -> list[dict[str, Any]]:
"""Create sonos favorites fixture."""
return load_json_value_fixture("homes.json", DOMAIN)
@pytest.fixture(name="generator")
def rehlko_generator_fixture() -> dict[str, Any]:
"""Create sonos favorites fixture."""
return load_json_value_fixture("generator.json", DOMAIN)
@pytest.fixture(name="rehlko_config_entry")
def rehlko_config_entry_fixture() -> MockConfigEntry:
"""Create a config entry fixture."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
},
unique_id=TEST_SUBJECT,
)
@pytest.fixture(name="rehlko_config_entry_with_refresh_token")
def rehlko_config_entry_with_refresh_token_fixture() -> MockConfigEntry:
"""Create a config entry fixture with refresh token."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
CONF_REFRESH_TOKEN: TEST_REFRESH_TOKEN,
},
unique_id=TEST_SUBJECT,
)
@pytest.fixture
async def mock_rehlko(
homes: list[dict[str, Any]],
generator: dict[str, Any],
):
"""Mock Rehlko instance."""
with (
patch("homeassistant.components.rehlko.AioKem", autospec=True) as mock_kem,
patch("homeassistant.components.rehlko.config_flow.AioKem", new=mock_kem),
):
client = mock_kem.return_value
client.get_homes = AsyncMock(return_value=homes)
client.get_generator_data = AsyncMock(return_value=generator)
client.authenticate = AsyncMock(return_value=None)
client.get_token_subject = Mock(return_value=TEST_SUBJECT)
client.get_refresh_token = AsyncMock(return_value=TEST_REFRESH_TOKEN)
client.set_refresh_token_callback = Mock()
client.set_retry_policy = Mock()
yield client
@pytest.fixture
async def load_rehlko_config_entry(
hass: HomeAssistant,
mock_rehlko: Mock,
rehlko_config_entry: MockConfigEntry,
) -> None:
"""Load the config entry."""
rehlko_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(rehlko_config_entry.entry_id)
await hass.async_block_till_done()

191
tests/components/rehlko/fixtures/generator.json Normal file
View File

@ -0,0 +1,191 @@
{
"device": {
"id": 12345,
"serialNumber": "123MGVHR4567",
"displayName": "Generator 1",
"deviceHost": "Oncue",
"hasAcceptedPrivacyPolicy": true,
"address": {
"lat": 41.3341111,
"long": -72.3333111,
"address1": "Highway 66",
"address2": null,
"city": "Somewhere",
"state": "CA",
"postalCode": "00000",
"country": "US"
},
"product": "Rdc2v4",
"productDisplayName": "RDC 2.4",
"controllerType": "RDC2 (Blue Board)",
"firmwareVersion": "3.4.5",
"currentFirmware": "RDC2.4 3.4.5",
"isConnected": true,
"lastConnectedTimestamp": "2025-04-14T09:30:17+00:00",
"deviceIpAddress": "1.1.1.1:2402",
"macAddress": "91:E1:20:63:10:00",
"status": "ReadyToRun",
"statusUpdateTimestamp": "2025-04-14T09:29:01+00:00",
"dealerOrgs": [
{
"id": 123,
"businessPartnerNo": "123456",
"name": "Generators R Us",
"e164PhoneNumber": "+199999999999",
"displayPhoneNumber": "(999) 999-9999",
"wizardStep": "OnboardingComplete",
"wizardComplete": true,
"address": {
"lat": null,
"long": null,
"address1": "Highway 66",
"address2": null,
"city": "Revisited",
"state": "CA",
"postalCode": "000000",
"country": null
},
"userCount": 4,
"technicianCount": 3,
"deviceCount": 71,
"adminEmails": ["admin@gmail.com"]
}
],
"alertCount": 0,
"model": "Model20KW",
"modelDisplayName": "20 KW",
"lastMaintenanceTimestamp": "2025-04-10T09:12:59",
"nextMaintenanceTimestamp": "2026-04-10T09:12:59",
"maintenancePeriodDays": 365,
"hasServiceAgreement": null,
"totalRuntimeHours": 120.2
},
"powerSource": "Utility",
"switchState": "Auto",
"coolingType": "Air",
"connectionType": "Unknown",
"serverIpAddress": "2.2.2.2",
"serviceAgreement": {
"hasServiceAgreement": null,
"beginTimestamp": null,
"term": null,
"termMonths": null,
"termDays": null
},
"exercise": {
"frequency": "Weekly",
"nextStartTimestamp": "2025-04-19T10:00:00",
"mode": "Unloaded",
"runningMode": null,
"durationMinutes": 20,
"lastStartTimestamp": "2025-04-12T14:00:00+00:00",
"lastEndTimestamp": "2025-04-12T14:19:59+00:00"
},
"lastRanTimestamp": "2025-04-12T14:00:00+00:00",
"totalRuntimeHours": 120.2,
"totalOperationHours": 33932.3,
"runtimeSinceLastMaintenanceHours": 0.3,
"remoteResetCounterSeconds": 0,
"addedBy": null,
"associatedUsers": ["pete.rage@rage.com"],
"controllerClockTimestamp": "2025-04-15T07:08:50",
"fuelType": "LiquidPropane",
"batteryVoltageV": 13.9,
"engineCoolantTempF": null,
"engineFrequencyHz": 0,
"engineSpeedRpm": 0,
"lubeOilTempF": 42.8,
"controllerTempF": 71.6,
"engineCompartmentTempF": null,
"engineOilPressurePsi": null,
"engineOilPressureOk": true,
"generatorLoadW": 0,
"generatorLoadPercent": 0,
"generatorVoltageAvgV": 0,
"setOutputVoltageV": 240,
"utilityVoltageV": 259.7,
"engineState": "Standby",
"engineStateDisplayNameEn": "Standby",
"loadShed": {
"isConnected": true,
"parameters": [
{
"definitionId": 1,
"displayName": "HVAC A",
"value": false,
"isReadOnly": false
},
{
"definitionId": 2,
"displayName": "HVAC B",
"value": false,
"isReadOnly": false
},
{
"definitionId": 3,
"displayName": "Load A",
"value": false,
"isReadOnly": false
},
{
"definitionId": 4,
"displayName": "Load B",
"value": false,
"isReadOnly": false
},
{
"definitionId": 5,
"displayName": "Load C",
"value": false,
"isReadOnly": false
},
{
"definitionId": 6,
"displayName": "Load D",
"value": false,
"isReadOnly": false
}
]
},
"pim": {
"isConnected": false,
"parameters": [
{
"definitionId": 7,
"displayName": "Digital Output B1 Value",
"value": false,
"isReadOnly": true
},
{
"definitionId": 8,
"displayName": "Digital Output B2 Value",
"value": false,
"isReadOnly": true
},
{
"definitionId": 9,
"displayName": "Digital Output B3 Value",
"value": false,
"isReadOnly": false
},
{
"definitionId": 10,
"displayName": "Digital Output B4 Value",
"value": false,
"isReadOnly": false
},
{
"definitionId": 11,
"displayName": "Digital Output B5 Value",
"value": false,
"isReadOnly": false
},
{
"definitionId": 12,
"displayName": "Digital Output B6 Value",
"value": false,
"isReadOnly": false
}
]
}
}

82
tests/components/rehlko/fixtures/homes.json Normal file
View File

@ -0,0 +1,82 @@
[
{
"id": 12345,
"name": "Generator 1",
"weatherCondition": "Mist",
"weatherTempF": 46.11200000000006,
"weatherTimePeriod": "Day",
"address": {
"lat": 41.334111,
"long": -72.3333111,
"address1": "Highway 66",
"address2": null,
"city": "Somewhere",
"state": "CA",
"postalCode": "000000",
"country": "US"
},
"devices": [
{
"id": 12345,
"serialNumber": "123MGVHR4567",
"displayName": "Generator 1",
"deviceHost": "Oncue",
"hasAcceptedPrivacyPolicy": true,
"address": {
"lat": 41.334111,
"long": -72.3333111,
"address1": "Highway 66",
"address2": null,
"city": "Somewhere",
"state": "CA",
"postalCode": "000000",
"country": "US"
},
"product": "Rdc2v4",
"productDisplayName": "RDC 2.4",
"controllerType": "RDC2 (Blue Board)",
"firmwareVersion": "3.4.5",
"currentFirmware": "RDC2.4 3.4.5",
"isConnected": true,
"lastConnectedTimestamp": "2025-04-14T09:30:17+00:00",
"deviceIpAddress": "1.1.1.1:2402",
"macAddress": "91:E1:20:63:10:00",
"status": "ReadyToRun",
"statusUpdateTimestamp": "2025-04-14T09:29:01+00:00",
"dealerOrgs": [
{
"id": 123,
"businessPartnerNo": "123456",
"name": "Generators R Us",
"e164PhoneNumber": "+199999999999",
"displayPhoneNumber": "(999) 999-9999",
"wizardStep": "OnboardingComplete",
"wizardComplete": true,
"address": {
"lat": null,
"long": null,
"address1": "Highway 66",
"address2": null,
"city": "Revisited",
"state": "CA",
"postalCode": "000000",
"country": null
},
"userCount": 4,
"technicianCount": 3,
"deviceCount": 71,
"adminEmails": ["admin@gmail.com"]
}
],
"alertCount": 0,
"model": "Model20KW",
"modelDisplayName": "20 KW",
"lastMaintenanceTimestamp": "2025-04-10T09:12:59",
"nextMaintenanceTimestamp": "2026-04-10T09:12:59",
"maintenancePeriodDays": 365,
"hasServiceAgreement": null,
"totalRuntimeHours": 120.2
}
]
}
]

876
tests/components/rehlko/snapshots/test_sensor.ambr Normal file
View File

@ -0,0 +1,876 @@
# serializer version: 1
# name: test_sensors[sensor.generator_1_battery_voltage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_battery_voltage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'Battery voltage',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'battery_voltage',
'unique_id': 'myemail@email.com_12345_batteryVoltageV',
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
})
# ---
# name: test_sensors[sensor.generator_1_battery_voltage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Generator 1 Battery voltage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_battery_voltage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '13.9',
})
# ---
# name: test_sensors[sensor.generator_1_controller_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_controller_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Controller temperature',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'controller_temperature',
'unique_id': 'myemail@email.com_12345_controllerTempF',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensors[sensor.generator_1_controller_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Generator 1 Controller temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_controller_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '22.0',
})
# ---
# name: test_sensors[sensor.generator_1_device_ip_address-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_device_ip_address',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Device IP address',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'device_ip_address',
'unique_id': 'myemail@email.com_12345_deviceIpAddress',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.generator_1_device_ip_address-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Generator 1 Device IP address',
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_device_ip_address',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1.1.1.1:2402',
})
# ---
# name: test_sensors[sensor.generator_1_engine_compartment_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_engine_compartment_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Engine compartment temperature',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'engine_compartment_temperature',
'unique_id': 'myemail@email.com_12345_engineCompartmentTempF',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensors[sensor.generator_1_engine_compartment_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Generator 1 Engine compartment temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_engine_compartment_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensors[sensor.generator_1_engine_coolant_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_engine_coolant_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Engine coolant temperature',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'engine_coolant_temperature',
'unique_id': 'myemail@email.com_12345_engineCoolantTempF',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensors[sensor.generator_1_engine_coolant_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Generator 1 Engine coolant temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_engine_coolant_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensors[sensor.generator_1_engine_frequency-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_engine_frequency',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.FREQUENCY: 'frequency'>,
'original_icon': None,
'original_name': 'Engine frequency',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'engine_frequency',
'unique_id': 'myemail@email.com_12345_engineFrequencyHz',
'unit_of_measurement': <UnitOfFrequency.HERTZ: 'Hz'>,
})
# ---
# name: test_sensors[sensor.generator_1_engine_frequency-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'frequency',
'friendly_name': 'Generator 1 Engine frequency',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfFrequency.HERTZ: 'Hz'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_engine_frequency',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.generator_1_engine_oil_pressure-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_engine_oil_pressure',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfPressure.KPA: 'kPa'>,
}),
}),
'original_device_class': <SensorDeviceClass.PRESSURE: 'pressure'>,
'original_icon': None,
'original_name': 'Engine oil pressure',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'engine_oil_pressure',
'unique_id': 'myemail@email.com_12345_engineOilPressurePsi',
'unit_of_measurement': <UnitOfPressure.KPA: 'kPa'>,
})
# ---
# name: test_sensors[sensor.generator_1_engine_oil_pressure-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'pressure',
'friendly_name': 'Generator 1 Engine oil pressure',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfPressure.KPA: 'kPa'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_engine_oil_pressure',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensors[sensor.generator_1_engine_speed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_engine_speed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Engine speed',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'engine_speed',
'unique_id': 'myemail@email.com_12345_engineSpeedRpm',
'unit_of_measurement': 'rpm',
})
# ---
# name: test_sensors[sensor.generator_1_engine_speed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Generator 1 Engine speed',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'rpm',
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_engine_speed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.generator_1_generator_load-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_generator_load',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Generator load',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'generator_load',
'unique_id': 'myemail@email.com_12345_generatorLoadW',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_sensors[sensor.generator_1_generator_load-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Generator 1 Generator load',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_generator_load',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.generator_1_generator_load_percentage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_generator_load_percentage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Generator load percentage',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'generator_load_percent',
'unique_id': 'myemail@email.com_12345_generatorLoadPercent',
'unit_of_measurement': '%',
})
# ---
# name: test_sensors[sensor.generator_1_generator_load_percentage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Generator 1 Generator load percentage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_generator_load_percentage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.generator_1_lube_oil_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_lube_oil_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Lube oil temperature',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'lube_oil_temperature',
'unique_id': 'myemail@email.com_12345_lubeOilTempF',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensors[sensor.generator_1_lube_oil_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Generator 1 Lube oil temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_lube_oil_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '6.0',
})
# ---
# name: test_sensors[sensor.generator_1_runtime_since_last_maintenance-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_runtime_since_last_maintenance',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Runtime since last maintenance',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'runtime_since_last_maintenance',
'unique_id': 'myemail@email.com_12345_runtimeSinceLastMaintenanceHours',
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
})
# ---
# name: test_sensors[sensor.generator_1_runtime_since_last_maintenance-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'Generator 1 Runtime since last maintenance',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_runtime_since_last_maintenance',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.3',
})
# ---
# name: test_sensors[sensor.generator_1_server_ip_address-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_server_ip_address',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Server IP address',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'server_ip_address',
'unique_id': 'myemail@email.com_12345_serverIpAddress',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.generator_1_server_ip_address-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Generator 1 Server IP address',
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_server_ip_address',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.2.2.2',
})
# ---
# name: test_sensors[sensor.generator_1_total_operation-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_total_operation',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Total operation',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'total_operation',
'unique_id': 'myemail@email.com_12345_totalOperationHours',
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
})
# ---
# name: test_sensors[sensor.generator_1_total_operation-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'Generator 1 Total operation',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_total_operation',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '33932.3',
})
# ---
# name: test_sensors[sensor.generator_1_total_runtime-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.generator_1_total_runtime',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Total runtime',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'total_runtime',
'unique_id': 'myemail@email.com_12345_totalRuntimeHours',
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
})
# ---
# name: test_sensors[sensor.generator_1_total_runtime-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'Generator 1 Total runtime',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_total_runtime',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '120.2',
})
# ---
# name: test_sensors[sensor.generator_1_utility_voltage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_utility_voltage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'Utility voltage',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'utility_voltage',
'unique_id': 'myemail@email.com_12345_utilityVoltageV',
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
})
# ---
# name: test_sensors[sensor.generator_1_utility_voltage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Generator 1 Utility voltage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_utility_voltage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '259.7',
})
# ---
# name: test_sensors[sensor.generator_1_voltage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.generator_1_voltage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'Voltage',
'platform': 'rehlko',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'generator_voltage_avg',
'unique_id': 'myemail@email.com_12345_generatorVoltageAvgV',
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
})
# ---
# name: test_sensors[sensor.generator_1_voltage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Generator 1 Voltage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
}),
'context': <ANY>,
'entity_id': 'sensor.generator_1_voltage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---

218
tests/components/rehlko/test_config_flow.py Normal file
View File

@ -0,0 +1,218 @@
"""Test the Rehlko config flow."""
from unittest.mock import AsyncMock
from aiokem import AuthenticationCredentialsError
import pytest
from homeassistant.components.rehlko import DOMAIN
from homeassistant.config_entries import SOURCE_DHCP, SOURCE_USER
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .conftest import TEST_EMAIL, TEST_PASSWORD, TEST_SUBJECT
from tests.common import MockConfigEntry
DHCP_DISCOVERY = DhcpServiceInfo(
ip="1.1.1.1",
hostname="KohlerGen",
macaddress="00146FAABBCC",
)
async def test_configure_entry(
hass: HomeAssistant, mock_rehlko: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test we can configure the entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_EMAIL.lower()
assert result["data"] == {
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
}
assert result["result"].unique_id == TEST_SUBJECT
assert mock_setup_entry.call_count == 1
@pytest.mark.parametrize(
("error", "conf_error"),
[
(AuthenticationCredentialsError, {CONF_PASSWORD: "invalid_auth"}),
(TimeoutError, {"base": "cannot_connect"}),
(Exception, {"base": "unknown"}),
],
)
async def test_configure_entry_exceptions(
hass: HomeAssistant,
mock_rehlko: AsyncMock,
error: Exception,
conf_error: dict[str, str],
mock_setup_entry: AsyncMock,
) -> None:
"""Test we handle a variety of exceptions and recover by adding new entry."""
# First try to authenticate and get an error
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_rehlko.authenticate.side_effect = error
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == conf_error
assert mock_setup_entry.call_count == 0
# Now try to authenticate again and succeed
# This should create a new entry
mock_rehlko.authenticate.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_EMAIL.lower()
assert result["data"] == {
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
}
assert result["result"].unique_id == TEST_SUBJECT
assert mock_setup_entry.call_count == 1
async def test_already_configured(
hass: HomeAssistant, rehlko_config_entry: MockConfigEntry, mock_rehlko: AsyncMock
) -> None:
"""Test if entry is already configured."""
rehlko_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_reauth(
hass: HomeAssistant,
rehlko_config_entry: MockConfigEntry,
mock_rehlko: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test reauth flow."""
rehlko_config_entry.add_to_hass(hass)
result = await rehlko_config_entry.start_reauth_flow(hass)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: TEST_PASSWORD + "new",
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert rehlko_config_entry.data[CONF_PASSWORD] == TEST_PASSWORD + "new"
assert mock_setup_entry.call_count == 1
async def test_reauth_exception(
hass: HomeAssistant,
rehlko_config_entry: MockConfigEntry,
mock_rehlko: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test reauth flow."""
rehlko_config_entry.add_to_hass(hass)
result = await rehlko_config_entry.start_reauth_flow(hass)
mock_rehlko.authenticate.side_effect = AuthenticationCredentialsError
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"password": "invalid_auth"}
mock_rehlko.authenticate.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: TEST_PASSWORD + "new",
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
async def test_dhcp_discovery(
hass: HomeAssistant, mock_rehlko: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test we can setup from dhcp discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: TEST_EMAIL,
CONF_PASSWORD: TEST_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_dhcp_discovery_already_set_up(
hass: HomeAssistant, rehlko_config_entry: MockConfigEntry, mock_rehlko: AsyncMock
) -> None:
"""Test DHCP discovery aborts if already set up."""
rehlko_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

85
tests/components/rehlko/test_sensor.py Normal file
View File

@ -0,0 +1,85 @@
"""Tests for the Rehlko sensors."""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.rehlko.coordinator import SCAN_INTERVAL_MINUTES
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
@pytest.fixture(name="platform_sensor", autouse=True)
async def platform_sensor_fixture():
"""Patch Rehlko to only load Sensor platform."""
with patch("homeassistant.components.rehlko.PLATFORMS", [Platform.SENSOR]):
yield
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
rehlko_config_entry: MockConfigEntry,
load_rehlko_config_entry: None,
) -> None:
"""Test the Rehlko sensors."""
await snapshot_platform(
hass, entity_registry, snapshot, rehlko_config_entry.entry_id
)
async def test_sensor_availability_device_disconnect(
hass: HomeAssistant,
generator: dict[str, Any],
mock_rehlko: AsyncMock,
load_rehlko_config_entry: None,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the Rehlko sensor availability when device is disconnected."""
state = hass.states.get("sensor.generator_1_battery_voltage")
assert state
assert state.state == "13.9"
generator["device"]["isConnected"] = False
# Move time to next update
freezer.tick(SCAN_INTERVAL_MINUTES)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("sensor.generator_1_battery_voltage")
assert state
assert state.state == STATE_UNAVAILABLE
async def test_sensor_availability_poll_failure(
hass: HomeAssistant,
mock_rehlko: AsyncMock,
load_rehlko_config_entry: None,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the Rehlko sensor availability when cloud poll fails."""
state = hass.states.get("sensor.generator_1_battery_voltage")
assert state
assert state.state == "13.9"
mock_rehlko.get_generator_data.side_effect = Exception("Test exception")
# Move time to next update
freezer.tick(SCAN_INTERVAL_MINUTES)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("sensor.generator_1_battery_voltage")
assert state
assert state.state == STATE_UNAVAILABLE