Compare commits

...

17 Commits

Author SHA1 Message Date
Paul Bottein d4fae9c33b Make battery sensor diagnostic 2026-06-12 22:03:09 +02:00
Paul Bottein ec4c75a3ac Remove extended status 2026-06-12 22:03:09 +02:00
Paul Bottein 3e82beff8e Add sensors to Yoto
Add the sensor platform: battery, card slot and day mode, plus power
source, SSID and Wi-Fi RSSI as diagnostics (RSSI disabled by default).

Move the online check to the base entity so all entities (including the
media player) report unavailable when the player is offline.
2026-06-12 22:03:09 +02:00
Franck Nijhof 7f0d86bd4f Avoid leaking Immich API key in error logs (#173541) 2026-06-12 21:07:37 +02:00
Franck Nijhof 96c5774bef Suppress InsecureKeyLengthWarning in HTML5 push notifications (#173551) 2026-06-12 19:04:06 +02:00
Franck Nijhof 6288905ca5 Disambiguate duplicate channel names in LG Netcast source list (#173560) 2026-06-12 19:03:55 +02:00
Franck Nijhof 4c1dbec599 Fix Yale Smart Living panic button unique_id for multiple hubs (#173547) 2026-06-12 18:57:44 +02:00
Franck Nijhof 2a0b5ca895 Sort aliases in LLM prompts for stable prefix caching (#173558) 2026-06-12 18:56:38 +02:00
Sarah Seidman 746c8dd908 Upgrade pydroplet to v2.4.0 (#173615) 2026-06-12 18:53:41 +02:00
Franck Nijhof dadfea4d62 Return enum values from config_entry_attr template function (#173554) 2026-06-12 18:49:42 +02:00
Franck Nijhof 53aef99921 Convert JPEG-incompatible image modes to RGB in image upload thumbnail generation (#173538) 2026-06-12 18:49:19 +02:00
Paul Bottein 88bd563a2c Bump yoto-api to 4.2.0 (#173606) 2026-06-12 17:43:16 +02:00
Mick Vleeshouwer c57c8fad16 Bump pyOverkiz to 2.0.1 (#173607) 2026-06-12 16:35:10 +03:00
Ashton 5932a11e0c Extend cloud system health with connection and certificate diagnostics (#171804)
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-12 09:01:08 -04:00
bkobus-bbx 080492c64d Add reactive energy sensors for Blebox energyMeter device (#173504) 2026-06-12 14:47:30 +02:00
bkobus-bbx 3b8689637a Add reauth flow and improve error handling for BleBox integration (#173268) 2026-06-12 14:46:47 +02:00
David Bishop 8cd97fc60e Extract setup_light helper for Govee local tests (#167846)
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-06-12 14:40:26 +02:00
48 changed files with 1166 additions and 382 deletions
+14 -8
View File
@@ -1,9 +1,7 @@
"""The BleBox devices integration."""
import logging
from blebox_uniapi.box import Box
from blebox_uniapi.error import Error
from blebox_uniapi.error import ConnectionError, Error, HttpError, UnauthorizedRequest
from blebox_uniapi.session import ApiHost
from homeassistant.const import (
@@ -14,14 +12,16 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryError,
ConfigEntryNotReady,
)
from .const import DEFAULT_SETUP_TIMEOUT
from .coordinator import BleBoxConfigEntry, BleBoxCoordinator
from .helpers import get_maybe_authenticated_session
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
@@ -50,9 +50,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: BleBoxConfigEntry) -> bo
try:
product = await Box.async_from_host(api_host)
except Error as ex:
_LOGGER.error("Identify failed at %s:%d (%s)", api_host.host, api_host.port, ex)
except UnauthorizedRequest as ex:
raise ConfigEntryAuthFailed from ex
except (
ConnectionError,
HttpError,
) as ex:
raise ConfigEntryNotReady from ex
except Error as ex:
raise ConfigEntryError from ex
coordinator = BleBoxCoordinator(hass, entry, product)
await coordinator.async_config_entry_first_refresh()
+61 -1
View File
@@ -1,5 +1,6 @@
"""Config flow for BleBox devices integration."""
from collections.abc import Mapping
import logging
from typing import Any
@@ -26,6 +27,7 @@ from .const import (
DEFAULT_PORT,
DEFAULT_SETUP_TIMEOUT,
DOMAIN,
INVALID_AUTH,
UNKNOWN,
UNSUPPORTED_VERSION,
)
@@ -46,6 +48,7 @@ STEP_SCHEMA = vol.Schema(
LOG_MSG = {
UNSUPPORTED_VERSION: "Outdated firmware",
CANNOT_CONNECT: "Failed to identify device",
INVALID_AUTH: "Authentication failed",
UNKNOWN: "Unknown error while identifying device",
}
@@ -87,7 +90,7 @@ class BleBoxConfigFlow(ConfigFlow, domain=DOMAIN):
)
except UnauthorizedRequest as ex:
return None, self.handle_step_exception(
ex, schema, host, port, CANNOT_CONNECT, _LOGGER.error, step_id
ex, schema, host, port, INVALID_AUTH, _LOGGER.error, step_id
)
except Error as ex:
return None, self.handle_step_exception(
@@ -115,6 +118,8 @@ class BleBoxConfigFlow(ConfigFlow, domain=DOMAIN):
try:
product = await Box.async_from_host(api_host)
except UnauthorizedRequest:
return self.async_abort(reason="authorization_required")
except UnsupportedBoxVersion:
return self.async_abort(reason="unsupported_device_version")
except UnsupportedBoxResponse:
@@ -246,3 +251,58 @@ class BleBoxConfigFlow(ConfigFlow, domain=DOMAIN):
reconfigure_entry,
data_updates=data_updates,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle reauthentication upon an API authentication error."""
self.context["title_placeholders"] = {
"name": self._get_reauth_entry().title,
"host": entry_data[CONF_HOST],
}
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauthentication confirmation."""
errors: dict[str, str] = {}
reauth_entry = self._get_reauth_entry()
host = reauth_entry.data[CONF_HOST]
port = reauth_entry.data[CONF_PORT]
if user_input is not None:
username = user_input.get(CONF_USERNAME)
password = user_input.get(CONF_PASSWORD)
websession = get_maybe_authenticated_session(self.hass, password, username)
api_host = ApiHost(
host, port, DEFAULT_SETUP_TIMEOUT, websession, self.hass.loop, _LOGGER
)
try:
await Box.async_from_host(api_host)
except UnauthorizedRequest:
errors["base"] = INVALID_AUTH
except Error:
errors["base"] = CANNOT_CONNECT
except RuntimeError:
errors["base"] = UNKNOWN
else:
return self.async_update_reload_and_abort(
reauth_entry,
data_updates={
CONF_USERNAME: username,
CONF_PASSWORD: password,
},
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(
{
vol.Inclusive(CONF_USERNAME, "auth"): str,
vol.Inclusive(CONF_PASSWORD, "auth"): str,
}
),
errors=errors,
description_placeholders={"address": f"{host}:{port}"},
)
+1
View File
@@ -7,6 +7,7 @@ DEFAULT_SETUP_TIMEOUT = 10
# translation strings
ADDRESS_ALREADY_CONFIGURED = "address_already_configured"
CANNOT_CONNECT = "cannot_connect"
INVALID_AUTH = "invalid_auth"
UNSUPPORTED_VERSION = "unsupported_version"
UNKNOWN = "unknown"
@@ -4,10 +4,11 @@ from datetime import timedelta
import logging
from blebox_uniapi.box import Box
from blebox_uniapi.error import Error
from blebox_uniapi.error import Error, UnauthorizedRequest
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
@@ -40,6 +41,8 @@ class BleBoxCoordinator(DataUpdateCoordinator[None]):
"""Fetch data from the BleBox device."""
try:
await self.box.async_update_data()
except UnauthorizedRequest as err:
raise ConfigEntryAuthFailed from err
except Error as err:
raise UpdateFailed(
translation_domain=DOMAIN,
+15
View File
@@ -24,6 +24,7 @@ from homeassistant.const import (
UnitOfEnergy,
UnitOfFrequency,
UnitOfPower,
UnitOfReactiveEnergy,
UnitOfReactivePower,
UnitOfSpeed,
UnitOfTemperature,
@@ -97,6 +98,20 @@ SENSOR_TYPES: tuple[BleBoxSensorEntityDescription, ...] = (
native_unit_of_measurement=LIGHT_LUX,
state_class=SensorStateClass.MEASUREMENT,
),
BleBoxSensorEntityDescription(
key="forwardReactiveEnergy",
translation_key="forward_reactive_energy",
device_class=SensorDeviceClass.REACTIVE_ENERGY,
native_unit_of_measurement=UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
),
BleBoxSensorEntityDescription(
key="reverseReactiveEnergy",
translation_key="reverse_reactive_energy",
device_class=SensorDeviceClass.REACTIVE_ENERGY,
native_unit_of_measurement=UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
),
BleBoxSensorEntityDescription(
key="forwardActiveEnergy",
translation_key="forward_active_energy",
+26 -1
View File
@@ -3,16 +3,33 @@
"abort": {
"address_already_configured": "A BleBox device is already configured at {address}.",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"authorization_required": "The BleBox device requires authentication.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unique_id_mismatch": "The device identifier does not match the previously configured device."
"unique_id_mismatch": "The device identifier does not match the previously configured device.",
"unsupported_device_response": "The BleBox device returned an unrecognized response.",
"unsupported_device_version": "[%key:component::blebox::config::error::unsupported_version%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"unsupported_version": "BleBox device has outdated firmware. Please upgrade it first."
},
"flow_title": "{name} ({host})",
"step": {
"reauth_confirm": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "The password for your BleBox device.",
"username": "The username for your BleBox device."
},
"description": "Enter credentials for the BleBox device at {address}.",
"title": "Reauthenticate your BleBox device"
},
"reconfigure": {
"data": {
"host": "[%key:common::config_flow::data::ip%]",
@@ -64,6 +81,10 @@
"current_n": { "name": "Current {index}" },
"forward_active_energy": { "name": "Forward active energy" },
"forward_active_energy_n": { "name": "Forward active energy {index}" },
"forward_reactive_energy": { "name": "Forward reactive energy" },
"forward_reactive_energy_n": {
"name": "Forward reactive energy {index}"
},
"frequency": { "name": "Frequency" },
"frequency_n": { "name": "Frequency {index}" },
"open_status": {
@@ -81,6 +102,10 @@
"reactive_power_n": { "name": "Reactive power {index}" },
"reverse_active_energy": { "name": "Reverse active energy" },
"reverse_active_energy_n": { "name": "Reverse active energy {index}" },
"reverse_reactive_energy": { "name": "Reverse reactive energy" },
"reverse_reactive_energy_n": {
"name": "Reverse reactive energy {index}"
},
"temperature": { "name": "Temperature" },
"temperature_n": { "name": "Temperature {index}" },
"voltage": { "name": "Voltage" },
@@ -31,12 +31,24 @@ async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
data["relayer_region"] = client.relayer_region
data["remote_enabled"] = client.prefs.remote_enabled
data["remote_connected"] = cloud.remote.is_connected
data["remote_server"] = cloud.remote.snitun_server
data["alexa_enabled"] = client.prefs.alexa_enabled
data["google_enabled"] = client.prefs.google_enabled
data["cloud_ice_servers_enabled"] = client.prefs.cloud_ice_servers_enabled
data["remote_server"] = cloud.remote.snitun_server
data["certificate_status"] = cloud.remote.certificate_status
data["instance_id"] = client.prefs.instance_id
data["iot_state"] = cloud.iot.state
data["iot_tries"] = cloud.iot.tries
if (cert := cloud.remote.certificate) is not None:
data["certificate_expire_date"] = cert.expire_date
data["certificate_fingerprint"] = cert.fingerprint
if cert.alternative_names:
data["certificate_alternative_names"] = cert.alternative_names
if (disconnect := cloud.iot.last_disconnect_reason) is not None:
data["iot_last_disconnect_clean"] = disconnect.clean
data["iot_last_disconnect_reason"] = disconnect.reason
data["can_reach_cert_server"] = system_health.async_check_can_reach_url(
hass, f"https://{cloud.acme_server}/directory"
@@ -7,6 +7,6 @@
"integration_type": "device",
"iot_class": "local_push",
"quality_scale": "bronze",
"requirements": ["pydroplet==2.3.4"],
"requirements": ["pydroplet==2.4.0"],
"zeroconf": ["_droplet._tcp.local."]
}
+7 -2
View File
@@ -9,10 +9,12 @@ import time
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urlparse
import uuid
import warnings
from aiohttp import ClientError, ClientResponse, ClientSession, web
from aiohttp.hdrs import AUTHORIZATION
import jwt
from jwt.warnings import InsecureKeyLengthWarning
from py_vapid import Vapid
from pywebpush import WebPusher, WebPushException, webpush_async
import voluptuous as vol
@@ -325,7 +327,8 @@ class HTML5PushCallbackView(HomeAssistantView):
if target_check.get(ATTR_TARGET) in self.registrations:
possible_target = self.registrations[target_check[ATTR_TARGET]]
key = possible_target["subscription"]["keys"]["auth"]
with suppress(jwt.exceptions.DecodeError):
with suppress(jwt.exceptions.DecodeError), warnings.catch_warnings():
warnings.simplefilter("ignore", InsecureKeyLengthWarning)
return jwt.decode(token, key, algorithms=["ES256", "HS256"])
return self.json_message(
@@ -585,7 +588,9 @@ def add_jwt(timestamp: int, target: str, tag: str, jwt_secret: str) -> str:
ATTR_TARGET: target,
ATTR_TAG: tag,
}
return jwt.encode(jwt_claims, jwt_secret)
with warnings.catch_warnings():
warnings.simplefilter("ignore", InsecureKeyLengthWarning)
return jwt.encode(jwt_claims, jwt_secret)
async def async_setup_entry(
@@ -248,7 +248,10 @@ def _generate_thumbnail_if_file_does_not_exist(
if not target_file.is_file():
image = ImageOps.exif_transpose(Image.open(original_path))
image.thumbnail(target_size)
image.save(target_path, format=content_type.partition("/")[-1])
save_format = content_type.partition("/")[-1]
if save_format == "jpeg" and image.mode not in ("RGB", "L", "CMYK"):
image = image.convert("RGB")
image.save(target_path, format=save_format)
def _validate_size_from_filename(filename: str) -> tuple[int, int]:
@@ -119,7 +119,7 @@ class ImmichDataUpdateCoordinator(DataUpdateCoordinator[ImmichData]):
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error",
translation_placeholders={"error": repr(err)},
translation_placeholders={"error": str(err)},
) from err
return ImmichData(
@@ -1,5 +1,6 @@
"""Support for LG TV running on NetCast 3 or 4."""
from collections import Counter
from datetime import datetime
from typing import TYPE_CHECKING, Any
@@ -133,13 +134,22 @@ class LgTVDevice(MediaPlayerEntity):
channel_list = client.query_data("channel_list")
if channel_list:
channel_names = []
channel_pairs = []
for channel in channel_list:
channel_name = channel.find("chname")
if channel_name is not None:
channel_names.append(str(channel_name.text))
self._sources = dict(zip(channel_names, channel_list, strict=False))
# sort source names by the major channel number
channel_pairs.append((str(channel_name.text), channel))
name_count = Counter(name for name, _ in channel_pairs)
self._sources = {}
for name, channel in channel_pairs:
if name_count[name] > 1:
major = channel.find("major")
if major is not None:
name = f"{name} ({major.text})"
self._sources[name] = channel
source_tuples = [
(k, source.find("major").text)
for k, source in self._sources.items()
@@ -13,7 +13,7 @@
"integration_type": "hub",
"iot_class": "local_polling",
"loggers": ["boto3", "botocore", "pyoverkiz", "s3transfer"],
"requirements": ["pyoverkiz[nexity]==2.0.0"],
"requirements": ["pyoverkiz[nexity]==2.0.1"],
"zeroconf": [
{
"name": "gateway*",
@@ -31,7 +31,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: YaleConfigEntry) -> boo
async def async_migrate_entry(hass: HomeAssistant, entry: YaleConfigEntry) -> bool:
"""Migrate old entry."""
LOGGER.debug("Migrating from version %s", entry.version)
LOGGER.debug("Migrating from version %s.%s", entry.version, entry.minor_version)
if entry.version == 1:
new_options = entry.options.copy()
@@ -55,6 +55,19 @@ async def async_migrate_entry(hass: HomeAssistant, entry: YaleConfigEntry) -> bo
del new_data[CONF_NAME]
hass.config_entries.async_update_entry(entry, data=new_data, minor_version=2)
LOGGER.debug("Migration to version %s successful", entry.version)
if entry.version == 2 and entry.minor_version == 2:
entity_reg = er.async_get(hass)
entries = er.async_entries_for_config_entry(entity_reg, entry.entry_id)
for entity in entries:
if entity.unique_id == "yale_smart_alarm-panic":
entity_reg.async_update_entity(
entity.entity_id,
new_unique_id=f"{entry.entry_id}-panic",
)
hass.config_entries.async_update_entry(entry, minor_version=3)
LOGGER.debug(
"Migration to version %s.%s successful", entry.version, entry.minor_version
)
return True
@@ -47,7 +47,7 @@ class YalePanicButton(YaleAlarmEntity, ButtonEntity):
"""Initialize the plug switch."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"yale_smart_alarm-{description.key}"
self._attr_unique_id = f"{coordinator.config_entry.entry_id}-{description.key}"
async def async_press(self) -> None:
"""Press the button."""
@@ -64,7 +64,7 @@ class YaleConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Yale integration."""
VERSION = 2
MINOR_VERSION = 2
MINOR_VERSION = 3
@staticmethod
@callback
+1 -1
View File
@@ -19,7 +19,7 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
from .const import DOMAIN
from .coordinator import YotoConfigEntry, YotoDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER]
PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: YotoConfigEntry) -> bool:
@@ -148,8 +148,6 @@ class YotoDataUpdateCoordinator(DataUpdateCoordinator[dict[str, YotoPlayer]]):
"""Ask each player to push a fresh status snapshot over MQTT."""
if not self.client.is_mqtt_connected:
return
# Fire-and-forget: the data/status response lands via the on_update
# callback later, which already triggers async_set_updated_data.
for device_id in list(self.client.players):
await self.client.request_player_status(device_id)
+5 -1
View File
@@ -43,4 +43,8 @@ class YotoEntity(CoordinatorEntity[YotoDataUpdateCoordinator]):
@property
def available(self) -> bool:
"""Return if the entity is available."""
return super().available and self._player_id in self.coordinator.data
return (
super().available
and self._player_id in self.coordinator.data
and bool(self.player.is_online)
)
+22
View File
@@ -0,0 +1,22 @@
{
"entity": {
"sensor": {
"card_insertion_state": {
"default": "mdi:card-bulleted-outline",
"state": {
"none": "mdi:card-bulleted-off-outline",
"physical": "mdi:card-bulleted",
"remote": "mdi:cast-audio",
"streaming": "mdi:radio-tower"
}
},
"day_mode": {
"default": "mdi:theme-light-dark",
"state": {
"day": "mdi:weather-sunny",
"night": "mdi:weather-night"
}
}
}
}
}
+1 -1
View File
@@ -10,5 +10,5 @@
"iot_class": "cloud_push",
"loggers": ["yoto_api"],
"quality_scale": "bronze",
"requirements": ["yoto-api==4.1.0"]
"requirements": ["yoto-api==4.2.0"]
}
@@ -82,11 +82,6 @@ class YotoMediaPlayer(YotoEntity, MediaPlayerEntity):
super().__init__(coordinator, player)
self._attr_unique_id = player.id
@property
def available(self) -> bool:
"""Return whether the player is reachable through the Yoto cloud."""
return super().available and bool(self.player.is_online)
@property
def state(self) -> MediaPlayerState:
"""Return the playback state."""
@@ -57,20 +57,14 @@ rules:
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category:
status: exempt
comment: Only the media_player entity ships in this PR; no diagnostic entities yet.
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: Only the media_player entity ships in this PR; no entities are disabled by default.
entity-translations:
status: exempt
comment: The media_player uses the device name; no translatable strings yet.
comment: No noisy or less popular entities; nothing is disabled by default.
entity-translations: done
exception-translations: done
icon-translations:
status: exempt
comment: No custom icon translations are needed yet.
icon-translations: done
reconfiguration-flow:
status: exempt
comment: Authorization is the only configuration; reauth covers re-linking the account.
+103
View File
@@ -0,0 +1,103 @@
"""Sensor platform for the Yoto integration."""
from collections.abc import Callable
from dataclasses import dataclass
from yoto_api import CardInsertionState, DayMode, YotoPlayer
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import PERCENTAGE, EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .coordinator import YotoConfigEntry, YotoDataUpdateCoordinator
from .entity import YotoEntity
PARALLEL_UPDATES = 0
def _enum_state(value: CardInsertionState | None) -> str | None:
"""Return an enum member as a lowercase string, or None if unset."""
return value.name.lower() if value is not None else None
def _day_mode_state(value: DayMode | None) -> str | None:
"""Return day/night, treating the firmware's UNKNOWN as unset."""
if value is None or value is DayMode.UNKNOWN:
return None
return value.name.lower()
@dataclass(frozen=True, kw_only=True)
class YotoSensorEntityDescription(SensorEntityDescription):
"""Describes a Yoto sensor entity."""
value_fn: Callable[[YotoPlayer], StateType]
SENSORS: tuple[YotoSensorEntityDescription, ...] = (
YotoSensorEntityDescription(
key="battery_level",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda player: player.status.battery_level_percentage,
),
YotoSensorEntityDescription(
key="card_insertion_state",
translation_key="card_insertion_state",
device_class=SensorDeviceClass.ENUM,
options=[state.name.lower() for state in CardInsertionState],
value_fn=lambda player: _enum_state(player.status.card_insertion_state),
),
YotoSensorEntityDescription(
key="day_mode",
translation_key="day_mode",
device_class=SensorDeviceClass.ENUM,
options=["day", "night"],
value_fn=lambda player: _day_mode_state(player.status.day_mode),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: YotoConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Yoto sensor platform."""
coordinator = entry.runtime_data
async_add_entities(
YotoSensor(coordinator, player, description)
for player in coordinator.client.players.values()
for description in SENSORS
)
class YotoSensor(YotoEntity, SensorEntity):
"""Representation of a Yoto player sensor."""
entity_description: YotoSensorEntityDescription
def __init__(
self,
coordinator: YotoDataUpdateCoordinator,
player: YotoPlayer,
description: YotoSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, player)
self.entity_description = description
self._attr_unique_id = f"{player.id}_{description.key}"
@property
def native_value(self) -> StateType:
"""Return the sensor value."""
return self.entity_description.value_fn(self.player)
@@ -36,6 +36,26 @@
}
}
},
"entity": {
"sensor": {
"card_insertion_state": {
"name": "Card slot",
"state": {
"none": "Empty",
"physical": "Physical card",
"remote": "Remote",
"streaming": "Streaming"
}
},
"day_mode": {
"name": "Day mode",
"state": {
"day": "Day",
"night": "Night"
}
}
}
},
"exceptions": {
"authentication_failed": {
"message": "Yoto credentials are no longer valid. Please reauthenticate your account."
+4 -4
View File
@@ -700,7 +700,7 @@ def _get_exposed_entities(
):
# Entity is in area
area_names.append(area_entry.name)
area_names.extend(area_entry.aliases)
area_names.extend(sorted(area_entry.aliases))
elif device_entry is not None:
# Check device area
if (
@@ -711,7 +711,7 @@ def _get_exposed_entities(
is not None
):
area_names.append(area_entry.name)
area_names.extend(area_entry.aliases)
area_names.extend(sorted(area_entry.aliases))
info: dict[str, Any] = {
"names": ", ".join(names),
@@ -962,9 +962,9 @@ def _get_cached_action_parameters(
aliases = er.async_get_entity_aliases(hass, entity_entry)
if aliases:
if description:
description = description + ". Aliases: " + str(list(aliases))
description = description + ". Aliases: " + str(sorted(aliases))
else:
description = "Aliases: " + str(list(aliases))
description = "Aliases: " + str(sorted(aliases))
parameters_cache.setdefault(domain, {})[action] = (description, parameters)
@@ -1,6 +1,7 @@
"""Config entry functions for Home Assistant templates."""
from collections.abc import Iterable
from enum import Enum
from typing import TYPE_CHECKING, Any
from homeassistant.exceptions import TemplateError
@@ -104,4 +105,6 @@ class ConfigEntryExtension(BaseTemplateExtension):
if config_entry is None:
return None
return getattr(config_entry, attr_name)
if isinstance(result := getattr(config_entry, attr_name), Enum):
return result.value
return result
+3 -3
View File
@@ -2120,7 +2120,7 @@ pydrawise==2026.4.0
pydroid-ipcam==3.0.0
# homeassistant.components.droplet
pydroplet==2.3.4
pydroplet==2.4.0
# homeassistant.components.ebox
pyebox==1.1.4
@@ -2438,7 +2438,7 @@ pyotgw==2.2.3
pyotp==2.9.0
# homeassistant.components.overkiz
pyoverkiz[nexity]==2.0.0
pyoverkiz[nexity]==2.0.1
# homeassistant.components.palazzetti
pypalazzetti==0.1.20
@@ -3436,7 +3436,7 @@ yeelightsunflower==0.0.10
yolink-api==0.6.5
# homeassistant.components.yoto
yoto-api==4.1.0
yoto-api==4.2.0
# homeassistant.components.youless
youless-api==2.2.0
+151 -43
View File
@@ -27,6 +27,20 @@ from .conftest import (
from tests.common import MockConfigEntry
@pytest.fixture(name="zeroconf_data")
def zeroconf_data_fixture() -> ZeroconfServiceInfo:
"""Return ZeroconfServiceInfo for a BleBox device."""
return ZeroconfServiceInfo(
ip_address=ip_address("172.100.123.4"),
ip_addresses=[ip_address("172.100.123.4")],
port=80,
hostname="bbx-bbtest123456.local.",
type="_bbxsrv._tcp.local.",
name="bbx-bbtest123456._bbxsrv._tcp.local.",
properties={"_raw": {}},
)
def create_valid_feature_mock(path="homeassistant.components.blebox.Products"):
"""Return a valid, complete BleBox feature mock."""
feature = mock_only_feature(
@@ -173,7 +187,7 @@ async def test_flow_with_auth_failure(hass: HomeAssistant, product_class_mock) -
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["errors"] == {"base": "cannot_connect"}
assert result["errors"] == {"base": "invalid_auth"}
async def test_async_setup(hass: HomeAssistant) -> None:
@@ -225,20 +239,14 @@ async def test_async_remove_entry(
assert config_entry.state is ConfigEntryState.NOT_LOADED
async def test_flow_with_zeroconf(hass: HomeAssistant) -> None:
async def test_flow_with_zeroconf(
hass: HomeAssistant, zeroconf_data: ZeroconfServiceInfo
) -> None:
"""Test setup from zeroconf discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
ip_address=ip_address("172.100.123.4"),
ip_addresses=[ip_address("172.100.123.4")],
port=80,
hostname="bbx-bbtest123456.local.",
type="_bbxsrv._tcp.local.",
name="bbx-bbtest123456._bbxsrv._tcp.local.",
properties={"_raw": {}},
),
data=zeroconf_data,
)
assert result["type"] is FlowResultType.FORM
@@ -252,7 +260,9 @@ async def test_flow_with_zeroconf(hass: HomeAssistant) -> None:
async def test_flow_with_zeroconf_when_already_configured(
hass: HomeAssistant, config_entry: MockConfigEntry
hass: HomeAssistant,
config_entry: MockConfigEntry,
zeroconf_data: ZeroconfServiceInfo,
) -> None:
"""Test behaviour if device already configured."""
config_entry.add_to_hass(hass)
@@ -267,22 +277,16 @@ async def test_flow_with_zeroconf_when_already_configured(
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
ip_address=ip_address("172.100.123.4"),
ip_addresses=[ip_address("172.100.123.4")],
port=80,
hostname="bbx-bbtest123456.local.",
type="_bbxsrv._tcp.local.",
name="bbx-bbtest123456._bbxsrv._tcp.local.",
properties={"_raw": {}},
),
data=zeroconf_data,
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "already_configured"
async def test_flow_with_zeroconf_when_device_unsupported(hass: HomeAssistant) -> None:
async def test_flow_with_zeroconf_when_device_unsupported(
hass: HomeAssistant, zeroconf_data: ZeroconfServiceInfo
) -> None:
"""Test behaviour when device is not supported."""
with patch(
"homeassistant.components.blebox.config_flow.Box.async_from_host",
@@ -291,25 +295,16 @@ async def test_flow_with_zeroconf_when_device_unsupported(hass: HomeAssistant) -
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
ip_address=ip_address("172.100.123.4"),
ip_addresses=[ip_address("172.100.123.4")],
port=80,
hostname="bbx-bbtest123456.local.",
type="_bbxsrv._tcp.local.",
name="bbx-bbtest123456._bbxsrv._tcp.local.",
properties={"_raw": {}},
),
data=zeroconf_data,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unsupported_device_version"
async def test_flow_with_zeroconf_when_device_response_unsupported(
hass: HomeAssistant,
hass: HomeAssistant, zeroconf_data: ZeroconfServiceInfo
) -> None:
"""Test behaviour when device returned unsupported response."""
with patch(
"homeassistant.components.blebox.config_flow.Box.async_from_host",
side_effect=blebox_uniapi.error.UnsupportedBoxResponse,
@@ -317,20 +312,29 @@ async def test_flow_with_zeroconf_when_device_response_unsupported(
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
ip_address=ip_address("172.100.123.4"),
ip_addresses=[ip_address("172.100.123.4")],
port=80,
hostname="bbx-bbtest123456.local.",
type="_bbxsrv._tcp.local.",
name="bbx-bbtest123456._bbxsrv._tcp.local.",
properties={"_raw": {}},
),
data=zeroconf_data,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unsupported_device_response"
async def test_flow_with_zeroconf_when_unauthorized(
hass: HomeAssistant, zeroconf_data: ZeroconfServiceInfo
) -> None:
"""Test behaviour when device requires authentication during zeroconf discovery."""
with patch(
"homeassistant.components.blebox.config_flow.Box.async_from_host",
side_effect=blebox_uniapi.error.UnauthorizedRequest,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=zeroconf_data,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "authorization_required"
def create_product_mock(unique_id: str = "abcd0123ef5678"):
"""Return a product mock with a given unique_id."""
product = create_autospec(blebox_uniapi.box.Box, True, True)
@@ -398,7 +402,7 @@ async def test_reconfigure_flow_unique_id_mismatch(
[
pytest.param(blebox_uniapi.error.Error, "cannot_connect", id="api_error"),
pytest.param(
blebox_uniapi.error.UnauthorizedRequest, "cannot_connect", id="auth_failure"
blebox_uniapi.error.UnauthorizedRequest, "invalid_auth", id="auth_failure"
),
pytest.param(
blebox_uniapi.error.UnsupportedBoxVersion,
@@ -442,3 +446,107 @@ async def test_reconfigure_flow_recovers_after_error(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
async def test_reauth_flow_works(
hass: HomeAssistant, config_entry: MockConfigEntry, product_class_mock
) -> None:
"""Test that reauth flow updates credentials and reloads."""
config_entry.add_to_hass(hass)
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with product_class_mock as box_class:
box_class.async_from_host = AsyncMock(
return_value=create_product_mock("abcd0123ef5678")
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{config_flow.CONF_USERNAME: "admin", config_flow.CONF_PASSWORD: "secret"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert config_entry.data[config_flow.CONF_USERNAME] == "admin"
assert config_entry.data[config_flow.CONF_PASSWORD] == "secret"
async def test_reauth_flow_works_without_credentials(
hass: HomeAssistant, product_class_mock
) -> None:
"""Test that reauth flow clears credentials when submitted without them."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
config_flow.CONF_HOST: "172.100.123.4",
config_flow.CONF_PORT: 80,
config_flow.CONF_USERNAME: "admin",
config_flow.CONF_PASSWORD: "secret",
},
)
config_entry.add_to_hass(hass)
result = await config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
with product_class_mock as box_class:
box_class.async_from_host = AsyncMock(
return_value=create_product_mock("abcd0123ef5678")
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert config_entry.data[config_flow.CONF_USERNAME] is None
assert config_entry.data[config_flow.CONF_PASSWORD] is None
@pytest.mark.parametrize(
("exception", "expected_error"),
[
pytest.param(
blebox_uniapi.error.UnauthorizedRequest, "invalid_auth", id="auth_failure"
),
pytest.param(blebox_uniapi.error.Error, "cannot_connect", id="api_error"),
pytest.param(RuntimeError, "unknown", id="runtime_error"),
],
)
async def test_reauth_flow_recovers_after_error(
hass: HomeAssistant,
config_entry: MockConfigEntry,
product_class_mock,
exception: type[Exception],
expected_error: str,
) -> None:
"""Test that reauth shows the correct error and allows a successful retry."""
config_entry.add_to_hass(hass)
result = await config_entry.start_reauth_flow(hass)
with product_class_mock as box_class:
box_class.async_from_host = AsyncMock(side_effect=exception)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{config_flow.CONF_USERNAME: "admin", config_flow.CONF_PASSWORD: "secret"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": expected_error}
with product_class_mock as box_class:
box_class.async_from_host = AsyncMock(
return_value=create_product_mock("abcd0123ef5678")
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{config_flow.CONF_USERNAME: "admin", config_flow.CONF_PASSWORD: "secret"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
+24 -20
View File
@@ -1,7 +1,5 @@
"""BleBox devices setup tests."""
import logging
import blebox_uniapi
import pytest
@@ -17,36 +15,42 @@ from .conftest import (
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("exception", "expected_state"),
[
(blebox_uniapi.error.ConnectionError, ConfigEntryState.SETUP_RETRY),
(blebox_uniapi.error.HttpError, ConfigEntryState.SETUP_RETRY),
(blebox_uniapi.error.UnsupportedBoxVersion, ConfigEntryState.SETUP_ERROR),
(blebox_uniapi.error.UnsupportedBoxResponse, ConfigEntryState.SETUP_ERROR),
(blebox_uniapi.error.UnauthorizedRequest, ConfigEntryState.SETUP_ERROR),
(blebox_uniapi.error.Error, ConfigEntryState.SETUP_ERROR),
],
)
async def test_setup_failure(
hass: HomeAssistant,
config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
exception: type[Exception],
expected_state: ConfigEntryState,
) -> None:
"""Test that setup failure is handled and logged."""
patch_product_identify(None, side_effect=blebox_uniapi.error.ClientError)
caplog.set_level(logging.ERROR)
"""Test that setup failures map to the correct config entry state."""
patch_product_identify(None, side_effect=exception)
await async_setup_config_entry(hass, config_entry)
assert "Identify failed at 172.100.123.4:80 ()" in caplog.text
assert config_entry.state is ConfigEntryState.SETUP_RETRY
assert config_entry.state is expected_state
async def test_setup_failure_on_connection(
async def test_setup_auth_failure_triggers_reauth(
hass: HomeAssistant,
config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that setup failure is handled and logged."""
patch_product_identify(None, side_effect=blebox_uniapi.error.ConnectionError)
caplog.set_level(logging.ERROR)
"""Test that UnauthorizedRequest during setup triggers a reauth flow."""
patch_product_identify(None, side_effect=blebox_uniapi.error.UnauthorizedRequest)
await async_setup_config_entry(hass, config_entry)
assert "Identify failed at 172.100.123.4:80 ()" in caplog.text
assert config_entry.state is ConfigEntryState.SETUP_RETRY
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert any(
f["handler"] == "blebox" and f["context"]["source"] == "reauth" for f in flows
)
async def test_unload_config_entry(
+1 -1
View File
@@ -70,7 +70,7 @@ async def cloud_fixture() -> AsyncGenerator[MagicMock]:
)
mock_cloud.auth = MagicMock(spec=CognitoAuth)
mock_cloud.iot = MagicMock(
spec=CloudIoT, last_disconnect_reason=None, state=STATE_CONNECTED
spec=CloudIoT, last_disconnect_reason=None, state=STATE_CONNECTED, tries=0
)
mock_cloud.voice = MagicMock(spec=Voice)
mock_cloud.files = MagicMock(spec=Files)
@@ -88,12 +88,14 @@
relayer_region | xx-earth-616
remote_enabled | True
remote_connected | False
remote_server | us-west-1
alexa_enabled | True
google_enabled | False
cloud_ice_servers_enabled | True
remote_server | us-west-1
certificate_status | ready
instance_id | 12345678901234567890
iot_state | connected
iot_tries | 0
can_reach_cert_server | Exception: Unexpected exception
can_reach_cloud_auth | Failed: unreachable
can_reach_cloud | ok
@@ -204,12 +206,14 @@
relayer_region | xx-earth-616
remote_enabled | True
remote_connected | False
remote_server | us-west-1
alexa_enabled | True
google_enabled | False
cloud_ice_servers_enabled | True
remote_server | us-west-1
certificate_status | ready
instance_id | 12345678901234567890
iot_state | connected
iot_tries | 0
can_reach_cert_server | Exception: Unexpected exception
can_reach_cloud_auth | Failed: unreachable
can_reach_cloud | ok
@@ -284,12 +288,14 @@
relayer_region | xx-earth-616
remote_enabled | True
remote_connected | False
remote_server | us-west-1
alexa_enabled | True
google_enabled | False
cloud_ice_servers_enabled | True
remote_server | us-west-1
certificate_status | ready
instance_id | 12345678901234567890
iot_state | connected
iot_tries | 0
can_reach_cert_server | Exception: Unexpected exception
can_reach_cloud_auth | Failed: unreachable
can_reach_cloud | ok
+71 -1
View File
@@ -2,15 +2,18 @@
import asyncio
from collections.abc import Callable, Coroutine
from datetime import datetime
from typing import Any
from unittest.mock import MagicMock
from aiohttp import ClientError
from hass_nabucasa.remote import CertificateStatus
from hass_nabucasa.iot_base import DisconnectReason
from hass_nabucasa.remote import Certificate, CertificateStatus
from homeassistant.components.cloud.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import UTC
from tests.common import get_system_health_info
from tests.test_util.aiohttp import AiohttpClientMocker
@@ -47,6 +50,11 @@ async def test_cloud_system_health(
cloud.remote.snitun_server = "us-west-1"
cloud.remote.certificate_status = CertificateStatus.READY
cloud.remote.certificate = None
cloud.remote.latency_by_location = {}
cloud.iot.state = "connected"
cloud.iot.tries = 0
cloud.iot.last_disconnect_reason = None
await cloud.client.async_system_message({"region": "xx-earth-616"})
await set_cloud_prefs(
@@ -80,4 +88,66 @@ async def test_cloud_system_health(
"can_reach_cloud_auth": {"type": "failed", "error": "unreachable"},
"can_reach_cloud": "ok",
"instance_id": cloud.client.prefs.instance_id,
"iot_state": "connected",
"iot_tries": 0,
}
async def test_cloud_system_health_with_cert_and_disconnect(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
cloud: MagicMock,
set_cloud_prefs: Callable[[dict[str, Any]], Coroutine[Any, Any, None]],
) -> None:
"""Test cloud system health with certificate details and disconnect reason."""
aioclient_mock.get("https://cloud.bla.com/status", text="")
aioclient_mock.get("https://cert-server/directory", text="")
aioclient_mock.get(
"https://cognito-idp.us-east-1.amazonaws.com/AAAA/.well-known/jwks.json",
text="",
)
assert await async_setup_component(hass, "system_health", {})
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
"user_pool_id": "AAAA",
"region": "us-east-1",
"acme_server": "cert-server",
"relayer_server": "cloud.bla.com",
},
},
)
await hass.async_block_till_done()
await cloud.login("test-user", "test-pass")
expire = datetime(2026, 8, 1, tzinfo=UTC)
cloud.remote.snitun_server = "eu-central-1"
cloud.remote.certificate_status = CertificateStatus.READY
cloud.remote.certificate = Certificate(
common_name="my-home.ui.nabu.casa",
expire_date=expire,
fingerprint="abc123def456",
alternative_names=["custom.example.com"],
)
cloud.iot.state = "connected"
cloud.iot.tries = 2
cloud.iot.last_disconnect_reason = DisconnectReason(
clean=False, reason="ping_timeout"
)
await set_cloud_prefs({"remote_enabled": True})
info = await get_system_health_info(hass, "cloud")
for key, val in info.items():
if asyncio.iscoroutine(val):
info[key] = await val
assert info["certificate_expire_date"] == expire
assert info["certificate_fingerprint"] == "abc123def456"
assert info["certificate_alternative_names"] == ["custom.example.com"]
assert info["iot_last_disconnect_clean"] is False
assert info["iot_last_disconnect_reason"] == "ping_timeout"
assert info["iot_tries"] == 2
+39 -1
View File
@@ -4,11 +4,15 @@ from asyncio import Event
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from govee_local_api import GoveeLightCapabilities, GoveeLightFeatures
from govee_local_api import GoveeDevice, GoveeLightCapabilities, GoveeLightFeatures
from govee_local_api.light_capabilities import COMMON_FEATURES, SCENE_CODES
import pytest
from homeassistant.components.govee_light_local.const import DOMAIN
from homeassistant.components.govee_light_local.coordinator import GoveeController
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture(name="mock_govee_api")
@@ -56,3 +60,37 @@ SCENE_CAPABILITIES: GoveeLightCapabilities = GoveeLightCapabilities(
segments=[],
scenes=SCENE_CODES,
)
async def setup_light(
hass: HomeAssistant,
mock_govee_api: AsyncMock,
capabilities: GoveeLightCapabilities = DEFAULT_CAPABILITIES,
*,
ip: str = "192.168.1.100",
fingerprint: str = "asdawdqwdqwd",
sku: str = "H615A",
) -> tuple[MockConfigEntry, GoveeDevice]:
"""Set up a single mocked Govee light device and return its entry and device.
The returned tuple lets tests that need to mutate the device after setup
(e.g. ``device.update(...)`` in availability tests) access the underlying
``GoveeDevice`` directly. Tests that only need the entry or neither can
discard the unused half with ``_``.
"""
device = GoveeDevice(
controller=mock_govee_api,
ip=ip,
fingerprint=fingerprint,
sku=sku,
capabilities=capabilities,
)
mock_govee_api.devices = [device]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry, device
+38 -235
View File
@@ -36,7 +36,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from .conftest import DEFAULT_CAPABILITIES, SCENE_CAPABILITIES
from .conftest import DEFAULT_CAPABILITIES, SCENE_CAPABILITIES, setup_light
from tests.common import MockConfigEntry, async_fire_time_changed
@@ -45,22 +45,7 @@ async def test_light_known_device(
hass: HomeAssistant, mock_govee_api: AsyncMock
) -> None:
"""Test adding a known device."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entry, _ = await setup_light(hass, mock_govee_api)
assert len(hass.states.async_all()) == 1
@@ -80,22 +65,14 @@ async def test_light_unknown_device(
hass: HomeAssistant, mock_govee_api: AsyncMock
) -> None:
"""Test adding an unknown device."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.101",
fingerprint="unkown_device",
sku="XYZK",
capabilities=ON_OFF_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await setup_light(
hass,
mock_govee_api,
ON_OFF_CAPABILITIES,
ip="192.168.1.101",
fingerprint="unkown_device",
sku="XYZK",
)
assert len(hass.states.async_all()) == 1
@@ -107,22 +84,8 @@ async def test_light_unknown_device(
async def test_light_remove(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
"""Test remove device."""
entry, _ = await setup_light(hass, mock_govee_api, fingerprint="asdawdqwdqwd1")
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd1",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("light.H615A") is not None
# Remove 1
@@ -199,22 +162,7 @@ async def test_light_setup_error(
async def test_light_on_off(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
"""Test light on and then off."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api)
assert len(hass.states.async_all()) == 1
@@ -233,7 +181,7 @@ async def test_light_on_off(hass: HomeAssistant, mock_govee_api: AsyncMock) -> N
light = hass.states.get("light.H615A")
assert light is not None
assert light.state == "on"
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], True)
mock_govee_api.turn_on_off.assert_awaited_with(device, True)
# Turn off
await hass.services.async_call(
@@ -247,7 +195,7 @@ async def test_light_on_off(hass: HomeAssistant, mock_govee_api: AsyncMock) -> N
light = hass.states.get("light.H615A")
assert light is not None
assert light.state == "off"
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], False)
mock_govee_api.turn_on_off.assert_awaited_with(device, False)
@pytest.mark.parametrize(
@@ -280,21 +228,7 @@ async def test_turn_on_call_order(
mock_call_kwargs: dict[str, Any],
) -> None:
"""Test that turn_on is called after set_brightness/set_color/set_preset."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=SCENE_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api, SCENE_CAPABILITIES)
assert len(hass.states.async_all()) == 1
@@ -312,32 +246,16 @@ async def test_turn_on_call_order(
mock_govee_api.assert_has_calls(
[
call.set_brightness(mock_govee_api.devices[0], 50),
getattr(call, mock_call)(
mock_govee_api.devices[0], *mock_call_args, **mock_call_kwargs
),
call.turn_on_off(mock_govee_api.devices[0], True),
call.set_brightness(device, 50),
getattr(call, mock_call)(device, *mock_call_args, **mock_call_kwargs),
call.turn_on_off(device, True),
]
)
async def test_light_brightness(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
"""Test changing brightness."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api)
assert len(hass.states.async_all()) == 1
@@ -356,7 +274,7 @@ async def test_light_brightness(hass: HomeAssistant, mock_govee_api: AsyncMock)
light = hass.states.get("light.H615A")
assert light is not None
assert light.state == "on"
mock_govee_api.set_brightness.assert_awaited_with(mock_govee_api.devices[0], 50)
mock_govee_api.set_brightness.assert_awaited_with(device, 50)
assert light.attributes[ATTR_BRIGHTNESS] == 127
await hass.services.async_call(
@@ -371,7 +289,7 @@ async def test_light_brightness(hass: HomeAssistant, mock_govee_api: AsyncMock)
assert light is not None
assert light.state == "on"
assert light.attributes[ATTR_BRIGHTNESS] == 255
mock_govee_api.set_brightness.assert_awaited_with(mock_govee_api.devices[0], 100)
mock_govee_api.set_brightness.assert_awaited_with(device, 100)
await hass.services.async_call(
LIGHT_DOMAIN,
@@ -385,26 +303,12 @@ async def test_light_brightness(hass: HomeAssistant, mock_govee_api: AsyncMock)
assert light is not None
assert light.state == "on"
assert light.attributes[ATTR_BRIGHTNESS] == 255
mock_govee_api.set_brightness.assert_awaited_with(mock_govee_api.devices[0], 100)
mock_govee_api.set_brightness.assert_awaited_with(device, 100)
async def test_light_color(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
"""Test changing color."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api)
assert len(hass.states.async_all()) == 1
@@ -427,7 +331,7 @@ async def test_light_color(hass: HomeAssistant, mock_govee_api: AsyncMock) -> No
assert light.attributes[ATTR_COLOR_MODE] == ColorMode.RGB
mock_govee_api.set_color.assert_awaited_with(
mock_govee_api.devices[0], rgb=(100, 255, 50), temperature=None
device, rgb=(100, 255, 50), temperature=None
)
await hass.services.async_call(
@@ -444,29 +348,12 @@ async def test_light_color(hass: HomeAssistant, mock_govee_api: AsyncMock) -> No
assert light.attributes[ATTR_COLOR_TEMP_KELVIN] == 4400
assert light.attributes[ATTR_COLOR_MODE] == ColorMode.COLOR_TEMP
mock_govee_api.set_color.assert_awaited_with(
mock_govee_api.devices[0], rgb=None, temperature=4400
)
mock_govee_api.set_color.assert_awaited_with(device, rgb=None, temperature=4400)
async def test_scene_on(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
"""Test turning on scene."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=SCENE_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api, SCENE_CAPABILITIES)
assert len(hass.states.async_all()) == 1
@@ -486,29 +373,14 @@ async def test_scene_on(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
assert light is not None
assert light.state == "on"
assert light.attributes[ATTR_EFFECT] == "sunrise"
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], True)
mock_govee_api.turn_on_off.assert_awaited_with(device, True)
async def test_scene_restore_rgb(
hass: HomeAssistant, mock_govee_api: AsyncMock
) -> None:
"""Test restore rgb color."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=SCENE_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api, SCENE_CAPABILITIES)
assert len(hass.states.async_all()) == 1
@@ -538,7 +410,7 @@ async def test_scene_restore_rgb(
assert light.state == "on"
assert light.attributes[ATTR_RGB_COLOR] == initial_color
assert light.attributes[ATTR_BRIGHTNESS] == 255
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], True)
mock_govee_api.turn_on_off.assert_awaited_with(device, True)
# Activate scene
await hass.services.async_call(
@@ -553,7 +425,7 @@ async def test_scene_restore_rgb(
assert light is not None
assert light.state == "on"
assert light.attributes[ATTR_EFFECT] == "sunrise"
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], True)
mock_govee_api.turn_on_off.assert_awaited_with(device, True)
# Deactivate scene
await hass.services.async_call(
@@ -576,22 +448,7 @@ async def test_scene_restore_temperature(
hass: HomeAssistant, mock_govee_api: AsyncMock
) -> None:
"""Test restore color temperature."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=SCENE_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api, SCENE_CAPABILITIES)
assert len(hass.states.async_all()) == 1
@@ -613,7 +470,7 @@ async def test_scene_restore_temperature(
assert light is not None
assert light.state == "on"
assert light.attributes[ATTR_COLOR_TEMP_KELVIN] == initial_color
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], True)
mock_govee_api.turn_on_off.assert_awaited_with(device, True)
# Activate scene
await hass.services.async_call(
@@ -628,7 +485,7 @@ async def test_scene_restore_temperature(
assert light is not None
assert light.state == "on"
assert light.attributes[ATTR_EFFECT] == "sunrise"
mock_govee_api.set_scene.assert_awaited_with(mock_govee_api.devices[0], "sunrise")
mock_govee_api.set_scene.assert_awaited_with(device, "sunrise")
# Deactivate scene
await hass.services.async_call(
@@ -650,20 +507,7 @@ async def test_update_callback_registered_and_triggers_state_update(
hass: HomeAssistant, mock_govee_api: AsyncMock
) -> None:
"""Test that update callback is registered and triggers state update."""
device = GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
mock_govee_api.devices = [device]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api)
assert device.update_callback is not None
@@ -685,20 +529,7 @@ async def test_update_callback_cleared_on_remove(
hass: HomeAssistant, mock_govee_api: AsyncMock
) -> None:
"""Test that update callback is cleared when entity is removed."""
device = GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
mock_govee_api.devices = [device]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entry, device = await setup_light(hass, mock_govee_api)
assert device.update_callback is not None
@@ -710,22 +541,7 @@ async def test_update_callback_cleared_on_remove(
async def test_scene_none(hass: HomeAssistant, mock_govee_api: AsyncMock) -> None:
"""Test turn on 'none' scene."""
mock_govee_api.devices = [
GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=SCENE_CAPABILITIES,
)
]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api, SCENE_CAPABILITIES)
assert len(hass.states.async_all()) == 1
@@ -755,7 +571,7 @@ async def test_scene_none(hass: HomeAssistant, mock_govee_api: AsyncMock) -> Non
assert light.state == "on"
assert light.attributes[ATTR_RGB_COLOR] == initial_color
assert light.attributes[ATTR_BRIGHTNESS] == 255
mock_govee_api.turn_on_off.assert_awaited_with(mock_govee_api.devices[0], True)
mock_govee_api.turn_on_off.assert_awaited_with(device, True)
# Activate scene
await hass.services.async_call(
@@ -808,20 +624,7 @@ async def test_device_availability(
timeout, goes unavailable past it, and recovers when a status response
refreshes ``lastseen``.
"""
device = GoveeDevice(
controller=mock_govee_api,
ip="192.168.1.100",
fingerprint="asdawdqwdqwd",
sku="H615A",
capabilities=DEFAULT_CAPABILITIES,
)
mock_govee_api.devices = [device]
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
_, device = await setup_light(hass, mock_govee_api)
state = hass.states.get("light.H615A")
assert state is not None
+10
View File
@@ -5,9 +5,11 @@ from http import HTTPStatus
import json
from typing import Any
from unittest.mock import AsyncMock, MagicMock, Mock, mock_open, patch
import warnings
from aiohttp import ClientError
from aiohttp.hdrs import AUTHORIZATION
import jwt.warnings
import pytest
from pywebpush import WebPushException
from syrupy.assertion import SnapshotAssertion
@@ -1289,3 +1291,11 @@ async def test_html5_dismiss_message(
"data": {"jwt": "JWT"},
**expected_payload,
}
def test_add_jwt_no_insecure_key_warning() -> None:
"""Test that add_jwt does not emit InsecureKeyLengthWarning for short keys."""
short_key = "c2hvcnRfa2V5X2hlcmU="
with warnings.catch_warnings():
warnings.simplefilter("error", jwt.warnings.InsecureKeyLengthWarning)
html5.add_jwt(1234567890, "device", "tag", short_key)
@@ -6,6 +6,8 @@ from unittest.mock import patch
from aiohttp import ClientSession, ClientWebSocketResponse
from freezegun.api import FrozenDateTimeFactory
from PIL import Image
import pytest
from homeassistant.components.image_upload import DOMAIN
from homeassistant.components.websocket_api import TYPE_RESULT
@@ -94,3 +96,57 @@ async def test_upload_image(
# Ensure removed from disk
assert not item_folder.is_dir()
@pytest.mark.parametrize(
("image_mode", "content_type"),
[
("RGBA", "image/jpeg"),
("LA", "image/jpeg"),
("P", "image/jpeg"),
],
)
async def test_upload_image_thumbnail_rgba_as_jpeg(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
hass_client: ClientSessionGenerator,
image_mode: str,
content_type: str,
) -> None:
"""Test thumbnail generation when image mode is incompatible with JPEG."""
now = dt_util.utcnow()
freezer.move_to(now)
with (
tempfile.TemporaryDirectory() as tempdir,
patch.object(hass.config, "path", return_value=tempdir),
):
assert await async_setup_component(hass, DOMAIN, {})
client: ClientSession = await hass_client()
with TEST_IMAGE.open("rb") as fp:
res = await client.post("/api/image/upload", data={"file": fp})
assert res.status == 200
item = await res.json()
image_id = item["id"]
tempdir = pathlib.Path(tempdir)
item_folder = tempdir / image_id
# Create an image file with the given mode to simulate the mismatch
original_path = item_folder / "original"
img = Image.new(image_mode, (300, 300))
img.save(original_path, format="png")
# Change the stored content_type to simulate the mismatch
hass.data[DOMAIN].data[image_id]["content_type"] = content_type
# Fetch the thumbnail; this should not raise an OSError
res = await client.get(f"/api/image/serve/{image_id}/256x256")
assert res.status == 200
assert (item_folder / "256x256").is_file()
# Verify the generated thumbnail is a valid JPEG
thumbnail = Image.open(item_folder / "256x256")
assert thumbnail.mode == "RGB"
+32
View File
@@ -2,8 +2,11 @@
from unittest.mock import Mock, patch
from aiohttp import ContentTypeError, RequestInfo
from multidict import CIMultiDict, CIMultiDictProxy
import pytest
from syrupy.assertion import SnapshotAssertion
from yarl import URL
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
@@ -43,3 +46,32 @@ async def test_admin_sensors(
assert hass.states.get("sensor.mock_title_videos_count") is None
assert hass.states.get("sensor.mock_title_disk_used_by_photos") is None
assert hass.states.get("sensor.mock_title_disk_used_by_videos") is None
async def test_update_error_does_not_leak_api_key(
hass: HomeAssistant,
mock_immich: Mock,
mock_config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that API key is not leaked in error logs on connection failure."""
await setup_integration(hass, mock_config_entry)
api_key = "SECRET_API_KEY_12345"
headers = CIMultiDictProxy(
CIMultiDict({"x-api-key": api_key, "Host": "example.com"})
)
request_info = RequestInfo(
url=URL("https://example.com/api/server/about"),
method="GET",
headers=headers,
real_url=URL("https://example.com/api/server/about"),
)
mock_immich.server.async_get_about_info.side_effect = ContentTypeError(
request_info, (), status=503, message="Service Unavailable"
)
await hass.config_entries.async_reload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert api_key not in caplog.text
@@ -0,0 +1,70 @@
"""Tests for LG Netcast media player platform."""
from collections.abc import Generator
from datetime import timedelta
from unittest.mock import MagicMock, patch
import xml.etree.ElementTree as ET
import pytest
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import MODEL_NAME, setup_lgnetcast
from tests.common import async_fire_time_changed
ENTITY_ID = f"{MP_DOMAIN}.{MODEL_NAME.lower()}"
def _make_channel(name: str, major: str) -> ET.Element:
"""Create a fake channel XML element."""
channel = ET.Element("data")
chname = ET.SubElement(channel, "chname")
chname.text = name
major_el = ET.SubElement(channel, "major")
major_el.text = major
return channel
@pytest.fixture(autouse=True)
def mock_lg_netcast() -> Generator[MagicMock]:
"""Mock LG Netcast library."""
with patch(
"homeassistant.components.lg_netcast.LgNetCastClient"
) as mock_client_class:
yield mock_client_class
async def test_source_list_duplicate_channel_names(
hass: HomeAssistant,
mock_lg_netcast: MagicMock,
) -> None:
"""Test that duplicate channel names are disambiguated in source list."""
client = mock_lg_netcast.return_value
client.get_volume.return_value = (20, False)
context_client = client.__enter__.return_value
channel_data = {
"cur_channel": None,
"channel_list": [
_make_channel("BBC One", "1"),
_make_channel("ITV", "3"),
_make_channel("BBC One", "101"),
],
}
context_client.query_data.side_effect = channel_data.get
await setup_lgnetcast(hass)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=60))
await hass.async_block_till_done(wait_background_tasks=True)
entity = hass.states.get(ENTITY_ID)
assert entity is not None
source_list = entity.attributes.get("source_list")
assert source_list is not None
assert len(source_list) == 3
assert "ITV" in source_list
assert "BBC One (1)" in source_list
assert "BBC One (101)" in source_list
@@ -46,7 +46,7 @@ async def load_config_entry(
entry_id="1",
unique_id="username",
version=2,
minor_version=2,
minor_version=3,
)
config_entry.add_to_hass(hass)
@@ -32,7 +32,7 @@
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'panic',
'unique_id': 'yale_smart_alarm-panic',
'unique_id': '1-panic',
'unit_of_measurement': None,
})
# ---
+46 -2
View File
@@ -27,7 +27,7 @@ async def test_setup_entry(
entry_id="1",
unique_id="username",
version=2,
minor_version=2,
minor_version=3,
)
entry.add_to_hass(hass)
@@ -87,7 +87,7 @@ async def test_migrate_entry(
assert entry.state is ConfigEntryState.LOADED
assert entry.version == 2
assert entry.minor_version == 2
assert entry.minor_version == 3
assert entry.data == ENTRY_CONFIG
assert entry.options == OPTIONS_CONFIG
@@ -95,3 +95,47 @@ async def test_migrate_entry(
lock = entity_registry.async_get(lock_entity_id)
assert lock.options == {"lock": {"default_code": "123456"}}
async def test_migrate_panic_button_unique_id(
hass: HomeAssistant,
get_client: Mock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test migration of panic button unique_id from v2.2 to v2.3."""
entry = MockConfigEntry(
title=ENTRY_CONFIG["username"],
domain=DOMAIN,
source=SOURCE_USER,
data=ENTRY_CONFIG,
options=OPTIONS_CONFIG,
entry_id="1",
unique_id="username",
version=2,
minor_version=2,
)
entry.add_to_hass(hass)
entity_registry.async_get_or_create(
"button",
DOMAIN,
"yale_smart_alarm-panic",
config_entry=entry,
)
with patch(
"homeassistant.components.yale_smart_alarm.coordinator.YaleSmartAlarmClient",
return_value=get_client,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert entry.version == 2
assert entry.minor_version == 3
migrated = entity_registry.async_get_entity_id("button", DOMAIN, "1-panic")
assert migrated is not None
old = entity_registry.async_get_entity_id(
"button", DOMAIN, "yale_smart_alarm-panic"
)
assert old is None
+8
View File
@@ -9,12 +9,15 @@ import jwt
import pytest
from yoto_api import (
Card,
CardInsertionState,
Chapter,
DayMode,
Device,
Group,
PlaybackEvent,
PlaybackStatus,
PlayerInfo,
PlayerStatus,
Track,
YotoPlayer,
)
@@ -88,6 +91,11 @@ def _build_player() -> YotoPlayer:
firmware_version="v2.17.5",
mac="aa:bb:cc:dd:ee:ff",
)
player.status = PlayerStatus(
battery_level_percentage=75,
card_insertion_state=CardInsertionState.PHYSICAL,
day_mode=DayMode.DAY,
)
player.last_event = PlaybackEvent(
player_id=PLAYER_ID,
playback_status=PlaybackStatus.PLAYING,
@@ -0,0 +1,180 @@
# serializer version: 1
# name: test_all_entities[sensor.nursery_yoto_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.nursery_yoto_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': 'yoto',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'player-test_battery_level',
'unit_of_measurement': '%',
})
# ---
# name: test_all_entities[sensor.nursery_yoto_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Nursery Yoto Battery',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.nursery_yoto_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '75',
})
# ---
# name: test_all_entities[sensor.nursery_yoto_card_slot-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'none',
'physical',
'remote',
'streaming',
]),
}),
'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.nursery_yoto_card_slot',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Card slot',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Card slot',
'platform': 'yoto',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'card_insertion_state',
'unique_id': 'player-test_card_insertion_state',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.nursery_yoto_card_slot-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Nursery Yoto Card slot',
'options': list([
'none',
'physical',
'remote',
'streaming',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.nursery_yoto_card_slot',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'physical',
})
# ---
# name: test_all_entities[sensor.nursery_yoto_day_mode-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'day',
'night',
]),
}),
'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.nursery_yoto_day_mode',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Day mode',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Day mode',
'platform': 'yoto',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'day_mode',
'unique_id': 'player-test_day_mode',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.nursery_yoto_day_mode-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Nursery Yoto Day mode',
'options': list([
'day',
'night',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.nursery_yoto_day_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'day',
})
# ---
+4 -3
View File
@@ -1,7 +1,7 @@
"""Tests for the Yoto media player platform."""
from typing import Any
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
@@ -22,7 +22,7 @@ from homeassistant.components.media_player import (
SERVICE_VOLUME_SET,
MediaPlayerState,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
@@ -68,7 +68,8 @@ async def test_entity_state(
) -> None:
"""Snapshot the media player entity state."""
freezer.move_to("2026-05-08T12:00:00+00:00")
await setup_integration(hass, mock_config_entry)
with patch("homeassistant.components.yoto.PLATFORMS", [Platform.MEDIA_PLAYER]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+49
View File
@@ -0,0 +1,49 @@
"""Tests for the Yoto sensor platform."""
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures("setup_credentials")
ENTITY_ID = "sensor.nursery_yoto_battery"
@pytest.mark.usefixtures("mock_yoto_client")
async def test_all_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Snapshot every Yoto sensor entity."""
with patch("homeassistant.components.yoto.PLATFORMS", [Platform.SENSOR]):
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_offline(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Sensors are unavailable while the player is offline."""
player = next(iter(mock_yoto_client.players.values()))
player.is_online = False
with patch("homeassistant.components.yoto.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNAVAILABLE
@@ -110,24 +110,37 @@ async def test_config_entry_id(
async def test_config_entry_attr(hass: HomeAssistant) -> None:
"""Test config entry attr."""
info = {
config_entry = MockConfigEntry(
domain="mock_light",
title="mock title",
source=config_entries.SOURCE_BLUETOOTH,
disabled_by=config_entries.ConfigEntryDisabler.USER,
pref_disable_polling=True,
)
config_entry.add_to_hass(hass)
expected = {
"domain": "mock_light",
"title": "mock title",
"source": config_entries.SOURCE_BLUETOOTH,
"disabled_by": config_entries.ConfigEntryDisabler.USER,
"pref_disable_polling": True,
"disabled_by": "user",
"pref_disable_polling": "True",
"state": "not_loaded",
}
config_entry = MockConfigEntry(**info)
config_entry.add_to_hass(hass)
info["state"] = config_entries.ConfigEntryState.NOT_LOADED
for key, value in info.items():
assert render(
hass,
"{{ config_entry_attr('" + config_entry.entry_id + "', '" + key + "') }}",
parse_result=False,
) == str(value)
for key, value in expected.items():
assert (
render(
hass,
"{{ config_entry_attr('"
+ config_entry.entry_id
+ "', '"
+ key
+ "') }}",
parse_result=False,
)
== value
)
for config_entry_id, key in (
(config_entry.entry_id, "invalid_key"),
+4 -4
View File
@@ -1234,7 +1234,7 @@ async def test_script_tool(
assert tool.name == "test_script"
assert (
tool.description
== "This is a test script. Aliases: ['script name', 'script alias']"
== "This is a test script. Aliases: ['script alias', 'script name']"
)
schema = {
vol.Required("beer", description="Number of beers"): cv.string,
@@ -1249,7 +1249,7 @@ async def test_script_tool(
assert hass.data[llm.ACTION_PARAMETERS_CACHE]["script"] == {
"test_script": (
"This is a test script. Aliases: ['script name', 'script alias']",
"This is a test script. Aliases: ['script alias', 'script name']",
vol.Schema(schema),
),
"script_with_no_fields": (
@@ -1358,14 +1358,14 @@ async def test_script_tool(
assert tool.name == "test_script"
assert (
tool.description
== "This is a new test script. Aliases: ['script name', 'script alias']"
== "This is a new test script. Aliases: ['script alias', 'script name']"
)
schema = {vol.Required("beer", description="Number of beers"): cv.string}
assert tool.parameters.schema == schema
assert hass.data[llm.ACTION_PARAMETERS_CACHE]["script"] == {
"test_script": (
"This is a new test script. Aliases: ['script name', 'script alias']",
"This is a new test script. Aliases: ['script alias', 'script name']",
vol.Schema(schema),
),
"script_with_no_fields": (