mirror of
https://github.com/home-assistant/core.git
synced 2026-10-05 21:17:41 +01:00
373 lines
13 KiB
Python
373 lines
13 KiB
Python
"""Tests for the Duco number platform."""
|
|
|
|
import asyncio
|
|
from dataclasses import replace
|
|
from unittest.mock import AsyncMock, call
|
|
|
|
from duco_connectivity import (
|
|
BypassSupplyTemperatureTarget,
|
|
DucoError,
|
|
DucoRateLimitError,
|
|
)
|
|
from freezegun.api import FrozenDateTimeFactory
|
|
import pytest
|
|
from syrupy.assertion import SnapshotAssertion
|
|
|
|
from homeassistant.components.number import DOMAIN as NUMBER_DOMAIN, SERVICE_SET_VALUE
|
|
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, Platform
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import entity_registry as er
|
|
from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
|
|
|
|
from . import async_fire_coordinator_update, setup_platform_integration
|
|
|
|
from tests.common import MockConfigEntry, snapshot_platform
|
|
|
|
_ZONE_1_ENTITY_ID = "number.living_bypass_target_1"
|
|
_ZONE_2_ENTITY_ID = "number.living_bypass_target_2"
|
|
_ZONE_8_ENTITY_ID = "number.living_bypass_target_8"
|
|
|
|
|
|
@pytest.fixture
|
|
async def init_integration(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> MockConfigEntry:
|
|
"""Set up only the number platform for testing."""
|
|
return await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
|
|
async def test_bypass_supply_temperature_target_numbers_support_all_exposed_zones(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test bypass target controls are created for all exposed zones."""
|
|
mock_bypass_supply_temperature_targets[8] = BypassSupplyTemperatureTarget(
|
|
zone_id=8,
|
|
value=22.0,
|
|
minimum=15.0,
|
|
increment=0.1,
|
|
maximum=25.0,
|
|
)
|
|
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
for entity_id in (
|
|
_ZONE_1_ENTITY_ID,
|
|
_ZONE_2_ENTITY_ID,
|
|
_ZONE_8_ENTITY_ID,
|
|
):
|
|
assert hass.states.get(entity_id) is not None
|
|
|
|
mock_duco_client.async_get_bypass_supply_temperature_targets.assert_awaited_once_with()
|
|
|
|
|
|
async def test_successful_write_does_not_recover_failed_coordinator(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test a successful write does not recover a failed coordinator."""
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
write_started = asyncio.Event()
|
|
release_write = asyncio.Event()
|
|
target = mock_bypass_supply_temperature_targets[1]
|
|
|
|
async def set_bypass_supply_temperature_target(
|
|
zone_id: int,
|
|
temperature: float,
|
|
*,
|
|
target: BypassSupplyTemperatureTarget,
|
|
) -> BypassSupplyTemperatureTarget:
|
|
updated_target = replace(target, zone_id=zone_id, value=temperature)
|
|
mock_bypass_supply_temperature_targets[zone_id] = updated_target
|
|
write_started.set()
|
|
await release_write.wait()
|
|
return updated_target
|
|
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.side_effect = (
|
|
set_bypass_supply_temperature_target
|
|
)
|
|
write_task = asyncio.create_task(
|
|
hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": 20.5},
|
|
blocking=True,
|
|
)
|
|
)
|
|
await write_started.wait()
|
|
|
|
mock_duco_client.async_get_nodes.side_effect = DucoError("Temporary update failure")
|
|
await mock_config_entry.runtime_data.async_refresh()
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
release_write.set()
|
|
await write_task
|
|
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.assert_awaited_once_with(
|
|
1, 20.5, target=target
|
|
)
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
|
|
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
|
|
async def test_bypass_supply_temperature_target_number_entities_state(
|
|
hass: HomeAssistant,
|
|
entity_registry: er.EntityRegistry,
|
|
mock_config_entry: MockConfigEntry,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test bypass supply temperature target number entity states."""
|
|
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_duco_client")
|
|
async def test_bypass_supply_temperature_targets_missing_skips_number_creation(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test no number entities are created when bypass targets are unavailable."""
|
|
mock_bypass_supply_temperature_targets.clear()
|
|
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
assert hass.states.get(_ZONE_1_ENTITY_ID) is None
|
|
assert hass.states.get(_ZONE_2_ENTITY_ID) is None
|
|
|
|
|
|
@pytest.mark.usefixtures("init_integration")
|
|
async def test_set_bypass_supply_temperature_target(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test consecutive bypass target writes update directly from their responses."""
|
|
target = mock_bypass_supply_temperature_targets[1]
|
|
|
|
for value in (20.5, 21.0):
|
|
await hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": value},
|
|
blocking=True,
|
|
)
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == str(value)
|
|
|
|
assert (
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.await_args_list
|
|
== [
|
|
call(1, 20.5, target=target),
|
|
call(1, 21.0, target=replace(target, value=20.5)),
|
|
]
|
|
)
|
|
mock_duco_client.async_get_bypass_supply_temperature_targets.assert_awaited_once_with()
|
|
assert mock_bypass_supply_temperature_targets[1] == replace(target, value=21.0)
|
|
|
|
|
|
async def test_set_bypass_supply_temperature_target_honors_increment_metadata(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test bypass target writes follow the API-provided increment metadata."""
|
|
mock_bypass_supply_temperature_targets[1] = replace(
|
|
mock_bypass_supply_temperature_targets[1],
|
|
minimum=10.0,
|
|
increment=0.5,
|
|
maximum=25.5,
|
|
)
|
|
target = mock_bypass_supply_temperature_targets[1]
|
|
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
await hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": 20.5},
|
|
blocking=True,
|
|
)
|
|
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.assert_awaited_once_with(
|
|
1, 20.5, target=target
|
|
)
|
|
|
|
with pytest.raises(
|
|
HomeAssistantError,
|
|
match="supported increment of 0.5 starting at 10.0",
|
|
):
|
|
await hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": 20.2},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
async def test_set_bypass_supply_temperature_target_in_fahrenheit_units(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test Fahrenheit service writes normalize to the nearest supported Celsius step."""
|
|
hass.config.units = US_CUSTOMARY_SYSTEM
|
|
mock_bypass_supply_temperature_targets[1] = replace(
|
|
mock_bypass_supply_temperature_targets[1],
|
|
minimum=10.0,
|
|
increment=0.5,
|
|
maximum=25.5,
|
|
)
|
|
target = mock_bypass_supply_temperature_targets[1]
|
|
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
await hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": 69.0},
|
|
blocking=True,
|
|
)
|
|
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.assert_awaited_once_with(
|
|
1, 20.5, target=target
|
|
)
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == "68.9"
|
|
|
|
|
|
async def test_set_bypass_supply_temperature_target_stays_within_maximum(
|
|
hass: HomeAssistant,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test normalization never rounds past a maximum that is not a whole step."""
|
|
hass.config.units = US_CUSTOMARY_SYSTEM
|
|
mock_bypass_supply_temperature_targets[1] = replace(
|
|
mock_bypass_supply_temperature_targets[1],
|
|
minimum=10.0,
|
|
increment=0.5,
|
|
maximum=24.8,
|
|
)
|
|
target = mock_bypass_supply_temperature_targets[1]
|
|
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
await hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": 76.6},
|
|
blocking=True,
|
|
)
|
|
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.assert_awaited_once_with(
|
|
1, 24.5, target=target
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_duco_client")
|
|
async def test_bypass_supply_temperature_target_becomes_unavailable_when_missing(
|
|
hass: HomeAssistant,
|
|
freezer: FrozenDateTimeFactory,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test a bypass target becomes unavailable when a bulk read omits it."""
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == "20.0"
|
|
|
|
updated_target = replace(mock_bypass_supply_temperature_targets.pop(1), value=20.5)
|
|
await async_fire_coordinator_update(hass, freezer)
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
mock_bypass_supply_temperature_targets[1] = updated_target
|
|
await async_fire_coordinator_update(hass, freezer)
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == "20.5"
|
|
|
|
|
|
async def test_bypass_supply_temperature_target_recovers_from_refresh_error(
|
|
hass: HomeAssistant,
|
|
freezer: FrozenDateTimeFactory,
|
|
mock_bypass_supply_temperature_targets: dict[int, BypassSupplyTemperatureTarget],
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_duco_client: AsyncMock,
|
|
) -> None:
|
|
"""Test a bypass target recovers after a transient refresh error."""
|
|
await setup_platform_integration(hass, mock_config_entry, [Platform.NUMBER])
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == "20.0"
|
|
|
|
mock_duco_client.async_get_bypass_supply_temperature_targets.side_effect = [
|
|
DucoError("Temporary bypass target failure"),
|
|
mock_bypass_supply_temperature_targets.copy(),
|
|
]
|
|
await async_fire_coordinator_update(hass, freezer)
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
await async_fire_coordinator_update(hass, freezer)
|
|
|
|
state = hass.states.get(_ZONE_1_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state == "20.0"
|
|
|
|
|
|
@pytest.mark.usefixtures("init_integration")
|
|
@pytest.mark.parametrize(
|
|
("exception", "match"),
|
|
[
|
|
pytest.param(
|
|
DucoError("Unexpected error"),
|
|
"Failed to set bypass supply target temperature",
|
|
id="duco_error",
|
|
),
|
|
pytest.param(DucoRateLimitError(), "daily write limit", id="rate_limit"),
|
|
],
|
|
)
|
|
async def test_set_bypass_supply_temperature_target_error(
|
|
hass: HomeAssistant,
|
|
mock_duco_client: AsyncMock,
|
|
exception: Exception,
|
|
match: str,
|
|
) -> None:
|
|
"""Test write failures raise translated Home Assistant errors."""
|
|
mock_duco_client.async_set_bypass_supply_temperature_target.side_effect = exception
|
|
|
|
with pytest.raises(HomeAssistantError, match=match):
|
|
await hass.services.async_call(
|
|
NUMBER_DOMAIN,
|
|
SERVICE_SET_VALUE,
|
|
{ATTR_ENTITY_ID: _ZONE_1_ENTITY_ID, "value": 20.5},
|
|
blocking=True,
|
|
)
|