Compare commits

..

10 Commits

Author SHA1 Message Date
copilot-swe-agent[bot] 8112a0ed90 Replace EdifierCommandSets with EdifierCommandSet in tests 2026-06-02 21:57:04 +00:00
Abílio Costa 2b0b3ccd24 Merge branch 'dev' into edifier_ir1 2026-06-02 22:43:49 +01:00
abmantis caa25c2069 Fix name 2026-06-02 22:03:12 +01:00
abmantis 7e9a50dca4 Import model mappings from lib 2026-06-02 19:51:05 +01:00
abmantis f0ff66b09a Update config flow IQS comment 2026-06-01 19:02:18 +01:00
abmantis e7482e9852 Merge branch 'dev' of github.com:home-assistant/core into edifier_ir1 2026-06-01 17:08:37 +01:00
abmantis d2b705a617 Rename var 2026-05-27 13:09:05 +01:00
abmantis 43cb41d396 Fix iot_class 2026-05-27 13:06:18 +01:00
abmantis dba52262f3 Merge branch 'dev' of github.com:home-assistant/core into edifier_ir1 2026-05-27 11:46:37 +01:00
abmantis c43155ed4b Add Edifier Infrared integration 2026-05-27 11:42:00 +01:00
70 changed files with 1003 additions and 3018 deletions
Generated
+2 -2
View File
@@ -453,6 +453,8 @@ CLAUDE.md @home-assistant/core
/tests/components/ecovacs/ @mib1185 @edenhaus @Augar
/homeassistant/components/ecowitt/ @pvizeli
/tests/components/ecowitt/ @pvizeli
/homeassistant/components/edifier_infrared/ @abmantis
/tests/components/edifier_infrared/ @abmantis
/homeassistant/components/efergy/ @tkdrob
/tests/components/efergy/ @tkdrob
/homeassistant/components/egardia/ @jeroenterheerdt
@@ -840,8 +842,6 @@ CLAUDE.md @home-assistant/core
/tests/components/imgw_pib/ @bieniu
/homeassistant/components/immich/ @mib1185
/tests/components/immich/ @mib1185
/homeassistant/components/imou/ @Imou-OpenPlatform
/tests/components/imou/ @Imou-OpenPlatform
/homeassistant/components/improv_ble/ @emontnemery
/tests/components/improv_ble/ @emontnemery
/homeassistant/components/incomfort/ @jbouwh
+5 -19
View File
@@ -2,18 +2,12 @@
import avea
from homeassistant.components.bluetooth import (
BluetoothReachabilityIntent,
async_address_reachability_diagnostics,
async_ble_device_from_address,
)
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import DOMAIN
type AveaConfigEntry = ConfigEntry[avea.Bulb]
PLATFORMS: list[Platform] = [Platform.LIGHT]
@@ -21,20 +15,12 @@ PLATFORMS: list[Platform] = [Platform.LIGHT]
async def async_setup_entry(hass: HomeAssistant, entry: AveaConfigEntry) -> bool:
"""Set up Avea from a config entry."""
address = entry.data[CONF_ADDRESS]
ble_device = async_ble_device_from_address(hass, address, connectable=True)
ble_device = async_ble_device_from_address(
hass, entry.data[CONF_ADDRESS], connectable=True
)
if not ble_device:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="device_not_found",
translation_placeholders={
"address": address,
"reason": async_address_reachability_diagnostics(
hass,
address.upper(),
BluetoothReachabilityIntent.CONNECTION,
),
},
f"Could not find Avea device with address {entry.data[CONF_ADDRESS]}"
)
entry.runtime_data = avea.Bulb(ble_device)
@@ -22,11 +22,6 @@
}
}
},
"exceptions": {
"device_not_found": {
"message": "Could not find Avea device with address {address}: {reason}"
}
},
"issues": {
"deprecated_yaml": {
"description": "[%key:component::homeassistant::issues::deprecated_yaml::description%]",
@@ -0,0 +1,18 @@
"""Edifier infrared integration for Home Assistant."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
PLATFORMS = [Platform.MEDIA_PLAYER]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Edifier IR from a config entry."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload an Edifier IR config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@@ -0,0 +1,77 @@
"""Config flow for Edifier infrared integration."""
from typing import Any
from infrared_protocols.codes.edifier.models import MODEL_TO_COMMAND_SET, EdifierModel
import voluptuous as vol
from homeassistant.components.infrared import (
DOMAIN as INFRARED_DOMAIN,
async_get_emitters,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_MODEL
from homeassistant.helpers.selector import (
EntitySelector,
EntitySelectorConfig,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import CONF_COMMAND_SET, CONF_INFRARED_ENTITY_ID, DOMAIN
class EdifierIrConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle config flow for Edifier IR."""
VERSION = 1
MINOR_VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step - select IR entity and speaker model."""
emitter_entity_ids = async_get_emitters(self.hass)
if not emitter_entity_ids:
return self.async_abort(reason="no_emitters")
if user_input is not None:
infrared_entity_id = user_input[CONF_INFRARED_ENTITY_ID]
model = EdifierModel(user_input[CONF_MODEL])
command_set = MODEL_TO_COMMAND_SET[model]
await self.async_set_unique_id(f"{command_set}_{infrared_entity_id}")
self._abort_if_unique_id_configured()
entity_name = infrared_entity_id
if state := self.hass.states.get(infrared_entity_id):
entity_name = state.name or infrared_entity_id
return self.async_create_entry(
title=f"Edifier {model.value} via {entity_name}",
data={
CONF_INFRARED_ENTITY_ID: infrared_entity_id,
CONF_MODEL: model.value,
CONF_COMMAND_SET: command_set.value,
},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_INFRARED_ENTITY_ID): EntitySelector(
EntitySelectorConfig(
domain=INFRARED_DOMAIN, include_entities=emitter_entity_ids
)
),
vol.Required(CONF_MODEL): SelectSelector(
SelectSelectorConfig(
options=[model.value for model in EdifierModel],
mode=SelectSelectorMode.DROPDOWN,
)
),
}
),
)
@@ -0,0 +1,19 @@
"""Constants for the Edifier infrared integration."""
from infrared_protocols.codes.edifier.r1280db import EdifierR1280DBCode
from infrared_protocols.codes.edifier.r1280t import EdifierR1280TCode
from infrared_protocols.codes.edifier.r1700bt import EdifierR1700BTCode
from infrared_protocols.codes.edifier.rc20g import EdifierRC20GCode
from infrared_protocols.codes.edifier.s360db import EdifierS360DBCode
DOMAIN = "edifier_infrared"
CONF_INFRARED_ENTITY_ID = "infrared_entity_id"
CONF_COMMAND_SET = "command_set"
type EdifierCode = (
EdifierR1700BTCode
| EdifierR1280DBCode
| EdifierR1280TCode
| EdifierS360DBCode
| EdifierRC20GCode
)
@@ -0,0 +1,27 @@
"""Common entity for Edifier infrared integration."""
from infrared_protocols.codes.edifier.models import EdifierModel
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
class EdifierIrEntity(Entity):
"""Edifier IR base entity providing common device info."""
_attr_has_entity_name = True
def __init__(
self, entry: ConfigEntry, model: EdifierModel, unique_id_suffix: str
) -> None:
"""Initialize Edifier IR entity."""
self._attr_unique_id = f"{entry.entry_id}_{unique_id_suffix}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
name=f"Edifier {model.value}",
manufacturer="Edifier",
model=model.value,
)
@@ -0,0 +1,11 @@
{
"domain": "edifier_infrared",
"name": "Edifier Infrared",
"codeowners": ["@abmantis"],
"config_flow": true,
"dependencies": ["infrared"],
"documentation": "https://www.home-assistant.io/integrations/edifier_infrared",
"integration_type": "device",
"iot_class": "assumed_state",
"quality_scale": "bronze"
}
@@ -0,0 +1,174 @@
"""Media player platform for Edifier infrared integration."""
from infrared_protocols.codes.edifier.models import EdifierCommandSet, EdifierModel
from infrared_protocols.codes.edifier.r1280db import EdifierR1280DBCode
from infrared_protocols.codes.edifier.r1280t import EdifierR1280TCode
from infrared_protocols.codes.edifier.r1700bt import EdifierR1700BTCode
from infrared_protocols.codes.edifier.rc20g import EdifierRC20GCode
from infrared_protocols.codes.edifier.s360db import EdifierS360DBCode
from homeassistant.components.infrared import InfraredEmitterConsumerEntity
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MODEL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_COMMAND_SET, CONF_INFRARED_ENTITY_ID, EdifierCode
from .entity import EdifierIrEntity
PARALLEL_UPDATES = 1
COMMAND_SET_COMMANDS: dict[
EdifierCommandSet,
dict[
MediaPlayerEntityFeature,
tuple[EdifierCode | tuple[EdifierCode, ...], ...],
],
] = {
EdifierCommandSet.R1700BT: {
MediaPlayerEntityFeature.TURN_ON: (EdifierR1700BTCode.POWER,),
MediaPlayerEntityFeature.TURN_OFF: (EdifierR1700BTCode.POWER,),
MediaPlayerEntityFeature.VOLUME_STEP: (
(EdifierR1700BTCode.VOLUME_UP,),
(EdifierR1700BTCode.VOLUME_DOWN,),
),
MediaPlayerEntityFeature.VOLUME_MUTE: (EdifierR1700BTCode.MUTE,),
MediaPlayerEntityFeature.PLAY: (EdifierR1700BTCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.PAUSE: (EdifierR1700BTCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.NEXT_TRACK: (EdifierR1700BTCode.FORWARD,),
MediaPlayerEntityFeature.PREVIOUS_TRACK: (EdifierR1700BTCode.BACK,),
},
EdifierCommandSet.R1280DB: {
MediaPlayerEntityFeature.TURN_ON: (EdifierR1280DBCode.POWER,),
MediaPlayerEntityFeature.TURN_OFF: (EdifierR1280DBCode.POWER,),
MediaPlayerEntityFeature.VOLUME_STEP: (
(EdifierR1280DBCode.VOLUME_UP,),
(EdifierR1280DBCode.VOLUME_DOWN,),
),
MediaPlayerEntityFeature.VOLUME_MUTE: (EdifierR1280DBCode.MUTE,),
MediaPlayerEntityFeature.PLAY: (EdifierR1280DBCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.PAUSE: (EdifierR1280DBCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.NEXT_TRACK: (EdifierR1280DBCode.FORWARD,),
MediaPlayerEntityFeature.PREVIOUS_TRACK: (EdifierR1280DBCode.BACK,),
},
EdifierCommandSet.R1280T: {
MediaPlayerEntityFeature.VOLUME_STEP: (
(EdifierR1280TCode.VOLUME_UP,),
(EdifierR1280TCode.VOLUME_DOWN,),
),
MediaPlayerEntityFeature.VOLUME_MUTE: (EdifierR1280TCode.MUTE,),
},
EdifierCommandSet.S360DB: {
MediaPlayerEntityFeature.TURN_ON: (EdifierS360DBCode.POWER,),
MediaPlayerEntityFeature.TURN_OFF: (EdifierS360DBCode.POWER,),
MediaPlayerEntityFeature.VOLUME_STEP: (
(EdifierS360DBCode.VOLUME_UP,),
(EdifierS360DBCode.VOLUME_DOWN,),
),
MediaPlayerEntityFeature.PLAY: (EdifierS360DBCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.PAUSE: (EdifierS360DBCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.NEXT_TRACK: (EdifierS360DBCode.NEXT,),
MediaPlayerEntityFeature.PREVIOUS_TRACK: (EdifierS360DBCode.PREVIOUS,),
},
EdifierCommandSet.RC20G: {
MediaPlayerEntityFeature.TURN_ON: (EdifierRC20GCode.POWER,),
MediaPlayerEntityFeature.TURN_OFF: (EdifierRC20GCode.POWER,),
MediaPlayerEntityFeature.VOLUME_STEP: (
(EdifierRC20GCode.VOLUME_UP_LEFT, EdifierRC20GCode.VOLUME_UP_RIGHT),
(EdifierRC20GCode.VOLUME_DOWN_LEFT, EdifierRC20GCode.VOLUME_DOWN_RIGHT),
),
MediaPlayerEntityFeature.VOLUME_MUTE: (EdifierRC20GCode.MUTE,),
MediaPlayerEntityFeature.PLAY: (EdifierRC20GCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.PAUSE: (EdifierRC20GCode.PLAY_PAUSE,),
MediaPlayerEntityFeature.NEXT_TRACK: (EdifierRC20GCode.FORWARD,),
MediaPlayerEntityFeature.PREVIOUS_TRACK: (EdifierRC20GCode.PREVIOUS,),
},
}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Edifier IR media player."""
infrared_entity_id = entry.data[CONF_INFRARED_ENTITY_ID]
command_set = EdifierCommandSet(entry.data[CONF_COMMAND_SET])
model = EdifierModel(entry.data[CONF_MODEL])
async_add_entities(
[EdifierIrMediaPlayer(entry, model, infrared_entity_id, command_set)]
)
class EdifierIrMediaPlayer(
EdifierIrEntity, InfraredEmitterConsumerEntity, MediaPlayerEntity
):
"""Edifier IR media player entity."""
_attr_name = None
_attr_assumed_state = True
_attr_device_class = MediaPlayerDeviceClass.SPEAKER
def __init__(
self,
entry: ConfigEntry,
model: EdifierModel,
infrared_entity_id: str,
command_set: EdifierCommandSet,
) -> None:
"""Initialize Edifier IR media player."""
super().__init__(entry, model, unique_id_suffix="media_player")
self._infrared_emitter_entity_id = infrared_entity_id
self._commands = COMMAND_SET_COMMANDS[command_set]
self._attr_state = MediaPlayerState.ON
self._attr_supported_features = MediaPlayerEntityFeature(0)
for feature in self._commands:
self._attr_supported_features |= feature
async def _send_codes(self, *codes: EdifierCode) -> None:
"""Send one or more IR commands."""
for code in codes:
await self._send_command(code.to_command())
async def async_turn_on(self) -> None:
"""Turn on the speaker."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.TURN_ON])
async def async_turn_off(self) -> None:
"""Turn off the speaker."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.TURN_OFF])
async def async_volume_up(self) -> None:
"""Send volume up command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.VOLUME_STEP][0])
async def async_volume_down(self) -> None:
"""Send volume down command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.VOLUME_STEP][1])
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.VOLUME_MUTE])
async def async_media_play(self) -> None:
"""Send play command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.PLAY])
async def async_media_pause(self) -> None:
"""Send pause command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.PAUSE])
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.NEXT_TRACK])
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
await self._send_codes(*self._commands[MediaPlayerEntityFeature.PREVIOUS_TRACK])
@@ -0,0 +1,114 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
This integration does not provide additional actions.
appropriate-polling:
status: exempt
comment: |
This integration does not poll.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data:
status: exempt
comment: |
This integration does not store runtime data.
test-before-configure: done
test-before-setup:
status: exempt
comment: |
This integration only proxies commands through an existing infrared
entity, so there is no separate connection to validate during setup.
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: |
This integration does not register custom actions.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow:
status: exempt
comment: |
This integration does not require authentication.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
This integration does not support discovery.
discovery:
status: exempt
comment: |
Discovery is not supported for infrared integrations.
docs-data-update:
status: exempt
comment: |
This integration does not fetch data from devices.
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: done
dynamic-devices:
status: exempt
comment: |
Each config entry creates a single device.
entity-category:
status: exempt
comment: |
The media player entity is the primary entity and does not need a category.
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: |
The media player entity is the primary entity and should be enabled by default.
entity-translations: done
exception-translations:
status: exempt
comment: |
This integration does not raise exceptions.
icon-translations: done
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: |
This integration does not have repairable issues.
stale-devices:
status: exempt
comment: |
Each config entry manages exactly one device.
# Platinum
async-dependency:
status: exempt
comment: |
This integration depends on infrared_protocols which provides only code
definitions with no I/O, so async dependency does not apply.
inject-websession:
status: exempt
comment: |
This integration does not make HTTP requests.
strict-typing: todo
@@ -0,0 +1,22 @@
{
"config": {
"abort": {
"already_configured": "This Edifier device has already been configured with this transmitter.",
"no_emitters": "No infrared transmitter entities found. Please set up an infrared device first."
},
"step": {
"user": {
"data": {
"infrared_entity_id": "IR transmitter",
"model": "Speaker model"
},
"data_description": {
"infrared_entity_id": "Select the infrared transmitter entity to use.",
"model": "Choose your Edifier speaker model from the list."
},
"description": "Configure your Edifier speaker for IR control.",
"title": "Set up Edifier IR speaker"
}
}
}
}
+2 -2
View File
@@ -1,6 +1,7 @@
"""Support for entities of the Evohome integration."""
from collections.abc import Mapping
from datetime import UTC, datetime
import logging
from typing import Any
@@ -13,7 +14,6 @@ from evohomeasync2.schemas.typedefs import DayOfWeekDhwT
from homeassistant.core import callback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt as dt_util
from .coordinator import EvoDataUpdateCoordinator
@@ -161,7 +161,7 @@ class EvoChild(EvoEntity):
or self._schedule is None
or (
(until := self._setpoints.get("next_sp_from")) is not None
and until < dt_util.utcnow()
and until < datetime.now(UTC) # pylint: disable=home-assistant-enforce-utcnow
)
): # must use self._setpoints, not self.setpoints
await get_schedule()
+3 -3
View File
@@ -1,6 +1,6 @@
"""Support for (EMEA/EU-based) Honeywell TCC systems."""
from datetime import datetime, timedelta
from datetime import UTC, datetime, timedelta
from typing import Any, NotRequired, TypedDict
from evohomeasync.auth import (
@@ -12,7 +12,6 @@ from evohomeasync2.auth import AbstractTokenManager
from homeassistant.core import HomeAssistant
from homeassistant.helpers.storage import Store
from homeassistant.util import dt as dt_util
from .const import STORAGE_KEY, STORAGE_VER
@@ -92,7 +91,8 @@ class TokenManager(AbstractTokenManager, AbstractSessionManager):
session_id_expires = session.get(SZ_SESSION_ID_EXPIRES)
if session_id_expires is None:
self._session_id_expires = dt_util.utcnow() + timedelta(minutes=15)
# pylint: disable-next=home-assistant-enforce-utcnow
self._session_id_expires = datetime.now(tz=UTC) + timedelta(minutes=15)
else:
self._session_id_expires = datetime.fromisoformat(session_id_expires)
@@ -21,5 +21,5 @@
"integration_type": "system",
"preview_features": { "winter_mode": {} },
"quality_scale": "internal",
"requirements": ["home-assistant-frontend==20260527.4"]
"requirements": ["home-assistant-frontend==20260527.2"]
}
-42
View File
@@ -1,42 +0,0 @@
"""Support for Imou devices."""
from pyimouapi.device import ImouDeviceManager
from pyimouapi.ha_device import ImouHaDeviceManager
from pyimouapi.openapi import ImouOpenApiClient
from homeassistant.core import HomeAssistant, callback
from .const import API_URLS, CONF_API_URL, CONF_APP_ID, CONF_APP_SECRET, PLATFORMS
from .coordinator import ImouConfigEntry, ImouDataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ImouConfigEntry) -> bool:
"""Set up Imou integration from a config entry."""
imou_client = ImouOpenApiClient(
entry.data[CONF_APP_ID],
entry.data[CONF_APP_SECRET],
API_URLS[entry.data[CONF_API_URL]],
)
device_manager = ImouDeviceManager(imou_client)
imou_device_manager = ImouHaDeviceManager(device_manager)
imou_coordinator = ImouDataUpdateCoordinator(hass, imou_device_manager, entry)
await imou_coordinator.async_config_entry_first_refresh()
entry.runtime_data = imou_coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# DataUpdateCoordinator schedules periodic refreshes only when it has
# listeners. With zero entities (e.g. an empty account at setup), register a
# no-op listener so polling continues and later devices are discovered via
# new_device_callbacks.
@callback
def _async_keep_polling() -> None:
"""Keep periodic polling when no entities are registered yet."""
entry.async_on_unload(imou_coordinator.async_add_listener(_async_keep_polling))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ImouConfigEntry) -> bool:
"""Handle removal of an entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
-109
View File
@@ -1,109 +0,0 @@
"""Support for Imou button controls."""
from pyimouapi.exceptions import ImouException
from pyimouapi.ha_device import ImouHaDevice
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import PTZ_MOVE_DURATION_MS, imou_device_identifier
from .coordinator import ImouConfigEntry, ImouDataUpdateCoordinator
from .entity import ImouEntity
PARALLEL_UPDATES = 1
# Button types
PARAM_RESTART_DEVICE = "restart_device"
PARAM_MUTE = "mute"
PARAM_PTZ_UP = "ptz_up"
PARAM_PTZ_DOWN = "ptz_down"
PARAM_PTZ_LEFT = "ptz_left"
PARAM_PTZ_RIGHT = "ptz_right"
BUTTON_TYPES = (
PARAM_RESTART_DEVICE,
PARAM_MUTE,
PARAM_PTZ_UP,
PARAM_PTZ_DOWN,
PARAM_PTZ_LEFT,
PARAM_PTZ_RIGHT,
)
PTZ_BUTTON_TYPES = (
PARAM_PTZ_UP,
PARAM_PTZ_DOWN,
PARAM_PTZ_LEFT,
PARAM_PTZ_RIGHT,
)
BUTTON_DEVICE_CLASS: dict[str, ButtonDeviceClass] = {
PARAM_RESTART_DEVICE: ButtonDeviceClass.RESTART,
}
def _iter_buttons(
coordinator: ImouDataUpdateCoordinator,
) -> list[tuple[str, ImouHaDevice]]:
"""Return (button_type, device) pairs for supported buttons."""
return [
(button_type, device)
for device in coordinator.devices
for button_type in device.buttons
if button_type in BUTTON_TYPES
]
async def async_setup_entry(
hass: HomeAssistant,
entry: ImouConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Imou button entities."""
coordinator = entry.runtime_data
def _add_buttons(new_devices: list[ImouHaDevice]) -> None:
device_keys = {imou_device_identifier(device) for device in new_devices}
async_add_entities(
ImouButton(coordinator, button_type, device)
for button_type, device in _iter_buttons(coordinator)
if imou_device_identifier(device) in device_keys
)
coordinator.new_device_callbacks.append(_add_buttons)
@callback
def _remove_new_device_callback() -> None:
if _add_buttons in coordinator.new_device_callbacks:
coordinator.new_device_callbacks.remove(_add_buttons)
entry.async_on_unload(_remove_new_device_callback)
_add_buttons(coordinator.devices)
class ImouButton(ImouEntity, ButtonEntity):
"""Imou button entity."""
def __init__(
self,
coordinator: ImouDataUpdateCoordinator,
entity_type: str,
device: ImouHaDevice,
) -> None:
"""Initialize the Imou button entity."""
super().__init__(coordinator, entity_type, device)
if device_class := BUTTON_DEVICE_CLASS.get(entity_type):
self._attr_device_class = device_class
self._attr_translation_key = None
async def async_press(self) -> None:
"""Handle button press."""
duration = PTZ_MOVE_DURATION_MS if self._entity_type in PTZ_BUTTON_TYPES else 0
try:
await self.coordinator.device_manager.async_press_button(
self.device,
self._entity_type,
duration,
)
except ImouException as e:
raise HomeAssistantError(str(e)) from e
@@ -1,80 +0,0 @@
"""Config flow for Imou."""
import logging
from typing import Any
from pyimouapi.exceptions import (
ConnectFailedException,
ImouException,
InvalidAppIdOrSecretException,
RequestFailedException,
)
from pyimouapi.openapi import ImouOpenApiClient
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.helpers.selector import (
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import API_URLS, CONF_API_URL, CONF_APP_ID, CONF_APP_SECRET, DOMAIN
_LOGGER = logging.getLogger(__name__)
class ImouConfigFlow(ConfigFlow, domain=DOMAIN):
"""Config flow for Imou integration."""
VERSION = 1
MINOR_VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step of the config flow."""
errors: dict[str, str] = {}
if user_input is not None:
await self.async_set_unique_id(user_input[CONF_APP_ID])
self._abort_if_unique_id_configured()
api_client = ImouOpenApiClient(
user_input[CONF_APP_ID],
user_input[CONF_APP_SECRET],
API_URLS[user_input[CONF_API_URL]],
)
try:
await api_client.async_get_token()
except InvalidAppIdOrSecretException:
errors["base"] = "invalid_auth"
except ConnectFailedException, RequestFailedException:
errors["base"] = "cannot_connect"
except ImouException as exception:
_LOGGER.debug("Imou error during config flow: %s", exception)
errors["base"] = "unknown"
else:
return self.async_create_entry(
title="Imou",
data={
CONF_APP_ID: user_input[CONF_APP_ID],
CONF_APP_SECRET: user_input[CONF_APP_SECRET],
CONF_API_URL: user_input[CONF_API_URL],
},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_APP_ID): str,
vol.Required(CONF_APP_SECRET): str,
vol.Required(CONF_API_URL, default="sg"): SelectSelector(
SelectSelectorConfig(
options=list(API_URLS),
translation_key="api_url",
mode=SelectSelectorMode.DROPDOWN,
)
),
}
),
errors=errors,
)
-39
View File
@@ -1,39 +0,0 @@
"""Constants."""
from pyimouapi.ha_device import ImouHaDevice
from homeassistant.const import Platform
DOMAIN = "imou"
def imou_device_identifier(device: ImouHaDevice) -> str:
"""Return a device registry identifier (device_id + channel when present)."""
if device.channel_id is not None:
return f"{device.device_id}_{device.channel_id}"
return device.device_id
# API URL region mapping
API_URLS: dict[str, str] = {
"sg": "openapi-sg.easy4ip.com",
"eu": "openapi-or.easy4ip.com",
"na": "openapi-fk.easy4ip.com",
"cn": "openapi.lechange.cn",
}
CONF_API_URL = "api_url"
CONF_APP_ID = "app_id"
CONF_APP_SECRET = "app_secret"
PARAM_STATUS = "status"
PARAM_STATE = "state"
# How long each PTZ button press moves the camera, in milliseconds (Imou cloud API).
PTZ_MOVE_DURATION_MS = 500
# Upper bound for a full coordinator refresh (device list + status for all devices).
UPDATE_TIMEOUT = 300
PLATFORMS = [Platform.BUTTON]
@@ -1,152 +0,0 @@
"""Provides the Imou DataUpdateCoordinator."""
import asyncio
from collections.abc import Callable
from datetime import timedelta
import logging
from pyimouapi.exceptions import ImouException
from pyimouapi.ha_device import ImouHaDevice, ImouHaDeviceManager
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, UPDATE_TIMEOUT, imou_device_identifier
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=120)
type ImouConfigEntry = ConfigEntry[ImouDataUpdateCoordinator]
class ImouDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""Data update coordinator for Imou devices."""
config_entry: ImouConfigEntry
def __init__(
self,
hass: HomeAssistant,
device_manager: ImouHaDeviceManager,
config_entry: ImouConfigEntry,
) -> None:
"""Initialize the Imou data update coordinator."""
super().__init__(
hass,
_LOGGER,
name="ImouDataUpdateCoordinator",
update_interval=SCAN_INTERVAL,
config_entry=config_entry,
always_update=True,
)
self._device_manager = device_manager
self.devices_by_key: dict[str, ImouHaDevice] = {}
self._devices_initialized = False
self.new_device_callbacks: list[Callable[[list[ImouHaDevice]], None]] = []
@property
def devices(self) -> list[ImouHaDevice]:
"""Return the list of devices."""
return list(self.devices_by_key.values())
@property
def device_manager(self) -> ImouHaDeviceManager:
"""Return the device manager."""
return self._device_manager
def get_device(self, device_key: str) -> ImouHaDevice | None:
"""Return the current device for device_key, if still on the account."""
return self.devices_by_key.get(device_key)
async def _async_update_data(self) -> None:
"""Update coordinator data."""
try:
async with asyncio.timeout(UPDATE_TIMEOUT):
fresh_devices = await self._device_manager.async_get_devices()
except TimeoutError as err:
raise UpdateFailed(f"Timeout while fetching data: {err}") from err
except ImouException as err:
raise UpdateFailed(f"Error fetching Imou devices: {err}") from err
fresh_by_key = {
imou_device_identifier(device): device for device in fresh_devices
}
self._async_add_remove_devices(fresh_by_key)
devices = list(self.devices_by_key.values())
if not devices:
return
try:
async with asyncio.timeout(UPDATE_TIMEOUT):
results = await asyncio.gather(
*(
self._device_manager.async_update_device_status(device)
for device in devices
),
return_exceptions=True,
)
except TimeoutError as err:
raise UpdateFailed(f"Timeout while fetching data: {err}") from err
failures: list[Exception] = []
for device, result in zip(devices, results, strict=True):
if isinstance(result, BaseException) and not isinstance(result, Exception):
# Propagate CancelledError and other BaseExceptions instead of
# swallowing them as a regular device failure.
raise result
if not isinstance(result, Exception):
continue
device_key = imou_device_identifier(device)
_LOGGER.warning(
"Error updating status for Imou device %s: %s",
device_key,
result,
)
failures.append(result)
if failures and len(failures) == len(devices):
raise UpdateFailed(
f"Error updating Imou devices: {failures[0]}"
) from failures[0]
def _async_add_remove_devices(self, fresh_by_key: dict[str, ImouHaDevice]) -> None:
"""Add new devices, remove devices no longer in the account.
This only tracks which devices exist on the account; per-device state
is updated in place by `async_update_device_status`, so devices that
remain on the account keep their existing object and are not replaced.
"""
if not self._devices_initialized:
self.devices_by_key = fresh_by_key
self._devices_initialized = True
return
current_keys = set(fresh_by_key)
known_keys = set(self.devices_by_key)
if current_keys == known_keys:
return
if removed_keys := known_keys - current_keys:
_LOGGER.debug("Removed Imou device(s): %s", ", ".join(removed_keys))
device_registry = dr.async_get(self.hass)
for device_key in removed_keys:
del self.devices_by_key[device_key]
if device := device_registry.async_get_device(
identifiers={(DOMAIN, device_key)}
):
device_registry.async_update_device(
device_id=device.id,
remove_config_entry_id=self.config_entry.entry_id,
)
if new_keys := current_keys - known_keys:
_LOGGER.debug("New Imou device(s) found: %s", ", ".join(new_keys))
new_devices = []
for device_key in new_keys:
self.devices_by_key[device_key] = fresh_by_key[device_key]
new_devices.append(fresh_by_key[device_key])
for callback in self.new_device_callbacks:
callback(new_devices)
-59
View File
@@ -1,59 +0,0 @@
"""An abstract class common to all Imou entities."""
from pyimouapi.ha_device import DeviceStatus, ImouHaDevice
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, PARAM_STATE, PARAM_STATUS, imou_device_identifier
from .coordinator import ImouDataUpdateCoordinator
class ImouEntity(CoordinatorEntity[ImouDataUpdateCoordinator]):
"""Base class for all Imou entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: ImouDataUpdateCoordinator,
entity_type: str,
device: ImouHaDevice,
) -> None:
"""Initialize the Imou entity."""
super().__init__(coordinator)
self._entity_type = entity_type
self._device_key = imou_device_identifier(device)
self._attr_unique_id = f"{self._device_key}${entity_type}"
self._attr_translation_key = entity_type
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._device_key)},
name=device.channel_name or device.device_name,
manufacturer=device.manufacturer,
model=device.model,
sw_version=device.swversion,
serial_number=device.device_id,
)
@property
def device(self) -> ImouHaDevice:
"""Return the live device from the coordinator.
Callers must guard with `available` first; accessing this for a device
that has left the account raises `KeyError`.
"""
return self.coordinator.devices_by_key[self._device_key]
@property
def available(self) -> bool:
"""Return if the entity is available."""
if (
not super().available
or self._device_key not in self.coordinator.devices_by_key
):
return False
if PARAM_STATUS not in self.device.sensors:
return False
return (
self.device.sensors[PARAM_STATUS][PARAM_STATE] != DeviceStatus.OFFLINE.value
)
-18
View File
@@ -1,18 +0,0 @@
{
"entity": {
"button": {
"ptz_down": {
"default": "mdi:arrow-down-bold"
},
"ptz_left": {
"default": "mdi:arrow-left-bold"
},
"ptz_right": {
"default": "mdi:arrow-right-bold"
},
"ptz_up": {
"default": "mdi:arrow-up-bold"
}
}
}
}
@@ -1,11 +0,0 @@
{
"domain": "imou",
"name": "Imou",
"codeowners": ["@Imou-OpenPlatform"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/imou",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["pyimouapi==1.2.7"]
}
@@ -1,73 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register custom actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not register custom actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: Entities do not explicitly subscribe to events.
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Cloud service integration, does not support discovery.
discovery:
status: exempt
comment: >-
Devices are reached via Imou Open Platform cloud APIs (App ID / secret). No
supported local discovery flow today; example cues if investigated later:
hostname `IPC-ABCD.imou.local`, MAC `aa:bb:cc:dd:ee:ff`.
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: done
entity-category: todo
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: done
exception-translations: todo
icon-translations: done
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: todo
inject-websession: todo
strict-typing: todo
@@ -1,56 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"api_url": "Server region",
"app_id": "App ID",
"app_secret": "App secret"
},
"data_description": {
"api_url": "Select the server region closest to your location",
"app_id": "The app ID obtained from the Imou cloud platform",
"app_secret": "The app secret obtained from the Imou cloud platform"
},
"title": "Log in to Imou cloud"
}
}
},
"entity": {
"button": {
"mute": {
"name": "Mute"
},
"ptz_down": {
"name": "PTZ down"
},
"ptz_left": {
"name": "PTZ left"
},
"ptz_right": {
"name": "PTZ right"
},
"ptz_up": {
"name": "PTZ up"
}
}
},
"selector": {
"api_url": {
"options": {
"cn": "China",
"eu": "Europe",
"na": "North America",
"sg": "Singapore (Asia-Pacific)"
}
}
}
}
@@ -8,7 +8,6 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .coordinator import OpenEVSEConfigEntry, OpenEVSEDataUpdateCoordinator
PLATFORMS = [Platform.NUMBER, Platform.SENSOR]
@@ -26,15 +25,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenEVSEConfigEntry) ->
try:
await charger.test_and_get()
except TimeoutError as ex:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="communication_error",
) from ex
raise ConfigEntryNotReady("Unable to connect to charger") from ex
except AuthenticationError as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from ex
raise ConfigEntryAuthFailed("Invalid credentials for charger") from ex
coordinator = OpenEVSEDataUpdateCoordinator(hass, entry, charger)
await coordinator.async_config_entry_first_refresh()
@@ -63,11 +63,7 @@ class OpenEVSEDataUpdateCoordinator(DataUpdateCoordinator[None]):
await self.charger.update()
except TimeoutError as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="communication_error",
f"Timeout communicating with charger: {error}"
) from error
except AuthenticationError as error:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from error
raise ConfigEntryAuthFailed("Invalid credentials for charger") from error
@@ -168,10 +168,10 @@
},
"exceptions": {
"authentication_error": {
"message": "Authentication failed"
"message": "Authentication failed while communicating with the charger."
},
"communication_error": {
"message": "Failed to communicate with the charger"
"message": "Failed to communicate with the charger."
},
"invalid_value": {
"message": "Value {value} is invalid for the charger."
@@ -2,7 +2,7 @@
import asyncio
import logging
from typing import Any
from typing import Any, cast
from awesomeversion import AwesomeVersion, AwesomeVersionException
from httpx import HTTPError, InvalidURL
@@ -41,8 +41,7 @@ def ensure_printer_is_supported(version: VersionInfo) -> None:
# Workaround to allow PrusaLink 0.7.2 on MK3 and MK2.5 that supports
# the 2.0.0 API, but doesn't advertise it yet
original_value = version.get("original")
original = original_value if isinstance(original_value, str) else ""
original = cast(str, version.get("original", ""))
if original.startswith(("PrusaLink I3MK3", "PrusaLink I3MK2")) and (
AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"])
):
+2 -2
View File
@@ -218,8 +218,8 @@ class SwitchBotBlindTiltEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_closed = (_tilt < self.CLOSED_DOWN_THRESHOLD) or (
_tilt > self.CLOSED_UP_THRESHOLD
)
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self._attr_is_opening = self.parsed_data["motionDirection"]["opening"]
self._attr_is_closing = self.parsed_data["motionDirection"]["closing"]
self.async_write_ha_state()
@@ -139,14 +139,12 @@
"device_tracker": {
"data": {
"device_id": "[%key:common::config_flow::data::device%]",
"in_zones": "Zones",
"latitude": "Latitude",
"longitude": "Longitude",
"name": "[%key:common::config_flow::data::name%]"
},
"data_description": {
"device_id": "[%key:component::template::common::device_id_description%]",
"in_zones": "Defines a template that returns a list of zones the device tracker is currently in. The template should return a list of zone entity IDs. If the device tracker is not in any zone, the template should return an empty list.",
"latitude": "Defines a template to get the latitude of the device tracker. Valid values are numbers between `-90` and `90`.",
"longitude": "Defines a template to get the longitude of the device tracker. Valid values are numbers between `-180` and `180`.",
"name": "[%key:common::config_flow::data::name%]"
@@ -717,13 +715,11 @@
"device_tracker": {
"data": {
"device_id": "[%key:common::config_flow::data::device%]",
"in_zones": "[%key:component::template::config::step::device_tracker::data::in_zones%]",
"latitude": "[%key:component::template::config::step::device_tracker::data::latitude%]",
"longitude": "[%key:component::template::config::step::device_tracker::data::longitude%]"
},
"data_description": {
"device_id": "[%key:component::template::common::device_id_description%]",
"in_zones": "[%key:component::template::config::step::device_tracker::data_description::in_zones%]",
"latitude": "[%key:component::template::config::step::device_tracker::data_description::latitude%]",
"longitude": "[%key:component::template::config::step::device_tracker::data_description::longitude%]"
},
@@ -29,7 +29,7 @@ rules:
docs-configuration-parameters:
status: exempt
comment: No options flow
docs-troubleshooting: done
docs-troubleshooting: todo
entity-category: done
entity-disabled-by-default: done
integration-owner: done
@@ -42,15 +42,13 @@ rules:
devices: done
diagnostics: todo
discovery: done
discovery-update-info:
status: exempt
comment: Integration is cloud-only; no local host info is stored on the config entry.
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-use-cases: done
discovery-update-info: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-use-cases: todo
dynamic-devices: todo
entity-translations: done
exception-translations: done
@@ -39,7 +39,6 @@ from homeassistant.helpers import (
entity,
target as target_helpers,
template,
trace,
)
from homeassistant.helpers.condition import (
async_from_config as async_condition_from_config,
@@ -1027,53 +1026,14 @@ async def handle_test_condition(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle test condition command."""
# Validating and instantiating the condition can fail on bad user input.
# Handle those errors here so they are reported to the client without being
# logged as unexpected errors by the default websocket error handler.
# Do static + dynamic validation of the condition
config = await async_validate_condition_config(hass, msg["condition"])
# Test the condition
condition = await async_condition_from_config(hass, config)
try:
# Do static + dynamic validation of the condition
config = await async_validate_condition_config(hass, msg["condition"])
condition = await async_condition_from_config(hass, config)
except vol.Invalid as err:
connection.send_error(msg["id"], const.ERR_INVALID_FORMAT, str(err))
return
except HomeAssistantError as err:
connection.send_error(
msg["id"],
const.ERR_HOME_ASSISTANT_ERROR,
str(err),
translation_domain=err.translation_domain,
translation_key=err.translation_key,
translation_placeholders=err.translation_placeholders,
connection.send_result(
msg["id"], {"result": condition.async_check(variables=msg.get("variables"))}
)
return
# Template errors (e.g. undefined variables) are recorded in the trace
# instead of being logged. Capture the trace and forward them to the client
# alongside the result.
condition_trace = trace.trace_get()
try:
with trace.suppress_template_error_logging():
check_result = condition.async_check(variables=msg.get("variables"))
except HomeAssistantError as err:
connection.send_error(
msg["id"],
const.ERR_HOME_ASSISTANT_ERROR,
str(err),
translation_domain=err.translation_domain,
translation_key=err.translation_key,
translation_placeholders=err.translation_placeholders,
)
else:
result: dict[str, Any] = {"result": check_result}
if template_errors := [
template_error
for elements in condition_trace.values()
for element in elements
for template_error in element.template_errors
]:
result["template_errors"] = template_errors
connection.send_result(msg["id"], result)
finally:
condition.async_unload()
@@ -1090,23 +1050,9 @@ async def handle_subscribe_condition(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle subscribe condition command."""
try:
condition_config = await async_validate_condition_config(hass, msg["condition"])
condition = await async_condition_from_config(hass, condition_config)
except vol.Invalid as err:
connection.send_error(msg["id"], const.ERR_INVALID_FORMAT, str(err))
return
except HomeAssistantError as err:
connection.send_error(
msg["id"],
const.ERR_HOME_ASSISTANT_ERROR,
str(err),
translation_domain=err.translation_domain,
translation_key=err.translation_key,
translation_placeholders=err.translation_placeholders,
)
return
condition_config = await async_validate_condition_config(hass, msg["condition"])
condition = await async_condition_from_config(hass, condition_config)
event_data: dict[str, Any] = {}
@callback
@@ -1115,24 +1061,10 @@ async def handle_subscribe_condition(
nonlocal event_data
new_event_data: dict[str, Any]
condition_trace = trace.trace_get()
try:
with trace.suppress_template_error_logging():
new_event_data = {"result": condition.async_check()}
new_event_data = {"result": condition.async_check()}
except HomeAssistantError as err:
new_event_data = {"error": str(err)}
# Template errors (e.g. undefined variables) are recorded in the trace
# instead of being logged. Forward them to the client so they are not
# lost, even when the condition still evaluated to a result.
if template_errors := [
template_error
for elements in condition_trace.values()
for element in elements
for template_error in element.template_errors
]:
new_event_data["template_errors"] = template_errors
if new_event_data == event_data:
return
event_data = new_event_data
@@ -92,9 +92,7 @@ class WebSocketHandler:
self._hass = hass
self._loop = hass.loop
self._request: web.Request = request
# decode_text=False so orjson decodes the raw TEXT bytes directly
# instead of decoding to str first and re-scanning.
self._wsock = web.WebSocketResponse(heartbeat=55, decode_text=False)
self._wsock = web.WebSocketResponse(heartbeat=55)
self._handle_task: asyncio.Task | None = None
self._writer_task: asyncio.Task | None = None
self._closing: bool = False
+2 -128
View File
@@ -4,15 +4,12 @@ from typing import Any, Unpack, cast
import voluptuous as vol
from homeassistant.components.device_tracker import ATTR_IN_ZONES
from homeassistant.const import (
ATTR_GPS_ACCURACY,
ATTR_LATITUDE,
ATTR_LONGITUDE,
CONF_ENTITY_ID,
CONF_FOR,
CONF_OPTIONS,
CONF_TARGET,
CONF_ZONE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
@@ -20,23 +17,15 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import ConditionErrorContainer, ConditionErrorMessage
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.automation import (
DomainSpec,
move_top_level_schema_fields_to_options,
)
from homeassistant.helpers.automation import move_top_level_schema_fields_to_options
from homeassistant.helpers.condition import (
ATTR_BEHAVIOR,
BEHAVIOR_ANY,
ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL,
Condition,
ConditionCheckParams,
ConditionConfig,
EntityConditionBase,
)
from homeassistant.helpers.typing import ConfigType
from . import in_zone
from .const import DOMAIN
_OPTIONS_SCHEMA_DICT: dict[vol.Marker, Any] = {
vol.Required(CONF_ENTITY_ID): cv.entity_ids,
@@ -160,126 +149,11 @@ class ZoneCondition(Condition):
return all_ok
_DOMAIN_SPECS: dict[str, DomainSpec] = {
"person": DomainSpec(value_source=ATTR_IN_ZONES),
"device_tracker": DomainSpec(value_source=ATTR_IN_ZONES),
}
_ZONE_CONDITION_SCHEMA = ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL.extend(
{
vol.Required(CONF_OPTIONS): {
vol.Required(CONF_ZONE): cv.entity_domain(DOMAIN),
},
}
)
class _ZoneTargetConditionBase(EntityConditionBase):
"""Base for zone-target conditions on person and device_tracker entities."""
_domain_specs = _DOMAIN_SPECS
_schema = _ZONE_CONDITION_SCHEMA
def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
"""Initialize the condition."""
super().__init__(hass, config)
assert config.options is not None
self._zone: str = config.options[CONF_ZONE]
def _in_target_zone(self, entity_state: State) -> bool:
"""Check if the entity is currently in the selected zone."""
in_zones = entity_state.attributes.get(ATTR_IN_ZONES) or ()
return self._zone in in_zones
class InZoneCondition(_ZoneTargetConditionBase):
"""Condition: targeted entity is in the selected zone."""
def is_valid_state(self, entity_state: State) -> bool:
"""Check that the entity is in the selected zone."""
return self._in_target_zone(entity_state)
class NotInZoneCondition(_ZoneTargetConditionBase):
"""Condition: targeted entity is not in the selected zone."""
def is_valid_state(self, entity_state: State) -> bool:
"""Check that the entity is not in the selected zone."""
return not self._in_target_zone(entity_state)
_OCCUPANCY_CONDITION_SCHEMA = vol.Schema(
{
vol.Required(CONF_OPTIONS, default={}): {
vol.Required(CONF_ZONE): cv.entity_domain("zone"),
vol.Optional(CONF_FOR): cv.positive_time_period,
},
}
)
class _ZoneOccupancyConditionBase(EntityConditionBase):
"""Base for zone occupancy conditions (single zone, no behavior)."""
_domain_specs = {"zone": DomainSpec()}
_schema = _OCCUPANCY_CONDITION_SCHEMA
@classmethod
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config and synthesize a target from the zone option.
We synthesize a target because we allow users to pick a single zone
to monitor, not a target.
"""
config = cast(ConfigType, cls._schema(config))
zone_entity_id: str = config[CONF_OPTIONS][CONF_ZONE]
config[CONF_TARGET] = {CONF_ENTITY_ID: [zone_entity_id]}
# `behavior` is needed by `EntityConditionBase.__init__`
config[CONF_OPTIONS][ATTR_BEHAVIOR] = BEHAVIOR_ANY
return config
@staticmethod
def _occupancy_count(entity_state: State) -> int | None:
"""Return the zone's persons-in-zone count; None if unparsable."""
try:
return int(entity_state.state)
except TypeError, ValueError:
return None
@classmethod
def _is_occupied(cls, entity_state: State) -> bool:
"""Return True if the zone has at least one occupant."""
count = cls._occupancy_count(entity_state)
return count is not None and count >= 1
class OccupancyIsDetectedCondition(_ZoneOccupancyConditionBase):
"""Condition: the selected zone is occupied."""
def is_valid_state(self, entity_state: State) -> bool:
"""Check that the zone is occupied."""
return self._is_occupied(entity_state)
class OccupancyIsNotDetectedCondition(_ZoneOccupancyConditionBase):
"""Condition: the selected zone is empty."""
def is_valid_state(self, entity_state: State) -> bool:
"""Check that the zone is empty (count == 0)."""
return self._occupancy_count(entity_state) == 0
CONDITIONS: dict[str, type[Condition]] = {
"_": ZoneCondition,
"in_zone": InZoneCondition,
"not_in_zone": NotInZoneCondition,
"occupancy_is_detected": OccupancyIsDetectedCondition,
"occupancy_is_not_detected": OccupancyIsNotDetectedCondition,
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the zone conditions."""
"""Return the sun conditions."""
return CONDITIONS
@@ -1,42 +0,0 @@
.condition_zone: &condition_zone
target:
entity:
domain:
- person
- device_tracker
fields:
behavior:
required: true
default: any
selector:
automation_behavior:
mode: condition
for:
required: true
default: 00:00:00
selector:
duration:
zone:
required: true
selector:
entity:
domain: zone
in_zone: *condition_zone
not_in_zone: *condition_zone
.condition_occupancy: &condition_occupancy
fields:
for:
required: true
default: 00:00:00
selector:
duration:
zone:
required: true
selector:
entity:
domain: zone
occupancy_is_detected: *condition_occupancy
occupancy_is_not_detected: *condition_occupancy
-14
View File
@@ -1,18 +1,4 @@
{
"conditions": {
"in_zone": {
"condition": "mdi:map-marker-check"
},
"not_in_zone": {
"condition": "mdi:map-marker-remove"
},
"occupancy_is_detected": {
"condition": "mdi:account-group"
},
"occupancy_is_not_detected": {
"condition": "mdi:account-off"
}
},
"services": {
"reload": {
"service": "mdi:reload"
@@ -1,74 +1,10 @@
{
"common": {
"condition_behavior_name": "Check when",
"condition_for_name": "For at least",
"condition_zone_description": "The zone to test against.",
"condition_zone_name": "Zone",
"trigger_behavior_name": "Trigger when",
"trigger_for_name": "For at least",
"trigger_zone_description": "The zone to trigger on.",
"trigger_zone_name": "Zone"
},
"conditions": {
"in_zone": {
"description": "Tests if one or more persons or device trackers are in a zone.",
"fields": {
"behavior": {
"name": "[%key:component::zone::common::condition_behavior_name%]"
},
"for": {
"name": "[%key:component::zone::common::condition_for_name%]"
},
"zone": {
"description": "[%key:component::zone::common::condition_zone_description%]",
"name": "[%key:component::zone::common::condition_zone_name%]"
}
},
"name": "Is in zone"
},
"not_in_zone": {
"description": "Tests if one or more persons or device trackers are not in a zone.",
"fields": {
"behavior": {
"name": "[%key:component::zone::common::condition_behavior_name%]"
},
"for": {
"name": "[%key:component::zone::common::condition_for_name%]"
},
"zone": {
"description": "[%key:component::zone::common::condition_zone_description%]",
"name": "[%key:component::zone::common::condition_zone_name%]"
}
},
"name": "Is not in zone"
},
"occupancy_is_detected": {
"description": "Tests if a zone is occupied.",
"fields": {
"for": {
"name": "[%key:component::zone::common::condition_for_name%]"
},
"zone": {
"description": "The zone to monitor.",
"name": "[%key:component::zone::common::condition_zone_name%]"
}
},
"name": "Zone occupancy is detected"
},
"occupancy_is_not_detected": {
"description": "Tests if a zone is empty.",
"fields": {
"for": {
"name": "[%key:component::zone::common::condition_for_name%]"
},
"zone": {
"description": "[%key:component::zone::conditions::occupancy_is_detected::fields::zone::description%]",
"name": "[%key:component::zone::common::condition_zone_name%]"
}
},
"name": "Zone occupancy is not detected"
}
},
"services": {
"reload": {
"description": "Reloads zones from the YAML-configuration.",
+2 -3
View File
@@ -43,7 +43,6 @@ from homeassistant.helpers.trigger import (
from homeassistant.helpers.typing import ConfigType
from . import condition
from .const import DOMAIN
EVENT_ENTER = "enter"
EVENT_LEAVE = "leave"
@@ -69,7 +68,7 @@ _LEGACY_TRIGGER_OPTIONS_SCHEMA = vol.Schema(
_ZONE_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA_WITH_BEHAVIOR.extend(
{
vol.Required(CONF_OPTIONS): {
vol.Required(CONF_ZONE): cv.entity_domain(DOMAIN),
vol.Required(CONF_ZONE): cv.entity_domain("zone"),
},
}
)
@@ -209,7 +208,7 @@ class LeftZoneTrigger(ZoneTriggerBase):
_OCCUPANCY_TRIGGER_SCHEMA = vol.Schema(
{
vol.Required(CONF_OPTIONS, default={}): {
vol.Required(CONF_ZONE): cv.entity_domain(DOMAIN),
vol.Required(CONF_ZONE): cv.entity_domain("zone"),
vol.Optional(CONF_FOR): cv.positive_time_period,
},
}
+1 -1
View File
@@ -186,6 +186,7 @@ FLOWS = {
"econet",
"ecovacs",
"ecowitt",
"edifier_infrared",
"edl21",
"efergy",
"egauge",
@@ -348,7 +349,6 @@ FLOWS = {
"imeon_inverter",
"imgw_pib",
"immich",
"imou",
"improv_ble",
"incomfort",
"indevolt",
+6 -6
View File
@@ -1654,6 +1654,12 @@
"config_flow": true,
"iot_class": "local_push"
},
"edifier_infrared": {
"name": "Edifier Infrared",
"integration_type": "device",
"config_flow": true,
"iot_class": "assumed_state"
},
"edimax": {
"name": "Edimax",
"integration_type": "hub",
@@ -3229,12 +3235,6 @@
"config_flow": true,
"iot_class": "local_polling"
},
"imou": {
"name": "Imou",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"improv_ble": {
"name": "Improv via BLE",
"integration_type": "device",
@@ -1953,7 +1953,6 @@ _SCRIPT_CHOOSE_SCHEMA = vol.Schema(
[
{
vol.Optional(CONF_ALIAS): string,
vol.Remove(CONF_NOTE): str, # Is only used in frontend
vol.Required(CONF_CONDITIONS): CONDITIONS_SCHEMA,
vol.Required(CONF_SEQUENCE): SCRIPT_SCHEMA,
}
@@ -24,11 +24,6 @@ from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_S
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import TemplateError
from homeassistant.helpers.singleton import singleton
from homeassistant.helpers.trace import (
suppress_template_error_logging_cv,
trace_stack_cv,
trace_stack_top,
)
from homeassistant.helpers.typing import TemplateVarsType
from homeassistant.util.async_ import run_callback_threadsafe
from homeassistant.util.hass_dict import HassKey
@@ -632,14 +627,6 @@ def make_logging_undefined(
return jinja2.StrictUndefined
def _log_with_logger(level: int, msg: str) -> None:
# Record the error on the active trace element so it is surfaced in the
# trace. Consumers such as the subscribe_condition websocket command can
# opt in to additionally suppress the (otherwise repeated) log entry.
if node := trace_stack_top(trace_stack_cv):
node.add_template_error(msg)
if suppress_template_error_logging_cv.get():
return
template, action = template_cv.get() or ("", "rendering or compiling")
_LOGGER.log(
level,
+2 -59
View File
@@ -5,7 +5,7 @@ from collections.abc import Callable, Coroutine, Generator
from contextlib import contextmanager
from contextvars import ContextVar
from functools import wraps
from typing import Any, Literal, overload
from typing import Any
from homeassistant.core import ServiceResponse
from homeassistant.util import dt as dt_util
@@ -22,7 +22,6 @@ class TraceElement:
"_error",
"_last_variables",
"_result",
"_template_errors",
"_timestamp",
"_variables",
"path",
@@ -36,7 +35,6 @@ class TraceElement:
self._error: BaseException | None = None
self.path: str = path
self._result: dict[str, Any] | None = None
self._template_errors: list[str] | None = None
self.reuse_by_child = False
self._timestamp = dt_util.utcnow()
@@ -56,23 +54,6 @@ class TraceElement:
"""Set error."""
self._error = ex
def add_template_error(self, msg: str) -> None:
"""Record a template error message.
Used to record template variable errors which would otherwise be logged
directly, so they are surfaced in the trace instead of spamming the log.
A single template render can emit more than one message, so they are
accumulated in a list.
"""
if self._template_errors is None:
self._template_errors = []
self._template_errors.append(msg)
@property
def template_errors(self) -> list[str]:
"""Return the recorded template error messages."""
return self._template_errors or []
def set_result(self, **kwargs: Any) -> None:
"""Set result."""
self._result = {**kwargs}
@@ -109,8 +90,6 @@ class TraceElement:
result["changed_variables"] = self._variables
if self._error is not None:
result["error"] = str(self._error) or self._error.__class__.__name__
if self._template_errors:
result["template_errors"] = self._template_errors
if self._result is not None:
result["result"] = self._result
return result
@@ -139,27 +118,6 @@ trace_id_cv: ContextVar[tuple[str, str] | None] = ContextVar(
script_execution_cv: ContextVar[StopReason | None] = ContextVar(
"script_execution_cv", default=None
)
# When set, template errors recorded on the active TraceElement are not also
# logged. Template errors are always recorded in the trace regardless.
suppress_template_error_logging_cv: ContextVar[bool] = ContextVar(
"suppress_template_error_logging_cv", default=False
)
@contextmanager
def suppress_template_error_logging() -> Generator[None]:
"""Suppress logging of template errors that are recorded in the trace.
Template errors are always recorded on the active trace element. Consumers
such as the subscribe_condition websocket command, which re-evaluate a
condition repeatedly and forward template errors to the client via the
trace, can use this to also stop the errors from spamming the log.
"""
token = suppress_template_error_logging_cv.set(True)
try:
yield
finally:
suppress_template_error_logging_cv.reset(token)
def trace_id_set(trace_id: tuple[str, str]) -> None:
@@ -231,23 +189,8 @@ def trace_append_element(
trace[path].append(trace_element)
@overload
def trace_get(clear: Literal[True] = True) -> dict[str, deque[TraceElement]]: ...
@overload
def trace_get(clear: Literal[False]) -> dict[str, deque[TraceElement]] | None: ...
def trace_get(clear: bool = True) -> dict[str, deque[TraceElement]] | None:
"""Return the current trace.
When clear is True the trace is reset and a fresh (empty) trace is
unconditionally returned.
When clear is False, the current trace is returned without modification
if it exists, otherwise None is returned.
"""
"""Return the current trace."""
if clear:
trace_clear()
return trace_cv.get()
+1 -1
View File
@@ -39,7 +39,7 @@ habluetooth==6.8.1
hass-nabucasa==2.2.0
hassil==3.5.0
home-assistant-bluetooth==2.0.0
home-assistant-frontend==20260527.4
home-assistant-frontend==20260527.2
home-assistant-intents==2026.6.1
httpx==0.28.1
ifaddr==0.2.0
+1 -1
View File
@@ -5,7 +5,7 @@ To update, run python3 -m script.hassfest
from typing import Final
FRONTEND_VERSION: Final[str] = "20260527.4"
FRONTEND_VERSION: Final[str] = "20260527.2"
MDI_ICONS: Final[set[str]] = {
"ab-testing",
+1 -4
View File
@@ -1269,7 +1269,7 @@ hole==0.9.0
holidays==0.97
# homeassistant.components.frontend
home-assistant-frontend==20260527.4
home-assistant-frontend==20260527.2
# homeassistant.components.conversation
home-assistant-intents==2026.6.1
@@ -2228,9 +2228,6 @@ pyialarm==2.2.0
# homeassistant.components.icloud
pyicloud==2.4.1
# homeassistant.components.imou
pyimouapi==1.2.7
# homeassistant.components.insteon
pyinsteon==1.6.4
+2 -2
View File
@@ -10,7 +10,7 @@
# ast-serialize is an internal mypy dependency
ast-serialize==0.3.0
astroid==4.0.4
coverage==7.14.1
coverage==7.14.0
freezegun==1.5.5
# librt is an internal mypy dependency
librt==0.11.0
@@ -22,7 +22,7 @@ pydantic==2.13.4
pylint==4.0.5
pylint-per-file-ignores==3.2.1
pipdeptree==2.26.1
pytest-asyncio==1.4.0
pytest-asyncio==1.3.0
pytest-aiohttp==1.1.0
pytest-cov==7.1.0
pytest-freezer==0.4.9
+3 -18
View File
@@ -2,11 +2,8 @@
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.avea.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
@@ -53,31 +50,19 @@ async def _setup_yaml_import(hass: HomeAssistant, bulbs: list[MagicMock]) -> Non
async def test_setup_entry_retries_when_ble_device_is_missing(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test setup retries when the Bluetooth device is unavailable."""
mock_config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.avea.async_ble_device_from_address",
return_value=None,
),
patch(
"homeassistant.components.avea.async_address_reachability_diagnostics",
return_value="mock reachability reason",
),
with patch(
"homeassistant.components.avea.async_ble_device_from_address",
return_value=None,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
assert (
"Could not find Avea device with address "
f"{mock_config_entry.data[CONF_ADDRESS]}: mock reachability reason"
in caplog.text
)
async def test_yaml_import_creates_entries_for_discovered_bulbs(
+10 -29
View File
@@ -1485,12 +1485,12 @@ async def _validate_condition_options(
options: dict[str, Any] | None,
*,
valid: bool,
supports_target: bool = True,
) -> None:
"""Assert that a condition accepts or rejects the given options."""
config: dict[str, Any] = {CONF_CONDITION: condition}
if supports_target:
config[CONF_TARGET] = {ATTR_LABEL_ID: "test_label"}
config: dict[str, Any] = {
CONF_CONDITION: condition,
CONF_TARGET: {ATTR_LABEL_ID: "test_label"},
}
if options is not None:
config[CONF_OPTIONS] = options
if valid:
@@ -1536,7 +1536,6 @@ async def assert_condition_options_supported(
*,
supports_behavior: bool,
supports_duration: bool,
supports_target: bool = True,
) -> None:
"""Assert which options a condition supports.
@@ -1556,15 +1555,9 @@ async def assert_condition_options_supported(
# Minimal config should always be valid
# If there are no base options, also test that options can be omitted or be empty
supports_empty = not bool(base_options)
await _validate_condition_options(
hass, condition, None, valid=supports_empty, supports_target=supports_target
)
await _validate_condition_options(
hass, condition, {}, valid=supports_empty, supports_target=supports_target
)
await _validate_condition_options(
hass, condition, base_options, valid=True, supports_target=supports_target
)
await _validate_condition_options(hass, condition, None, valid=supports_empty)
await _validate_condition_options(hass, condition, {}, valid=supports_empty)
await _validate_condition_options(hass, condition, base_options, valid=True)
def _merge(extra: dict[str, Any]) -> dict[str, Any]:
return {**(base_options or {}), **extra}
@@ -1572,30 +1565,18 @@ async def assert_condition_options_supported(
# Behavior
for behavior in ("any", "all"):
await _validate_condition_options(
hass,
condition,
_merge({"behavior": behavior}),
valid=supports_behavior,
supports_target=supports_target,
hass, condition, _merge({"behavior": behavior}), valid=supports_behavior
)
# Duration
for for_value in ({"seconds": 5}, "00:00:05", 5):
await _validate_condition_options(
hass,
condition,
_merge({"for": for_value}),
valid=supports_duration,
supports_target=supports_target,
hass, condition, _merge({"for": for_value}), valid=supports_duration
)
# Unknown option should always be rejected
await _validate_condition_options(
hass,
condition,
_merge({"unknown_option": True}),
valid=False,
supports_target=supports_target,
hass, condition, _merge({"unknown_option": True}), valid=False
)
@@ -0,0 +1 @@
"""Tests for the Edifier Infrared integration."""
@@ -0,0 +1,99 @@
"""Common fixtures for the Edifier Infrared tests."""
from collections.abc import Generator
from unittest.mock import patch
from infrared_protocols.codes.edifier.models import EdifierCommandSet, EdifierModel
import pytest
from homeassistant.components.edifier_infrared import PLATFORMS
from homeassistant.components.edifier_infrared.const import (
CONF_COMMAND_SET,
CONF_INFRARED_ENTITY_ID,
DOMAIN,
)
from homeassistant.const import CONF_MODEL, Platform
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.infrared import (
EMITTER_ENTITY_ID as MOCK_INFRARED_EMITTER_ENTITY_ID,
)
from tests.components.infrared.common import MockInfraredEmitterEntity
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
entry_id="01JTEST0000000000000000000",
title="Edifier R1700BT via Test IR emitter",
data={
CONF_INFRARED_ENTITY_ID: MOCK_INFRARED_EMITTER_ENTITY_ID,
CONF_MODEL: EdifierModel.R1700BT.value,
CONF_COMMAND_SET: EdifierCommandSet.R1700BT.value,
},
unique_id=f"r1700bt_{MOCK_INFRARED_EMITTER_ENTITY_ID}",
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Return platforms to set up."""
return PLATFORMS
@pytest.fixture
def mock_edifier_code_to_command() -> Generator[None]:
"""Patch Edifier *Code.to_command to return the code enum directly.
This allows tests to assert on the high-level code enum value
rather than the raw NEC timings.
"""
with (
patch(
"infrared_protocols.codes.edifier.r1700bt.EdifierR1700BTCode.to_command",
autospec=True,
side_effect=lambda self: self,
),
patch(
"infrared_protocols.codes.edifier.r1280db.EdifierR1280DBCode.to_command",
autospec=True,
side_effect=lambda self: self,
),
patch(
"infrared_protocols.codes.edifier.r1280t.EdifierR1280TCode.to_command",
autospec=True,
side_effect=lambda self: self,
),
patch(
"infrared_protocols.codes.edifier.s360db.EdifierS360DBCode.to_command",
autospec=True,
side_effect=lambda self: self,
),
patch(
"infrared_protocols.codes.edifier.rc20g.EdifierRC20GCode.to_command",
autospec=True,
side_effect=lambda self: self,
),
):
yield
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_infrared_emitter_entity: MockInfraredEmitterEntity,
mock_edifier_code_to_command: None,
platforms: list[Platform],
) -> MockConfigEntry:
"""Set up the Edifier Infrared integration for testing."""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.edifier_infrared.PLATFORMS", platforms):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
@@ -0,0 +1,55 @@
# serializer version: 1
# name: test_entities[media_player.edifier_r1700bt-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'media_player',
'entity_category': None,
'entity_id': 'media_player.edifier_r1700bt',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <MediaPlayerDeviceClass.SPEAKER: 'speaker'>,
'original_icon': None,
'original_name': None,
'platform': 'edifier_infrared',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 17849>,
'translation_key': None,
'unique_id': '01JTEST0000000000000000000_media_player',
'unit_of_measurement': None,
})
# ---
# name: test_entities[media_player.edifier_r1700bt-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'assumed_state': True,
'device_class': 'speaker',
'friendly_name': 'Edifier R1700BT',
'supported_features': <MediaPlayerEntityFeature: 17849>,
}),
'context': <ANY>,
'entity_id': 'media_player.edifier_r1700bt',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
@@ -0,0 +1,131 @@
"""Tests for the Edifier Infrared config flow."""
from infrared_protocols.codes.edifier.models import EdifierCommandSet, EdifierModel
import pytest
from homeassistant.components.edifier_infrared.const import (
CONF_COMMAND_SET,
CONF_INFRARED_ENTITY_ID,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_MODEL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
from tests.components.infrared import EMITTER_ENTITY_ID
@pytest.mark.parametrize(
("model", "expected_command_set"),
[
(EdifierModel.R1700BT, EdifierCommandSet.R1700BT),
(EdifierModel.R1280DB, EdifierCommandSet.R1280DB),
(EdifierModel.R1280T, EdifierCommandSet.R1280T),
(EdifierModel.S360DB, EdifierCommandSet.S360DB),
(EdifierModel.RC20G, EdifierCommandSet.RC20G),
],
)
@pytest.mark.usefixtures("mock_infrared_emitter_entity")
async def test_user_flow_success(
hass: HomeAssistant,
model: EdifierModel,
expected_command_set: EdifierCommandSet,
) -> None:
"""Test successful user config flow for each command set."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_INFRARED_ENTITY_ID: EMITTER_ENTITY_ID,
CONF_MODEL: model.value,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == f"Edifier {model.value} via Test IR emitter"
assert result["data"] == {
CONF_INFRARED_ENTITY_ID: EMITTER_ENTITY_ID,
CONF_MODEL: model.value,
CONF_COMMAND_SET: expected_command_set.value,
}
assert (
result["result"].unique_id
== f"{expected_command_set.value}_{EMITTER_ENTITY_ID}"
)
@pytest.mark.usefixtures("mock_infrared_emitter_entity")
async def test_user_flow_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test user flow aborts when entry is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_INFRARED_ENTITY_ID: EMITTER_ENTITY_ID,
CONF_MODEL: EdifierModel.R1700BT.value,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("init_infrared")
async def test_user_flow_no_emitters(hass: HomeAssistant) -> None:
"""Test user flow aborts when no infrared emitters exist."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_emitters"
@pytest.mark.usefixtures("mock_infrared_emitter_entity")
@pytest.mark.parametrize(
("entity_name", "expected_title"),
[
(None, "Edifier R1700BT via Test IR emitter"),
("Living room IR", "Edifier R1700BT via Living room IR"),
],
)
async def test_user_flow_title_from_entity_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
entity_name: str | None,
expected_title: str,
) -> None:
"""Test config entry title uses the entity name."""
entity_registry.async_update_entity(EMITTER_ENTITY_ID, name=entity_name)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_INFRARED_ENTITY_ID: EMITTER_ENTITY_ID,
CONF_MODEL: EdifierModel.R1700BT.value,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == expected_title
@@ -0,0 +1,19 @@
"""Tests for the Edifier Infrared integration setup."""
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_setup_and_unload_entry(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
"""Test setting up and unloading a config entry."""
entry = init_integration
assert entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
@@ -0,0 +1,143 @@
"""Tests for the Edifier Infrared media player platform."""
from infrared_protocols.codes.edifier.models import EdifierCommandSet, EdifierModel
from infrared_protocols.codes.edifier.r1700bt import EdifierR1700BTCode
from infrared_protocols.codes.edifier.rc20g import EdifierRC20GCode
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.edifier_infrared.const import (
CONF_COMMAND_SET,
CONF_INFRARED_ENTITY_ID,
DOMAIN,
)
from homeassistant.components.media_player import (
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_UP,
)
from homeassistant.const import ATTR_ENTITY_ID, CONF_MODEL, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
from tests.components.common import assert_availability_follows_source_entity
from tests.components.infrared import EMITTER_ENTITY_ID
from tests.components.infrared.common import MockInfraredEmitterEntity
MEDIA_PLAYER_ENTITY_ID = "media_player.edifier_r1700bt"
@pytest.fixture
def platforms() -> list[Platform]:
"""Return platforms to set up."""
return [Platform.MEDIA_PLAYER]
@pytest.mark.usefixtures("init_integration")
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the media player entity is created with correct attributes."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("service", "service_data", "expected_code"),
[
(SERVICE_TURN_ON, {}, EdifierR1700BTCode.POWER),
(SERVICE_TURN_OFF, {}, EdifierR1700BTCode.POWER),
(SERVICE_VOLUME_UP, {}, EdifierR1700BTCode.VOLUME_UP),
(SERVICE_VOLUME_DOWN, {}, EdifierR1700BTCode.VOLUME_DOWN),
(SERVICE_VOLUME_MUTE, {"is_volume_muted": True}, EdifierR1700BTCode.MUTE),
(SERVICE_MEDIA_PLAY, {}, EdifierR1700BTCode.PLAY_PAUSE),
(SERVICE_MEDIA_PAUSE, {}, EdifierR1700BTCode.PLAY_PAUSE),
(SERVICE_MEDIA_NEXT_TRACK, {}, EdifierR1700BTCode.FORWARD),
(SERVICE_MEDIA_PREVIOUS_TRACK, {}, EdifierR1700BTCode.BACK),
],
)
@pytest.mark.usefixtures("init_integration")
async def test_media_player_action_sends_correct_code(
hass: HomeAssistant,
mock_infrared_emitter_entity: MockInfraredEmitterEntity,
service: str,
service_data: dict[str, bool],
expected_code: EdifierR1700BTCode,
) -> None:
"""Test each media player action sends the correct IR code."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: MEDIA_PLAYER_ENTITY_ID, **service_data},
blocking=True,
)
assert len(mock_infrared_emitter_entity.send_command_calls) == 1
assert mock_infrared_emitter_entity.send_command_calls[0] == expected_code
@pytest.mark.parametrize(
"mock_config_entry",
[
MockConfigEntry(
domain=DOMAIN,
entry_id="01JTEST0000000000000000001",
title="Edifier RC20G via Test IR emitter",
data={
CONF_INFRARED_ENTITY_ID: EMITTER_ENTITY_ID,
CONF_MODEL: EdifierModel.RC20G.value,
CONF_COMMAND_SET: EdifierCommandSet.RC20G.value,
},
unique_id=f"rc20g_{EMITTER_ENTITY_ID}",
)
],
)
@pytest.mark.parametrize(
("service", "expected_codes"),
[
(
SERVICE_VOLUME_UP,
(EdifierRC20GCode.VOLUME_UP_LEFT, EdifierRC20GCode.VOLUME_UP_RIGHT),
),
(
SERVICE_VOLUME_DOWN,
(EdifierRC20GCode.VOLUME_DOWN_LEFT, EdifierRC20GCode.VOLUME_DOWN_RIGHT),
),
],
)
@pytest.mark.usefixtures("init_integration")
async def test_rc20g_volume_sends_left_and_right_codes(
hass: HomeAssistant,
mock_infrared_emitter_entity: MockInfraredEmitterEntity,
service: str,
expected_codes: tuple[EdifierRC20GCode, ...],
) -> None:
"""Test that RC20G volume up/down send both left and right channel codes."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: "media_player.edifier_rc20g"},
blocking=True,
)
assert tuple(mock_infrared_emitter_entity.send_command_calls) == expected_codes
@pytest.mark.usefixtures("init_integration")
async def test_media_player_availability_follows_ir_entity(
hass: HomeAssistant,
) -> None:
"""Test media player becomes unavailable when IR entity is unavailable."""
await assert_availability_follows_source_entity(
hass, MEDIA_PLAYER_ENTITY_ID, EMITTER_ENTITY_ID
)
-1
View File
@@ -1 +0,0 @@
"""Tests for the Imou integration."""
-89
View File
@@ -1,89 +0,0 @@
"""Test configuration and fixtures for Imou integration."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from pyimouapi.ha_device import ImouHaDevice
import pytest
from homeassistant.components.imou.const import CONF_APP_ID, DOMAIN
from homeassistant.core import HomeAssistant
from .const import CONFIG_ENTRY_DATA, DEFAULT_MOCK_DEVICES
from tests.common import MockConfigEntry
PATCH_IMOU_OPENAPI_CLIENT = "homeassistant.components.imou.ImouOpenApiClient"
PATCH_CONFIG_FLOW_IMOU_OPENAPI_CLIENT = (
"homeassistant.components.imou.config_flow.ImouOpenApiClient"
)
PATCH_IMOU_HA_DEVICE_MANAGER = "homeassistant.components.imou.ImouHaDeviceManager"
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="Imou",
domain=DOMAIN,
data=CONFIG_ENTRY_DATA,
unique_id=CONFIG_ENTRY_DATA[CONF_APP_ID],
entry_id="test_entry_id",
)
@pytest.fixture
def mock_imou_openapi_client() -> Generator[AsyncMock]:
"""Mock ImouOpenApiClient for config flow and setup entry."""
with (
patch(
PATCH_IMOU_OPENAPI_CLIENT,
autospec=True,
) as mock_client,
patch(
PATCH_CONFIG_FLOW_IMOU_OPENAPI_CLIENT,
new=mock_client,
),
):
yield mock_client.return_value
@pytest.fixture
def imou_mock_devices(request: pytest.FixtureRequest) -> list[ImouHaDevice]:
"""Devices returned by ImouHaDeviceManager.async_get_devices (override via indirect)."""
return getattr(request, "param", DEFAULT_MOCK_DEVICES)
@pytest.fixture
def mock_imou_ha_device_manager(
imou_mock_devices: list[ImouHaDevice],
) -> Generator[MagicMock]:
"""Mock ImouHaDeviceManager with a default device list."""
with patch(PATCH_IMOU_HA_DEVICE_MANAGER, autospec=True) as mock_manager:
device_manager = mock_manager.return_value
device_manager.async_get_devices.return_value = imou_mock_devices
yield device_manager
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry so config flow tests do not load the full integration."""
with patch(
"homeassistant.components.imou.async_setup_entry",
return_value=True,
) as mock_setup:
yield mock_setup
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_imou_openapi_client: AsyncMock,
mock_imou_ha_device_manager: MagicMock,
) -> MagicMock:
"""Set up Imou with mocked library clients; returns the HA device manager mock."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_imou_ha_device_manager
-95
View File
@@ -1,95 +0,0 @@
"""Constants for the Imou tests."""
from pyimouapi.ha_device import DeviceStatus, ImouHaDevice
from homeassistant.components.imou.button import (
PARAM_MUTE,
PARAM_PTZ_UP,
PARAM_RESTART_DEVICE,
)
from homeassistant.components.imou.const import (
CONF_API_URL,
CONF_APP_ID,
CONF_APP_SECRET,
PARAM_STATE,
PARAM_STATUS,
)
TEST_APP_ID = "test_app_id"
TEST_APP_SECRET = "test_app_secret"
TEST_API_URL = "sg"
USER_INPUT = {
CONF_APP_ID: TEST_APP_ID,
CONF_APP_SECRET: TEST_APP_SECRET,
CONF_API_URL: TEST_API_URL,
}
CONFIG_ENTRY_DATA = {
CONF_APP_ID: TEST_APP_ID,
CONF_APP_SECRET: TEST_APP_SECRET,
CONF_API_URL: TEST_API_URL,
}
UNKNOWN_BUTTON_KEY = "legacy_unknown_button"
def create_online_device(
device_id: str,
name: str,
*,
channel_id: str | None = None,
button_keys: tuple[str, ...] = (),
) -> ImouHaDevice:
"""Build an online ImouHaDevice for tests."""
return create_device(
device_id,
name,
channel_id=channel_id,
button_keys=button_keys,
status=DeviceStatus.ONLINE,
)
def create_offline_device(
device_id: str,
name: str,
*,
channel_id: str | None = None,
button_keys: tuple[str, ...] = (),
) -> ImouHaDevice:
"""Build an offline ImouHaDevice for tests."""
return create_device(
device_id,
name,
channel_id=channel_id,
button_keys=button_keys,
status=DeviceStatus.OFFLINE,
)
def create_device(
device_id: str,
name: str,
*,
channel_id: str | None = None,
button_keys: tuple[str, ...] = (),
status: DeviceStatus = DeviceStatus.ONLINE,
) -> ImouHaDevice:
"""Build an ImouHaDevice for tests."""
device = ImouHaDevice(device_id, name, "Imou", "m1", "1.0")
if channel_id is not None:
device.set_channel_id(channel_id)
for key in button_keys:
device._buttons[key] = {}
device._sensors[PARAM_STATUS] = {PARAM_STATE: status.value}
return device
DEFAULT_MOCK_DEVICES = [
create_online_device(
"d1",
"Device 1",
button_keys=(PARAM_MUTE, PARAM_PTZ_UP, PARAM_RESTART_DEVICE),
),
]
@@ -1,152 +0,0 @@
# serializer version: 1
# name: test_button_entities_snapshot[button.device_1_mute-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.device_1_mute',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Mute',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Mute',
'platform': 'imou',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'mute',
'unique_id': 'd1$mute',
'unit_of_measurement': None,
})
# ---
# name: test_button_entities_snapshot[button.device_1_mute-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device 1 Mute',
}),
'context': <ANY>,
'entity_id': 'button.device_1_mute',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button_entities_snapshot[button.device_1_ptz_up-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.device_1_ptz_up',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'PTZ up',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'PTZ up',
'platform': 'imou',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ptz_up',
'unique_id': 'd1$ptz_up',
'unit_of_measurement': None,
})
# ---
# name: test_button_entities_snapshot[button.device_1_ptz_up-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device 1 PTZ up',
}),
'context': <ANY>,
'entity_id': 'button.device_1_ptz_up',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button_entities_snapshot[button.device_1_restart-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.device_1_restart',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Restart',
'options': dict({
}),
'original_device_class': <ButtonDeviceClass.RESTART: 'restart'>,
'original_icon': None,
'original_name': 'Restart',
'platform': 'imou',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'd1$restart_device',
'unit_of_measurement': None,
})
# ---
# name: test_button_entities_snapshot[button.device_1_restart-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'restart',
'friendly_name': 'Device 1 Restart',
}),
'context': <ANY>,
'entity_id': 'button.device_1_restart',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
-220
View File
@@ -1,220 +0,0 @@
"""Tests for Imou button platform."""
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from pyimouapi.exceptions import ImouException
from pyimouapi.ha_device import DeviceStatus, ImouHaDevice
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.components.imou.button import PARAM_MUTE, PARAM_PTZ_UP
from homeassistant.components.imou.const import (
PARAM_STATE,
PARAM_STATUS,
PTZ_MOVE_DURATION_MS,
)
from homeassistant.components.imou.coordinator import SCAN_INTERVAL
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .const import UNKNOWN_BUTTON_KEY, create_online_device
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
@pytest.mark.usefixtures("init_integration")
async def test_button_entities_snapshot(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
) -> None:
"""Snapshot button entities created from the default mock device list."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
"imou_mock_devices",
[
[
create_online_device(
"d1",
"Device 1",
button_keys=(UNKNOWN_BUTTON_KEY, PARAM_MUTE),
)
]
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_setup_ignores_unknown_button_types(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Unknown button keys from the API are not turned into entities."""
registry = er.async_get(hass)
entries = er.async_entries_for_config_entry(registry, mock_config_entry.entry_id)
assert len(entries) == 1
assert entries[0].translation_key == PARAM_MUTE
@pytest.mark.usefixtures("init_integration")
async def test_press_button_via_service(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
init_integration: MagicMock,
) -> None:
"""Pressing a button calls the vendor library through the coordinator."""
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
mute_entry = next(e for e in entries if e.translation_key == PARAM_MUTE)
entity_id = mute_entry.entity_id
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
init_integration.async_press_button.assert_awaited_once()
call = init_integration.async_press_button.await_args
assert call is not None
assert call.args[1] == PARAM_MUTE
assert call.args[2] == 0
@pytest.mark.usefixtures("init_integration")
async def test_press_ptz_button_passes_move_duration(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
init_integration: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""PTZ buttons pass the configured move duration to the vendor library."""
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
ptz_entry = next(e for e in entries if e.translation_key == PARAM_PTZ_UP)
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: ptz_entry.entity_id},
blocking=True,
)
init_integration.async_press_button.assert_awaited_once()
call = init_integration.async_press_button.await_args
assert call is not None
assert call.args[1] == PARAM_PTZ_UP
assert call.args[2] == PTZ_MOVE_DURATION_MS
@pytest.mark.usefixtures("init_integration")
async def test_press_button_service_propagates_api_error(
hass: HomeAssistant,
init_integration: MagicMock,
) -> None:
"""Imou API errors from async_press_button surface to the service call."""
init_integration.async_press_button.side_effect = ImouException("cloud failure")
entity_id = hass.states.async_all("button")[0].entity_id
with pytest.raises(HomeAssistantError, match="cloud failure"):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
@pytest.mark.parametrize(
"imou_mock_devices",
[
[
create_online_device(
"d1",
"Device 1",
button_keys=(PARAM_MUTE,),
)
]
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_press_unavailable_offline_device_via_service(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
init_integration: MagicMock,
) -> None:
"""Pressing an offline device does not call the vendor library."""
mute_entry = next(
entry
for entry in er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
if entry.unique_id == "d1$mute"
)
async def set_device_offline(device: ImouHaDevice) -> None:
device._sensors[PARAM_STATUS] = {PARAM_STATE: DeviceStatus.OFFLINE.value}
mock_imou_ha_device_manager.async_update_device_status.side_effect = (
set_device_offline
)
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(mute_entry.entity_id).state == STATE_UNAVAILABLE
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: mute_entry.entity_id},
blocking=True,
)
init_integration.async_press_button.assert_not_called()
@pytest.mark.usefixtures("init_integration")
async def test_entities_removed_when_device_leaves_account(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Button entities are removed when the device is no longer on the account."""
mute_entry = next(
entry
for entry in er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
if entry.unique_id == "d1$mute"
)
assert hass.states.get(mute_entry.entity_id).state != STATE_UNAVAILABLE
mock_imou_ha_device_manager.async_get_devices.return_value = []
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert (
er.async_entries_for_config_entry(entity_registry, mock_config_entry.entry_id)
== []
)
assert hass.states.get(mute_entry.entity_id) is None
-152
View File
@@ -1,152 +0,0 @@
"""Tests for the Imou config flow."""
from unittest.mock import AsyncMock
from pyimouapi.exceptions import (
ConnectFailedException,
ImouException,
InvalidAppIdOrSecretException,
RequestFailedException,
)
import pytest
from homeassistant.components.imou.const import (
CONF_API_URL,
CONF_APP_ID,
CONF_APP_SECRET,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import TEST_APP_ID, TEST_APP_SECRET, USER_INPUT
from tests.common import MockConfigEntry
async def test_user_flow_success(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_imou_openapi_client: AsyncMock,
) -> None:
"""Test successful user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Imou"
assert result["data"] == USER_INPUT
assert result["result"].unique_id == USER_INPUT[CONF_APP_ID]
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_flow_duplicate_entry(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_imou_openapi_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test duplicate entry is aborted."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("side_effect", "expected_error"),
[
(ConnectFailedException("fail"), "cannot_connect"),
(RequestFailedException("fail"), "cannot_connect"),
(InvalidAppIdOrSecretException("fail"), "invalid_auth"),
(ImouException("fail"), "unknown"),
],
)
async def test_user_flow_exception_then_recover(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_imou_openapi_client: AsyncMock,
side_effect: Exception,
expected_error: str,
) -> None:
"""Errors map to stable keys; clearing the failure allows completing the flow."""
mock_imou_openapi_client.async_get_token.side_effect = side_effect
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert "errors" in result
assert result["errors"]["base"] == expected_error
mock_imou_openapi_client.async_get_token.reset_mock(side_effect=True)
recover = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=USER_INPUT,
)
assert recover["type"] is FlowResultType.CREATE_ENTRY
assert recover["title"] == "Imou"
assert recover["data"] == USER_INPUT
assert recover["result"].unique_id == USER_INPUT[CONF_APP_ID]
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize("region", ["sg", "eu", "na", "cn"])
async def test_user_flow_success_per_region(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_imou_openapi_client: AsyncMock,
region: str,
) -> None:
"""Each supported API region can complete the config flow."""
user_input = {
CONF_APP_ID: f"{TEST_APP_ID}_{region}",
CONF_APP_SECRET: TEST_APP_SECRET,
CONF_API_URL: region,
}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=user_input,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Imou"
assert result["data"] == user_input
assert result["result"].unique_id == user_input[CONF_APP_ID]
-354
View File
@@ -1,354 +0,0 @@
"""Tests for the Imou init."""
from unittest.mock import AsyncMock, MagicMock
from freezegun.api import FrozenDateTimeFactory
from pyimouapi.exceptions import ImouException
from pyimouapi.ha_device import DeviceStatus, ImouHaDevice
import pytest
from homeassistant.components.imou.button import PARAM_MUTE, PARAM_PTZ_UP
from homeassistant.components.imou.const import DOMAIN, PARAM_STATE, PARAM_STATUS
from homeassistant.components.imou.coordinator import SCAN_INTERVAL
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .const import DEFAULT_MOCK_DEVICES, create_offline_device, create_online_device
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.usefixtures("mock_imou_openapi_client", "mock_imou_ha_device_manager")
async def test_setup_and_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
init_integration: MagicMock,
) -> None:
"""Test loading and unloading the config entry."""
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.usefixtures("mock_imou_openapi_client", "mock_imou_ha_device_manager")
async def test_setup_entry_failed_on_refresh(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: AsyncMock,
) -> None:
"""Device fetch failure during coordinator setup surfaces as setup retry."""
mock_imou_ha_device_manager.async_get_devices.side_effect = RuntimeError(
"Setup failed"
)
mock_config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
@pytest.mark.usefixtures("init_integration")
async def test_device_registry_identifiers(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Device registry uses channel-aware identifiers from the default mock devices."""
registry = dr.async_get(hass)
devices = dr.async_entries_for_config_entry(registry, mock_config_entry.entry_id)
assert len(devices) == 1
assert (DOMAIN, "d1") in devices[0].identifiers
@pytest.mark.parametrize(
"imou_mock_devices",
[
[
create_online_device(
"dev-1",
"Cam",
channel_id="ch9",
button_keys=(PARAM_MUTE,),
),
create_online_device(
"dev-1",
"Cam",
channel_id="ch10",
button_keys=(PARAM_MUTE,),
),
]
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_multiple_channels_create_separate_devices(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Each channel gets its own device and button entities in the registries."""
devices = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
device_ids_by_key = {
next(iter(device.identifiers))[1]: device.id for device in devices
}
assert set(device_ids_by_key) == {"dev-1_ch9", "dev-1_ch10"}
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entries) == 2
assert {entry.unique_id for entry in entries} == {
"dev-1_ch9$mute",
"dev-1_ch10$mute",
}
for entry in entries:
assert entry.translation_key == PARAM_MUTE
device_key = entry.unique_id.split("$", 1)[0]
assert entry.device_id == device_ids_by_key[device_key]
state = hass.states.get(entry.entity_id)
assert state is not None
assert state.state != STATE_UNAVAILABLE
@pytest.mark.parametrize("imou_mock_devices", [[]], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_coordinator_adds_entities_after_initial_empty_device_list(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Devices added after an empty first refresh still get entities via callbacks."""
assert (
len(
er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
)
== 0
)
mock_imou_ha_device_manager.async_get_devices.return_value = DEFAULT_MOCK_DEVICES
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entries) == 3
assert {entry.unique_id for entry in entries} == {
"d1$mute",
"d1$ptz_up",
"d1$restart_device",
}
@pytest.mark.usefixtures("init_integration")
async def test_coordinator_adds_entities_for_new_device(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""A device added to the Imou account is discovered on the next coordinator refresh."""
assert (
len(
er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
)
== 3
)
mock_imou_ha_device_manager.async_get_devices.return_value = [
*DEFAULT_MOCK_DEVICES,
create_online_device("d2", "Device 2", button_keys=(PARAM_PTZ_UP,)),
]
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert len(entries) == 4
assert "d2$ptz_up" in {entry.unique_id for entry in entries}
ptz_entry = next(entry for entry in entries if entry.unique_id == "d2$ptz_up")
assert hass.states.get(ptz_entry.entity_id).state != STATE_UNAVAILABLE
devices = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert len(devices) == 2
device_keys = {next(iter(device.identifiers))[1] for device in devices}
assert device_keys == {"d1", "d2"}
@pytest.mark.parametrize(
"imou_mock_devices",
[
[
create_online_device("d1", "Device 1", button_keys=(PARAM_MUTE,)),
create_online_device("d2", "Device 2", button_keys=(PARAM_PTZ_UP,)),
]
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_coordinator_removes_device_updates_registries(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""A removed device is dropped from the device and entity registries."""
assert (
len(
dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
)
== 2
)
entries_before = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert {entry.unique_id for entry in entries_before} == {
"d1$mute",
"d2$ptz_up",
}
mock_imou_ha_device_manager.async_get_devices.return_value = DEFAULT_MOCK_DEVICES
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
devices = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert len(devices) == 1
assert (DOMAIN, "d1") in devices[0].identifiers
entries_after = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert {entry.unique_id for entry in entries_after} == {"d1$mute"}
mute_entry = next(entry for entry in entries_after if entry.unique_id == "d1$mute")
assert hass.states.get(mute_entry.entity_id).state != STATE_UNAVAILABLE
@pytest.mark.parametrize(
"imou_mock_devices",
[
[
create_online_device(
"d1",
"Device 1",
button_keys=(PARAM_MUTE,),
)
]
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_offline_device_marked_unavailable_after_refresh(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
entity_registry: er.EntityRegistry,
) -> None:
"""An offline device reported on refresh marks button entities unavailable."""
mute_entry = next(
entry
for entry in er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
if entry.unique_id == "d1$mute"
)
assert hass.states.get(mute_entry.entity_id).state != STATE_UNAVAILABLE
async def set_device_offline(device: ImouHaDevice) -> None:
device._sensors[PARAM_STATUS] = {PARAM_STATE: DeviceStatus.OFFLINE.value}
mock_imou_ha_device_manager.async_update_device_status.side_effect = (
set_device_offline
)
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(mute_entry.entity_id).state == STATE_UNAVAILABLE
@pytest.mark.usefixtures("init_integration")
async def test_coordinator_update_fails_when_all_devices_fail(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
entity_registry: er.EntityRegistry,
) -> None:
"""When every device status update fails, the coordinator update fails."""
mute_entry = next(
entry
for entry in er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
if entry.unique_id == "d1$mute"
)
assert hass.states.get(mute_entry.entity_id).state != STATE_UNAVAILABLE
mock_imou_ha_device_manager.async_update_device_status.side_effect = ImouException(
"cloud failure"
)
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert mock_config_entry.runtime_data.last_update_success is False
assert hass.states.get(mute_entry.entity_id).state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
"imou_mock_devices",
[
[
create_offline_device(
"d1",
"Device 1",
button_keys=(PARAM_MUTE,),
)
]
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_offline_device_unavailable_at_setup(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""An offline device marks button entities unavailable via the state machine."""
mute_entry = next(
entry
for entry in er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
if entry.unique_id == "d1$mute"
)
assert hass.states.get(mute_entry.entity_id).state == STATE_UNAVAILABLE
+2 -12
View File
@@ -79,24 +79,14 @@ async def test_set_value(
"authentication_error",
None,
),
(
TimeoutError("timed out"),
HomeAssistantError,
"communication_error",
None,
),
(TimeoutError("timed out"), HomeAssistantError, "communication_error", None),
(
ServerTimeoutError("timed out"),
HomeAssistantError,
"communication_error",
None,
),
(
ParseJSONError("bad json"),
HomeAssistantError,
"communication_error",
None,
),
(ParseJSONError("bad json"), HomeAssistantError, "communication_error", None),
(
UnsupportedFeature("old firmware"),
HomeAssistantError,
-31
View File
@@ -392,37 +392,6 @@ async def test_blindtilt_controlling(
assert state.attributes[ATTR_CURRENT_TILT_POSITION] == 50
async def test_blindtilt_idle_advertisement(
hass: HomeAssistant, mock_entry_factory: Callable[[str], MockConfigEntry]
) -> None:
"""Test blindtilt handles BLE advertisement without motionDirection."""
inject_bluetooth_service_info(hass, WOBLINDTILT_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="blind_tilt")
entry.add_to_hass(hass)
with patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotBlindTilt.get_basic_info",
new=AsyncMock(return_value={}),
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_id = "cover.test_name"
address = "AA:BB:CC:DD:EE:FF"
service_data = b"x\x00*"
manufacturer_data = b"\xfbgA`\x98\xe8\x1d%F\x12\x85"
inject_bluetooth_service_info(
hass, make_advertisement(address, manufacturer_data, service_data)
)
await hass.async_block_till_done()
# Should not crash; entity should still exist
state = hass.states.get(entity_id)
assert state is not None
async def test_roller_shade_setup(
hass: HomeAssistant, mock_entry_factory: Callable[[str], MockConfigEntry]
) -> None:
+4 -4
View File
@@ -1,5 +1,6 @@
"""The tests for the Template button platform."""
import datetime as dt
from typing import Any
from freezegun.api import FrozenDateTimeFactory
@@ -23,7 +24,6 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.util import dt as dt_util
from .conftest import (
ConfigurationStyle,
@@ -145,7 +145,7 @@ async def test_missing_emtpy_press_action_config(
"""Test: missing optional template is ok."""
_verify(hass, STATE_UNKNOWN)
now = dt_util.utcnow()
now = dt.datetime.now(dt.UTC) # pylint: disable=home-assistant-enforce-utcnow
freezer.move_to(now)
await hass.services.async_call(
BUTTON_DOMAIN,
@@ -196,7 +196,7 @@ async def test_device_class_option(
TEST_BUTTON.entity_id,
)
now = dt_util.utcnow()
now = dt.datetime.now(dt.UTC) # pylint: disable=home-assistant-enforce-utcnow
freezer.move_to(now)
await hass.services.async_call(
BUTTON_DOMAIN,
@@ -252,7 +252,7 @@ async def test_options_that_are_templates(
_verify(hass, STATE_UNKNOWN, expected_attributes)
now = dt_util.utcnow()
now = dt.datetime.now(dt.UTC) # pylint: disable=home-assistant-enforce-utcnow
freezer.move_to(now)
await hass.services.async_call(
BUTTON_DOMAIN,
+11 -256
View File
@@ -2821,131 +2821,6 @@ async def test_test_condition(
assert msg["result"]["result"] is False
@pytest.mark.parametrize(
("value_template", "expected_template_errors"),
[
("{{ no_such_variable }}", ["'no_such_variable' is undefined"]),
# A single render emitting multiple errors forwards all of them
("{{ foo }}{{ bar }}", ["'foo' is undefined", "'bar' is undefined"]),
],
)
async def test_test_condition_template_error(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
caplog: pytest.LogCaptureFixture,
value_template: str,
expected_template_errors: list[str],
) -> None:
"""Test template errors are forwarded in the result without being logged."""
caplog.set_level(logging.WARNING)
await websocket_client.send_json_auto_id(
{
"type": "test_condition",
"condition": {"condition": "template", "value_template": value_template},
}
)
msg = await websocket_client.receive_json()
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert msg["result"] == {
"result": False,
"template_errors": expected_template_errors,
}
assert "Template variable" not in caplog.text
@pytest.mark.parametrize(
("condition", "expected_error"),
[
# Missing mandatory config, raised by async_validate_condition_config
(
{"condition": "sun"},
{
"code": "invalid_format",
"message": (
"must contain at least one of before, after. for dictionary value "
"@ data['options']"
),
},
),
# Failing enabled template, raised by async_condition_from_config
(
{
"condition": "template",
"value_template": "{{ true }}",
"enabled": "{{ 1 / 0 }}",
},
{
"code": "home_assistant_error",
"message": (
"Error rendering condition enabled template: "
"ZeroDivisionError: division by zero"
),
},
),
],
)
async def test_test_condition_config_error(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
caplog: pytest.LogCaptureFixture,
condition: dict,
expected_error: dict,
) -> None:
"""Test condition config errors are reported to the client without logging."""
caplog.set_level(logging.ERROR)
await websocket_client.send_json_auto_id(
{"type": "test_condition", "condition": condition}
)
msg = await websocket_client.receive_json()
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"] == expected_error
# The expected error is not logged by the default websocket error handler
assert "Error handling message" not in caplog.text
async def test_test_condition_check_error_not_logged(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test errors raised while checking the condition are not logged.
The condition is valid and instantiates fine, but checking it raises (here
the entity does not exist). The error is reported to the client without
being logged by the default websocket error handler.
"""
caplog.set_level(logging.ERROR)
await websocket_client.send_json_auto_id(
{
"type": "test_condition",
"condition": {
"condition": "state",
"entity_id": "hello.world",
"state": "paulus",
},
}
)
msg = await websocket_client.receive_json()
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"] == {
"code": "home_assistant_error",
"message": "In 'state':\n In 'state' condition: unknown entity hello.world",
}
assert "Error handling message" not in caplog.text
async def test_subscribe_condition(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
@@ -2993,83 +2868,6 @@ async def test_subscribe_condition(
}
@pytest.mark.parametrize(
("value_template", "expected_event"),
[
# Undefined variable used in a way that raises: forwarded as an error,
# with the underlying template error included.
(
"{{ trigger.to_state.attributes.event_type == 'double_press' }}",
{
"error": "In 'template' condition: UndefinedError: 'trigger' is undefined",
"template_errors": ["'trigger' is undefined"],
},
),
# Undefined variable used in a way that only warns: the condition still
# evaluates to a result, but the template error is forwarded alongside it.
(
"{{ no_such_variable }}",
{"result": False, "template_errors": ["'no_such_variable' is undefined"]},
),
# A single render emitting multiple errors forwards all of them.
(
"{{ foo }}{{ bar }}",
{
"result": False,
"template_errors": ["'foo' is undefined", "'bar' is undefined"],
},
),
],
)
async def test_subscribe_condition_template_error(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
value_template: str,
expected_event: dict[str, Any],
) -> None:
"""Test template errors are forwarded as events and don't spam the log."""
caplog.set_level(logging.WARNING)
await websocket_client.send_json_auto_id(
{
"type": "subscribe_condition",
"condition": {
"condition": "template",
"value_template": value_template,
},
}
)
msg = await websocket_client.receive_json()
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
subscription_id = msg["id"]
msg = await websocket_client.receive_json()
assert msg == {
"id": subscription_id,
"type": "event",
"event": expected_event,
}
# Let the condition be evaluated a few more times
for _ in range(5):
freezer.tick(1.1)
await hass.async_block_till_done()
# The unchanged result/error is not re-sent; a ping is the next message
await websocket_client.send_json_auto_id({"type": "ping"})
msg = await websocket_client.receive_json()
assert msg["type"] == "pong"
# The template error is forwarded, not logged
assert "Template variable warning" not in caplog.text
assert "Template variable error" not in caplog.text
@pytest.mark.parametrize(
("condition", "expected_error"),
[
@@ -3094,6 +2892,17 @@ async def test_subscribe_condition_template_error(
),
},
),
# Validated by async_validate_condition_config
(
{"condition": "sun"},
{
"code": "invalid_format",
"message": (
"must contain at least one of before, after. for dictionary value "
"@ data['options']. Got None"
),
},
),
],
)
async def test_subscribe_condition_error(
@@ -3115,60 +2924,6 @@ async def test_subscribe_condition_error(
assert msg["error"] == expected_error
@pytest.mark.parametrize(
("condition", "expected_error"),
[
# Missing mandatory config, raised by async_validate_condition_config
(
{"condition": "sun"},
{
"code": "invalid_format",
"message": (
"must contain at least one of before, after. for dictionary value "
"@ data['options']"
),
},
),
# Failing enabled template, raised by async_condition_from_config
(
{
"condition": "template",
"value_template": "{{ true }}",
"enabled": "{{ 1 / 0 }}",
},
{
"code": "home_assistant_error",
"message": (
"Error rendering condition enabled template: "
"ZeroDivisionError: division by zero"
),
},
),
],
)
async def test_subscribe_condition_config_error(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
caplog: pytest.LogCaptureFixture,
condition: dict,
expected_error: dict,
) -> None:
"""Test condition config errors are reported to the client without logging."""
caplog.set_level(logging.ERROR)
await websocket_client.send_json_auto_id(
{"type": "subscribe_condition", "condition": condition}
)
msg = await websocket_client.receive_json()
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"] == expected_error
# The expected error is not logged by the default websocket error handler
assert "Error handling message" not in caplog.text
async def test_execute_script(
hass: HomeAssistant, websocket_client: MockHAClientWebSocket
) -> None:
-337
View File
@@ -1,29 +1,12 @@
"""The tests for the location condition."""
from datetime import timedelta
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
import voluptuous as vol
from homeassistant.components.zone import condition as zone_condition
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConditionError
from homeassistant.helpers import condition, config_validation as cv
from tests.components.common import (
ConditionStateDescription,
assert_condition_behavior_all,
assert_condition_behavior_any,
assert_condition_options_supported,
parametrize_condition_states_all,
parametrize_condition_states_any,
parametrize_target_entities,
target_entities,
)
async def test_zone_raises(hass: HomeAssistant) -> None:
"""Test that zone raises ConditionError on errors."""
@@ -223,323 +206,3 @@ async def test_multiple_zones(hass: HomeAssistant) -> None:
{"friendly_name": "person", "latitude": 50.1, "longitude": 20.1},
)
assert not test.async_check()
# --- New-style zone condition tests ---
ZONE_HOME = "zone.home"
ZONE_WORK = "zone.work"
IN_ZONES_HOME = {"in_zones": [ZONE_HOME]}
IN_ZONES_WORK = {"in_zones": [ZONE_WORK]}
IN_ZONES_NONE: dict[str, list[str]] = {"in_zones": []}
TARGET_ZONE = ZONE_HOME
@pytest.mark.parametrize(
(
"condition_key",
"base_options",
"supports_behavior",
"supports_duration",
"supports_target",
),
[
("zone.in_zone", {"zone": TARGET_ZONE}, True, True, True),
("zone.not_in_zone", {"zone": TARGET_ZONE}, True, True, True),
("zone.occupancy_is_detected", {"zone": ZONE_HOME}, False, True, False),
("zone.occupancy_is_not_detected", {"zone": ZONE_HOME}, False, True, False),
],
)
async def test_zone_condition_options_validation(
hass: HomeAssistant,
condition_key: str,
base_options: dict[str, Any] | None,
supports_behavior: bool,
supports_duration: bool,
supports_target: bool,
) -> None:
"""Test that zone conditions support the expected options."""
await assert_condition_options_supported(
hass,
condition_key,
base_options,
supports_behavior=supports_behavior,
supports_duration=supports_duration,
supports_target=supports_target,
)
@pytest.mark.parametrize(
("condition_key", "config"),
[
(
"zone.in_zone",
{"target": {"entity_id": "person.alice"}, "options": {"zone": "light.x"}},
),
(
"zone.not_in_zone",
{"target": {"entity_id": "person.alice"}, "options": {"zone": "light.x"}},
),
(
"zone.occupancy_is_detected",
{"options": {"zone": "light.x"}},
),
(
"zone.occupancy_is_not_detected",
{"options": {"zone": "light.x"}},
),
],
)
async def test_zone_condition_rejects_non_zone_entity_id(
hass: HomeAssistant, condition_key: str, config: dict[str, Any]
) -> None:
"""Test that the zone option must reference entities in the zone domain."""
with pytest.raises(vol.Invalid):
await condition.async_validate_condition_config(
hass,
{"condition": condition_key, **config},
)
@pytest.fixture
async def target_zone_entities(
hass: HomeAssistant, domain: str
) -> dict[str, list[str]]:
"""Create multiple zone-trackable entities associated with different targets."""
return await target_entities(hass, domain, domain_excluded="sensor")
# `in_zone` is True for states where the entity carries the target zone in
# `in_zones`; `not_in_zone` flips the relation.
_ZONE_CONDITION_STATES_ANY = [
*parametrize_condition_states_any(
condition="zone.in_zone",
condition_options={"zone": TARGET_ZONE},
target_states=[
("home", IN_ZONES_HOME),
],
other_states=[
("not_home", IN_ZONES_NONE),
("Work", IN_ZONES_WORK),
],
excluded_entities_from_other_domain=True,
),
*parametrize_condition_states_any(
condition="zone.not_in_zone",
condition_options={"zone": TARGET_ZONE},
target_states=[
("not_home", IN_ZONES_NONE),
("Work", IN_ZONES_WORK),
],
other_states=[
("home", IN_ZONES_HOME),
],
excluded_entities_from_other_domain=True,
),
]
_ZONE_CONDITION_STATES_ALL = [
*parametrize_condition_states_all(
condition="zone.in_zone",
condition_options={"zone": TARGET_ZONE},
target_states=[
("home", IN_ZONES_HOME),
],
other_states=[
("not_home", IN_ZONES_NONE),
("Work", IN_ZONES_WORK),
],
excluded_entities_from_other_domain=True,
),
*parametrize_condition_states_all(
condition="zone.not_in_zone",
condition_options={"zone": TARGET_ZONE},
target_states=[
("not_home", IN_ZONES_NONE),
("Work", IN_ZONES_WORK),
],
other_states=[
("home", IN_ZONES_HOME),
],
excluded_entities_from_other_domain=True,
),
]
def _parametrize_zone_target_entities() -> list[tuple[dict[str, Any], str, int, str]]:
"""Parametrize target entities for all supported zone condition domains."""
return [
(*params, domain)
for domain in ("person", "device_tracker")
for params in parametrize_target_entities(domain)
]
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target", "domain"),
_parametrize_zone_target_entities(),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
_ZONE_CONDITION_STATES_ANY,
)
async def test_zone_condition_behavior_any(
hass: HomeAssistant,
target_zone_entities: dict[str, list[str]],
condition_target_config: dict[str, Any],
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test zone conditions under behavior=any."""
await assert_condition_behavior_any(
hass,
target_entities=target_zone_entities,
condition_target_config=condition_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
condition=condition,
condition_options=condition_options,
states=states,
)
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target", "domain"),
_parametrize_zone_target_entities(),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
_ZONE_CONDITION_STATES_ALL,
)
async def test_zone_condition_behavior_all(
hass: HomeAssistant,
target_zone_entities: dict[str, list[str]],
condition_target_config: dict[str, Any],
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test zone conditions under behavior=all."""
await assert_condition_behavior_all(
hass,
target_entities=target_zone_entities,
condition_target_config=condition_target_config,
entity_id=entity_id,
entities_in_target=entities_in_target,
condition=condition,
condition_options=condition_options,
states=states,
)
async def test_in_zone_condition_for_attribute_only_change(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test `for:` anchors to in_zones updates, not state.state changes.
A person already "home" who enters an overlapping zone (e.g. zone.coffee)
keeps state.state == "home" while in_zones grows. `for: 5m` on
in_zone(zone.coffee) must start counting from when in_zones changed, not
from the (older) last state.state transition.
"""
coffee_zone = "zone.coffee"
# Person at home but not yet in the coffee zone.
hass.states.async_set(
"person.alice",
"home",
{"in_zones": [ZONE_HOME]},
)
await hass.async_block_till_done()
# Time passes — state.state's last_changed sits 10 minutes in the past.
freezer.tick(timedelta(minutes=10))
config = await condition.async_validate_condition_config(
hass,
{
"condition": "zone.in_zone",
"target": {"entity_id": "person.alice"},
"options": {"zone": coffee_zone, "for": {"minutes": 5}},
},
)
test = await condition.async_from_config(hass, config)
# in_zones gains the coffee zone; state.state stays "home", so last_changed
# is untouched and only last_updated advances.
hass.states.async_set(
"person.alice",
"home",
{"in_zones": [ZONE_HOME, coffee_zone]},
)
await hass.async_block_till_done()
# Just entered; `for: 5m` must not be satisfied yet. (Without value_source
# set on the DomainSpec, the anchor would be last_changed from 10 minutes
# ago and this would incorrectly evaluate to True.)
assert test.async_check() is False
# After the duration elapses, the condition is satisfied.
freezer.tick(timedelta(minutes=6))
assert test.async_check() is True
# --- Zone occupancy condition tests ---
@pytest.mark.parametrize(
("condition_key", "zone_state", "expected"),
[
# occupancy_is_detected — true when count >= 1
pytest.param("zone.occupancy_is_detected", "1", True, id="detected_1"),
pytest.param("zone.occupancy_is_detected", "3", True, id="detected_3"),
pytest.param("zone.occupancy_is_detected", "0", False, id="detected_0"),
pytest.param(
"zone.occupancy_is_detected",
STATE_UNAVAILABLE,
False,
id="detected_unavailable",
),
pytest.param(
"zone.occupancy_is_detected", STATE_UNKNOWN, False, id="detected_unknown"
),
# occupancy_is_not_detected — true only when count == 0
pytest.param("zone.occupancy_is_not_detected", "0", True, id="empty_0"),
pytest.param("zone.occupancy_is_not_detected", "1", False, id="empty_1"),
pytest.param("zone.occupancy_is_not_detected", "3", False, id="empty_3"),
# Unavailable / unknown are not "empty" — they're indeterminate.
pytest.param(
"zone.occupancy_is_not_detected",
STATE_UNAVAILABLE,
False,
id="empty_unavailable",
),
pytest.param(
"zone.occupancy_is_not_detected",
STATE_UNKNOWN,
False,
id="empty_unknown",
),
],
)
async def test_zone_occupancy_condition_evaluates(
hass: HomeAssistant,
condition_key: str,
zone_state: str,
expected: bool,
) -> None:
"""Test occupancy conditions evaluate against the zone's integer state."""
hass.states.async_set(ZONE_HOME, zone_state)
await hass.async_block_till_done()
config = await condition.async_validate_condition_config(
hass, {"condition": condition_key, "options": {"zone": ZONE_HOME}}
)
test = await condition.async_from_config(hass, config)
assert test.async_check() is expected
-86
View File
@@ -2205,92 +2205,6 @@ async def test_condition_template_error(hass: HomeAssistant) -> None:
test.async_check()
@pytest.mark.parametrize(
("value_template", "expectation", "expected_template_errors", "expected_result"),
[
# Undefined variable used in a way that raises (e.g. attribute access)
(
"{{ trigger.to_state.attributes.event_type == 'double_press' }}",
pytest.raises(ConditionError),
["'trigger' is undefined"],
{},
),
# Undefined variable used in a way that only warns
(
"{{ no_such_variable }}",
does_not_raise(),
["'no_such_variable' is undefined"],
{"result": False, "entities": []},
),
# A single render can emit more than one message
(
"{{ foo }}{{ bar }}",
does_not_raise(),
["'foo' is undefined", "'bar' is undefined"],
{"result": False, "entities": []},
),
],
)
async def test_condition_template_error_traced_not_logged(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
value_template: str,
expectation: AbstractContextManager,
expected_template_errors: list[str],
expected_result: dict[str, Any],
) -> None:
"""Test template errors are added to the trace and not logged when opted in.
The subscribe_condition websocket command re-evaluates a condition every
second and opts in via trace.suppress_template_error_logging(). Template
variable errors are then recorded in the trace without being logged.
"""
caplog.set_level(logging.WARNING)
config = {"condition": "template", "value_template": value_template}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
with expectation, trace.suppress_template_error_logging():
test.async_check()
# The template errors are recorded in the trace...
condition_trace = trace.trace_get(clear=False)
trace.trace_clear()
trace_element = condition_trace[""][0]
assert trace_element.template_errors == expected_template_errors
assert (trace_element._result or {}) == expected_result
# ...and not logged
assert "Template variable" not in caplog.text
async def test_condition_template_error_logged_without_opt_in(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test template errors are logged when suppression is not opted in.
The error is always recorded in the trace, but unless the consumer opts in
via trace.suppress_template_error_logging() it is also logged as usual.
"""
caplog.set_level(logging.WARNING)
config = {"condition": "template", "value_template": "{{ no_such_variable }}"}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
assert test.async_check() is False
# Recorded in the trace...
condition_trace = trace.trace_get(clear=False)
trace.trace_clear()
assert condition_trace[""][0].template_errors == ["'no_such_variable' is undefined"]
# ...and also logged
assert "Template variable warning: 'no_such_variable' is undefined" in caplog.text
async def test_condition_template_invalid_results(hass: HomeAssistant) -> None:
"""Test template condition render false with invalid results."""
config = {"condition": "template", "value_template": "{{ 'string' }}"}
-36
View File
@@ -2084,39 +2084,3 @@ def test_base_schemas_reject_invalid_note(
"""Test that script, condition, trigger base schemas reject non-string notes."""
with pytest.raises(vol.Invalid):
validator({**base_config, "note": invalid_note})
_CHOOSE_OPTION_BASE_CONFIG = {
"conditions": [
{"condition": "state", "entity_id": "sun.sun", "state": "above_horizon"}
],
"sequence": [{"action": "test.foo"}],
}
@pytest.mark.usefixtures("hass")
def test_choose_option_accepts_note() -> None:
"""Test that the note field is accepted and stripped from a choose option."""
validated = cv.script_action(
{"choose": [{**_CHOOSE_OPTION_BASE_CONFIG, "note": "Single line"}]}
)
assert "note" not in validated["choose"][0]
@pytest.mark.parametrize(
"invalid_note",
[
pytest.param(None, id="none"),
pytest.param(42, id="int"),
pytest.param(True, id="bool"),
pytest.param([], id="list"),
pytest.param({}, id="dict"),
],
)
@pytest.mark.usefixtures("hass")
def test_choose_option_rejects_invalid_note(invalid_note: Any) -> None:
"""Test that choose option schemas reject non-string notes."""
with pytest.raises(vol.Invalid):
cv.script_action(
{"choose": [{**_CHOOSE_OPTION_BASE_CONFIG, "note": invalid_note}]}
)
+4 -20
View File
@@ -868,8 +868,7 @@ async def test_delay_template_invalid(
{
"error": (
"offset should be format 'HH:MM', 'HH:MM:SS' or 'HH:MM:SS.F'"
),
"template_errors": ["'invalid_delay' is undefined"],
)
}
],
},
@@ -934,12 +933,7 @@ async def test_delay_template_complex_invalid(
assert_action_trace(
{
"0": [{"result": {"event": "test_event", "event_data": {}}}],
"1": [
{
"error": "expected float for dictionary value @ data['seconds']",
"template_errors": ["'invalid_delay' is undefined"],
}
],
"1": [{"error": "expected float for dictionary value @ data['seconds']"}],
},
expected_script_execution="aborted",
)
@@ -2652,12 +2646,7 @@ async def test_repeat_for_each_invalid_template(
assert_action_trace(
{
"0": [
{
"error": "Repeat 'for_each' must be a list of items",
"template_errors": ["'Muhaha' is undefined"],
}
],
"0": [{"error": "Repeat 'for_each' must be a list of items"}],
},
expected_script_execution="aborted",
)
@@ -2726,12 +2715,7 @@ async def test_repeat_condition_warning(
expected_trace[f"0/repeat/{condition}/0"] = [
{"error": "In 'numeric_state':\n " + expected_error}
]
expected_trace[f"0/repeat/{condition}/0/entity_id/0"] = [
{
"error": expected_error,
"template_errors": ["'unassigned_variable' is undefined"],
}
]
expected_trace[f"0/repeat/{condition}/0/entity_id/0"] = [{"error": expected_error}]
assert_action_trace(expected_trace)