mirror of
https://github.com/home-assistant/core.git
synced 2026-06-12 20:21:40 +02:00
Compare commits
13 Commits
yoto_time_entity
...
dev
| Author | SHA1 | Date | |
|---|---|---|---|
| 96c5774bef | |||
| 6288905ca5 | |||
| 4c1dbec599 | |||
| 2a0b5ca895 | |||
| 746c8dd908 | |||
| dadfea4d62 | |||
| 53aef99921 | |||
| 88bd563a2c | |||
| c57c8fad16 | |||
| 5932a11e0c | |||
| 080492c64d | |||
| 3b8689637a | |||
| 8cd97fc60e |
@@ -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()
|
||||
|
||||
@@ -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}"},
|
||||
)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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."]
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Generated
+3
-3
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
# ---
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -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": (
|
||||
|
||||
Reference in New Issue
Block a user