Compare commits

...

13 Commits

Author SHA1 Message Date
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
34 changed files with 737 additions and 359 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]:
@@ -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
@@ -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"]
}
+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"
@@ -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
@@ -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": (