evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

2024.10.4 (#129181)

Browse source
This commit is contained in:
Franck Nijhof 2024-10-25 21:27:01 +02:00 • committed by GitHub
commit d31995f878
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
36 changed files with 1071 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
homeassistant/components/awair/manifest.json
View file

@ -3,11 +3,6 @@
"name": "Awair",
"codeowners": ["@ahayworth", "@danielsjf"],
"config_flow": true,
"dhcp": [
{
"macaddress": "70886B1*"
}
],
"documentation": "https://www.home-assistant.io/integrations/awair",
"iot_class": "local_polling",
"loggers": ["python_awair"],

93
homeassistant/components/comelit/diagnostics.py Normal file
View file

@ -0,0 +1,93 @@
"""Diagnostics support for Comelit integration."""
from __future__ import annotations
from typing import Any
from aiocomelit import (
ComelitSerialBridgeObject,
ComelitVedoAreaObject,
ComelitVedoZoneObject,
)
from aiocomelit.const import BRIDGE
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PIN, CONF_TYPE
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .coordinator import ComelitBaseCoordinator
TO_REDACT = {CONF_PIN}
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator: ComelitBaseCoordinator = hass.data[DOMAIN][entry.entry_id]
dev_list: list[dict[str, Any]] = []
dev_type_list: list[dict[int, Any]] = []
for dev_type in coordinator.data:
dev_type_list = []
for sensor_data in coordinator.data[dev_type].values():
if isinstance(sensor_data, ComelitSerialBridgeObject):
dev_type_list.append(
{
sensor_data.index: {
"name": sensor_data.name,
"status": sensor_data.status,
"human_status": sensor_data.human_status,
"protected": sensor_data.protected,
"val": sensor_data.val,
"zone": sensor_data.zone,
"power": sensor_data.power,
"power_unit": sensor_data.power_unit,
}
}
)
if isinstance(sensor_data, ComelitVedoAreaObject):
dev_type_list.append(
{
sensor_data.index: {
"name": sensor_data.name,
"human_status": sensor_data.human_status.value,
"p1": sensor_data.p1,
"p2": sensor_data.p2,
"ready": sensor_data.ready,
"armed": sensor_data.armed,
"alarm": sensor_data.alarm,
"alarm_memory": sensor_data.alarm_memory,
"sabotage": sensor_data.sabotage,
"anomaly": sensor_data.anomaly,
"in_time": sensor_data.in_time,
"out_time": sensor_data.out_time,
}
}
)
if isinstance(sensor_data, ComelitVedoZoneObject):
dev_type_list.append(
{
sensor_data.index: {
"name": sensor_data.name,
"human_status": sensor_data.human_status.value,
"status": sensor_data.status,
"status_api": sensor_data.status_api,
}
}
)
dev_list.append({dev_type: dev_type_list})
return {
"entry": async_redact_data(entry.as_dict(), TO_REDACT),
"type": entry.data.get(CONF_TYPE, BRIDGE),
"device_info": {
"last_update success": coordinator.last_update_success,
"last_exception": repr(coordinator.last_exception),
"devices": dev_list,
},
}

2
homeassistant/components/devolo_home_network/device_tracker.py
View file

@ -51,7 +51,7 @@ async def async_setup_entry(
)
)
tracked.add(station.mac_address)
async_add_entities(new_entities)
async_add_entities(new_entities)
@callback
def restore_entities() -> None:

6
homeassistant/components/devolo_home_network/entity.py
View file

@ -9,6 +9,7 @@ from devolo_plc_api.device_api import (
)
from devolo_plc_api.plcnet_api import DataRate, LogicalNetwork
from homeassistant.const import ATTR_CONNECTIONS
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import (
@ -45,7 +46,6 @@ class DevoloEntity(Entity):
self._attr_device_info = DeviceInfo(
configuration_url=f"http://{self.device.ip}",
connections={(CONNECTION_NETWORK_MAC, self.device.mac)},
identifiers={(DOMAIN, str(self.device.serial_number))},
manufacturer="devolo",
model=self.device.product,
@ -53,6 +53,10 @@ class DevoloEntity(Entity):
serial_number=self.device.serial_number,
sw_version=self.device.firmware_version,
)
if self.device.mac:
self._attr_device_info[ATTR_CONNECTIONS] = {
(CONNECTION_NETWORK_MAC, self.device.mac)
}
self._attr_translation_key = self.entity_description.key
self._attr_unique_id = (
f"{self.device.serial_number}_{self.entity_description.key}"

2
homeassistant/components/duotecno/manifest.json
View file

@ -7,5 +7,5 @@
"iot_class": "local_push",
"loggers": ["pyduotecno", "pyduotecno-node", "pyduotecno-unit"],
"quality_scale": "silver",
"requirements": ["pyDuotecno==2024.10.0"]
"requirements": ["pyDuotecno==2024.10.1"]
}

2
homeassistant/components/evohome/__init__.py
View file

@ -223,7 +223,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
config[DOMAIN][CONF_PASSWORD],
)
except evo.AuthenticationFailed as err:
except (evo.AuthenticationFailed, evo.RequestFailed) as err:
handle_evo_exception(err)
return False

2
homeassistant/components/frontend/manifest.json
View file

@ -20,5 +20,5 @@
"documentation": "https://www.home-assistant.io/integrations/frontend",
"integration_type": "system",
"quality_scale": "internal",
"requirements": ["home-assistant-frontend==20241002.3"]
"requirements": ["home-assistant-frontend==20241002.4"]
}

22
homeassistant/components/honeywell/climate.py
View file

@ -49,6 +49,10 @@ from .const import (
RETRY,
)
MODE_PERMANENT_HOLD = 2
MODE_TEMPORARY_HOLD = 1
MODE_HOLD = {MODE_PERMANENT_HOLD, MODE_TEMPORARY_HOLD}
ATTR_FAN_ACTION = "fan_action"
ATTR_PERMANENT_HOLD = "permanent_hold"
@ -175,6 +179,7 @@ class HoneywellUSThermostat(ClimateEntity):
self._cool_away_temp = cool_away_temp
self._heat_away_temp = heat_away_temp
self._away = False
self._away_hold = False
self._retry = 0
self._attr_unique_id = str(device.deviceid)
@ -323,11 +328,15 @@ class HoneywellUSThermostat(ClimateEntity):
@property
def preset_mode(self) -> str | None:
"""Return the current preset mode, e.g., home, away, temp."""
if self._away:
if self._away and self._is_hold():
self._away_hold = True
return PRESET_AWAY
if self._is_permanent_hold():
if self._is_hold():
return PRESET_HOLD
# Someone has changed the stat manually out of hold in away mode
if self._away and self._away_hold:
self._away = False
self._away_hold = False
return PRESET_NONE
@property
@ -335,10 +344,15 @@ class HoneywellUSThermostat(ClimateEntity):
"""Return the fan setting."""
return HW_FAN_MODE_TO_HA.get(self._device.fan_mode)
def _is_hold(self) -> bool:
heat_status = self._device.raw_ui_data.get("StatusHeat", 0)
cool_status = self._device.raw_ui_data.get("StatusCool", 0)
return heat_status in MODE_HOLD or cool_status in MODE_HOLD
def _is_permanent_hold(self) -> bool:
heat_status = self._device.raw_ui_data.get("StatusHeat", 0)
cool_status = self._device.raw_ui_data.get("StatusCool", 0)
return heat_status == 2 or cool_status == 2
return MODE_PERMANENT_HOLD in (heat_status, cool_status)
async def _set_temperature(self, **kwargs) -> None:
"""Set new target temperature."""

2
homeassistant/components/media_extractor/manifest.json
View file

@ -8,6 +8,6 @@
"iot_class": "calculated",
"loggers": ["yt_dlp"],
"quality_scale": "internal",
"requirements": ["yt-dlp==2024.10.07"],
"requirements": ["yt-dlp==2024.10.22"],
"single_config_entry": true
}

2
homeassistant/components/nyt_games/manifest.json
View file

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/nyt_games",
"integration_type": "service",
"iot_class": "cloud_polling",
"requirements": ["nyt_games==0.4.3"]
"requirements": ["nyt_games==0.4.4"]
}

2
homeassistant/components/nyt_games/sensor.py
View file

@ -139,7 +139,7 @@ CONNECTIONS_SENSORS: tuple[NYTGamesConnectionsSensorEntityDescription, ...] = (
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfTime.DAYS,
device_class=SensorDeviceClass.DURATION,
value_fn=lambda connections: connections.current_streak,
value_fn=lambda connections: connections.max_streak,
),
)

2
homeassistant/components/ring/manifest.json
View file

@ -14,5 +14,5 @@
"iot_class": "cloud_polling",
"loggers": ["ring_doorbell"],
"quality_scale": "silver",
"requirements": ["ring-doorbell==0.9.6"]
"requirements": ["ring-doorbell==0.9.8"]
}

2
homeassistant/components/tibber/manifest.json
View file

@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["tibber"],
"quality_scale": "silver",
"requirements": ["pyTibber==0.30.2"]
"requirements": ["pyTibber==0.30.3"]
}

47
homeassistant/components/vodafone_station/diagnostics.py Normal file
View file

@ -0,0 +1,47 @@
"""Diagnostics support for Vodafone Station."""
from __future__ import annotations
from typing import Any
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .coordinator import VodafoneStationRouter
TO_REDACT = {CONF_USERNAME, CONF_PASSWORD}
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator: VodafoneStationRouter = hass.data[DOMAIN][entry.entry_id]
sensors_data = coordinator.data.sensors
return {
"entry": async_redact_data(entry.as_dict(), TO_REDACT),
"device_info": {
"sys_model_name": sensors_data.get("sys_model_name"),
"sys_firmware_version": sensors_data["sys_firmware_version"],
"sys_hardware_version": sensors_data["sys_hardware_version"],
"sys_cpu_usage": sensors_data["sys_cpu_usage"][:-1],
"sys_memory_usage": sensors_data["sys_memory_usage"][:-1],
"sys_reboot_cause": sensors_data["sys_reboot_cause"],
"last_update success": coordinator.last_update_success,
"last_exception": coordinator.last_exception,
"client_devices": [
{
"hostname": device_info.device.name,
"connection_type": device_info.device.connection_type,
"connected": device_info.device.connected,
"type": device_info.device.type,
}
for _, device_info in coordinator.data.devices.items()
],
},
}

41
homeassistant/components/vodafone_station/sensor.py
View file

@ -23,25 +23,42 @@ from .const import _LOGGER, DOMAIN, LINE_TYPES
from .coordinator import VodafoneStationRouter
NOT_AVAILABLE: list = ["", "N/A", "0.0.0.0"]
UPTIME_DEVIATION = 30
@dataclass(frozen=True, kw_only=True)
class VodafoneStationEntityDescription(SensorEntityDescription):
"""Vodafone Station entity description."""
value: Callable[[Any, Any], Any] = (
lambda coordinator, key: coordinator.data.sensors[key]
value: Callable[[Any, Any, Any], Any] = (
lambda coordinator, last_value, key: coordinator.data.sensors[key]
)
is_suitable: Callable[[dict], bool] = lambda val: True
def _calculate_uptime(coordinator: VodafoneStationRouter, key: str) -> datetime:
def _calculate_uptime(
coordinator: VodafoneStationRouter,
last_value: datetime | None,
key: str,
) -> datetime:
"""Calculate device uptime."""
return coordinator.api.convert_uptime(coordinator.data.sensors[key])
delta_uptime = coordinator.api.convert_uptime(coordinator.data.sensors[key])
if (
not last_value
or abs((delta_uptime - last_value).total_seconds()) > UPTIME_DEVIATION
):
return delta_uptime
return last_value
def _line_connection(coordinator: VodafoneStationRouter, key: str) -> str | None:
def _line_connection(
coordinator: VodafoneStationRouter,
last_value: str | None,
key: str,
) -> str | None:
"""Identify line type."""
value = coordinator.data.sensors
@ -126,14 +143,18 @@ SENSOR_TYPES: Final = (
translation_key="sys_cpu_usage",
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
value=lambda coordinator, key: float(coordinator.data.sensors[key][:-1]),
value=lambda coordinator, last_value, key: float(
coordinator.data.sensors[key][:-1]
),
),
VodafoneStationEntityDescription(
key="sys_memory_usage",
translation_key="sys_memory_usage",
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
value=lambda coordinator, key: float(coordinator.data.sensors[key][:-1]),
value=lambda coordinator, last_value, key: float(
coordinator.data.sensors[key][:-1]
),
),
VodafoneStationEntityDescription(
key="sys_reboot_cause",
@ -178,10 +199,12 @@ class VodafoneStationSensorEntity(
self.entity_description = description
self._attr_device_info = coordinator.device_info
self._attr_unique_id = f"{coordinator.serial_number}_{description.key}"
self._old_state = None
@property
def native_value(self) -> StateType:
"""Sensor value."""
return self.entity_description.value(
self.coordinator, self.entity_description.key
self._old_state = self.entity_description.value(
self.coordinator, self._old_state, self.entity_description.key
)
return self._old_state

2
homeassistant/const.py
View file

@ -24,7 +24,7 @@ if TYPE_CHECKING:
APPLICATION_NAME: Final = "HomeAssistant"
MAJOR_VERSION: Final = 2024
MINOR_VERSION: Final = 10
PATCH_VERSION: Final = "3"
PATCH_VERSION: Final = "4"
__short_version__: Final = f"{MAJOR_VERSION}.{MINOR_VERSION}"
__version__: Final = f"{__short_version__}.{PATCH_VERSION}"
REQUIRED_PYTHON_VER: Final[tuple[int, int, int]] = (3, 12, 0)

4
homeassistant/generated/dhcp.py
View file

@ -37,10 +37,6 @@ DHCP: Final[list[dict[str, str | bool]]] = [
"hostname": "august*",
"macaddress": "E076D0*",
},
{
"domain": "awair",
"macaddress": "70886B1*",
},
{
"domain": "axis",
"registered_devices": True,

5
homeassistant/helpers/llm.py
View file

@ -177,11 +177,6 @@ class APIInstance:
else:
raise HomeAssistantError(f'Tool "{tool_input.tool_name}" not found')
tool_input = ToolInput(
tool_name=tool_input.tool_name,
tool_args=tool.parameters(tool_input.tool_args),
)
return await tool.async_call(self.api.hass, tool_input, self.llm_context)

2
homeassistant/package_constraints.txt
View file

@ -32,7 +32,7 @@ habluetooth==3.4.0
hass-nabucasa==0.81.1
hassil==1.7.4
home-assistant-bluetooth==1.13.0
home-assistant-frontend==20241002.3
home-assistant-frontend==20241002.4
home-assistant-intents==2024.10.2
httpx==0.27.2
ifaddr==0.2.0

33
homeassistant/util/timeout.py
View file

@ -16,7 +16,7 @@ from .async_ import run_callback_threadsafe
ZONE_GLOBAL = "global"
class _State(str, enum.Enum):
class _State(enum.Enum):
"""States of a task."""
INIT = "INIT"
@ -160,11 +160,16 @@ class _GlobalTaskContext:
self._wait_zone: asyncio.Event = asyncio.Event()
self._state: _State = _State.INIT
self._cool_down: float = cool_down
self._cancelling = 0
async def __aenter__(self) -> Self:
self._manager.global_tasks.append(self)
self._start_timer()
self._state = _State.ACTIVE
# Remember if the task was already cancelling
# so when we __aexit__ we can decide if we should
# raise asyncio.TimeoutError or let the cancellation propagate
self._cancelling = self._task.cancelling()
return self
async def __aexit__(
@ -177,7 +182,15 @@ class _GlobalTaskContext:
self._manager.global_tasks.remove(self)
# Timeout on exit
if exc_type is asyncio.CancelledError and self.state == _State.TIMEOUT:
if exc_type is asyncio.CancelledError and self.state is _State.TIMEOUT:
# The timeout was hit, and the task was cancelled
# so we need to uncancel the task since the cancellation
# should not leak out of the context manager
if self._task.uncancel() > self._cancelling:
# If the task was already cancelling don't raise
# asyncio.TimeoutError and instead return None
# to allow the cancellation to propagate
return None
raise TimeoutError
self._state = _State.EXIT
@ -266,6 +279,7 @@ class _ZoneTaskContext:
self._time_left: float = timeout
self._expiration_time: float | None = None
self._timeout_handler: asyncio.Handle | None = None
self._cancelling = 0
@property
def state(self) -> _State:
@ -280,6 +294,11 @@ class _ZoneTaskContext:
if self._zone.freezes_done:
self._start_timer()
# Remember if the task was already cancelling
# so when we __aexit__ we can decide if we should
# raise asyncio.TimeoutError or let the cancellation propagate
self._cancelling = self._task.cancelling()
return self
async def __aexit__(
@ -292,7 +311,15 @@ class _ZoneTaskContext:
self._stop_timer()
# Timeout on exit
if exc_type is asyncio.CancelledError and self.state == _State.TIMEOUT:
if exc_type is asyncio.CancelledError and self.state is _State.TIMEOUT:
# The timeout was hit, and the task was cancelled
# so we need to uncancel the task since the cancellation
# should not leak out of the context manager
if self._task.uncancel() > self._cancelling:
# If the task was already cancelling don't raise
# asyncio.TimeoutError and instead return None
# to allow the cancellation to propagate
return None
raise TimeoutError
self._state = _State.EXIT

2
pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "homeassistant"
version = "2024.10.3"
version = "2024.10.4"
license = {text = "Apache-2.0"}
description = "Open-source home automation platform running on Python 3."
readme = "README.rst"

12
requirements_all.txt
View file

@ -1117,7 +1117,7 @@ hole==0.8.0
holidays==0.58
# homeassistant.components.frontend
home-assistant-frontend==20241002.3
home-assistant-frontend==20241002.4
# homeassistant.components.conversation
home-assistant-intents==2024.10.2
@ -1484,7 +1484,7 @@ numato-gpio==0.13.0
numpy==1.26.4
# homeassistant.components.nyt_games
nyt_games==0.4.3
nyt_games==0.4.4
# homeassistant.components.oasa_telematics
oasatelematics==0.3
@ -1710,7 +1710,7 @@ pyCEC==0.5.2
pyControl4==1.2.0
# homeassistant.components.duotecno
pyDuotecno==2024.10.0
pyDuotecno==2024.10.1
# homeassistant.components.electrasmart
pyElectra==1.2.4
@ -1728,7 +1728,7 @@ pyRFXtrx==0.31.1
pySDCP==1
# homeassistant.components.tibber
pyTibber==0.30.2
pyTibber==0.30.3
# homeassistant.components.dlink
pyW215==0.7.0
@ -2543,7 +2543,7 @@ rfk101py==0.0.1
rflink==0.0.66
# homeassistant.components.ring
ring-doorbell==0.9.6
ring-doorbell==0.9.8
# homeassistant.components.fleetgo
ritassist==0.9.2
@ -3032,7 +3032,7 @@ youless-api==2.1.2
youtubeaio==1.1.5
# homeassistant.components.media_extractor
yt-dlp==2024.10.07
yt-dlp==2024.10.22
# homeassistant.components.zamg
zamg==0.3.6

12
requirements_test_all.txt
View file

@ -943,7 +943,7 @@ hole==0.8.0
holidays==0.58
# homeassistant.components.frontend
home-assistant-frontend==20241002.3
home-assistant-frontend==20241002.4
# homeassistant.components.conversation
home-assistant-intents==2024.10.2
@ -1232,7 +1232,7 @@ numato-gpio==0.13.0
numpy==1.26.4
# homeassistant.components.nyt_games
nyt_games==0.4.3
nyt_games==0.4.4
# homeassistant.components.google
oauth2client==4.1.3
@ -1396,7 +1396,7 @@ pyCEC==0.5.2
pyControl4==1.2.0
# homeassistant.components.duotecno
pyDuotecno==2024.10.0
pyDuotecno==2024.10.1
# homeassistant.components.electrasmart
pyElectra==1.2.4
@ -1405,7 +1405,7 @@ pyElectra==1.2.4
pyRFXtrx==0.31.1
# homeassistant.components.tibber
pyTibber==0.30.2
pyTibber==0.30.3
# homeassistant.components.dlink
pyW215==0.7.0
@ -2025,7 +2025,7 @@ reolink-aio==0.9.11
rflink==0.0.66
# homeassistant.components.ring
ring-doorbell==0.9.6
ring-doorbell==0.9.8
# homeassistant.components.roku
rokuecp==0.19.3
@ -2415,7 +2415,7 @@ youless-api==2.1.2
youtubeaio==1.1.5
# homeassistant.components.media_extractor
yt-dlp==2024.10.07
yt-dlp==2024.10.22
# homeassistant.components.zamg
zamg==0.3.6

79
tests/components/comelit/const.py
View file

@ -1,6 +1,19 @@
"""Common stuff for Comelit SimpleHome tests."""
from aiocomelit.const import VEDO
from aiocomelit import ComelitVedoAreaObject, ComelitVedoZoneObject
from aiocomelit.api import ComelitSerialBridgeObject
from aiocomelit.const import (
CLIMATE,
COVER,
IRRIGATION,
LIGHT,
OTHER,
SCENARIO,
VEDO,
WATT,
AlarmAreaState,
AlarmZoneState,
)
from homeassistant.components.comelit.const import DOMAIN
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PIN, CONF_PORT, CONF_TYPE
@ -27,3 +40,67 @@ MOCK_USER_BRIDGE_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
MOCK_USER_VEDO_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][1]
FAKE_PIN = 5678
BRIDGE_DEVICE_QUERY = {
CLIMATE: {},
COVER: {
0: ComelitSerialBridgeObject(
index=0,
name="Cover0",
status=0,
human_status="closed",
type="cover",
val=0,
protected=0,
zone="Open space",
power=0.0,
power_unit=WATT,
)
},
LIGHT: {
0: ComelitSerialBridgeObject(
index=0,
name="Light0",
status=0,
human_status="off",
type="light",
val=0,
protected=0,
zone="Bathroom",
power=0.0,
power_unit=WATT,
)
},
OTHER: {},
IRRIGATION: {},
SCENARIO: {},
}
VEDO_DEVICE_QUERY = {
"aree": {
0: ComelitVedoAreaObject(
index=0,
name="Area0",
p1=True,
p2=False,
ready=False,
armed=False,
alarm=False,
alarm_memory=False,
sabotage=False,
anomaly=False,
in_time=False,
out_time=False,
human_status=AlarmAreaState.UNKNOWN,
)
},
"zone": {
0: ComelitVedoZoneObject(
index=0,
name="Zone0",
status_api="0x000",
status=0,
human_status=AlarmZoneState.REST,
)
},
}

144
tests/components/comelit/snapshots/test_diagnostics.ambr Normal file
View file

@ -0,0 +1,144 @@
# serializer version: 1
# name: test_entry_diagnostics_bridge
dict({
'device_info': dict({
'devices': list([
dict({
'clima': list([
]),
}),
dict({
'shutter': list([
dict({
'0': dict({
'human_status': 'closed',
'name': 'Cover0',
'power': 0.0,
'power_unit': 'W',
'protected': 0,
'status': 0,
'val': 0,
'zone': 'Open space',
}),
}),
]),
}),
dict({
'light': list([
dict({
'0': dict({
'human_status': 'off',
'name': 'Light0',
'power': 0.0,
'power_unit': 'W',
'protected': 0,
'status': 0,
'val': 0,
'zone': 'Bathroom',
}),
}),
]),
}),
dict({
'other': list([
]),
}),
dict({
'irrigation': list([
]),
}),
dict({
'scenario': list([
]),
}),
]),
'last_exception': 'None',
'last_update success': True,
}),
'entry': dict({
'data': dict({
'host': 'fake_host',
'pin': '**REDACTED**',
'port': 80,
}),
'disabled_by': None,
'discovery_keys': dict({
}),
'domain': 'comelit',
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'Mock Title',
'unique_id': None,
'version': 1,
}),
'type': 'Serial bridge',
})
# ---
# name: test_entry_diagnostics_vedo
dict({
'device_info': dict({
'devices': list([
dict({
'aree': list([
dict({
'0': dict({
'alarm': False,
'alarm_memory': False,
'anomaly': False,
'armed': False,
'human_status': 'unknown',
'in_time': False,
'name': 'Area0',
'out_time': False,
'p1': True,
'p2': False,
'ready': False,
'sabotage': False,
}),
}),
]),
}),
dict({
'zone': list([
dict({
'0': dict({
'human_status': 'rest',
'name': 'Zone0',
'status': 0,
'status_api': '0x000',
}),
}),
]),
}),
]),
'last_exception': 'None',
'last_update success': True,
}),
'entry': dict({
'data': dict({
'host': 'fake_vedo_host',
'pin': '**REDACTED**',
'port': 8080,
'type': 'Vedo system',
}),
'disabled_by': None,
'discovery_keys': dict({
}),
'domain': 'comelit',
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'Mock Title',
'unique_id': None,
'version': 1,
}),
'type': 'Vedo system',
})
# ---

81
tests/components/comelit/test_diagnostics.py Normal file
View file

@ -0,0 +1,81 @@
"""Tests for Comelit Simplehome diagnostics platform."""
from __future__ import annotations
from unittest.mock import patch
from syrupy import SnapshotAssertion
from syrupy.filters import props
from homeassistant.components.comelit.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from .const import (
BRIDGE_DEVICE_QUERY,
MOCK_USER_BRIDGE_DATA,
MOCK_USER_VEDO_DATA,
VEDO_DEVICE_QUERY,
)
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
async def test_entry_diagnostics_bridge(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test Bridge config entry diagnostics."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_BRIDGE_DATA)
entry.add_to_hass(hass)
with (
patch("aiocomelit.api.ComeliteSerialBridgeApi.login"),
patch(
"aiocomelit.api.ComeliteSerialBridgeApi.get_all_devices",
return_value=BRIDGE_DEVICE_QUERY,
),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == snapshot(
exclude=props(
"entry_id",
"created_at",
"modified_at",
)
)
async def test_entry_diagnostics_vedo(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test Vedo System config entry diagnostics."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_VEDO_DATA)
entry.add_to_hass(hass)
with (
patch("aiocomelit.api.ComelitVedoApi.login"),
patch(
"aiocomelit.api.ComelitVedoApi.get_all_areas_and_zones",
return_value=VEDO_DEVICE_QUERY,
),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == snapshot(
exclude=props(
"entry_id",
"created_at",
"modified_at",
)
)

2
tests/components/devolo_home_network/mock.py
View file

@ -50,7 +50,7 @@ class MockDevice(Device):
self, session_instance: httpx.AsyncClient | None = None
) -> None:
"""Give a mocked device the needed properties."""
self.mac = DISCOVERY_INFO.properties["PlcMacAddress"]
self.mac = DISCOVERY_INFO.properties["PlcMacAddress"] if self.plcnet else None
self.mt_number = DISCOVERY_INFO.properties["MT"]
self.product = DISCOVERY_INFO.properties["Product"]
self.serial_number = DISCOVERY_INFO.properties["SN"]

34
tests/components/devolo_home_network/snapshots/test_init.ambr
View file

@ -1,5 +1,5 @@
# serializer version: 1
# name: test_setup_entry
# name: test_setup_entry[mock_device]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
@ -35,3 +35,35 @@
'via_device_id': None,
})
# ---
# name: test_setup_entry[mock_repeater_device]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': 'http://192.0.2.1',
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'devolo_home_network',
'1234567890',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'devolo',
'model': 'dLAN pro 1200+ WiFi ac',
'model_id': '2730',
'name': 'Mock Title',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '1234567890',
'suggested_area': None,
'sw_version': '5.6.1',
'via_device_id': None,
})
# ---

5
tests/components/devolo_home_network/test_init.py
View file

@ -27,13 +27,16 @@ from .mock import MockDevice
from tests.common import MockConfigEntry
@pytest.mark.parametrize("device", ["mock_device", "mock_repeater_device"])
async def test_setup_entry(
hass: HomeAssistant,
mock_device: MockDevice,
device: str,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
request: pytest.FixtureRequest,
) -> None:
"""Test setup entry."""
mock_device: MockDevice = request.getfixturevalue(device)
entry = configure_integration(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()

117
tests/components/evohome/test_init.py
View file

@ -2,11 +2,19 @@
from __future__ import annotations
from http import HTTPStatus
import logging
from unittest.mock import patch
from evohomeasync2 import exceptions as exc
from evohomeasync2.broker import _ERR_MSG_LOOKUP_AUTH, _ERR_MSG_LOOKUP_BASE
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.evohome import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import setup_evohome
from .const import TEST_INSTALLS
@ -28,3 +36,112 @@ async def test_entities(
await setup_evohome(hass, config, install=install)
assert hass.states.async_all() == snapshot
SETUP_FAILED_ANTICIPATED = (
"homeassistant.setup",
logging.ERROR,
"Setup failed for 'evohome': Integration failed to initialize.",
)
SETUP_FAILED_UNEXPECTED = (
"homeassistant.setup",
logging.ERROR,
"Error during setup of component evohome",
)
AUTHENTICATION_FAILED = (
"homeassistant.components.evohome.helpers",
logging.ERROR,
"Failed to authenticate with the vendor's server. Check your username"
" and password. NB: Some special password characters that work"
" correctly via the website will not work via the web API. Message"
" is: ",
)
REQUEST_FAILED_NONE = (
"homeassistant.components.evohome.helpers",
logging.WARNING,
"Unable to connect with the vendor's server. "
"Check your network and the vendor's service status page. "
"Message is: ",
)
REQUEST_FAILED_503 = (
"homeassistant.components.evohome.helpers",
logging.WARNING,
"The vendor says their server is currently unavailable. "
"Check the vendor's service status page",
)
REQUEST_FAILED_429 = (
"homeassistant.components.evohome.helpers",
logging.WARNING,
"The vendor's API rate limit has been exceeded. "
"If this message persists, consider increasing the scan_interval",
)
REQUEST_FAILED_LOOKUP = {
None: [
REQUEST_FAILED_NONE,
SETUP_FAILED_ANTICIPATED,
],
HTTPStatus.SERVICE_UNAVAILABLE: [
REQUEST_FAILED_503,
SETUP_FAILED_ANTICIPATED,
],
HTTPStatus.TOO_MANY_REQUESTS: [
REQUEST_FAILED_429,
SETUP_FAILED_ANTICIPATED,
],
}
@pytest.mark.parametrize(
"status", [*sorted([*_ERR_MSG_LOOKUP_AUTH, HTTPStatus.BAD_GATEWAY]), None]
)
async def test_authentication_failure_v2(
hass: HomeAssistant,
config: dict[str, str],
status: HTTPStatus,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test failure to setup an evohome-compatible system.
In this instance, the failure occurs in the v2 API.
"""
with patch("evohomeasync2.broker.Broker.get") as mock_fcn:
mock_fcn.side_effect = exc.AuthenticationFailed("", status=status)
with caplog.at_level(logging.WARNING):
result = await async_setup_component(hass, DOMAIN, {DOMAIN: config})
assert result is False
assert caplog.record_tuples == [
AUTHENTICATION_FAILED,
SETUP_FAILED_ANTICIPATED,
]
@pytest.mark.parametrize(
"status", [*sorted([*_ERR_MSG_LOOKUP_BASE, HTTPStatus.BAD_GATEWAY]), None]
)
async def test_client_request_failure_v2(
hass: HomeAssistant,
config: dict[str, str],
status: HTTPStatus,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test failure to setup an evohome-compatible system.
In this instance, the failure occurs in the v2 API.
"""
with patch("evohomeasync2.broker.Broker.get") as mock_fcn:
mock_fcn.side_effect = exc.RequestFailed("", status=status)
with caplog.at_level(logging.WARNING):
result = await async_setup_component(hass, DOMAIN, {DOMAIN: config})
assert result is False
assert caplog.record_tuples == REQUEST_FAILED_LOOKUP.get(
status, [SETUP_FAILED_UNEXPECTED]
)

59
tests/components/honeywell/test_climate.py
View file

@ -5,6 +5,7 @@ from unittest.mock import MagicMock
from aiohttp import ClientConnectionError
import aiosomecomfort
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
@ -29,6 +30,8 @@ from homeassistant.components.climate import (
)
from homeassistant.components.honeywell.climate import (
DOMAIN,
MODE_PERMANENT_HOLD,
MODE_TEMPORARY_HOLD,
PRESET_HOLD,
RETRY,
SCAN_INTERVAL,
@ -1207,3 +1210,59 @@ async def test_unique_id(
await init_integration(hass, config_entry)
entity_entry = entity_registry.async_get(f"climate.{device.name}")
assert entity_entry.unique_id == str(device.deviceid)
async def test_preset_mode(
hass: HomeAssistant,
device: MagicMock,
config_entry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test mode settings properly reflected."""
await init_integration(hass, config_entry)
entity_id = f"climate.{device.name}"
device.raw_ui_data["StatusHeat"] = 3
device.raw_ui_data["StatusCool"] = 3
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == PRESET_NONE
device.raw_ui_data["StatusHeat"] = MODE_TEMPORARY_HOLD
device.raw_ui_data["StatusCool"] = MODE_TEMPORARY_HOLD
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == PRESET_HOLD
device.raw_ui_data["StatusHeat"] = MODE_PERMANENT_HOLD
device.raw_ui_data["StatusCool"] = MODE_PERMANENT_HOLD
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == PRESET_HOLD
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: entity_id, ATTR_PRESET_MODE: PRESET_AWAY},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == PRESET_AWAY
device.raw_ui_data["StatusHeat"] = 3
device.raw_ui_data["StatusCool"] = 3
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == PRESET_NONE

2
tests/components/nyt_games/snapshots/test_sensor.ambr
View file

@ -98,7 +98,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
'state': '2',
})
# ---
# name: test_all_entities[sensor.connections_last_played-entry]

97
tests/components/vodafone_station/const.py
View file

@ -1,5 +1,7 @@
"""Common stuff for Vodafone Station tests."""
from aiovodafone.api import VodafoneStationDevice
from homeassistant.components.vodafone_station.const import DOMAIN
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
@ -16,3 +18,98 @@ MOCK_CONFIG = {
}
MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
DEVICE_DATA_QUERY = {
"xx:xx:xx:xx:xx:xx": VodafoneStationDevice(
connected=True,
connection_type="wifi",
ip_address="192.168.1.10",
name="WifiDevice0",
mac="xx:xx:xx:xx:xx:xx",
type="laptop",
wifi="2.4G",
)
}
SENSOR_DATA_QUERY = {
"sys_serial_number": "M123456789",
"sys_firmware_version": "XF6_4.0.05.04",
"sys_bootloader_version": "0220",
"sys_hardware_version": "RHG3006 v1",
"omci_software_version": "\t\t1.0.0.1_41032\t\t\n",
"sys_uptime": "12:16:41",
"sys_cpu_usage": "97%",
"sys_reboot_cause": "Web Reboot",
"sys_memory_usage": "51.94%",
"sys_wireless_driver_version": "17.10.188.75;17.10.188.75",
"sys_wireless_driver_version_5g": "17.10.188.75;17.10.188.75",
"vf_internet_key_online_since": "",
"vf_internet_key_ip_addr": "0.0.0.0",
"vf_internet_key_system": "0.0.0.0",
"vf_internet_key_mode": "Auto",
"sys_voip_version": "v02.01.00_01.13a\n",
"sys_date_time": "20.10.2024 | 03:44 pm",
"sys_build_time": "Sun Jun 23 17:55:49 CST 2024\n",
"sys_model_name": "RHG3006",
"inter_ip_address": "1.1.1.1",
"inter_gateway": "1.1.1.2",
"inter_primary_dns": "1.1.1.3",
"inter_secondary_dns": "1.1.1.4",
"inter_firewall": "601036",
"inter_wan_ip_address": "1.1.1.1",
"inter_ipv6_link_local_address": "",
"inter_ipv6_link_global_address": "",
"inter_ipv6_gateway": "",
"inter_ipv6_prefix_delegation": "",
"inter_ipv6_dns_address1": "",
"inter_ipv6_dns_address2": "",
"lan_ip_network": "192.168.0.1/24",
"lan_default_gateway": "192.168.0.1",
"lan_subnet_address_subnet1": "",
"lan_mac_address": "11:22:33:44:55:66",
"lan_dhcp_server": "601036",
"lan_dhcpv6_server": "601036",
"lan_router_advertisement": "601036",
"lan_ipv6_default_gateway": "fe80::1",
"lan_port1_switch_mode": "1301722",
"lan_port2_switch_mode": "1301722",
"lan_port3_switch_mode": "1301722",
"lan_port4_switch_mode": "1301722",
"lan_port1_switch_speed": "10",
"lan_port2_switch_speed": "100",
"lan_port3_switch_speed": "1000",
"lan_port4_switch_speed": "1000",
"lan_port1_switch_status": "1301724",
"lan_port2_switch_status": "1301724",
"lan_port3_switch_status": "1301724",
"lan_port4_switch_status": "1301724",
"wifi_status": "601036",
"wifi_name": "Wifi-Main-Network",
"wifi_mac_address": "AA:BB:CC:DD:EE:FF",
"wifi_security": "401027",
"wifi_channel": "8",
"wifi_bandwidth": "573",
"guest_wifi_status": "601037",
"guest_wifi_name": "Wifi-Guest",
"guest_wifi_mac_addr": "AA:BB:CC:DD:EE:GG",
"guest_wifi_security": "401027",
"guest_wifi_channel": "N/A",
"guest_wifi_ip": "192.168.2.1",
"guest_wifi_subnet_addr": "255.255.255.0",
"guest_wifi_dhcp_server": "192.168.2.1",
"wifi_status_5g": "601036",
"wifi_name_5g": "Wifi-Main-Network",
"wifi_mac_address_5g": "AA:BB:CC:DD:EE:HH",
"wifi_security_5g": "401027",
"wifi_channel_5g": "36",
"wifi_bandwidth_5g": "4803",
"guest_wifi_status_5g": "601037",
"guest_wifi_name_5g": "Wifi-Guest",
"guest_wifi_mac_addr_5g": "AA:BB:CC:DD:EE:II",
"guest_wifi_channel_5g": "N/A",
"guest_wifi_security_5g": "401027",
"guest_wifi_ip_5g": "192.168.2.1",
"guest_wifi_subnet_addr_5g": "255.255.255.0",
"guest_wifi_dhcp_server_5g": "192.168.2.1",
}

43
tests/components/vodafone_station/snapshots/test_diagnostics.ambr Normal file
View file

@ -0,0 +1,43 @@
# serializer version: 1
# name: test_entry_diagnostics
dict({
'device_info': dict({
'client_devices': list([
dict({
'connected': True,
'connection_type': 'wifi',
'hostname': 'WifiDevice0',
'type': 'laptop',
}),
]),
'last_exception': None,
'last_update success': True,
'sys_cpu_usage': '97',
'sys_firmware_version': 'XF6_4.0.05.04',
'sys_hardware_version': 'RHG3006 v1',
'sys_memory_usage': '51.94',
'sys_model_name': 'RHG3006',
'sys_reboot_cause': 'Web Reboot',
}),
'entry': dict({
'data': dict({
'host': 'fake_host',
'password': '**REDACTED**',
'username': '**REDACTED**',
}),
'disabled_by': None,
'discovery_keys': dict({
}),
'domain': 'vodafone_station',
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'Mock Title',
'unique_id': None,
'version': 1,
}),
})
# ---

51
tests/components/vodafone_station/test_diagnostics.py Normal file
View file

@ -0,0 +1,51 @@
"""Tests for Vodafone Station diagnostics platform."""
from __future__ import annotations
from unittest.mock import patch
from syrupy import SnapshotAssertion
from syrupy.filters import props
from homeassistant.components.vodafone_station.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from .const import DEVICE_DATA_QUERY, MOCK_USER_DATA, SENSOR_DATA_QUERY
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
async def test_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test config entry diagnostics."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_DATA)
entry.add_to_hass(hass)
with (
patch("aiovodafone.api.VodafoneStationSercommApi.login"),
patch(
"aiovodafone.api.VodafoneStationSercommApi.get_devices_data",
return_value=DEVICE_DATA_QUERY,
),
patch(
"aiovodafone.api.VodafoneStationSercommApi.get_sensor_data",
return_value=SENSOR_DATA_QUERY,
),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == snapshot(
exclude=props(
"entry_id",
"created_at",
"modified_at",
)
)

114
tests/util/test_timeout.py
View file

@ -146,6 +146,62 @@ async def test_simple_global_timeout_freeze_with_executor_job(
await hass.async_add_executor_job(time.sleep, 0.3)
async def test_simple_global_timeout_does_not_leak_upward(
hass: HomeAssistant,
) -> None:
"""Test a global timeout does not leak upward."""
timeout = TimeoutManager()
current_task = asyncio.current_task()
assert current_task is not None
cancelling_inside_timeout = None
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
async with timeout.async_timeout(0.1):
cancelling_inside_timeout = current_task.cancelling()
await asyncio.sleep(0.3)
assert cancelling_inside_timeout == 0
# After the context manager exits, the task should no longer be cancelling
assert current_task.cancelling() == 0
async def test_simple_global_timeout_does_swallow_cancellation(
hass: HomeAssistant,
) -> None:
"""Test a global timeout does not swallow cancellation."""
timeout = TimeoutManager()
current_task = asyncio.current_task()
assert current_task is not None
cancelling_inside_timeout = None
async def task_with_timeout() -> None:
nonlocal cancelling_inside_timeout
new_task = asyncio.current_task()
assert new_task is not None
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
cancelling_inside_timeout = new_task.cancelling()
async with timeout.async_timeout(0.1):
await asyncio.sleep(0.3)
# After the context manager exits, the task should no longer be cancelling
assert current_task.cancelling() == 0
task = asyncio.create_task(task_with_timeout())
await asyncio.sleep(0)
task.cancel()
assert task.cancelling() == 1
assert cancelling_inside_timeout == 0
# Cancellation should not leak into the current task
assert current_task.cancelling() == 0
# Cancellation should not be swallowed if the task is cancelled
# and it also times out
await asyncio.sleep(0)
with pytest.raises(asyncio.CancelledError):
await task
assert task.cancelling() == 1
async def test_simple_global_timeout_freeze_reset() -> None:
"""Test a simple global timeout freeze reset."""
timeout = TimeoutManager()
@ -166,6 +222,62 @@ async def test_simple_zone_timeout() -> None:
await asyncio.sleep(0.3)
async def test_simple_zone_timeout_does_not_leak_upward(
hass: HomeAssistant,
) -> None:
"""Test a zone timeout does not leak upward."""
timeout = TimeoutManager()
current_task = asyncio.current_task()
assert current_task is not None
cancelling_inside_timeout = None
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
async with timeout.async_timeout(0.1, "test"):
cancelling_inside_timeout = current_task.cancelling()
await asyncio.sleep(0.3)
assert cancelling_inside_timeout == 0
# After the context manager exits, the task should no longer be cancelling
assert current_task.cancelling() == 0
async def test_simple_zone_timeout_does_swallow_cancellation(
hass: HomeAssistant,
) -> None:
"""Test a zone timeout does not swallow cancellation."""
timeout = TimeoutManager()
current_task = asyncio.current_task()
assert current_task is not None
cancelling_inside_timeout = None
async def task_with_timeout() -> None:
nonlocal cancelling_inside_timeout
new_task = asyncio.current_task()
assert new_task is not None
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
async with timeout.async_timeout(0.1, "test"):
cancelling_inside_timeout = current_task.cancelling()
await asyncio.sleep(0.3)
# After the context manager exits, the task should no longer be cancelling
assert current_task.cancelling() == 0
task = asyncio.create_task(task_with_timeout())
await asyncio.sleep(0)
task.cancel()
assert task.cancelling() == 1
# Cancellation should not leak into the current task
assert cancelling_inside_timeout == 0
assert current_task.cancelling() == 0
# Cancellation should not be swallowed if the task is cancelled
# and it also times out
await asyncio.sleep(0)
with pytest.raises(asyncio.CancelledError):
await task
assert task.cancelling() == 1
async def test_multiple_zone_timeout() -> None:
"""Test a simple zone timeout."""
timeout = TimeoutManager()
@ -327,7 +439,7 @@ async def test_simple_zone_timeout_freeze_without_timeout_exeption() -> None:
await asyncio.sleep(0.4)
async def test_simple_zone_timeout_zone_with_timeout_exeption() -> None:
async def test_simple_zone_timeout_zone_with_timeout_exception() -> None:
"""Test a simple zone timeout freeze on a zone that does not have a timeout set."""
timeout = TimeoutManager()