Add new Vitesy integration (#181865)

This commit is contained in:
Simone Chemelli
2026-09-30 07:53:43 +02:00
committed by GitHub
parent beb3ed3b75
commit 22091fb803
22 changed files with 1468 additions and 0 deletions
+1
View File
@@ -648,6 +648,7 @@ homeassistant.components.velbus.*
homeassistant.components.velux.*
homeassistant.components.victron_gx.*
homeassistant.components.vistapool.*
homeassistant.components.vitesy.*
homeassistant.components.vivotek.*
homeassistant.components.vizio.*
homeassistant.components.vlc_telnet.*
Generated
+2
View File
@@ -2052,6 +2052,8 @@ CLAUDE.md @home-assistant/core
/tests/components/vilfo/ @ManneW
/homeassistant/components/vistapool/ @fdebrus
/tests/components/vistapool/ @fdebrus
/homeassistant/components/vitesy/ @chemelli74
/tests/components/vitesy/ @chemelli74
/homeassistant/components/vivotek/ @HarlemSquirrel
/tests/components/vivotek/ @HarlemSquirrel
/homeassistant/components/vizio/ @raman325
@@ -0,0 +1,25 @@
"""The Vitesy integration."""
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .coordinator import VitesyConfigEntry, VitesyDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: VitesyConfigEntry) -> bool:
"""Set up Vitesy from a config entry."""
coordinator = VitesyDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: VitesyConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@@ -0,0 +1,77 @@
"""Config flow for the Vitesy integration."""
from typing import Any, override
from aiovitesy.api import VitesyApi
from aiovitesy.exceptions import CannotAuthenticate, VitesyError
import probatio
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import DOMAIN
STEP_USER_DATA_SCHEMA = probatio.Schema(
{
probatio.Required(CONF_EMAIL): TextSelector(
TextSelectorConfig(
type=TextSelectorType.EMAIL,
autocomplete="username",
)
),
probatio.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD,
autocomplete="current-password",
)
),
}
)
async def _validate_credentials(hass: HomeAssistant, email: str, password: str) -> str:
"""Authenticate and return the account's stable user id."""
api = VitesyApi(email, password, async_get_clientsession(hass))
await api.login()
user = await api.get_user()
user_id = user.get("id")
if not isinstance(user_id, (str, int)):
raise VitesyError(f"Profile response is missing a valid user id: {user_id!r}")
return str(user_id)
class VitesyConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Vitesy."""
@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:
email = user_input[CONF_EMAIL]
try:
user_id = await _validate_credentials(
self.hass, email, user_input[CONF_PASSWORD]
)
except CannotAuthenticate:
errors["base"] = "invalid_auth"
except VitesyError:
errors["base"] = "cannot_connect"
else:
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=email, data=user_input)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
+8
View File
@@ -0,0 +1,8 @@
"""Constants for the Vitesy integration."""
import logging
from typing import Final
DOMAIN: Final = "vitesy"
LOGGER: Final = logging.getLogger(__package__)
@@ -0,0 +1,81 @@
"""Coordinator for the Vitesy integration."""
from collections.abc import Iterator
from contextlib import contextmanager
from datetime import timedelta
from typing import override
from aiovitesy.api import VitesyApi, VitesyDevice
from aiovitesy.exceptions import CannotAuthenticate, VitesyError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, LOGGER
type VitesyConfigEntry = ConfigEntry[VitesyDataUpdateCoordinator]
UPDATE_INTERVAL = timedelta(minutes=5)
@contextmanager
def _translate_errors(*, auth_recoverable: bool = False) -> Iterator[None]:
"""Convert aiovitesy errors into coordinator setup/update failures.
On a refresh, auth_recoverable must be True: the config flow has no
reauth step yet, so raising ConfigEntryAuthFailed there would stop
polling forever with no way to recover.
"""
try:
yield
except CannotAuthenticate as err:
if auth_recoverable:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
) from err
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
) from err
except VitesyError as err:
raise UpdateFailed(
translation_domain=DOMAIN, translation_key="update_failed"
) from err
class VitesyDataUpdateCoordinator(DataUpdateCoordinator[dict[str, VitesyDevice]]):
"""Fetch state for every device of a Vitesy Hub account."""
config_entry: VitesyConfigEntry
def __init__(self, hass: HomeAssistant, config_entry: VitesyConfigEntry) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=UPDATE_INTERVAL,
)
self.api = VitesyApi(
config_entry.data[CONF_EMAIL],
config_entry.data[CONF_PASSWORD],
async_get_clientsession(hass),
)
@override
async def _async_setup(self) -> None:
"""Authenticate against Vitesy Hub before the first refresh."""
with _translate_errors():
await self.api.login()
@override
async def _async_update_data(self) -> dict[str, VitesyDevice]:
"""Fetch the latest state for every device in the account."""
with _translate_errors(auth_recoverable=True):
return await self.api.get_all_devices()
+51
View File
@@ -0,0 +1,51 @@
"""Base entity for the Vitesy integration."""
from typing import override
from aiovitesy.api import VitesyDevice
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import VitesyDataUpdateCoordinator
class VitesyEntity(CoordinatorEntity[VitesyDataUpdateCoordinator]):
"""Base class for all Vitesy entities."""
_attr_has_entity_name = True
def __init__(
self, coordinator: VitesyDataUpdateCoordinator, device_id: str
) -> None:
"""Initialize the entity for a single Vitesy device."""
super().__init__(coordinator)
self._device_id = device_id
device = self.device
# The Vitesy Hub device id is the device's MAC address.
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device_id)},
connections={(CONNECTION_NETWORK_MAC, device_id)},
manufacturer="Vitesy",
model=device.device_type or None,
model_id=device.model or None,
name=device.name,
sw_version=device.firmware_version or None,
)
@property
def device(self) -> VitesyDevice:
"""Return this entity's device from the latest coordinator data."""
return self.coordinator.data[self._device_id]
@property
@override
def available(self) -> bool:
"""Return True when the device is present and reporting as connected."""
return (
super().available
and self._device_id in self.coordinator.data
and self.device.connected
)
@@ -0,0 +1,12 @@
{
"entity": {
"sensor": {
"air_quality_score": {
"default": "mdi:air-filter"
},
"door_openings": {
"default": "mdi:door-open"
}
}
}
}
@@ -0,0 +1,12 @@
{
"domain": "vitesy",
"name": "Vitesy",
"codeowners": ["@chemelli74"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/vitesy",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["aiovitesy"],
"quality_scale": "bronze",
"requirements": ["aiovitesy==1.1.0"]
}
@@ -0,0 +1,78 @@
rules:
# Bronze
action-setup:
status: exempt
comment: The integration does not register any custom 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: The integration does not register any custom actions.
docs-conditions:
status: exempt
comment: The integration does not register any conditions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-triggers:
status: exempt
comment: The integration does not register any 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: The integration does not register any custom actions.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
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: The integration is cloud based and has no discovery.
discovery:
status: exempt
comment: The integration is cloud based and has no discovery.
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: The integration has no repairable issues.
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done
+190
View File
@@ -0,0 +1,190 @@
"""Sensor platform for the Vitesy integration."""
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from typing import override
from aiovitesy.api import VitesyDevice
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util import dt as dt_util
from .const import LOGGER
from .coordinator import VitesyConfigEntry, VitesyDataUpdateCoordinator
from .entity import VitesyEntity
PARALLEL_UPDATES = 0
# Groups within a measurement payload.
_SENSORS_DATA = "sensors_data"
_STATUS_DATA = "status_data"
# Reading ids within those groups (Shelfy).
_FRIDGE_TEMPERATURE = "TMP01-SY"
_DOOR_OPENINGS = "DOC-SY"
_DOOR_OPEN_DURATION = "DOT-SY"
_BATTERY = "battery"
def _reading(device: VitesyDevice, group: str, reading_id: str) -> float | None:
"""Return a numeric value from a measurement group entry by its id."""
for entry in device.measurement.get(group, ()):
if entry.get("id") == reading_id:
value = entry.get("value")
value = value.get("avg") if isinstance(value, dict) else value
if isinstance(value, (int, float)):
return value
LOGGER.warning(
"Ignoring non-numeric value for reading %s on %s: %r",
reading_id,
device.name,
value,
)
return None
return None
def _air_quality_score(device: VitesyDevice) -> float | None:
"""Return the air quality score.
The Hub reports it as a 0-1 fraction; Vitesy's canonical scale is 0-100.
"""
score = device.measurement.get("score")
return score * 100 if score is not None else None
def _maintenance_due(component: str) -> Callable[[VitesyDevice], datetime | None]:
"""Return a value function for a maintenance component's due date."""
def _value(device: VitesyDevice) -> datetime | None:
due_date = device.maintenance.get(component, {}).get("due_date")
if not due_date:
return None
try:
return dt_util.parse_datetime(due_date, raise_on_error=True)
except ValueError:
LOGGER.warning(
"Ignoring unparsable %s due date for %s: %s",
component,
device.name,
due_date,
)
return None
return _value
@dataclass(frozen=True, kw_only=True)
class VitesySensorEntityDescription(SensorEntityDescription):
"""Describes a Vitesy sensor entity."""
value_fn: Callable[[VitesyDevice], StateType | datetime]
SENSORS: tuple[VitesySensorEntityDescription, ...] = (
VitesySensorEntityDescription(
key="air_quality_score",
translation_key="air_quality_score",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=0,
value_fn=_air_quality_score,
),
VitesySensorEntityDescription(
key="fridge_temperature",
translation_key="fridge_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda device: _reading(device, _SENSORS_DATA, _FRIDGE_TEMPERATURE),
),
VitesySensorEntityDescription(
key="door_openings",
translation_key="door_openings",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda device: _reading(device, _SENSORS_DATA, _DOOR_OPENINGS),
),
VitesySensorEntityDescription(
key="door_open_duration",
translation_key="door_open_duration",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.SECONDS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda device: _reading(device, _SENSORS_DATA, _DOOR_OPEN_DURATION),
),
VitesySensorEntityDescription(
key="battery",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda device: _reading(device, _STATUS_DATA, _BATTERY),
),
VitesySensorEntityDescription(
key="filter_change_due",
translation_key="filter_change_due",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=_maintenance_due("filter"),
),
VitesySensorEntityDescription(
key="fridge_cleaning_due",
translation_key="fridge_cleaning_due",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=_maintenance_due("fridge"),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: VitesyConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Vitesy sensors from a config entry."""
coordinator = entry.runtime_data
async_add_entities(
VitesySensor(coordinator, device_id, description)
for device_id, device in coordinator.data.items()
for description in SENSORS
if description.value_fn(device) is not None
)
class VitesySensor(VitesyEntity, SensorEntity):
"""Representation of a Vitesy sensor."""
entity_description: VitesySensorEntityDescription
def __init__(
self,
coordinator: VitesyDataUpdateCoordinator,
device_id: str,
description: VitesySensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, device_id)
self.entity_description = description
self._attr_unique_id = f"{device_id}_{description.key}"
@property
@override
def native_value(self) -> StateType | datetime:
"""Return the current value of the sensor."""
return self.entity_description.value_fn(self.device)
@@ -0,0 +1,53 @@
{
"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%]"
},
"step": {
"user": {
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"email": "The email address of your Vitesy Hub account.",
"password": "The password of your Vitesy Hub account."
}
}
}
},
"entity": {
"sensor": {
"air_quality_score": {
"name": "Air quality score"
},
"door_open_duration": {
"name": "Door open duration"
},
"door_openings": {
"name": "Door openings"
},
"filter_change_due": {
"name": "Filter change due"
},
"fridge_cleaning_due": {
"name": "Fridge cleaning due"
},
"fridge_temperature": {
"name": "Fridge temperature"
}
}
},
"exceptions": {
"invalid_auth": {
"message": "Vitesy Hub credentials are no longer valid."
},
"update_failed": {
"message": "Error communicating with Vitesy Hub, check logs for details"
}
}
}
+1
View File
@@ -874,6 +874,7 @@ FLOWS = {
"victron_remote_monitoring",
"vilfo",
"vistapool",
"vitesy",
"vivotek",
"vizio",
"vlc_telnet",
@@ -8056,6 +8056,12 @@
"config_flow": true,
"iot_class": "cloud_push"
},
"vitesy": {
"name": "Vitesy",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"vivotek": {
"name": "VIVOTEK",
"integration_type": "device",
Generated
+10
View File
@@ -6241,6 +6241,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.vitesy.*]
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.vivotek.*]
check_untyped_defs = true
disallow_incomplete_defs = true
+3
View File
@@ -470,6 +470,9 @@ aiounifi==96
# homeassistant.components.usb
aiousbwatcher==1.1.2
# homeassistant.components.vitesy
aiovitesy==1.1.0
# homeassistant.components.vlc_telnet
aiovlc==0.5.1
+13
View File
@@ -0,0 +1,13 @@
"""Tests for the Vitesy integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the Vitesy integration for testing."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
+94
View File
@@ -0,0 +1,94 @@
"""Common fixtures for the Vitesy tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from aiovitesy.api import VitesyDevice
import pytest
from homeassistant.components.vitesy.const import DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from tests.common import MockConfigEntry
DEVICE_ID = "80:65:99:34:F9:B4"
EMAIL = "test@example.com"
USER_ID = "u1"
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.vitesy.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_devices() -> dict[str, VitesyDevice]:
"""Return a mocked Vitesy account holding a single device."""
return {
DEVICE_ID: VitesyDevice(
device_id=DEVICE_ID,
name="Kitchen Shelfy",
model="SH02AA02",
device_type="SHELFY-R1",
firmware_version="1.2.3",
connected=True,
program_id="eco-s1",
data={"id": DEVICE_ID},
measurement={
"score": 0.49045833333333333,
"timestamp": "2026-09-10T17:19:38+02:00",
"sensors_data": [
{
"id": "TMP01-SY",
"value": {"avg": 9.59, "min": 9.59, "max": 9.59},
},
{"id": "DOC-SY", "value": {"avg": 11, "min": 11, "max": 11}},
{"id": "DOT-SY", "value": {"avg": 205, "min": 205, "max": 205}},
],
"status_data": [
{"id": "battery", "value": {"avg": 27, "min": 27, "max": 27}},
{"id": "charging", "value": False},
{"id": "mode", "value": "eco"},
{"id": "firmware", "value": "1.2.3"},
],
},
maintenance={
"filter": {"due_date": "2026-08-02T06:45:07.623Z"},
"fridge": {"due_date": "2026-10-31T06:45:07.623Z"},
},
programs={},
)
}
@pytest.fixture
def mock_vitesy_client(
mock_devices: dict[str, VitesyDevice],
) -> Generator[AsyncMock]:
"""Mock the aiovitesy client used by the coordinator and the config flow."""
with (
patch(
"homeassistant.components.vitesy.coordinator.VitesyApi", autospec=True
) as mock_client,
patch("homeassistant.components.vitesy.config_flow.VitesyApi", new=mock_client),
):
client = mock_client.return_value
client.login = AsyncMock(return_value=None)
client.get_user = AsyncMock(return_value={"id": USER_ID})
client.get_all_devices = AsyncMock(return_value=mock_devices)
yield client
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
title=EMAIL,
data={CONF_EMAIL: EMAIL, CONF_PASSWORD: "hunter2"},
unique_id=USER_ID,
)
@@ -0,0 +1,384 @@
# serializer version: 1
# name: test_all_entities[sensor.kitchen_shelfy_air_quality_score-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.kitchen_shelfy_air_quality_score',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Air quality score',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Air quality score',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'air_quality_score',
'unique_id': '80:65:99:34:F9:B4_air_quality_score',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_air_quality_score-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Air quality score',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: '%',
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_air_quality_score',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '49.0458333333333',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_battery-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': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.kitchen_shelfy_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '80:65:99:34:F9:B4_battery',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'battery',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Battery',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: '%',
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '27',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_door_open_duration-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.kitchen_shelfy_door_open_duration',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Door open duration',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Door open duration',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'door_open_duration',
'unique_id': '80:65:99:34:F9:B4_door_open_duration',
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_door_open_duration-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'duration',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Door open duration',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTime.SECONDS: 's'>,
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_door_open_duration',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '205',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_door_openings-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.kitchen_shelfy_door_openings',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Door openings',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Door openings',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'door_openings',
'unique_id': '80:65:99:34:F9:B4_door_openings',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_door_openings-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Door openings',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_door_openings',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '11',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_filter_change_due-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.kitchen_shelfy_filter_change_due',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Filter change due',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Filter change due',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'filter_change_due',
'unique_id': '80:65:99:34:F9:B4_filter_change_due',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_filter_change_due-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'timestamp',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Filter change due',
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_filter_change_due',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2026-08-02T06:45:07+00:00',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_fridge_cleaning_due-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.kitchen_shelfy_fridge_cleaning_due',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Fridge cleaning due',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Fridge cleaning due',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'fridge_cleaning_due',
'unique_id': '80:65:99:34:F9:B4_fridge_cleaning_due',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_fridge_cleaning_due-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'timestamp',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Fridge cleaning due',
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_fridge_cleaning_due',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2026-10-31T06:45:07+00:00',
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_fridge_temperature-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.kitchen_shelfy_fridge_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Fridge temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Fridge temperature',
'platform': 'vitesy',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'fridge_temperature',
'unique_id': '80:65:99:34:F9:B4_fridge_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_all_entities[sensor.kitchen_shelfy_fridge_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'temperature',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Kitchen Shelfy Fridge temperature',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.kitchen_shelfy_fridge_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '9.59',
})
# ---
+114
View File
@@ -0,0 +1,114 @@
"""Test the Vitesy config flow."""
from unittest.mock import AsyncMock
from aiovitesy.exceptions import CannotAuthenticate, CannotConnect, GenericResponseError
import pytest
from homeassistant.components.vitesy.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import EMAIL, USER_ID
from tests.common import MockConfigEntry
USER_INPUT = {CONF_EMAIL: EMAIL, CONF_PASSWORD: "hunter2"}
async def test_full_flow(
hass: HomeAssistant, mock_vitesy_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test the happy path creates an entry keyed on the account id."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == EMAIL
assert result["data"] == USER_INPUT
assert result["result"].unique_id == USER_ID
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("side_effect", "reason"),
[
(CannotAuthenticate, "invalid_auth"),
(CannotConnect, "cannot_connect"),
(GenericResponseError, "cannot_connect"),
],
)
async def test_form_errors_then_recovers(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_setup_entry: AsyncMock,
side_effect: type[Exception],
reason: str,
) -> None:
"""Test each login failure surfaces an error and the flow can recover."""
mock_vitesy_client.login.side_effect = side_effect
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": reason}
mock_vitesy_client.login.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == USER_INPUT
async def test_form_invalid_user_id(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test a profile response without a usable id surfaces a recoverable error."""
mock_vitesy_client.get_user.return_value = {"id": None}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
mock_vitesy_client.get_user.return_value = {"id": USER_ID}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == USER_INPUT
async def test_already_configured(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the same account cannot be added twice."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
+99
View File
@@ -0,0 +1,99 @@
"""Test the Vitesy integration setup and teardown."""
from unittest.mock import AsyncMock
from aiovitesy.exceptions import CannotAuthenticate, CannotConnect, VitesyError
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.vitesy.coordinator import UPDATE_INTERVAL
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_setup_and_unload(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test a config entry loads and unloads cleanly."""
await setup_integration(hass, mock_config_entry)
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
async def test_setup_auth_failure(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test rejected credentials put the entry in an error state."""
mock_vitesy_client.login.side_effect = CannotAuthenticate
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
async def test_setup_connection_failure_is_retried(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test a transient connection error triggers a setup retry."""
mock_vitesy_client.login.side_effect = CannotConnect
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_update_failure_marks_entities_unavailable(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a failed refresh marks the entities unavailable."""
await setup_integration(hass, mock_config_entry)
mock_vitesy_client.get_all_devices.side_effect = VitesyError("boom")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
hass.states.get("sensor.kitchen_shelfy_air_quality_score").state
== STATE_UNAVAILABLE
)
async def test_update_auth_failure(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test rejected credentials during a refresh keep the coordinator polling."""
await setup_integration(hass, mock_config_entry)
mock_vitesy_client.get_all_devices.side_effect = CannotAuthenticate
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert not mock_config_entry.runtime_data.last_update_success
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_vitesy_client.get_all_devices.side_effect = None
mock_vitesy_client.get_all_devices.return_value = {}
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_config_entry.runtime_data.last_update_success
+154
View File
@@ -0,0 +1,154 @@
"""Test the Vitesy sensor platform."""
from typing import Any
from unittest.mock import AsyncMock
from aiovitesy.api import VitesyDevice
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.vitesy.coordinator import UPDATE_INTERVAL
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from .conftest import DEVICE_ID
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
AIR_QUALITY_SCORE = "sensor.kitchen_shelfy_air_quality_score"
FILTER_CHANGE_DUE = "sensor.kitchen_shelfy_filter_change_due"
FRIDGE_TEMPERATURE = "sensor.kitchen_shelfy_fridge_temperature"
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_sensor_unavailable_when_device_disconnected(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_devices: dict[str, VitesyDevice],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sensors go unavailable when the device reports as disconnected."""
await setup_integration(hass, mock_config_entry)
assert hass.states.get(AIR_QUALITY_SCORE).state == "49.0458333333333"
mock_devices[DEVICE_ID].connected = False
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(AIR_QUALITY_SCORE).state == STATE_UNAVAILABLE
async def test_air_quality_score_without_value(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_devices: dict[str, VitesyDevice],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the score sensor reports unknown when the measurement drops it."""
await setup_integration(hass, mock_config_entry)
mock_devices[DEVICE_ID].measurement = {
**mock_devices[DEVICE_ID].measurement,
"score": None,
}
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(AIR_QUALITY_SCORE).state == STATE_UNKNOWN
@pytest.mark.parametrize(
("attribute", "update", "entity_id", "warning"),
[
pytest.param(
"measurement",
{"sensors_data": [{"id": "TMP01-SY", "value": {"avg": "n/a"}}]},
FRIDGE_TEMPERATURE,
"Ignoring non-numeric value for reading TMP01-SY on Kitchen Shelfy: 'n/a'",
id="reading_non_numeric_avg",
),
pytest.param(
"measurement",
{"sensors_data": [{"id": "TMP01-SY", "value": "n/a"}]},
FRIDGE_TEMPERATURE,
"Ignoring non-numeric value for reading TMP01-SY on Kitchen Shelfy: 'n/a'",
id="reading_non_numeric_scalar",
),
pytest.param(
"maintenance",
{"filter": {"due_date": "not-a-date"}},
FILTER_CHANGE_DUE,
"Ignoring unparsable filter due date for Kitchen Shelfy: not-a-date",
id="due_date_unparsable",
),
],
)
async def test_sensor_unknown_on_invalid_value(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_devices: dict[str, VitesyDevice],
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
attribute: str,
update: dict[str, Any],
entity_id: str,
warning: str,
) -> None:
"""Test a sensor reports unknown and warns when its value turns invalid."""
await setup_integration(hass, mock_config_entry)
assert hass.states.get(entity_id).state != STATE_UNKNOWN
device = mock_devices[DEVICE_ID]
setattr(device, attribute, {**getattr(device, attribute), **update})
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNKNOWN
assert warning in caplog.text
async def test_sensors_absent_from_measurement_are_not_created(
hass: HomeAssistant,
mock_vitesy_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_devices: dict[str, VitesyDevice],
) -> None:
"""Test only the readings the device actually reports become entities."""
device = mock_devices[DEVICE_ID]
device.measurement = {"score": 0.5}
device.maintenance = {}
await setup_integration(hass, mock_config_entry)
assert hass.states.get(AIR_QUALITY_SCORE) is not None
for absent in (
"battery",
"fridge_temperature",
"door_openings",
"door_open_duration",
"filter_change_due",
"fridge_cleaning_due",
):
assert hass.states.get(f"sensor.kitchen_shelfy_{absent}") is None