Add ENGIE Belgium integration (#179094)

This commit is contained in:
Daan Vervacke
2026-10-02 22:18:06 +01:00
committed by GitHub
parent 6528860ce1
commit cb0b8a7650
21 changed files with 3521 additions and 0 deletions
+1
View File
@@ -204,6 +204,7 @@ homeassistant.components.energieleser.*
homeassistant.components.energy.*
homeassistant.components.energyid.*
homeassistant.components.energyzero.*
homeassistant.components.engie_be.*
homeassistant.components.enigma2.*
homeassistant.components.enphase_envoy.*
homeassistant.components.eq3btsmart.*
Generated
+2
View File
@@ -513,6 +513,8 @@ CLAUDE.md @home-assistant/core
/tests/components/energyid/ @JrtPec @Molier
/homeassistant/components/energyzero/ @klaasnicolaas
/tests/components/energyzero/ @klaasnicolaas
/homeassistant/components/engie_be/ @DaanVervacke
/tests/components/engie_be/ @DaanVervacke
/homeassistant/components/enigma2/ @autinerd
/tests/components/enigma2/ @autinerd
/homeassistant/components/enphase_envoy/ @bdraco @cgarwood @catsmanac
@@ -0,0 +1,96 @@
"""The ENGIE Belgium integration."""
import asyncio
from dataclasses import dataclass
from aioengiebelgium import EngieBeClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_REFRESH_TOKEN
from .coordinator import EngieBePricesCoordinator, EngieBeRelationsCoordinator
_PLATFORMS: list[Platform] = [Platform.SENSOR]
@dataclass
class EngieBeHouseholdCoordinators:
"""Per-household coordinators."""
prices: EngieBePricesCoordinator
@dataclass
class EngieBeRuntimeData:
"""Runtime data for the ENGIE Belgium integration."""
relations: EngieBeRelationsCoordinator
households: dict[str, EngieBeHouseholdCoordinators]
type EngieBeConfigEntry = ConfigEntry[EngieBeRuntimeData]
async def async_setup_entry(hass: HomeAssistant, entry: EngieBeConfigEntry) -> bool:
"""Set up ENGIE Belgium from a config entry."""
async def _persist_tokens(access_token: str, refresh_token: str) -> None:
"""Persist rotated tokens to the config entry."""
if (
entry.data[CONF_ACCESS_TOKEN] == access_token
and entry.data[CONF_REFRESH_TOKEN] == refresh_token
):
return
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
CONF_ACCESS_TOKEN: access_token,
CONF_REFRESH_TOKEN: refresh_token,
},
)
client = EngieBeClient(
session=async_get_clientsession(hass),
access_token=entry.data[CONF_ACCESS_TOKEN],
refresh_token=entry.data[CONF_REFRESH_TOKEN],
on_token_refresh=_persist_tokens,
)
relations = EngieBeRelationsCoordinator(hass, entry, client)
await relations.async_config_entry_first_refresh()
device_registry = dr.async_get(hass)
@callback
def _async_create_household(ban: str) -> EngieBeHouseholdCoordinators:
"""Build the coordinators for a business agreement and register its device."""
household = EngieBeHouseholdCoordinators(
prices=EngieBePricesCoordinator(
hass, entry, client, ban, relations.data[ban]
)
)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id, **household.prices.device_info
)
return household
households = {ban: _async_create_household(ban) for ban in relations.data}
await asyncio.gather(
*(household.prices.async_refresh() for household in households.values())
)
entry.runtime_data = EngieBeRuntimeData(relations=relations, households=households)
await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: EngieBeConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
@@ -0,0 +1,147 @@
"""Config flow for the ENGIE Belgium integration."""
from typing import Any, override
from aioengiebelgium import (
AuthFlow,
EngieBeAuthenticationError,
EngieBeClient,
EngieBeCommunicationError,
EngieBeError,
EngieBeMfaError,
MfaMethod,
)
import probatio
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers import selector
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import CONF_MFA_METHOD, CONF_REFRESH_TOKEN, DOMAIN, USER_MANAGEMENT_URL
_MFA_METHOD_SELECTOR = selector.SelectSelector(
selector.SelectSelectorConfig(
options=[method.value for method in MfaMethod],
translation_key="mfa_method",
)
)
_CODE_SCHEMA = probatio.Schema({probatio.Required("code"): str})
class EngieBeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for ENGIE Belgium."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._email: str | None = None
self._password: str | None = None
self._mfa_method: MfaMethod = MfaMethod.SMS
self._auth_flow: AuthFlow | None = None
self._client: EngieBeClient | None = None
async def _async_start_authentication(
self, user_input: dict[str, Any]
) -> dict[str, str]:
"""Start authentication with ENGIE Belgium and return any form errors."""
self._email = user_input[CONF_EMAIL]
self._password = user_input[CONF_PASSWORD]
self._mfa_method = MfaMethod(user_input[CONF_MFA_METHOD])
session = async_create_clientsession(self.hass)
client = EngieBeClient(session=session)
self._client = client
try:
self._auth_flow = await client.async_start_authentication(
self._email,
self._password,
self._mfa_method,
auth_session=session,
)
except EngieBeAuthenticationError:
return {"base": "invalid_auth"}
except EngieBeCommunicationError:
return {"base": "cannot_connect"}
except EngieBeError:
return {"base": "unknown"}
return {}
async def _async_submit_mfa(
self, code: str
) -> tuple[dict[str, str], tuple[str, str] | None]:
"""Submit the MFA code and return (errors, tokens)."""
assert self._auth_flow is not None
try:
tokens = await self._auth_flow.async_submit_mfa(code)
except EngieBeMfaError:
return {"base": "invalid_mfa_code"}, None
except EngieBeAuthenticationError:
return {"base": "invalid_auth"}, None
except EngieBeCommunicationError:
return {"base": "cannot_connect"}, None
except EngieBeError:
return {"base": "unknown"}, None
return {}, tokens
@override
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 = await self._async_start_authentication(user_input)
if not errors:
return await self.async_step_mfa()
return self.async_show_form(
step_id="user",
data_schema=probatio.Schema(
{
probatio.Required(CONF_EMAIL): str,
probatio.Required(CONF_PASSWORD): str,
probatio.Required(
CONF_MFA_METHOD, default=MfaMethod.SMS.value
): _MFA_METHOD_SELECTOR,
}
),
description_placeholders={"user_management_url": USER_MANAGEMENT_URL},
errors=errors,
)
async def async_step_mfa(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the MFA code entry step."""
errors: dict[str, str] = {}
if user_input is not None and self._email is not None:
errors, tokens = await self._async_submit_mfa(user_input["code"])
if not errors and tokens is not None:
access_token, refresh_token = tokens
assert self._client is not None
subject = self._client.subject
if subject is None:
return self.async_show_form(
step_id="mfa",
data_schema=_CODE_SCHEMA,
errors={"base": "invalid_auth"},
)
await self.async_set_unique_id(subject)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self._email,
data={
CONF_EMAIL: self._email,
CONF_MFA_METHOD: self._mfa_method.value,
CONF_ACCESS_TOKEN: access_token,
CONF_REFRESH_TOKEN: refresh_token,
},
)
return self.async_show_form(
step_id="mfa", data_schema=_CODE_SCHEMA, errors=errors
)
@@ -0,0 +1,18 @@
"""Constants for the ENGIE Belgium integration."""
from datetime import timedelta
import logging
DOMAIN = "engie_be"
ATTRIBUTION = "Data provided by ENGIE Belgium"
LOGGER = logging.getLogger(__package__)
CONF_MFA_METHOD = "mfa_method"
CONF_REFRESH_TOKEN = "refresh_token"
USER_MANAGEMENT_URL = (
"https://www.engie.be/nl/energiedesk/usermanagement/manage-access/"
)
PRICES_SCAN_INTERVAL = timedelta(hours=1)
@@ -0,0 +1,194 @@
"""DataUpdateCoordinator for the ENGIE Belgium integration."""
import asyncio
from dataclasses import dataclass
from datetime import date
from typing import TYPE_CHECKING, override
from aioengiebelgium import (
BusinessAgreement,
EngieBeClient,
EngieBeError,
PricePeriod,
PriceSlot,
bare_ean,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import DOMAIN, LOGGER, PRICES_SCAN_INTERVAL
if TYPE_CHECKING:
from . import EngieBeConfigEntry
_DIRECTIONS = ("offtake", "injection")
_DIRECTION_PREFIXES = ("OFFTAKE_", "INJECTION_")
_BLENDED_SLOT_CODE = "EN"
def _mask(identifier: str) -> str:
"""Mask an account/meter identifier down to its last four characters."""
return f"…{identifier[-4:]}"
def normalize_slot_code(raw_code: str) -> str:
"""Strip a redundant direction prefix from a raw time-of-use slot code."""
for prefix in _DIRECTION_PREFIXES:
idx = raw_code.rfind(prefix)
if idx != -1:
return raw_code[idx + len(prefix) :]
return raw_code
def _current_period(
periods: tuple[PricePeriod, ...], today: date
) -> PricePeriod | None:
"""Return the price period covering today, if any."""
for period in periods:
if period.contains(today):
return period
return None
@dataclass
class EngieBePricesData:
"""Pre-processed price lookup for one business agreement."""
slots: dict[tuple[str, str, str], PriceSlot]
eans: tuple[str, ...]
class EngieBeRelationsCoordinator(DataUpdateCoordinator[dict[str, BusinessAgreement]]):
"""Coordinator that tracks the account's active business agreements."""
config_entry: EngieBeConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: EngieBeConfigEntry,
client: EngieBeClient,
) -> None:
"""Initialize the relations coordinator."""
super().__init__(
hass,
LOGGER,
config_entry=config_entry,
name=f"{DOMAIN}_relations",
)
self.client = client
@override
async def _async_update_data(self) -> dict[str, BusinessAgreement]:
"""Fetch the account's active business agreements."""
try:
relations = await self.client.async_get_customer_account_relations()
except EngieBeError as err:
raise UpdateFailed(str(err)) from err
agreements = {
agreement.business_agreement_number: agreement
for account in relations.accounts
for agreement in account.customer_account.business_agreements
if agreement.active
}
if not agreements:
LOGGER.debug("No active business agreements found")
return agreements
class EngieBePricesCoordinator(DataUpdateCoordinator[EngieBePricesData]):
"""Coordinator that fetches energy prices for one business agreement."""
config_entry: EngieBeConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: EngieBeConfigEntry,
client: EngieBeClient,
ban: str,
agreement: BusinessAgreement,
) -> None:
"""Initialize the prices coordinator for one business agreement."""
super().__init__(
hass,
LOGGER,
config_entry=config_entry,
name=f"{DOMAIN}_prices_{_mask(ban)}",
update_interval=PRICES_SCAN_INTERVAL,
)
self.client = client
self.ban = ban
self.agreement = agreement
self.ean_energy_types: dict[str, str | None] = {}
device_name = (
agreement.consumption_address.format()
if agreement.consumption_address is not None
else ""
) or ban
self.device_info = DeviceInfo(
identifiers={(DOMAIN, ban)},
entry_type=DeviceEntryType.SERVICE,
manufacturer="ENGIE Belgium",
name=device_name,
)
@override
async def _async_update_data(self) -> EngieBePricesData:
"""Fetch this household's prices and pre-process them into a slot lookup."""
try:
prices = await self.client.async_get_prices(self.ban)
except EngieBeError as err:
raise UpdateFailed(str(err)) from err
new_eans = list(
dict.fromkeys(
ean_prices.ean
for ean_prices in prices.items
if bare_ean(ean_prices.ean) not in self.ean_energy_types
)
)
if new_eans:
service_points = await asyncio.gather(
*(self.client.async_get_service_point(ean) for ean in new_eans),
return_exceptions=True,
)
for ean, service_point_result in zip(new_eans, service_points, strict=True):
if isinstance(service_point_result, EngieBeError):
LOGGER.debug(
"Fetching service point for %s failed: %s",
_mask(bare_ean(ean)),
service_point_result,
)
continue
if isinstance(service_point_result, BaseException):
raise service_point_result
self.ean_energy_types.update(service_point_result.ean_energy_types)
self.ean_energy_types.setdefault(bare_ean(ean), None)
brussels = dt_util.get_time_zone("Europe/Brussels")
assert brussels is not None
today = dt_util.now(brussels).date()
slots: dict[tuple[str, str, str], PriceSlot] = {}
for ean_prices in prices.items:
period = _current_period(ean_prices.periods, today)
if period is None:
continue
for direction in _DIRECTIONS:
direction_slots = (
period.offtake if direction == "offtake" else period.injection
)
for slot in direction_slots:
normalized = normalize_slot_code(slot.time_of_use_slot_code)
if normalized == _BLENDED_SLOT_CODE:
continue
slots[ean_prices.ean, direction, slot.time_of_use_slot_code] = slot
return EngieBePricesData(
slots=slots,
eans=tuple(ean_prices.ean for ean_prices in prices.items),
)
@@ -0,0 +1,156 @@
{
"entity": {
"sensor": {
"electricity_price_injection": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_excl_vat": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_offpeak": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_offpeak_excl_vat": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_peak": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_peak_excl_vat": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_slot": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_slot_excl_vat": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_superoffpeak": {
"default": "mdi:cash-plus"
},
"electricity_price_injection_superoffpeak_excl_vat": {
"default": "mdi:cash-plus"
},
"electricity_price_offtake": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_excl_vat": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_offpeak": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_offpeak_excl_vat": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_peak": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_peak_excl_vat": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_slot": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_slot_excl_vat": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_superoffpeak": {
"default": "mdi:cash-minus"
},
"electricity_price_offtake_superoffpeak_excl_vat": {
"default": "mdi:cash-minus"
},
"energy_price_injection": {
"default": "mdi:cash-plus"
},
"energy_price_injection_excl_vat": {
"default": "mdi:cash-plus"
},
"energy_price_injection_offpeak": {
"default": "mdi:cash-plus"
},
"energy_price_injection_offpeak_excl_vat": {
"default": "mdi:cash-plus"
},
"energy_price_injection_peak": {
"default": "mdi:cash-plus"
},
"energy_price_injection_peak_excl_vat": {
"default": "mdi:cash-plus"
},
"energy_price_injection_slot": {
"default": "mdi:cash-plus"
},
"energy_price_injection_slot_excl_vat": {
"default": "mdi:cash-plus"
},
"energy_price_injection_superoffpeak": {
"default": "mdi:cash-plus"
},
"energy_price_injection_superoffpeak_excl_vat": {
"default": "mdi:cash-plus"
},
"energy_price_offtake": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_excl_vat": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_offpeak": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_offpeak_excl_vat": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_peak": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_peak_excl_vat": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_slot": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_slot_excl_vat": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_superoffpeak": {
"default": "mdi:cash-minus"
},
"energy_price_offtake_superoffpeak_excl_vat": {
"default": "mdi:cash-minus"
},
"gas_price_offtake": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_excl_vat": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_offpeak": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_offpeak_excl_vat": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_peak": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_peak_excl_vat": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_slot": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_slot_excl_vat": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_superoffpeak": {
"default": "mdi:meter-gas"
},
"gas_price_offtake_superoffpeak_excl_vat": {
"default": "mdi:meter-gas"
}
}
}
}
@@ -0,0 +1,11 @@
{
"domain": "engie_be",
"name": "ENGIE Belgium",
"codeowners": ["@DaanVervacke"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/engie_be",
"integration_type": "service",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["aioengiebelgium==0.4.3"]
}
@@ -0,0 +1,78 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration provides no actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration provides no actions.
docs-conditions:
status: exempt
comment: Integration provides no conditions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-triggers:
status: exempt
comment: Integration provides no triggers.
entity-event-setup: done
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: Integration provides no actions.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: Integration provides no options flow.
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Cloud account-bound integration; no discovery surface.
discovery:
status: exempt
comment: Cloud account-bound integration; no LAN/mDNS/USB surface to discover.
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: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done
+166
View File
@@ -0,0 +1,166 @@
"""Sensor platform for the ENGIE Belgium integration."""
from collections import Counter
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, override
from aioengiebelgium import bare_ean
from homeassistant.components.sensor import SensorEntity, SensorStateClass
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTRIBUTION
from .coordinator import (
EngieBePricesCoordinator,
EngieBePricesData,
normalize_slot_code,
)
if TYPE_CHECKING:
from . import EngieBeConfigEntry
PARALLEL_UPDATES = 0
_UNIT = "EUR/kWh"
_SLOT_CODE_SUFFIXES = {
"TOTAL_HOURS": "",
"PEAK": "_peak",
"OFFPEAK": "_offpeak",
"SUPEROFFPEAK": "_superoffpeak",
}
_FALLBACK_SLOT_SUFFIX = "_slot"
_ENERGY_TYPE_KEYS = {"ELECTRICITY": "electricity", "GAS": "gas"}
_FALLBACK_TYPE_KEY = "energy"
def _energy_type_key(ean: str, ean_energy_types: Mapping[str, str | None]) -> str:
"""Resolve the translation-key type dimension for an EAN."""
energy_type = (ean_energy_types.get(bare_ean(ean)) or "").upper()
return _ENERGY_TYPE_KEYS.get(energy_type, _FALLBACK_TYPE_KEY)
def _duplicate_type_eans(
eans: Iterable[str], ean_energy_types: Mapping[str, str | None]
) -> set[str]:
"""Return the EANs whose energy type recurs more than once in a household."""
type_keys = {ean: _energy_type_key(ean, ean_energy_types) for ean in eans}
counts = Counter(type_keys.values())
return {ean for ean, type_key in type_keys.items() if counts[type_key] > 1}
async def async_setup_entry(
hass: HomeAssistant,
entry: EngieBeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the sensor platform."""
runtime_data = entry.runtime_data
known_unique_ids: set[str] = set()
@callback
def _async_add_new_entities() -> None:
"""Add price sensors for any household/EAN/slot combination not yet known."""
new_entities: list[EngieBePriceSensor] = []
for ban, household in runtime_data.households.items():
prices_data: EngieBePricesData | None = household.prices.data
if prices_data is None:
continue
duplicate_eans = _duplicate_type_eans(
prices_data.eans, household.prices.ean_energy_types
)
for ean, direction, slot_code in prices_data.slots:
type_key = _energy_type_key(ean, household.prices.ean_energy_types)
ean_suffix = bare_ean(ean)[-4:] if ean in duplicate_eans else ""
for excl_vat in (False, True):
entity = EngieBePriceSensor(
household.prices,
business_agreement_number=ban,
ean=ean,
direction=direction,
slot_code=slot_code,
excl_vat=excl_vat,
type_key=type_key,
ean_suffix=ean_suffix,
)
if entity.unique_id not in known_unique_ids:
new_entities.append(entity)
if new_entities:
known_unique_ids.update(
unique_id for entity in new_entities if (unique_id := entity.unique_id)
)
async_add_entities(new_entities)
for household in runtime_data.households.values():
entry.async_on_unload(
household.prices.async_add_listener(_async_add_new_entities)
)
_async_add_new_entities()
class EngieBePriceSensor(CoordinatorEntity[EngieBePricesCoordinator], SensorEntity):
"""Representation of an ENGIE Belgium energy price sensor."""
_attr_has_entity_name = True
_attr_attribution = ATTRIBUTION
_attr_native_unit_of_measurement = _UNIT
_attr_state_class = SensorStateClass.MEASUREMENT
_attr_suggested_display_precision = 6
def __init__(
self,
coordinator: EngieBePricesCoordinator,
*,
business_agreement_number: str,
ean: str,
direction: str,
slot_code: str,
excl_vat: bool,
type_key: str,
ean_suffix: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._attr_device_info = coordinator.device_info
self._ean = ean
self._direction = direction
self._slot_code = slot_code
self._excl_vat = excl_vat
unique_id = f"{business_agreement_number}_{ean}_{direction}_{slot_code}"
self._attr_unique_id = f"{unique_id}_excl_vat" if excl_vat else unique_id
normalized_slot_code = normalize_slot_code(slot_code)
suffix = _SLOT_CODE_SUFFIXES.get(normalized_slot_code, _FALLBACK_SLOT_SUFFIX)
translation_key = f"{type_key}_price_{direction}{suffix}"
if excl_vat:
translation_key = f"{translation_key}_excl_vat"
if ean_suffix:
translation_key = f"{translation_key}_with_ean"
self._attr_translation_key = translation_key
translation_placeholders: dict[str, str] = {}
if ean_suffix:
translation_placeholders["ean_suffix"] = ean_suffix
if suffix == _FALLBACK_SLOT_SUFFIX:
translation_placeholders["slot_code"] = normalized_slot_code.lower()
self._attr_translation_placeholders = translation_placeholders
if excl_vat:
self._attr_entity_registry_enabled_default = False
@property
@override
def available(self) -> bool:
"""Return True only when this entity's slot is present in the current data."""
return (
super().available
and (self._ean, self._direction, self._slot_code)
in self.coordinator.data.slots
)
@property
@override
def native_value(self) -> float | None:
"""Return the current price."""
slot = self.coordinator.data.slots[self._ean, self._direction, self._slot_code]
return slot.price_value_excl_vat if self._excl_vat else slot.price_value
@@ -0,0 +1,349 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_mfa_code": "Invalid verification code. Please try again.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"mfa": {
"data": {
"code": "Verification code"
},
"data_description": {
"code": "The verification code ENGIE Belgium just sent you."
},
"description": "Enter the 6-digit verification code that was sent to you."
},
"user": {
"data": {
"email": "[%key:common::config_flow::data::email%]",
"mfa_method": "Two-factor authentication method",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"email": "Your ENGIE Belgium account email address.",
"mfa_method": "Choose how you want to receive your verification code.",
"password": "Your ENGIE Belgium account password."
},
"description": "Enter your ENGIE Belgium credentials and two-factor authentication method. **A dedicated ENGIE account is required**: signing into the same account from engie.be or the ENGIE Smart App appears to revoke the integration's session and force you to re-authenticate. Create a separate user via the [ENGIE user management page]({user_management_url}). New accounts have 2FA enabled by default."
}
}
},
"entity": {
"sensor": {
"electricity_price_injection": {
"name": "Electricity injection price"
},
"electricity_price_injection_excl_vat": {
"name": "Electricity injection price excl. VAT"
},
"electricity_price_injection_excl_vat_with_ean": {
"name": "Electricity injection price excl. VAT ({ean_suffix})"
},
"electricity_price_injection_offpeak": {
"name": "Electricity off-peak injection price"
},
"electricity_price_injection_offpeak_excl_vat": {
"name": "Electricity off-peak injection price excl. VAT"
},
"electricity_price_injection_offpeak_excl_vat_with_ean": {
"name": "Electricity off-peak injection price excl. VAT ({ean_suffix})"
},
"electricity_price_injection_offpeak_with_ean": {
"name": "Electricity off-peak injection price ({ean_suffix})"
},
"electricity_price_injection_peak": {
"name": "Electricity peak injection price"
},
"electricity_price_injection_peak_excl_vat": {
"name": "Electricity peak injection price excl. VAT"
},
"electricity_price_injection_peak_excl_vat_with_ean": {
"name": "Electricity peak injection price excl. VAT ({ean_suffix})"
},
"electricity_price_injection_peak_with_ean": {
"name": "Electricity peak injection price ({ean_suffix})"
},
"electricity_price_injection_slot": {
"name": "Electricity injection price ({slot_code})"
},
"electricity_price_injection_slot_excl_vat": {
"name": "Electricity injection price excl. VAT ({slot_code})"
},
"electricity_price_injection_slot_excl_vat_with_ean": {
"name": "Electricity injection price excl. VAT ({slot_code}) ({ean_suffix})"
},
"electricity_price_injection_slot_with_ean": {
"name": "Electricity injection price ({slot_code}) ({ean_suffix})"
},
"electricity_price_injection_superoffpeak": {
"name": "Electricity super off-peak injection price"
},
"electricity_price_injection_superoffpeak_excl_vat": {
"name": "Electricity super off-peak injection price excl. VAT"
},
"electricity_price_injection_superoffpeak_excl_vat_with_ean": {
"name": "Electricity super off-peak injection price excl. VAT ({ean_suffix})"
},
"electricity_price_injection_superoffpeak_with_ean": {
"name": "Electricity super off-peak injection price ({ean_suffix})"
},
"electricity_price_injection_with_ean": {
"name": "Electricity injection price ({ean_suffix})"
},
"electricity_price_offtake": {
"name": "Electricity offtake price"
},
"electricity_price_offtake_excl_vat": {
"name": "Electricity offtake price excl. VAT"
},
"electricity_price_offtake_excl_vat_with_ean": {
"name": "Electricity offtake price excl. VAT ({ean_suffix})"
},
"electricity_price_offtake_offpeak": {
"name": "Electricity off-peak offtake price"
},
"electricity_price_offtake_offpeak_excl_vat": {
"name": "Electricity off-peak offtake price excl. VAT"
},
"electricity_price_offtake_offpeak_excl_vat_with_ean": {
"name": "Electricity off-peak offtake price excl. VAT ({ean_suffix})"
},
"electricity_price_offtake_offpeak_with_ean": {
"name": "Electricity off-peak offtake price ({ean_suffix})"
},
"electricity_price_offtake_peak": {
"name": "Electricity peak offtake price"
},
"electricity_price_offtake_peak_excl_vat": {
"name": "Electricity peak offtake price excl. VAT"
},
"electricity_price_offtake_peak_excl_vat_with_ean": {
"name": "Electricity peak offtake price excl. VAT ({ean_suffix})"
},
"electricity_price_offtake_peak_with_ean": {
"name": "Electricity peak offtake price ({ean_suffix})"
},
"electricity_price_offtake_slot": {
"name": "Electricity offtake price ({slot_code})"
},
"electricity_price_offtake_slot_excl_vat": {
"name": "Electricity offtake price excl. VAT ({slot_code})"
},
"electricity_price_offtake_slot_excl_vat_with_ean": {
"name": "Electricity offtake price excl. VAT ({slot_code}) ({ean_suffix})"
},
"electricity_price_offtake_slot_with_ean": {
"name": "Electricity offtake price ({slot_code}) ({ean_suffix})"
},
"electricity_price_offtake_superoffpeak": {
"name": "Electricity super off-peak offtake price"
},
"electricity_price_offtake_superoffpeak_excl_vat": {
"name": "Electricity super off-peak offtake price excl. VAT"
},
"electricity_price_offtake_superoffpeak_excl_vat_with_ean": {
"name": "Electricity super off-peak offtake price excl. VAT ({ean_suffix})"
},
"electricity_price_offtake_superoffpeak_with_ean": {
"name": "Electricity super off-peak offtake price ({ean_suffix})"
},
"electricity_price_offtake_with_ean": {
"name": "Electricity offtake price ({ean_suffix})"
},
"energy_price_injection": {
"name": "Injection price"
},
"energy_price_injection_excl_vat": {
"name": "Injection price excl. VAT"
},
"energy_price_injection_excl_vat_with_ean": {
"name": "Injection price excl. VAT ({ean_suffix})"
},
"energy_price_injection_offpeak": {
"name": "Off-peak injection price"
},
"energy_price_injection_offpeak_excl_vat": {
"name": "Off-peak injection price excl. VAT"
},
"energy_price_injection_offpeak_excl_vat_with_ean": {
"name": "Off-peak injection price excl. VAT ({ean_suffix})"
},
"energy_price_injection_offpeak_with_ean": {
"name": "Off-peak injection price ({ean_suffix})"
},
"energy_price_injection_peak": {
"name": "Peak injection price"
},
"energy_price_injection_peak_excl_vat": {
"name": "Peak injection price excl. VAT"
},
"energy_price_injection_peak_excl_vat_with_ean": {
"name": "Peak injection price excl. VAT ({ean_suffix})"
},
"energy_price_injection_peak_with_ean": {
"name": "Peak injection price ({ean_suffix})"
},
"energy_price_injection_slot": {
"name": "Injection price ({slot_code})"
},
"energy_price_injection_slot_excl_vat": {
"name": "Injection price excl. VAT ({slot_code})"
},
"energy_price_injection_slot_excl_vat_with_ean": {
"name": "Injection price excl. VAT ({slot_code}) ({ean_suffix})"
},
"energy_price_injection_slot_with_ean": {
"name": "Injection price ({slot_code}) ({ean_suffix})"
},
"energy_price_injection_superoffpeak": {
"name": "Super off-peak injection price"
},
"energy_price_injection_superoffpeak_excl_vat": {
"name": "Super off-peak injection price excl. VAT"
},
"energy_price_injection_superoffpeak_excl_vat_with_ean": {
"name": "Super off-peak injection price excl. VAT ({ean_suffix})"
},
"energy_price_injection_superoffpeak_with_ean": {
"name": "Super off-peak injection price ({ean_suffix})"
},
"energy_price_injection_with_ean": {
"name": "Injection price ({ean_suffix})"
},
"energy_price_offtake": {
"name": "Offtake price"
},
"energy_price_offtake_excl_vat": {
"name": "Offtake price excl. VAT"
},
"energy_price_offtake_excl_vat_with_ean": {
"name": "Offtake price excl. VAT ({ean_suffix})"
},
"energy_price_offtake_offpeak": {
"name": "Off-peak offtake price"
},
"energy_price_offtake_offpeak_excl_vat": {
"name": "Off-peak offtake price excl. VAT"
},
"energy_price_offtake_offpeak_excl_vat_with_ean": {
"name": "Off-peak offtake price excl. VAT ({ean_suffix})"
},
"energy_price_offtake_offpeak_with_ean": {
"name": "Off-peak offtake price ({ean_suffix})"
},
"energy_price_offtake_peak": {
"name": "Peak offtake price"
},
"energy_price_offtake_peak_excl_vat": {
"name": "Peak offtake price excl. VAT"
},
"energy_price_offtake_peak_excl_vat_with_ean": {
"name": "Peak offtake price excl. VAT ({ean_suffix})"
},
"energy_price_offtake_peak_with_ean": {
"name": "Peak offtake price ({ean_suffix})"
},
"energy_price_offtake_slot": {
"name": "Offtake price ({slot_code})"
},
"energy_price_offtake_slot_excl_vat": {
"name": "Offtake price excl. VAT ({slot_code})"
},
"energy_price_offtake_slot_excl_vat_with_ean": {
"name": "Offtake price excl. VAT ({slot_code}) ({ean_suffix})"
},
"energy_price_offtake_slot_with_ean": {
"name": "Offtake price ({slot_code}) ({ean_suffix})"
},
"energy_price_offtake_superoffpeak": {
"name": "Super off-peak offtake price"
},
"energy_price_offtake_superoffpeak_excl_vat": {
"name": "Super off-peak offtake price excl. VAT"
},
"energy_price_offtake_superoffpeak_excl_vat_with_ean": {
"name": "Super off-peak offtake price excl. VAT ({ean_suffix})"
},
"energy_price_offtake_superoffpeak_with_ean": {
"name": "Super off-peak offtake price ({ean_suffix})"
},
"energy_price_offtake_with_ean": {
"name": "Offtake price ({ean_suffix})"
},
"gas_price_offtake": {
"name": "Gas offtake price"
},
"gas_price_offtake_excl_vat": {
"name": "Gas offtake price excl. VAT"
},
"gas_price_offtake_excl_vat_with_ean": {
"name": "Gas offtake price excl. VAT ({ean_suffix})"
},
"gas_price_offtake_offpeak": {
"name": "Gas off-peak offtake price"
},
"gas_price_offtake_offpeak_excl_vat": {
"name": "Gas off-peak offtake price excl. VAT"
},
"gas_price_offtake_offpeak_excl_vat_with_ean": {
"name": "Gas off-peak offtake price excl. VAT ({ean_suffix})"
},
"gas_price_offtake_offpeak_with_ean": {
"name": "Gas off-peak offtake price ({ean_suffix})"
},
"gas_price_offtake_peak": {
"name": "Gas peak offtake price"
},
"gas_price_offtake_peak_excl_vat": {
"name": "Gas peak offtake price excl. VAT"
},
"gas_price_offtake_peak_excl_vat_with_ean": {
"name": "Gas peak offtake price excl. VAT ({ean_suffix})"
},
"gas_price_offtake_peak_with_ean": {
"name": "Gas peak offtake price ({ean_suffix})"
},
"gas_price_offtake_slot": {
"name": "Gas offtake price ({slot_code})"
},
"gas_price_offtake_slot_excl_vat": {
"name": "Gas offtake price excl. VAT ({slot_code})"
},
"gas_price_offtake_slot_excl_vat_with_ean": {
"name": "Gas offtake price excl. VAT ({slot_code}) ({ean_suffix})"
},
"gas_price_offtake_slot_with_ean": {
"name": "Gas offtake price ({slot_code}) ({ean_suffix})"
},
"gas_price_offtake_superoffpeak": {
"name": "Gas super off-peak offtake price"
},
"gas_price_offtake_superoffpeak_excl_vat": {
"name": "Gas super off-peak offtake price excl. VAT"
},
"gas_price_offtake_superoffpeak_excl_vat_with_ean": {
"name": "Gas super off-peak offtake price excl. VAT ({ean_suffix})"
},
"gas_price_offtake_superoffpeak_with_ean": {
"name": "Gas super off-peak offtake price ({ean_suffix})"
},
"gas_price_offtake_with_ean": {
"name": "Gas offtake price ({ean_suffix})"
}
}
},
"selector": {
"mfa_method": {
"options": {
"email": "Email",
"sms": "SMS"
}
}
}
}
+1
View File
@@ -215,6 +215,7 @@ FLOWS = {
"energieleser",
"energyid",
"energyzero",
"engie_be",
"enigma2",
"enocean",
"enphase_envoy",
@@ -1873,6 +1873,12 @@
"iot_class": "cloud_polling",
"single_config_entry": true
},
"engie_be": {
"name": "ENGIE Belgium",
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling"
},
"enigma2": {
"name": "Enigma2 (OpenWebif)",
"integration_type": "device",
Generated
+10
View File
@@ -1798,6 +1798,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.engie_be.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.enigma2.*]
check_untyped_defs = true
disallow_incomplete_defs = true
+3
View File
@@ -259,6 +259,9 @@ aioelectricitymaps==1.1.1
# homeassistant.components.emonitor
aioemonitor==1.0.5
# homeassistant.components.engie_be
aioengiebelgium==0.4.3
# homeassistant.components.esphome
aioesphomeapi==46.6.0
+1
View File
@@ -0,0 +1 @@
"""Tests for the ENGIE Belgium integration."""
+202
View File
@@ -0,0 +1,202 @@
"""Common fixtures for the ENGIE Belgium tests."""
from collections.abc import Generator
from datetime import date
from unittest.mock import AsyncMock, MagicMock, patch
from aioengiebelgium import (
AccountRelation,
AuthFlow,
BusinessAgreement,
ConsumptionAddress,
CustomerAccount,
CustomerAccountRelations,
EanPrices,
PricePeriod,
PriceSlot,
PricesResponse,
ServicePoint,
bare_ean,
)
import pytest
from homeassistant.components.engie_be.const import (
CONF_MFA_METHOD,
CONF_REFRESH_TOKEN,
DOMAIN,
)
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_EMAIL
from tests.common import MockConfigEntry
EMAIL = "user@example.com"
PASSWORD = "hunter2"
SUBJECT = "auth0|69f5b418f3be21cc2ede9c98"
BAN = "000000000001"
BAN_2 = "000000000002"
OFFTAKE_ONLY_EAN = "541448820000000001_ID1"
OFFTAKE_INJECTION_EAN = "541448820000000002_ID1"
_SERVICE_POINT_ENERGY_TYPES = {
bare_ean(OFFTAKE_ONLY_EAN): "GAS",
bare_ean(OFFTAKE_INJECTION_EAN): "ELECTRICITY",
}
def _business_agreement(ban: str, *, with_address: bool = True) -> BusinessAgreement:
"""Build a single active business agreement."""
return BusinessAgreement(
business_agreement_number=ban,
active=True,
consumption_address=ConsumptionAddress(
street="Main street",
house_number="1",
postal_code="1000",
city="Brussels",
)
if with_address
else None,
)
def build_relations(*bans: str, with_address: bool = True) -> CustomerAccountRelations:
"""Build a customer-account-relations response with the given active BANs."""
bans = bans or (BAN,)
return CustomerAccountRelations(
accounts=(
AccountRelation(
id="account-1",
admin=True,
customer_account=CustomerAccount(
customer_account_number="can-1",
business_agreements=tuple(
_business_agreement(ban, with_address=with_address)
for ban in bans
),
),
),
)
)
def build_prices(
*,
valid_from: date | None = date(2000, 1, 1),
valid_to: date | None = date(2099, 12, 31),
) -> PricesResponse:
"""Build a prices response with an offtake-only EAN and a dual-direction EAN."""
return PricesResponse(
items=(
EanPrices(
ean=OFFTAKE_ONLY_EAN,
periods=(
PricePeriod(
valid_from=valid_from,
valid_to=valid_to,
vat_tariff=6.0,
offtake=(
PriceSlot(
time_of_use_slot_code="TOTAL_HOURS",
price_value=0.123456,
price_value_excl_vat=0.116468,
),
),
),
),
),
EanPrices(
ean=OFFTAKE_INJECTION_EAN,
periods=(
PricePeriod(
valid_from=valid_from,
valid_to=valid_to,
vat_tariff=6.0,
offtake=(
PriceSlot(
time_of_use_slot_code="S_TOU1_OFFTAKE_PEAK",
price_value=0.18,
price_value_excl_vat=0.169811,
),
PriceSlot(
time_of_use_slot_code="EN",
price_value=0.12,
price_value_excl_vat=0.113208,
),
PriceSlot(
time_of_use_slot_code="S_TOU1_OFFTAKE_WEEKEND",
price_value=0.15,
price_value_excl_vat=0.141509,
),
),
injection=(
PriceSlot(
time_of_use_slot_code="S_TOU1_INJECTION_PEAK",
price_value=0.05,
price_value_excl_vat=0.047170,
),
),
),
),
),
)
)
def build_service_point(ean: str) -> ServicePoint:
"""Build a service point response for a queried (possibly suffixed) EAN."""
bare = bare_ean(ean)
return ServicePoint(ean_energy_types={bare: _SERVICE_POINT_ENERGY_TYPES[bare]})
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
title=EMAIL,
unique_id=SUBJECT,
data={
CONF_EMAIL: EMAIL,
CONF_MFA_METHOD: "sms",
CONF_ACCESS_TOKEN: "access-token",
CONF_REFRESH_TOKEN: "refresh-token",
},
)
@pytest.fixture
def mock_engie_client(mock_auth_flow: MagicMock) -> Generator[MagicMock]:
"""Mock the EngieBeClient class constructed by the integration and config flow."""
with (
patch(
"homeassistant.components.engie_be.EngieBeClient", autospec=True
) as mock_client_class,
patch(
"homeassistant.components.engie_be.config_flow.EngieBeClient",
new=mock_client_class,
),
):
client = mock_client_class.return_value
client.async_get_customer_account_relations.return_value = build_relations()
client.async_get_prices.return_value = build_prices()
client.async_get_service_point.side_effect = build_service_point
client.async_start_authentication.return_value = mock_auth_flow
yield mock_client_class
@pytest.fixture
def mock_auth_flow() -> MagicMock:
"""Return a mock in-progress authentication flow."""
auth_flow = MagicMock(spec=AuthFlow)
auth_flow.async_submit_mfa = AsyncMock(
return_value=("new-access-token", "new-refresh-token")
)
return auth_flow
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.engie_be.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@@ -0,0 +1,495 @@
# serializer version: 1
# name: test_sensors[Main street 1, 1000 Brussels-device]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': <DeviceEntryType.SERVICE: 'service'>,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'engie_be',
'000000000001',
),
}),
'labels': set({
}),
'manufacturer': 'ENGIE Belgium',
'model': None,
'model_id': None,
'name': 'Main street 1, 1000 Brussels',
'name_by_user': None,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_offtake_price_excl_vat_weekend-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_electricity_offtake_price_excl_vat_weekend',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electricity offtake price excl. VAT (weekend)',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electricity offtake price excl. VAT (weekend)',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electricity_price_offtake_slot_excl_vat',
'unique_id': '000000000001_541448820000000002_ID1_offtake_S_TOU1_OFFTAKE_WEEKEND_excl_vat',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_offtake_price_excl_vat_weekend-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Electricity offtake price excl. VAT (weekend)',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_electricity_offtake_price_excl_vat_weekend',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.141509',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_offtake_price_weekend-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_electricity_offtake_price_weekend',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electricity offtake price (weekend)',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electricity offtake price (weekend)',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electricity_price_offtake_slot',
'unique_id': '000000000001_541448820000000002_ID1_offtake_S_TOU1_OFFTAKE_WEEKEND',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_offtake_price_weekend-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Electricity offtake price (weekend)',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_electricity_offtake_price_weekend',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.15',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_injection_price-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_electricity_peak_injection_price',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electricity peak injection price',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electricity peak injection price',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electricity_price_injection_peak',
'unique_id': '000000000001_541448820000000002_ID1_injection_S_TOU1_INJECTION_PEAK',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_injection_price-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Electricity peak injection price',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_electricity_peak_injection_price',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.05',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_injection_price_excl_vat-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_electricity_peak_injection_price_excl_vat',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electricity peak injection price excl. VAT',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electricity peak injection price excl. VAT',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electricity_price_injection_peak_excl_vat',
'unique_id': '000000000001_541448820000000002_ID1_injection_S_TOU1_INJECTION_PEAK_excl_vat',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_injection_price_excl_vat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Electricity peak injection price excl. VAT',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_electricity_peak_injection_price_excl_vat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.04717',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_offtake_price-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_electricity_peak_offtake_price',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electricity peak offtake price',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electricity peak offtake price',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electricity_price_offtake_peak',
'unique_id': '000000000001_541448820000000002_ID1_offtake_S_TOU1_OFFTAKE_PEAK',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_offtake_price-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Electricity peak offtake price',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_electricity_peak_offtake_price',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.18',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_offtake_price_excl_vat-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_electricity_peak_offtake_price_excl_vat',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electricity peak offtake price excl. VAT',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electricity peak offtake price excl. VAT',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electricity_price_offtake_peak_excl_vat',
'unique_id': '000000000001_541448820000000002_ID1_offtake_S_TOU1_OFFTAKE_PEAK_excl_vat',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_electricity_peak_offtake_price_excl_vat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Electricity peak offtake price excl. VAT',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_electricity_peak_offtake_price_excl_vat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.169811',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_gas_offtake_price-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_gas_offtake_price',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Gas offtake price',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Gas offtake price',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'gas_price_offtake',
'unique_id': '000000000001_541448820000000001_ID1_offtake_TOTAL_HOURS',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_gas_offtake_price-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Gas offtake price',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_gas_offtake_price',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.123456',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_gas_offtake_price_excl_vat-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: '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.main_street_1_1000_brussels_gas_offtake_price_excl_vat',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Gas offtake price excl. VAT',
'options': dict({
'sensor': dict({
'suggested_display_precision': 6,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Gas offtake price excl. VAT',
'platform': 'engie_be',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'gas_price_offtake_excl_vat',
'unique_id': '000000000001_541448820000000001_ID1_offtake_TOTAL_HOURS_excl_vat',
'unit_of_measurement': 'EUR/kWh',
})
# ---
# name: test_sensors[sensor.main_street_1_1000_brussels_gas_offtake_price_excl_vat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by ENGIE Belgium',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Main street 1, 1000 Brussels Gas offtake price excl. VAT',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'EUR/kWh',
}),
'context': <ANY>,
'entity_id': 'sensor.main_street_1_1000_brussels_gas_offtake_price_excl_vat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.116468',
})
# ---
@@ -0,0 +1,241 @@
"""Test the ENGIE Belgium config flow."""
from unittest.mock import AsyncMock, MagicMock, patch
from aioengiebelgium import (
EngieBeAuthenticationError,
EngieBeCommunicationError,
EngieBeError,
EngieBeMfaError,
)
import pytest
from homeassistant import config_entries
from homeassistant.components.engie_be.const import CONF_MFA_METHOD, DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .conftest import EMAIL, PASSWORD, SUBJECT
from tests.common import MockConfigEntry
USER_INPUT = {
CONF_EMAIL: EMAIL,
CONF_PASSWORD: PASSWORD,
CONF_MFA_METHOD: "sms",
}
async def test_full_flow(
hass: HomeAssistant,
mock_engie_client: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the full user config flow."""
mock_engie_client.return_value.subject = SUBJECT
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
sentinel_session = MagicMock()
with patch(
"homeassistant.components.engie_be.config_flow.async_create_clientsession",
return_value=sentinel_session,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "mfa"
call_kwargs = (
mock_engie_client.return_value.async_start_authentication.call_args.kwargs
)
assert call_kwargs["auth_session"] is sentinel_session
assert call_kwargs["auth_session"] is not async_get_clientsession(hass)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == EMAIL
assert result["data"][CONF_EMAIL] == EMAIL
assert result["data"]["access_token"] == "new-access-token"
assert result["data"]["refresh_token"] == "new-refresh-token"
assert CONF_PASSWORD not in result["data"]
assert result["result"].unique_id == SUBJECT
@pytest.mark.parametrize(
("exception", "error"),
[
(EngieBeCommunicationError("boom"), "cannot_connect"),
(EngieBeAuthenticationError("boom"), "invalid_auth"),
(EngieBeError("boom"), "unknown"),
],
)
async def test_user_step_errors(
hass: HomeAssistant,
mock_engie_client: MagicMock,
mock_setup_entry: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test recoverable errors on the user step."""
mock_engie_client.return_value.subject = SUBJECT
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_engie_client.return_value.async_start_authentication.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": error}
mock_engie_client.return_value.async_start_authentication.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "mfa"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_setup_entry.called
@pytest.mark.parametrize(
("exception", "error"),
[
(EngieBeMfaError("bad code"), "invalid_mfa_code"),
(EngieBeAuthenticationError("boom"), "invalid_auth"),
(EngieBeCommunicationError("boom"), "cannot_connect"),
(EngieBeError("boom"), "unknown"),
],
)
async def test_mfa_submit_errors_recovery(
hass: HomeAssistant,
mock_engie_client: MagicMock,
mock_auth_flow: MagicMock,
mock_setup_entry: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test recoverable errors on the MFA submit step."""
mock_engie_client.return_value.subject = SUBJECT
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["step_id"] == "mfa"
mock_auth_flow.async_submit_mfa.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "000000"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "mfa"
assert result["errors"] == {"base": error}
mock_auth_flow.async_submit_mfa.side_effect = None
mock_auth_flow.async_submit_mfa.return_value = ("access", "refresh")
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_setup_entry.called
async def test_already_configured(
hass: HomeAssistant,
mock_engie_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test aborting when the JWT subject is already configured."""
mock_engie_client.return_value.subject = SUBJECT
mock_config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(mock_config_entry, unique_id=SUBJECT)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "mfa"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_two_accounts_get_distinct_entries(
hass: HomeAssistant,
mock_engie_client: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test that two distinct JWT subjects create two distinct entries."""
other_subject = "auth0|7a3c9e21b48f5d0261af3d77"
for subject in (SUBJECT, other_subject):
mock_engie_client.return_value.subject = subject
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["step_id"] == "mfa"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 2
assert {entry.unique_id for entry in entries} == {SUBJECT, other_subject}
async def test_jwt_subject_missing_shows_form_error(
hass: HomeAssistant,
mock_engie_client: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test that a missing JWT subject shows a form error on the MFA step."""
mock_engie_client.return_value.subject = None
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["step_id"] == "mfa"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "mfa"
assert result["errors"] == {"base": "invalid_auth"}
mock_engie_client.return_value.subject = SUBJECT
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"code": "123456"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_setup_entry.called
+346
View File
@@ -0,0 +1,346 @@
"""Test the ENGIE Belgium integration setup."""
from unittest.mock import AsyncMock, MagicMock
from aioengiebelgium import (
AccountRelation,
CustomerAccount,
CustomerAccountRelations,
EngieBeAuthenticationError,
EngieBeCommunicationError,
PricesResponse,
)
import pytest
from homeassistant.components.engie_be.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import BAN, BAN_2, OFFTAKE_ONLY_EAN, build_prices, build_relations
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("mock_engie_client")
async def test_setup_and_unload(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful setup and unload of a config entry."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
"side_effect",
[
pytest.param(EngieBeCommunicationError("boom"), id="communication_error"),
pytest.param(EngieBeAuthenticationError("boom"), id="auth_error"),
],
)
async def test_setup_relations_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
side_effect: Exception,
) -> None:
"""Test setup handles a failure of the customer-account-relations fetch."""
mock_engie_client.return_value.async_get_customer_account_relations.side_effect = (
side_effect
)
mock_config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
assert not hass.config_entries.flow.async_progress()
@pytest.mark.parametrize(
"side_effect",
[
pytest.param(EngieBeCommunicationError("boom"), id="communication_error"),
pytest.param(EngieBeAuthenticationError("boom"), id="auth_error"),
],
)
async def test_setup_prices_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
side_effect: Exception,
) -> None:
"""Test setup succeeds with a device but no entities when the initial prices fetch fails."""
mock_engie_client.return_value.async_get_prices.side_effect = side_effect
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, BAN), mock_config_entry.entry_id
)
is not None
)
assert not er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
async def test_partial_first_refresh(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test setup comes up when only one of two BANs succeeds on the first refresh."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
build_relations(BAN, BAN_2)
)
mock_engie_client.return_value.async_get_prices.side_effect = [
build_prices(),
EngieBeCommunicationError("boom"),
]
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert (
entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
)
is not None
)
assert (
entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN_2}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
)
is None
)
assert "Error fetching" in caplog.text
assert BAN_2 not in caplog.text
assert BAN_2[-4:] in caplog.text
async def test_all_households_loaded_despite_prices_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup succeeds with devices but no entities when every household's first refresh fails."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
build_relations(BAN, BAN_2)
)
mock_engie_client.return_value.async_get_prices.side_effect = (
EngieBeCommunicationError("boom")
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
device_entries = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert len(device_entries) == 2
assert not er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
async def test_setup_without_active_agreements(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup succeeds without devices or entities when there are no active business agreements."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
CustomerAccountRelations(
accounts=(
AccountRelation(
id="account-1",
admin=True,
customer_account=CustomerAccount(
customer_account_number="can-1",
business_agreements=(),
),
),
)
)
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert not dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert not er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
async def test_device_created_for_household_without_prices(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test a household device is registered even when its prices are empty."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
build_relations(BAN, BAN_2)
)
mock_engie_client.return_value.async_get_prices.side_effect = [
build_prices(),
PricesResponse(items=()),
]
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entries = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert len(device_entries) == 2
assert all(
device.entry_type is dr.DeviceEntryType.SERVICE for device in device_entries
)
ban_2_device = device_registry.async_get_device_by_identifier(
(DOMAIN, BAN_2), mock_config_entry.entry_id
)
assert ban_2_device is not None
assert not er.async_entries_for_device(entity_registry, ban_2_device.id)
ban_device = device_registry.async_get_device_by_identifier(
(DOMAIN, BAN), mock_config_entry.entry_id
)
assert ban_device is not None
assert er.async_entries_for_device(entity_registry, ban_device.id)
async def test_token_refresh_skips_unchanged_tokens(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
) -> None:
"""Test the on_token_refresh callback is a no-op when tokens are unchanged."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
listener = AsyncMock()
mock_config_entry.add_update_listener(listener)
on_token_refresh = mock_engie_client.call_args.kwargs["on_token_refresh"]
await on_token_refresh("access-token", "refresh-token")
await hass.async_block_till_done()
listener.assert_not_called()
assert mock_config_entry.data["access_token"] == "access-token"
assert mock_config_entry.data["refresh_token"] == "refresh-token"
async def test_token_refresh_persists_tokens(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
) -> None:
"""Test the on_token_refresh callback persists rotated tokens to entry.data."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
listener = AsyncMock()
mock_config_entry.add_update_listener(listener)
on_token_refresh = mock_engie_client.call_args.kwargs["on_token_refresh"]
await on_token_refresh("rotated-access", "rotated-refresh")
await hass.async_block_till_done()
listener.assert_called_once()
assert mock_config_entry.data["access_token"] == "rotated-access"
assert mock_config_entry.data["refresh_token"] == "rotated-refresh"
assert mock_config_entry.data["email"] == "user@example.com"
assert mock_config_entry.data["mfa_method"] == "sms"
async def test_unexpected_service_point_error_does_not_prevent_setup(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test an unexpected non-EngieBeError from a service-point lookup leaves the household without entities but still loads the entry."""
mock_engie_client.return_value.async_get_service_point.side_effect = ValueError(
"boom"
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert "Unexpected error fetching" in caplog.text
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, BAN), mock_config_entry.entry_id
)
is not None
)
assert not er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
async def test_household_is_a_single_service_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test a household's EANs share one SERVICE device instead of per-meter devices."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entries = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert len(device_entries) == 1
household_device = device_entries[0]
assert household_device.identifiers == {(DOMAIN, BAN)}
assert household_device.entry_type is dr.DeviceEntryType.SERVICE
assert household_device.via_device_id is None
+998
View File
@@ -0,0 +1,998 @@
"""Test the ENGIE Belgium sensor platform."""
from collections.abc import Callable, Mapping
import dataclasses
from datetime import date, timedelta
from unittest.mock import MagicMock
from aioengiebelgium import (
AccountRelation,
BusinessAgreement,
ConsumptionAddress,
CustomerAccount,
CustomerAccountRelations,
EanPrices,
EngieBeAuthenticationError,
EngieBeCommunicationError,
EngieBeError,
PricePeriod,
PriceSlot,
PricesResponse,
ServicePoint,
bare_ean,
)
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.engie_be.const import DOMAIN, PRICES_SCAN_INTERVAL
from homeassistant.components.engie_be.coordinator import EngieBePricesData
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import (
BAN,
BAN_2,
OFFTAKE_INJECTION_EAN,
OFFTAKE_ONLY_EAN,
build_prices,
build_relations,
)
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
def _prices_by_ban(
mapping: Mapping[str, PricesResponse | EngieBeError],
) -> Callable[[str], PricesResponse]:
"""Return an async_get_prices side_effect that resolves independently per BAN."""
def _side_effect(ban: str) -> PricesResponse:
result = mapping[ban]
if isinstance(result, EngieBeError):
raise result
return result
return _side_effect
@pytest.mark.usefixtures("mock_engie_client", "entity_registry_enabled_by_default")
async def test_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the energy-price sensors for an offtake-only and a dual-direction EAN."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entries = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
for device_entry in sorted(device_entries, key=lambda entry: entry.id):
assert device_entry == snapshot(name=f"{device_entry.name}-device")
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.usefixtures("mock_engie_client")
async def test_slot_aware_naming(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test slot-aware translation keys drive friendly names while unique_ids keep the raw slot code."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
total_hours_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
)
assert total_hours_entity_id is not None
total_hours_state = hass.states.get(total_hours_entity_id)
assert total_hours_state is not None
assert "total_hours" not in total_hours_state.name.lower()
peak_unique_id = f"{BAN}_{OFFTAKE_INJECTION_EAN}_offtake_S_TOU1_OFFTAKE_PEAK"
peak_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, peak_unique_id
)
assert peak_entity_id is not None
peak_entry = entity_registry.async_get(peak_entity_id)
assert peak_entry is not None
assert peak_entry.unique_id == peak_unique_id
peak_state = hass.states.get(peak_entity_id)
assert peak_state is not None
assert peak_state.name.endswith("Electricity peak offtake price")
fallback_unique_id = f"{BAN}_{OFFTAKE_INJECTION_EAN}_offtake_S_TOU1_OFFTAKE_WEEKEND"
fallback_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, fallback_unique_id
)
assert fallback_entity_id is not None
fallback_state = hass.states.get(fallback_entity_id)
assert fallback_state is not None
assert fallback_state.name.endswith("Electricity offtake price (weekend)")
assert (
entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{OFFTAKE_INJECTION_EAN}_offtake_EN"
)
is None
)
@pytest.mark.usefixtures("mock_engie_client")
async def test_excl_vat_disabled_by_default(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test excl-VAT price sensors are disabled by default, incl-VAT ones are not."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
excl_vat_entries = [
entry for entry in entity_entries if entry.unique_id.endswith("_excl_vat")
]
incl_vat_entries = [
entry for entry in entity_entries if not entry.unique_id.endswith("_excl_vat")
]
assert excl_vat_entries
assert incl_vat_entries
assert all(
entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION
for entry in excl_vat_entries
)
assert all(entry.disabled_by is None for entry in incl_vat_entries)
SECOND_ELECTRICITY_EAN = "541448820000000003_ID1"
def _single_period_prices(ean: str) -> EanPrices:
"""Build a single-period, offtake-only EanPrices for the given EAN."""
return EanPrices(
ean=ean,
periods=(
PricePeriod(
valid_from=date(2000, 1, 1),
valid_to=date(2099, 12, 31),
vat_tariff=6.0,
offtake=(
PriceSlot(
time_of_use_slot_code="TOTAL_HOURS",
price_value=0.2,
price_value_excl_vat=0.19,
),
),
),
),
)
async def test_duplicate_type_suffix(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test entities get a distinct last-4 EAN suffix for two same-type household EANs."""
eans = (OFFTAKE_INJECTION_EAN, SECOND_ELECTRICITY_EAN)
mock_engie_client.return_value.async_get_prices.return_value = PricesResponse(
items=tuple(_single_period_prices(ean) for ean in eans)
)
mock_engie_client.return_value.async_get_service_point.side_effect = lambda ean: (
ServicePoint(ean_energy_types={bare_ean(ean): "ELECTRICITY"})
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
names: dict[str, str] = {}
for ean in eans:
entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{ean}_offtake_TOTAL_HOURS"
)
assert entity_id is not None
state = hass.states.get(entity_id)
assert state is not None
names[ean] = state.name
for ean in eans:
assert names[ean].endswith(f"({bare_ean(ean)[-4:]})")
assert names[OFFTAKE_INJECTION_EAN] != names[SECOND_ELECTRICITY_EAN]
@pytest.mark.parametrize(
"exception",
[
pytest.param(EngieBeCommunicationError("boom"), id="communication_error"),
pytest.param(EngieBeAuthenticationError("boom"), id="auth_error"),
],
)
async def test_type_fallback_when_service_point_lookup_fails(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
exception: Exception,
) -> None:
"""Test a failed service-point lookup falls back to the untyped entity name."""
mock_engie_client.return_value.async_get_service_point.side_effect = [
exception,
ServicePoint(ean_energy_types={bare_ean(OFFTAKE_INJECTION_EAN): "ELECTRICITY"}),
]
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
)
assert entity_id is not None
state = hass.states.get(entity_id)
assert state is not None
assert state.name.endswith("Offtake price")
def _empty_address_relations() -> CustomerAccountRelations:
"""Build relations for a single BAN with an all-empty consumption address."""
return CustomerAccountRelations(
accounts=(
AccountRelation(
id="account-1",
admin=True,
customer_account=CustomerAccount(
customer_account_number="can-1",
business_agreements=(
BusinessAgreement(
business_agreement_number=BAN,
active=True,
consumption_address=ConsumptionAddress(
street="",
house_number="",
postal_code="",
city="",
),
),
),
),
),
)
)
@pytest.mark.parametrize(
"relations",
[
pytest.param(build_relations(with_address=False), id="no-address"),
pytest.param(_empty_address_relations(), id="empty-address"),
],
)
async def test_device_name_falls_back_to_ban(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
device_registry: dr.DeviceRegistry,
relations: CustomerAccountRelations,
) -> None:
"""Test the device name falls back to the bare BAN without a usable consumption address."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
relations
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
household_device = device_registry.async_get_device_by_identifier(
(DOMAIN, BAN), mock_config_entry.entry_id
)
assert household_device is not None
assert household_device.name == BAN
@pytest.mark.parametrize(
("valid_from", "valid_to"),
[
pytest.param(date(2026, 8, 1), date(2026, 9, 1), id="inside-window"),
pytest.param(date(2026, 8, 13), date(2026, 9, 1), id="first-day-inclusive"),
pytest.param(
date(2026, 8, 1),
date(2026, 9, 1),
id="date-objects",
),
],
)
async def test_period_window_creates_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
valid_from: date | None,
valid_to: date | None,
) -> None:
"""Test sensors are created with a numeric state for a period covering today."""
freezer.move_to("2026-08-13T12:00:00+02:00")
mock_engie_client.return_value.async_get_prices.return_value = build_prices(
valid_from=valid_from, valid_to=valid_to
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
enabled_entries = [
entity_entry
for entity_entry in entity_entries
if entity_entry.disabled_by is None
]
assert enabled_entries
for entity_entry in enabled_entries:
state = hass.states.get(entity_entry.entity_id)
assert state is not None
float(state.state)
async def test_period_window_uses_brussels_date(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test period selection uses the Brussels calendar date."""
freezer.move_to("2026-08-12T22:30:00+00:00")
mock_engie_client.return_value.async_get_prices.return_value = build_prices(
valid_from=date(2026, 8, 13), valid_to=date(2026, 8, 14)
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
@pytest.mark.parametrize(
("valid_from", "valid_to"),
[
pytest.param(date(2026, 8, 1), date(2026, 8, 13), id="last-day-exclusive"),
pytest.param(date(2026, 1, 1), date(2026, 2, 1), id="expired-window"),
pytest.param(None, date(2026, 9, 1), id="malformed-dates"),
],
)
async def test_period_window_skips_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
valid_from: date | None,
valid_to: date | None,
) -> None:
"""Test no sensors are created when no period covers today (locks in exclusive valid_to)."""
freezer.move_to("2026-08-13T12:00:00+02:00")
mock_engie_client.return_value.async_get_prices.return_value = build_prices(
valid_from=valid_from, valid_to=valid_to
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert entity_entries == []
async def test_period_window_refresh_makes_existing_sensors_unavailable(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test existing sensors become unavailable after the period expires."""
freezer.move_to("2026-08-12T12:00:00+02:00")
mock_engie_client.return_value.async_get_prices.return_value = build_prices(
valid_from=date(2026, 8, 1), valid_to=date(2026, 8, 13)
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
enabled_entity_entries = [
entity_entry
for entity_entry in entity_entries
if entity_entry.disabled_by is None
]
assert enabled_entity_entries
freezer.tick(timedelta(days=1, seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
for entity_entry in enabled_entity_entries:
assert entity_registry.async_get(entity_entry.entity_id) is not None
state = hass.states.get(entity_entry.entity_id)
assert state is not None
assert state.state == STATE_UNAVAILABLE
def _degraded_slots_empty() -> EngieBePricesData:
"""Return prices data with no slots or EANs at all."""
return EngieBePricesData(slots={}, eans=())
def _degraded_period_expired() -> EngieBePricesData:
"""Return prices data whose only period for the EAN has expired."""
return EngieBePricesData(slots={}, eans=(OFFTAKE_ONLY_EAN,))
def _degraded_slot_code_absent() -> EngieBePricesData:
"""Return prices data whose current period has no matching offtake slot."""
return EngieBePricesData(
slots={
(OFFTAKE_ONLY_EAN, "offtake", "OTHER_CODE"): PriceSlot(
time_of_use_slot_code="OTHER_CODE",
price_value=0.5,
price_value_excl_vat=0.5,
)
},
eans=(OFFTAKE_ONLY_EAN,),
)
def _degraded_ean_absent() -> EngieBePricesData:
"""Return prices data whose slots belong to a different EAN entirely."""
return EngieBePricesData(
slots={
(OFFTAKE_INJECTION_EAN, "offtake", "TOTAL_HOURS"): PriceSlot(
time_of_use_slot_code="TOTAL_HOURS",
price_value=0.5,
price_value_excl_vat=0.5,
)
},
eans=(OFFTAKE_INJECTION_EAN,),
)
@pytest.mark.usefixtures("mock_engie_client")
@pytest.mark.parametrize(
"degraded_data",
[
pytest.param(_degraded_slots_empty, id="slots-empty"),
pytest.param(_degraded_period_expired, id="period-expired"),
pytest.param(_degraded_slot_code_absent, id="slot-code-absent"),
pytest.param(_degraded_ean_absent, id="ean-absent"),
],
)
async def test_available_degrades_for_missing_slot(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
degraded_data: Callable[[], EngieBePricesData],
) -> None:
"""Test a missing slot key makes the entity unavailable for every degraded-data branch."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
)
assert entity_id is not None
coordinator = mock_config_entry.runtime_data.households[BAN].prices
coordinator.async_set_updated_data(degraded_data())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_recovering_ban_adds_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a BAN that fails on the first refresh gets its sensors once it recovers."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
build_relations(BAN, BAN_2)
)
mock_engie_client.return_value.async_get_prices.side_effect = [
build_prices(),
EngieBeCommunicationError("boom"),
]
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
ban_2_unique_id = f"{BAN_2}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
assert (
entity_registry.async_get_entity_id("sensor", DOMAIN, ban_2_unique_id) is None
)
mock_engie_client.return_value.async_get_prices.side_effect = [
build_prices(),
build_prices(),
]
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
ban_2_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, ban_2_unique_id
)
assert ban_2_entity_id is not None
entity_entry = entity_registry.async_get(ban_2_entity_id)
assert entity_entry is not None
ban_2_device = device_registry.async_get_device_by_identifier(
(DOMAIN, BAN_2), mock_config_entry.entry_id
)
assert ban_2_device is not None
assert entity_entry.device_id == ban_2_device.id
async def test_period_gap_heals(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sensors appear once today enters a price period that was a gap at setup."""
freezer.move_to("2026-08-13T12:00:00+02:00")
mock_engie_client.return_value.async_get_prices.return_value = build_prices(
valid_from=date(2026, 8, 14), valid_to=date(2026, 9, 1)
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert (
er.async_entries_for_config_entry(entity_registry, mock_config_entry.entry_id)
== []
)
freezer.tick(timedelta(days=1, seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
def _add_offtake_slot(
prices: PricesResponse, ean: str, slot: PriceSlot
) -> PricesResponse:
"""Return a copy of prices with an extra offtake slot added to one EAN's periods."""
return PricesResponse(
items=tuple(
dataclasses.replace(
ean_prices,
periods=tuple(
dataclasses.replace(period, offtake=(*period.offtake, slot))
for period in ean_prices.periods
),
)
if ean_prices.ean == ean
else ean_prices
for ean_prices in prices.items
)
)
async def test_new_slot_code_appears(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a new time-of-use slot code adds exactly its incl/excl VAT sensor pair."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
count_before = len(
er.async_entries_for_config_entry(entity_registry, mock_config_entry.entry_id)
)
extra_slot = PriceSlot(
time_of_use_slot_code="S_TOU1_OFFTAKE_NIGHT",
price_value=0.09,
price_value_excl_vat=0.084906,
)
mock_engie_client.return_value.async_get_prices.return_value = _add_offtake_slot(
build_prices(), OFFTAKE_INJECTION_EAN, extra_slot
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
count_after = len(
er.async_entries_for_config_entry(entity_registry, mock_config_entry.entry_id)
)
assert count_after == count_before + 2
incl_vat_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN}_{OFFTAKE_INJECTION_EAN}_offtake_S_TOU1_OFFTAKE_NIGHT"
)
excl_vat_id = entity_registry.async_get_entity_id(
"sensor",
DOMAIN,
f"{BAN}_{OFFTAKE_INJECTION_EAN}_offtake_S_TOU1_OFFTAKE_NIGHT_excl_vat",
)
assert incl_vat_id is not None
assert excl_vat_id is not None
async def test_no_duplicate_entities_on_repeated_refresh(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test refreshing with identical data does not add duplicate entities."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
count_before = len(
er.async_entries_for_config_entry(entity_registry, mock_config_entry.entry_id)
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
count_after = len(
er.async_entries_for_config_entry(entity_registry, mock_config_entry.entry_id)
)
assert count_after == count_before
async def test_late_service_point_failure_falls_back(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a recovering BAN whose new EAN fails its service-point lookup falls back."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
build_relations(BAN, BAN_2)
)
ban_2_prices = PricesResponse(
items=(_single_period_prices(SECOND_ELECTRICITY_EAN),)
)
mock_engie_client.return_value.async_get_prices.side_effect = [
build_prices(),
EngieBeCommunicationError("boom"),
]
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{BAN: build_prices(), BAN_2: ban_2_prices}
)
mock_engie_client.return_value.async_get_service_point.side_effect = (
EngieBeCommunicationError("boom")
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{BAN_2}_{SECOND_ELECTRICITY_EAN}_offtake_TOTAL_HOURS"
)
assert entity_id is not None
state = hass.states.get(entity_id)
assert state is not None
assert state.name.endswith("Offtake price")
def _prices_entity_id(entity_registry: er.EntityRegistry, ban: str) -> str:
"""Return the entity_id for a BAN's offtake-only price sensor."""
unique_id = f"{ban}_{OFFTAKE_ONLY_EAN}_offtake_TOTAL_HOURS"
entity_id = entity_registry.async_get_entity_id("sensor", DOMAIN, unique_id)
assert entity_id is not None
return entity_id
def _single_ban_prices_response(ean: str, price_value: float) -> PricesResponse:
"""Build a prices response with one offtake slot for the given EAN and price."""
return PricesResponse(
items=(
EanPrices(
ean=ean,
periods=(
PricePeriod(
valid_from=date(2000, 1, 1),
valid_to=date(2099, 12, 31),
vat_tariff=6.0,
offtake=(
PriceSlot(
time_of_use_slot_code="TOTAL_HOURS",
price_value=price_value,
price_value_excl_vat=price_value,
),
),
),
),
),
)
)
async def _setup_two_bans(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
) -> None:
"""Set up a config entry with two business agreements and initial prices."""
mock_engie_client.return_value.async_get_customer_account_relations.return_value = (
build_relations(BAN, BAN_2)
)
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{
BAN: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.1),
BAN_2: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.2),
}
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
async def test_single_ban_transient_failure_goes_unavailable(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a transient failure for one BAN makes only that BAN's sensor unavailable."""
await _setup_two_bans(hass, mock_config_entry, mock_engie_client)
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{
BAN: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.3),
BAN_2: EngieBeCommunicationError("boom"),
}
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
state_ban = hass.states.get(_prices_entity_id(entity_registry, BAN))
state_ban_2 = hass.states.get(_prices_entity_id(entity_registry, BAN_2))
assert state_ban is not None
assert state_ban_2 is not None
assert state_ban.state == "0.3"
assert state_ban_2.state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
"exception",
[
pytest.param(EngieBeCommunicationError("boom"), id="communication_error"),
pytest.param(EngieBeAuthenticationError("boom"), id="auth_error"),
],
)
async def test_all_bans_fail(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
exception: Exception,
) -> None:
"""Test all BANs failing on refresh makes every sensor unavailable."""
await _setup_two_bans(hass, mock_config_entry, mock_engie_client)
mock_engie_client.return_value.async_get_prices.side_effect = exception
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
state_ban = hass.states.get(_prices_entity_id(entity_registry, BAN))
state_ban_2 = hass.states.get(_prices_entity_id(entity_registry, BAN_2))
assert state_ban is not None
assert state_ban_2 is not None
assert state_ban.state == STATE_UNAVAILABLE
assert state_ban_2.state == STATE_UNAVAILABLE
assert not hass.config_entries.flow.async_progress()
async def test_unexpected_exception_is_not_swallowed(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test an unexpected non-EngieBeError exception is logged instead of being swallowed."""
await _setup_two_bans(hass, mock_config_entry, mock_engie_client)
mock_engie_client.return_value.async_get_prices.side_effect = ValueError("boom")
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Unexpected error fetching" in caplog.text
async def test_failure_and_recovery_are_logged_once(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test a failing BAN logs one error, stays quiet on repeat failures, then logs one recovery."""
await _setup_two_bans(hass, mock_config_entry, mock_engie_client)
caplog.set_level("DEBUG", logger="homeassistant.components.engie_be")
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{
BAN: EngieBeCommunicationError("boom"),
BAN_2: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.2),
}
)
caplog.clear()
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert caplog.text.count("Error fetching") == 1
assert "recovered" not in caplog.text
caplog.clear()
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{
BAN: EngieBeCommunicationError("boom"),
BAN_2: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.3),
}
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Error fetching" not in caplog.text
caplog.clear()
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{
BAN: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.4),
BAN_2: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.5),
}
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert caplog.text.count("recovered") == 1
state_ban = hass.states.get(_prices_entity_id(entity_registry, BAN))
state_ban_2 = hass.states.get(_prices_entity_id(entity_registry, BAN_2))
assert state_ban is not None
assert state_ban_2 is not None
assert state_ban.state == "0.4"
assert state_ban_2.state == "0.5"
async def test_recovery_after_all_bans_fail(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sensors recover with fresh values after all BANs previously failed."""
await _setup_two_bans(hass, mock_config_entry, mock_engie_client)
mock_engie_client.return_value.async_get_prices.side_effect = (
EngieBeCommunicationError("boom")
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
mock_engie_client.return_value.async_get_prices.side_effect = _prices_by_ban(
{
BAN: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.4),
BAN_2: _single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.5),
}
)
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
state_ban = hass.states.get(_prices_entity_id(entity_registry, BAN))
state_ban_2 = hass.states.get(_prices_entity_id(entity_registry, BAN_2))
assert state_ban is not None
assert state_ban_2 is not None
assert state_ban.state == "0.4"
assert state_ban_2.state == "0.5"
async def test_service_point_transient_failure_is_retried(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test a failed service-point fetch is retried on the next refresh cycle."""
caplog.set_level("DEBUG", logger="homeassistant.components.engie_be")
mock_engie_client.return_value.async_get_prices.return_value = (
_single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.1)
)
mock_engie_client.return_value.async_get_service_point.side_effect = [
EngieBeCommunicationError("boom"),
ServicePoint(ean_energy_types={bare_ean(OFFTAKE_ONLY_EAN): "GAS"}),
]
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert "Fetching service point for …0001 failed" in caplog.text
assert "…_ID1" not in caplog.text
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert mock_engie_client.return_value.async_get_service_point.call_count == 2
async def test_service_point_success_without_ean_is_cached_once(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_engie_client: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a successful fetch omitting the EAN is cached once and not retried."""
mock_engie_client.return_value.async_get_prices.return_value = (
_single_ban_prices_response(OFFTAKE_ONLY_EAN, 0.1)
)
mock_engie_client.return_value.async_get_service_point.side_effect = lambda ean: (
ServicePoint(ean_energy_types={})
)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(PRICES_SCAN_INTERVAL + timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert mock_engie_client.return_value.async_get_service_point.call_count == 1