mirror of
https://github.com/home-assistant/core.git
synced 2026-04-16 14:46:15 +02:00
Compare commits
9 Commits
ariel-pyth
...
joostlek-p
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2fe26c3ddb | ||
|
|
206c498027 | ||
|
|
0ac62b241e | ||
|
|
4ba123a1a8 | ||
|
|
8b8b39c1b7 | ||
|
|
5b70e5f829 | ||
|
|
4f8e7125d4 | ||
|
|
baf5e32c59 | ||
|
|
0f0ceaace2 |
@@ -7,23 +7,31 @@ to speed up the process.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Container, Iterable, Sequence
|
||||
from datetime import timedelta
|
||||
from functools import lru_cache, partial
|
||||
from typing import Any
|
||||
from functools import lru_cache
|
||||
from typing import Any, override
|
||||
|
||||
from jwt import DecodeError, PyJWS, PyJWT
|
||||
from jwt import DecodeError, PyJWK, PyJWS, PyJWT
|
||||
from jwt.algorithms import AllowedPublicKeys
|
||||
from jwt.types import Options
|
||||
|
||||
from homeassistant.util.json import json_loads
|
||||
|
||||
JWT_TOKEN_CACHE_SIZE = 16
|
||||
MAX_TOKEN_SIZE = 8192
|
||||
|
||||
_VERIFY_KEYS = ("signature", "exp", "nbf", "iat", "aud", "iss", "sub", "jti")
|
||||
|
||||
_VERIFY_OPTIONS: dict[str, Any] = {f"verify_{key}": True for key in _VERIFY_KEYS} | {
|
||||
"require": []
|
||||
}
|
||||
_NO_VERIFY_OPTIONS = {f"verify_{key}": False for key in _VERIFY_KEYS}
|
||||
_NO_VERIFY_OPTIONS = Options(
|
||||
verify_signature=False,
|
||||
verify_exp=False,
|
||||
verify_nbf=False,
|
||||
verify_iat=False,
|
||||
verify_aud=False,
|
||||
verify_iss=False,
|
||||
verify_sub=False,
|
||||
verify_jti=False,
|
||||
require=[],
|
||||
)
|
||||
|
||||
|
||||
class _PyJWSWithLoadCache(PyJWS):
|
||||
@@ -38,9 +46,6 @@ class _PyJWSWithLoadCache(PyJWS):
|
||||
return super()._load(jwt)
|
||||
|
||||
|
||||
_jws = _PyJWSWithLoadCache()
|
||||
|
||||
|
||||
@lru_cache(maxsize=JWT_TOKEN_CACHE_SIZE)
|
||||
def _decode_payload(json_payload: str) -> dict[str, Any]:
|
||||
"""Decode the payload from a JWS dictionary."""
|
||||
@@ -56,21 +61,12 @@ def _decode_payload(json_payload: str) -> dict[str, Any]:
|
||||
class _PyJWTWithVerify(PyJWT):
|
||||
"""PyJWT with a fast decode implementation."""
|
||||
|
||||
def decode_payload(
|
||||
self, jwt: str, key: str, options: dict[str, Any], algorithms: list[str]
|
||||
) -> dict[str, Any]:
|
||||
"""Decode a JWT's payload."""
|
||||
if len(jwt) > MAX_TOKEN_SIZE:
|
||||
# Avoid caching impossible tokens
|
||||
raise DecodeError("Token too large")
|
||||
return _decode_payload(
|
||||
_jws.decode_complete(
|
||||
jwt=jwt,
|
||||
key=key,
|
||||
algorithms=algorithms,
|
||||
options=options,
|
||||
)["payload"]
|
||||
)
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the PyJWT instance."""
|
||||
# We require exp and iat claims to be present
|
||||
super().__init__(Options(require=["exp", "iat"]))
|
||||
# Override the _jws instance with our cached version
|
||||
self._jws = _PyJWSWithLoadCache()
|
||||
|
||||
def verify_and_decode(
|
||||
self,
|
||||
@@ -79,37 +75,70 @@ class _PyJWTWithVerify(PyJWT):
|
||||
algorithms: list[str],
|
||||
issuer: str | None = None,
|
||||
leeway: float | timedelta = 0,
|
||||
options: dict[str, Any] | None = None,
|
||||
options: Options | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Verify a JWT's signature and claims."""
|
||||
merged_options = {**_VERIFY_OPTIONS, **(options or {})}
|
||||
payload = self.decode_payload(
|
||||
return self.decode(
|
||||
jwt=jwt,
|
||||
key=key,
|
||||
options=merged_options,
|
||||
algorithms=algorithms,
|
||||
)
|
||||
# These should never be missing since we verify them
|
||||
# but this is an additional safeguard to make sure
|
||||
# nothing slips through.
|
||||
assert "exp" in payload, "exp claim is required"
|
||||
assert "iat" in payload, "iat claim is required"
|
||||
self._validate_claims(
|
||||
payload=payload,
|
||||
options=merged_options,
|
||||
issuer=issuer,
|
||||
leeway=leeway,
|
||||
options=options,
|
||||
)
|
||||
return payload
|
||||
|
||||
@override
|
||||
def decode(
|
||||
self,
|
||||
jwt: str | bytes,
|
||||
key: AllowedPublicKeys | PyJWK | str | bytes = "",
|
||||
algorithms: Sequence[str] | None = None,
|
||||
options: Options | None = None,
|
||||
verify: bool | None = None,
|
||||
detached_payload: bytes | None = None,
|
||||
audience: str | Iterable[str] | None = None,
|
||||
subject: str | None = None,
|
||||
issuer: str | Container[str] | None = None,
|
||||
leeway: float | timedelta = 0,
|
||||
**kwargs: Any,
|
||||
) -> dict[str, Any]:
|
||||
"""Decode a JWT, verifying the signature and claims."""
|
||||
if len(jwt) > MAX_TOKEN_SIZE:
|
||||
# Avoid caching impossible tokens
|
||||
raise DecodeError("Token too large")
|
||||
return super().decode(
|
||||
jwt=jwt,
|
||||
key=key,
|
||||
algorithms=algorithms,
|
||||
options=options,
|
||||
verify=verify,
|
||||
detached_payload=detached_payload,
|
||||
audience=audience,
|
||||
subject=subject,
|
||||
issuer=issuer,
|
||||
leeway=leeway,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
@override
|
||||
def _decode_payload(self, decoded: dict[str, Any]) -> dict[str, Any]:
|
||||
return _decode_payload(decoded["payload"])
|
||||
|
||||
|
||||
_jwt = _PyJWTWithVerify()
|
||||
verify_and_decode = _jwt.verify_and_decode
|
||||
unverified_hs256_token_decode = lru_cache(maxsize=JWT_TOKEN_CACHE_SIZE)(
|
||||
partial(
|
||||
_jwt.decode_payload, key="", algorithms=["HS256"], options=_NO_VERIFY_OPTIONS
|
||||
|
||||
|
||||
@lru_cache(maxsize=JWT_TOKEN_CACHE_SIZE)
|
||||
def unverified_hs256_token_decode(jwt: str) -> dict[str, Any]:
|
||||
"""Decode a JWT without verifying the signature."""
|
||||
return _jwt.decode(
|
||||
jwt=jwt,
|
||||
key="",
|
||||
algorithms=["HS256"],
|
||||
options=_NO_VERIFY_OPTIONS,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"unverified_hs256_token_decode",
|
||||
|
||||
@@ -7,5 +7,5 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/camera",
|
||||
"integration_type": "entity",
|
||||
"quality_scale": "internal",
|
||||
"requirements": ["PyTurboJPEG==1.8.3"]
|
||||
"requirements": ["PyTurboJPEG==2.2.0"]
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["fritzconnection"],
|
||||
"quality_scale": "gold",
|
||||
"requirements": ["fritzconnection[qr]==1.15.1", "xmltodict==1.0.2"],
|
||||
"requirements": ["fritzconnection[qr]==1.15.1", "xmltodict==1.0.4"],
|
||||
"ssdp": [
|
||||
{
|
||||
"st": "urn:schemas-upnp-org:device:fritzbox:1"
|
||||
|
||||
@@ -7,5 +7,5 @@
|
||||
"integration_type": "service",
|
||||
"iot_class": "cloud_polling",
|
||||
"quality_scale": "platinum",
|
||||
"requirements": ["imgw_pib==2.0.2"]
|
||||
"requirements": ["imgw_pib==2.1.0"]
|
||||
}
|
||||
|
||||
@@ -7,5 +7,5 @@
|
||||
"integration_type": "hub",
|
||||
"iot_class": "cloud_push",
|
||||
"quality_scale": "bronze",
|
||||
"requirements": ["pyaxencoapi==1.0.6"]
|
||||
"requirements": ["pyaxencoapi==1.0.7"]
|
||||
}
|
||||
|
||||
@@ -5,5 +5,5 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/rest",
|
||||
"integration_type": "service",
|
||||
"iot_class": "local_polling",
|
||||
"requirements": ["jsonpath==0.82.2", "xmltodict==1.0.2"]
|
||||
"requirements": ["jsonpath==0.82.2", "xmltodict==1.0.4"]
|
||||
}
|
||||
|
||||
@@ -5,5 +5,5 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/startca",
|
||||
"iot_class": "cloud_polling",
|
||||
"quality_scale": "legacy",
|
||||
"requirements": ["xmltodict==1.0.2"]
|
||||
"requirements": ["xmltodict==1.0.4"]
|
||||
}
|
||||
|
||||
@@ -7,5 +7,5 @@
|
||||
"integration_type": "system",
|
||||
"iot_class": "local_push",
|
||||
"quality_scale": "internal",
|
||||
"requirements": ["PyTurboJPEG==1.8.3", "av==16.0.1", "numpy==2.3.2"]
|
||||
"requirements": ["PyTurboJPEG==2.2.0", "av==16.0.1", "numpy==2.3.2"]
|
||||
}
|
||||
|
||||
@@ -5,5 +5,5 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/ted5000",
|
||||
"iot_class": "local_polling",
|
||||
"quality_scale": "legacy",
|
||||
"requirements": ["xmltodict==1.0.2"]
|
||||
"requirements": ["xmltodict==1.0.4"]
|
||||
}
|
||||
|
||||
@@ -5,5 +5,5 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/zestimate",
|
||||
"iot_class": "cloud_polling",
|
||||
"quality_scale": "legacy",
|
||||
"requirements": ["xmltodict==1.0.2"]
|
||||
"requirements": ["xmltodict==1.0.4"]
|
||||
}
|
||||
|
||||
@@ -804,6 +804,6 @@ def _decode_jwt(hass: HomeAssistant, encoded: str) -> dict[str, Any] | None:
|
||||
return None
|
||||
|
||||
try:
|
||||
return jwt.decode(encoded, secret, algorithms=["HS256"]) # type: ignore[no-any-return]
|
||||
return jwt.decode(encoded, secret, algorithms=["HS256"])
|
||||
except jwt.InvalidTokenError:
|
||||
return None
|
||||
|
||||
@@ -38,13 +38,13 @@ ha-ffmpeg==3.2.2
|
||||
habluetooth==6.0.0
|
||||
hass-nabucasa==2.2.0
|
||||
hassil==3.5.0
|
||||
home-assistant-bluetooth==1.13.1
|
||||
home-assistant-bluetooth==2.0.0
|
||||
home-assistant-frontend==20260325.7
|
||||
home-assistant-intents==2026.3.24
|
||||
httpx==0.28.1
|
||||
ifaddr==0.2.0
|
||||
Jinja2==3.1.6
|
||||
lru-dict==1.3.0
|
||||
lru-dict==1.4.1
|
||||
mutagen==1.47.0
|
||||
openai==2.21.0
|
||||
orjson==3.11.8
|
||||
@@ -53,13 +53,13 @@ paho-mqtt==2.1.0
|
||||
Pillow==12.2.0
|
||||
propcache==0.4.1
|
||||
psutil-home-assistant==0.0.1
|
||||
PyJWT==2.10.1
|
||||
PyJWT==2.12.1
|
||||
pymicro-vad==1.0.1
|
||||
PyNaCl==1.6.2
|
||||
pyOpenSSL==26.0.0
|
||||
pyspeex-noise==1.0.2
|
||||
python-slugify==8.0.4
|
||||
PyTurboJPEG==1.8.3
|
||||
PyTurboJPEG==2.2.0
|
||||
PyYAML==6.0.3
|
||||
requests==2.33.1
|
||||
securetar==2026.4.1
|
||||
|
||||
@@ -51,11 +51,11 @@ dependencies = [
|
||||
# When bumping httpx, please check the version pins of
|
||||
# httpcore, anyio, and h11 in gen_requirements_all
|
||||
"httpx==0.28.1",
|
||||
"home-assistant-bluetooth==1.13.1",
|
||||
"home-assistant-bluetooth==2.0.0",
|
||||
"ifaddr==0.2.0",
|
||||
"Jinja2==3.1.6",
|
||||
"lru-dict==1.3.0",
|
||||
"PyJWT==2.10.1",
|
||||
"lru-dict==1.4.1",
|
||||
"PyJWT==2.12.1",
|
||||
# PyJWT has loose dependency. We want the latest one.
|
||||
"cryptography==46.0.7",
|
||||
"Pillow==12.2.0",
|
||||
|
||||
8
requirements.txt
generated
8
requirements.txt
generated
@@ -26,25 +26,25 @@ fnv-hash-fast==2.0.2
|
||||
ha-ffmpeg==3.2.2
|
||||
hass-nabucasa==2.2.0
|
||||
hassil==3.5.0
|
||||
home-assistant-bluetooth==1.13.1
|
||||
home-assistant-bluetooth==2.0.0
|
||||
home-assistant-intents==2026.3.24
|
||||
httpx==0.28.1
|
||||
ifaddr==0.2.0
|
||||
infrared-protocols==1.2.0
|
||||
Jinja2==3.1.6
|
||||
lru-dict==1.3.0
|
||||
lru-dict==1.4.1
|
||||
mutagen==1.47.0
|
||||
orjson==3.11.8
|
||||
packaging>=23.1
|
||||
Pillow==12.2.0
|
||||
propcache==0.4.1
|
||||
psutil-home-assistant==0.0.1
|
||||
PyJWT==2.10.1
|
||||
PyJWT==2.12.1
|
||||
pymicro-vad==1.0.1
|
||||
pyOpenSSL==26.0.0
|
||||
pyspeex-noise==1.0.2
|
||||
python-slugify==8.0.4
|
||||
PyTurboJPEG==1.8.3
|
||||
PyTurboJPEG==2.2.0
|
||||
PyYAML==6.0.3
|
||||
requests==2.33.1
|
||||
securetar==2026.4.1
|
||||
|
||||
8
requirements_all.txt
generated
8
requirements_all.txt
generated
@@ -96,7 +96,7 @@ PyTransportNSW==0.1.1
|
||||
|
||||
# homeassistant.components.camera
|
||||
# homeassistant.components.stream
|
||||
PyTurboJPEG==1.8.3
|
||||
PyTurboJPEG==2.2.0
|
||||
|
||||
# homeassistant.components.vicare
|
||||
PyViCare==2.59.0
|
||||
@@ -1313,7 +1313,7 @@ ihcsdk==2.8.5
|
||||
imeon_inverter_api==0.4.0
|
||||
|
||||
# homeassistant.components.imgw_pib
|
||||
imgw_pib==2.0.2
|
||||
imgw_pib==2.1.0
|
||||
|
||||
# homeassistant.components.incomfort
|
||||
incomfort-client==0.7.0
|
||||
@@ -1980,7 +1980,7 @@ pyatv==0.17.0
|
||||
pyaussiebb==0.1.5
|
||||
|
||||
# homeassistant.components.myneomitis
|
||||
pyaxencoapi==1.0.6
|
||||
pyaxencoapi==1.0.7
|
||||
|
||||
# homeassistant.components.balboa
|
||||
pybalboa==1.1.3
|
||||
@@ -3334,7 +3334,7 @@ xknxproject==3.8.2
|
||||
# homeassistant.components.startca
|
||||
# homeassistant.components.ted5000
|
||||
# homeassistant.components.zestimate
|
||||
xmltodict==1.0.2
|
||||
xmltodict==1.0.4
|
||||
|
||||
# homeassistant.components.xs1
|
||||
xs1-api-client==3.0.0
|
||||
|
||||
8
requirements_test_all.txt
generated
8
requirements_test_all.txt
generated
@@ -93,7 +93,7 @@ PyTransportNSW==0.1.1
|
||||
|
||||
# homeassistant.components.camera
|
||||
# homeassistant.components.stream
|
||||
PyTurboJPEG==1.8.3
|
||||
PyTurboJPEG==2.2.0
|
||||
|
||||
# homeassistant.components.vicare
|
||||
PyViCare==2.59.0
|
||||
@@ -1165,7 +1165,7 @@ igloohome-api==0.1.1
|
||||
imeon_inverter_api==0.4.0
|
||||
|
||||
# homeassistant.components.imgw_pib
|
||||
imgw_pib==2.0.2
|
||||
imgw_pib==2.1.0
|
||||
|
||||
# homeassistant.components.incomfort
|
||||
incomfort-client==0.7.0
|
||||
@@ -1717,7 +1717,7 @@ pyatv==0.17.0
|
||||
pyaussiebb==0.1.5
|
||||
|
||||
# homeassistant.components.myneomitis
|
||||
pyaxencoapi==1.0.6
|
||||
pyaxencoapi==1.0.7
|
||||
|
||||
# homeassistant.components.balboa
|
||||
pybalboa==1.1.3
|
||||
@@ -2828,7 +2828,7 @@ xknxproject==3.8.2
|
||||
# homeassistant.components.startca
|
||||
# homeassistant.components.ted5000
|
||||
# homeassistant.components.zestimate
|
||||
xmltodict==1.0.2
|
||||
xmltodict==1.0.4
|
||||
|
||||
# homeassistant.components.yale_smart_alarm
|
||||
yalesmartalarmclient==0.4.3
|
||||
|
||||
@@ -41,6 +41,7 @@ PACKAGE_CHECK_VERSION_RANGE = {
|
||||
"pymodbus": "Custom",
|
||||
"pytz": "CalVer",
|
||||
"requests": "SemVer",
|
||||
"serialx": "SemVer",
|
||||
"typing-extensions": "SemVer",
|
||||
"urllib3": "SemVer",
|
||||
"yarl": "SemVer",
|
||||
|
||||
@@ -6,12 +6,6 @@ import pytest
|
||||
from homeassistant.auth import jwt_wrapper
|
||||
|
||||
|
||||
async def test_all_default_options_are_in_verify_options() -> None:
|
||||
"""Test that all default options in _VERIFY_OPTIONS."""
|
||||
for option in jwt_wrapper._PyJWTWithVerify._get_default_options():
|
||||
assert option in jwt_wrapper._VERIFY_OPTIONS
|
||||
|
||||
|
||||
async def test_reject_access_token_with_impossible_large_size() -> None:
|
||||
"""Test rejecting access tokens with impossible sizes."""
|
||||
with pytest.raises(jwt.DecodeError):
|
||||
|
||||
@@ -21,6 +21,8 @@ HYDROLOGICAL_DATA = HydrologicalData(
|
||||
water_temperature=SensorData(name="Water Temperature", value=10.8),
|
||||
flood_alarm=None,
|
||||
flood_warning=None,
|
||||
ice_phenomenon=SensorData(name="Ice Phenomenon", value=20),
|
||||
ice_phenomenon_measurement_date=datetime(2024, 4, 27, 10, 0, tzinfo=UTC),
|
||||
water_level_measurement_date=datetime(2024, 4, 27, 10, 0, tzinfo=UTC),
|
||||
water_temperature_measurement_date=datetime(2024, 4, 27, 10, 10, tzinfo=UTC),
|
||||
water_flow=SensorData(name="Water Flow", value=123.45),
|
||||
|
||||
@@ -41,6 +41,12 @@
|
||||
'valid_to': '2024-04-28T11:00:00+00:00',
|
||||
'value': 'rapid_water_level_rise',
|
||||
}),
|
||||
'ice_phenomenon': dict({
|
||||
'name': 'Ice Phenomenon',
|
||||
'unit': None,
|
||||
'value': 20,
|
||||
}),
|
||||
'ice_phenomenon_measurement_date': '2024-04-27T10:00:00+00:00',
|
||||
'latitude': None,
|
||||
'longitude': None,
|
||||
'river': 'River Name',
|
||||
|
||||
@@ -28,6 +28,8 @@ from homeassistant.const import (
|
||||
CONF_ENTITY_ID,
|
||||
CONF_OPTIONS,
|
||||
CONF_TARGET,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
UnitOfTemperature,
|
||||
@@ -56,6 +58,7 @@ from homeassistant.helpers.condition import (
|
||||
async_validate_condition_config,
|
||||
make_entity_numerical_condition,
|
||||
make_entity_numerical_condition_with_unit,
|
||||
make_entity_state_condition,
|
||||
)
|
||||
from homeassistant.helpers.template import Template
|
||||
from homeassistant.helpers.typing import UNDEFINED, ConfigType, UndefinedType
|
||||
@@ -3939,3 +3942,188 @@ async def test_numerical_condition_with_unit_behavior(
|
||||
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
|
||||
)
|
||||
assert test(hass) is False
|
||||
|
||||
|
||||
async def _setup_state_condition(
|
||||
hass: HomeAssistant,
|
||||
entity_ids: str | list[str],
|
||||
states: str | bool | set[str | bool],
|
||||
condition_options: dict[str, Any] | None = None,
|
||||
domain_specs: Mapping[str, DomainSpec] | None = None,
|
||||
) -> condition.ConditionCheckerType:
|
||||
"""Set up a state condition via a mock platform and return the checker."""
|
||||
condition_cls = make_entity_state_condition(
|
||||
domain_specs or _DEFAULT_DOMAIN_SPECS,
|
||||
states,
|
||||
)
|
||||
|
||||
async def async_get_conditions(
|
||||
hass: HomeAssistant,
|
||||
) -> dict[str, type[Condition]]:
|
||||
return {"_": condition_cls}
|
||||
|
||||
mock_integration(hass, MockModule("test"))
|
||||
mock_platform(
|
||||
hass, "test.condition", Mock(async_get_conditions=async_get_conditions)
|
||||
)
|
||||
|
||||
if isinstance(entity_ids, str):
|
||||
entity_ids = [entity_ids]
|
||||
|
||||
config: dict[str, Any] = {
|
||||
CONF_CONDITION: "test",
|
||||
CONF_TARGET: {CONF_ENTITY_ID: entity_ids},
|
||||
CONF_OPTIONS: condition_options or {},
|
||||
}
|
||||
|
||||
config = await async_validate_condition_config(hass, config)
|
||||
test = await condition.async_from_config(hass, config)
|
||||
assert test is not None
|
||||
return test
|
||||
|
||||
|
||||
async def test_state_condition_single_entity(hass: HomeAssistant) -> None:
|
||||
"""Test state condition with a single entity."""
|
||||
test = await _setup_state_condition(
|
||||
hass, entity_ids="test.entity_1", states=STATE_ON
|
||||
)
|
||||
|
||||
hass.states.async_set("test.entity_1", STATE_ON)
|
||||
assert test(hass) is True
|
||||
|
||||
hass.states.async_set("test.entity_1", STATE_OFF)
|
||||
assert test(hass) is False
|
||||
|
||||
|
||||
async def test_state_condition_multiple_target_states(hass: HomeAssistant) -> None:
|
||||
"""Test state condition matching any of multiple target states."""
|
||||
test = await _setup_state_condition(
|
||||
hass, entity_ids="test.entity_1", states={"on", "heat"}
|
||||
)
|
||||
|
||||
hass.states.async_set("test.entity_1", "on")
|
||||
assert test(hass) is True
|
||||
|
||||
hass.states.async_set("test.entity_1", "heat")
|
||||
assert test(hass) is True
|
||||
|
||||
hass.states.async_set("test.entity_1", "off")
|
||||
assert test(hass) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"state_value",
|
||||
[STATE_UNAVAILABLE, STATE_UNKNOWN],
|
||||
)
|
||||
async def test_state_condition_unavailable_unknown(
|
||||
hass: HomeAssistant, state_value: str
|
||||
) -> None:
|
||||
"""Test state condition with unavailable/unknown entities.
|
||||
|
||||
Uses three entities: entity_1 is on, entity_2 is unavailable/unknown,
|
||||
entity_3 varies. Unavailable/unknown entities are excluded from
|
||||
evaluation, so:
|
||||
- behavior any: passes if at least one *available* entity matches
|
||||
- behavior all: passes if all *available* entities match
|
||||
"""
|
||||
# Single entity: unavailable/unknown → False
|
||||
test_single = await _setup_state_condition(
|
||||
hass, entity_ids="test.entity_1", states=STATE_ON
|
||||
)
|
||||
hass.states.async_set("test.entity_1", state_value)
|
||||
assert test_single(hass) is False
|
||||
|
||||
# behavior any: entity_1=on, entity_2=unavailable, entity_3=off
|
||||
# → True (entity_1 matches, entity_2 is skipped)
|
||||
test_any = await _setup_state_condition(
|
||||
hass,
|
||||
entity_ids=["test.entity_1", "test.entity_2", "test.entity_3"],
|
||||
states=STATE_ON,
|
||||
condition_options={ATTR_BEHAVIOR: BEHAVIOR_ANY},
|
||||
)
|
||||
hass.states.async_set("test.entity_1", STATE_ON)
|
||||
hass.states.async_set("test.entity_2", state_value)
|
||||
hass.states.async_set("test.entity_3", STATE_OFF)
|
||||
assert test_any(hass) is True
|
||||
|
||||
# behavior any: entity_1=off, entity_2=unavailable, entity_3=off
|
||||
# → False (no available entity matches)
|
||||
hass.states.async_set("test.entity_1", STATE_OFF)
|
||||
assert test_any(hass) is False
|
||||
|
||||
# behavior all: entity_1=on, entity_2=unavailable, entity_3=on
|
||||
# → True (all *available* entities match, entity_2 is skipped)
|
||||
test_all = await _setup_state_condition(
|
||||
hass,
|
||||
entity_ids=["test.entity_1", "test.entity_2", "test.entity_3"],
|
||||
states=STATE_ON,
|
||||
condition_options={ATTR_BEHAVIOR: BEHAVIOR_ALL},
|
||||
)
|
||||
hass.states.async_set("test.entity_1", STATE_ON)
|
||||
hass.states.async_set("test.entity_2", state_value)
|
||||
hass.states.async_set("test.entity_3", STATE_ON)
|
||||
assert test_all(hass) is True
|
||||
|
||||
# behavior all: entity_1=on, entity_2=unavailable, entity_3=off
|
||||
# → False (entity_3 is available and doesn't match)
|
||||
hass.states.async_set("test.entity_3", STATE_OFF)
|
||||
assert test_all(hass) is False
|
||||
|
||||
|
||||
async def test_state_condition_entity_not_found(hass: HomeAssistant) -> None:
|
||||
"""Test state condition when entity does not exist."""
|
||||
test = await _setup_state_condition(
|
||||
hass, entity_ids="test.nonexistent", states=STATE_ON
|
||||
)
|
||||
|
||||
# Entity doesn't exist — condition should be false
|
||||
assert test(hass) is False
|
||||
|
||||
|
||||
async def test_state_condition_attribute_value_source(hass: HomeAssistant) -> None:
|
||||
"""Test state condition reads from attribute when value_source is set."""
|
||||
test = await _setup_state_condition(
|
||||
hass,
|
||||
entity_ids="test.entity_1",
|
||||
states="heat",
|
||||
domain_specs={"test": DomainSpec(value_source="hvac_action")},
|
||||
)
|
||||
|
||||
hass.states.async_set("test.entity_1", "on", {"hvac_action": "heat"})
|
||||
assert test(hass) is True
|
||||
|
||||
hass.states.async_set("test.entity_1", "on", {"hvac_action": "idle"})
|
||||
assert test(hass) is False
|
||||
|
||||
# Missing attribute
|
||||
hass.states.async_set("test.entity_1", "on", {})
|
||||
assert test(hass) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("behavior", "one_match_expected"),
|
||||
[(BEHAVIOR_ANY, True), (BEHAVIOR_ALL, False)],
|
||||
)
|
||||
async def test_state_condition_behavior(
|
||||
hass: HomeAssistant, behavior: str, one_match_expected: bool
|
||||
) -> None:
|
||||
"""Test state condition with behavior any/all."""
|
||||
test = await _setup_state_condition(
|
||||
hass,
|
||||
entity_ids=["test.entity_1", "test.entity_2"],
|
||||
states=STATE_ON,
|
||||
condition_options={ATTR_BEHAVIOR: behavior},
|
||||
)
|
||||
|
||||
# Both on → True for any and all
|
||||
hass.states.async_set("test.entity_1", STATE_ON)
|
||||
hass.states.async_set("test.entity_2", STATE_ON)
|
||||
assert test(hass) is True
|
||||
|
||||
# Only one on → depends on behavior
|
||||
hass.states.async_set("test.entity_2", STATE_OFF)
|
||||
assert test(hass) is one_match_expected
|
||||
|
||||
# Neither on → False for any and all
|
||||
hass.states.async_set("test.entity_1", STATE_OFF)
|
||||
assert test(hass) is False
|
||||
|
||||
Reference in New Issue
Block a user