Refactor Whirlpool sensor tests (#142437)

This commit is contained in:
Abílio Costa 2025-04-10 15:47:28 +01:00 committed by GitHub
parent a5013cddd5
commit a26cdef427
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 747 additions and 385 deletions

View File

@ -55,15 +55,12 @@ WASHER_DRYER_MACHINE_STATE = {
MachineState.SystemInit: "system_initialize",
}
WASHER_DRYER_CYCLE_FUNC = [
(WasherDryer.get_cycle_status_filling, "cycle_filling"),
(WasherDryer.get_cycle_status_rinsing, "cycle_rinsing"),
(WasherDryer.get_cycle_status_sensing, "cycle_sensing"),
(WasherDryer.get_cycle_status_soaking, "cycle_soaking"),
(WasherDryer.get_cycle_status_spinning, "cycle_spinning"),
(WasherDryer.get_cycle_status_washing, "cycle_washing"),
]
STATE_CYCLE_FILLING = "cycle_filling"
STATE_CYCLE_RINSING = "cycle_rinsing"
STATE_CYCLE_SENSING = "cycle_sensing"
STATE_CYCLE_SOAKING = "cycle_soaking"
STATE_CYCLE_SPINNING = "cycle_spinning"
STATE_CYCLE_WASHING = "cycle_washing"
STATE_DOOR_OPEN = "door_open"
@ -76,9 +73,18 @@ def washer_dryer_state(washer_dryer: WasherDryer) -> str | None:
machine_state = washer_dryer.get_machine_state()
if machine_state == MachineState.RunningMainCycle:
for func, cycle_name in WASHER_DRYER_CYCLE_FUNC:
if func(washer_dryer):
return cycle_name
if washer_dryer.get_cycle_status_filling():
return STATE_CYCLE_FILLING
if washer_dryer.get_cycle_status_rinsing():
return STATE_CYCLE_RINSING
if washer_dryer.get_cycle_status_sensing():
return STATE_CYCLE_SENSING
if washer_dryer.get_cycle_status_soaking():
return STATE_CYCLE_SOAKING
if washer_dryer.get_cycle_status_spinning():
return STATE_CYCLE_SPINNING
if washer_dryer.get_cycle_status_washing():
return STATE_CYCLE_WASHING
return WASHER_DRYER_MACHINE_STATE.get(machine_state)
@ -90,11 +96,16 @@ class WhirlpoolSensorEntityDescription(SensorEntityDescription):
value_fn: Callable[[Appliance], str | None]
WASHER_DRYER_STATE_OPTIONS = (
list(WASHER_DRYER_MACHINE_STATE.values())
+ [value for _, value in WASHER_DRYER_CYCLE_FUNC]
+ [STATE_DOOR_OPEN]
)
WASHER_DRYER_STATE_OPTIONS = [
*WASHER_DRYER_MACHINE_STATE.values(),
STATE_CYCLE_FILLING,
STATE_CYCLE_RINSING,
STATE_CYCLE_SENSING,
STATE_CYCLE_SOAKING,
STATE_CYCLE_SPINNING,
STATE_CYCLE_WASHING,
STATE_DOOR_OPEN,
]
WASHER_SENSORS: tuple[WhirlpoolSensorEntityDescription, ...] = (
WhirlpoolSensorEntityDescription(
@ -221,9 +232,7 @@ class WasherDryerTimeSensor(WhirlpoolEntity, RestoreSensor):
if machine_state is MachineState.RunningMainCycle:
self._running = True
new_timestamp = now + timedelta(seconds=self._wd.get_time_remaining())
if self._value is None or (
isinstance(self._value, datetime)
and abs(new_timestamp - self._value) > timedelta(seconds=60)

View File

@ -44,6 +44,7 @@
"entity": {
"sensor": {
"washer_state": {
"name": "State",
"state": {
"standby": "[%key:common::state::standby%]",
"setting": "Setting",
@ -74,6 +75,7 @@
}
},
"dryer_state": {
"name": "[%key:component::whirlpool::entity::sensor::washer_state::name%]",
"state": {
"standby": "[%key:common::state::standby%]",
"setting": "[%key:component::whirlpool::entity::sensor::washer_state::state::setting%]",

View File

@ -1,8 +1,11 @@
"""Tests for the Whirlpool Sixth Sense integration."""
from syrupy import SnapshotAssertion
from homeassistant.components.whirlpool.const import CONF_BRAND, DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import EntityRegistry
from tests.common import MockConfigEntry
@ -32,3 +35,17 @@ async def init_integration_with_entry(
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
def snapshot_whirlpool_entities(
hass: HomeAssistant,
entity_registry: EntityRegistry,
snapshot: SnapshotAssertion,
platform: Platform,
) -> None:
"""Snapshot Whirlpool entities."""
entities = hass.states.async_all(platform)
for entity_state in entities:
entity_entry = entity_registry.async_get(entity_state.entity_id)
assert entity_entry == snapshot(name=f"{entity_entry.entity_id}-entry")
assert entity_state == snapshot(name=f"{entity_entry.entity_id}-state")

View File

@ -4,11 +4,10 @@ from unittest import mock
from unittest.mock import AsyncMock, MagicMock
import pytest
import whirlpool
import whirlpool.aircon
from whirlpool import aircon, washerdryer
from whirlpool.backendselector import Brand, Region
from .const import MOCK_SAID1, MOCK_SAID2, MOCK_SAID3, MOCK_SAID4
from .const import MOCK_SAID1, MOCK_SAID2
@pytest.fixture(
@ -49,7 +48,7 @@ def fixture_mock_auth_api():
@pytest.fixture(name="mock_appliances_manager_api", autouse=True)
def fixture_mock_appliances_manager_api(
mock_aircon1_api, mock_aircon2_api, mock_sensor1_api, mock_sensor2_api
mock_aircon1_api, mock_aircon2_api, mock_washer_api, mock_dryer_api
):
"""Set up AppliancesManager fixture."""
with (
@ -69,8 +68,8 @@ def fixture_mock_appliances_manager_api(
mock_aircon2_api,
]
mock_appliances_manager.return_value.washer_dryers = [
mock_sensor1_api,
mock_sensor2_api,
mock_washer_api,
mock_dryer_api,
]
yield mock_appliances_manager
@ -100,8 +99,8 @@ def get_aircon_mock(said):
mock_aircon.appliance_info.model_number = "12345"
mock_aircon.get_online.return_value = True
mock_aircon.get_power_on.return_value = True
mock_aircon.get_mode.return_value = whirlpool.aircon.Mode.Cool
mock_aircon.get_fanspeed.return_value = whirlpool.aircon.FanSpeed.Auto
mock_aircon.get_mode.return_value = aircon.Mode.Cool
mock_aircon.get_fanspeed.return_value = aircon.FanSpeed.Auto
mock_aircon.get_current_temp.return_value = 15
mock_aircon.get_temp.return_value = 20
mock_aircon.get_current_humidity.return_value = 80
@ -141,53 +140,64 @@ def fixture_mock_aircon_api_instances(mock_aircon1_api, mock_aircon2_api):
yield mock_aircon_api
def get_sensor_mock(said: str, data_model: str):
"""Get a mock of a sensor."""
mock_sensor = mock.Mock(said=said)
mock_sensor.name = f"WasherDryer {said}"
mock_sensor.register_attr_callback = MagicMock()
mock_sensor.appliance_info.data_model = data_model
mock_sensor.appliance_info.category = "washer_dryer"
mock_sensor.appliance_info.model_number = "12345"
mock_sensor.get_online.return_value = True
mock_sensor.get_machine_state.return_value = (
whirlpool.washerdryer.MachineState.Standby
@pytest.fixture
def mock_washer_api():
"""Get a mock of a washer."""
mock_washer = mock.Mock(said="said_washer")
mock_washer.name = "Washer"
mock_washer.fetch_data = AsyncMock()
mock_washer.register_attr_callback = MagicMock()
mock_washer.appliance_info.data_model = "washer"
mock_washer.appliance_info.category = "washer_dryer"
mock_washer.appliance_info.model_number = "12345"
mock_washer.get_online.return_value = True
mock_washer.get_machine_state.return_value = (
washerdryer.MachineState.RunningMainCycle
)
mock_sensor.get_door_open.return_value = False
mock_sensor.get_dispense_1_level.return_value = 3
mock_sensor.get_time_remaining.return_value = 3540
mock_sensor.get_cycle_status_filling.return_value = False
mock_sensor.get_cycle_status_rinsing.return_value = False
mock_sensor.get_cycle_status_sensing.return_value = False
mock_sensor.get_cycle_status_soaking.return_value = False
mock_sensor.get_cycle_status_spinning.return_value = False
mock_sensor.get_cycle_status_washing.return_value = False
mock_washer.get_door_open.return_value = False
mock_washer.get_dispense_1_level.return_value = 3
mock_washer.get_time_remaining.return_value = 3540
mock_washer.get_cycle_status_filling.return_value = False
mock_washer.get_cycle_status_rinsing.return_value = False
mock_washer.get_cycle_status_sensing.return_value = False
mock_washer.get_cycle_status_soaking.return_value = False
mock_washer.get_cycle_status_spinning.return_value = False
mock_washer.get_cycle_status_washing.return_value = False
return mock_sensor
return mock_washer
@pytest.fixture(name="mock_sensor1_api", autouse=False)
def fixture_mock_sensor1_api():
"""Set up sensor API fixture."""
return get_sensor_mock(MOCK_SAID3, "washer")
@pytest.fixture
def mock_dryer_api():
"""Get a mock of a dryer."""
mock_dryer = mock.Mock(said="said_dryer")
mock_dryer.name = "Dryer"
mock_dryer.fetch_data = AsyncMock()
mock_dryer.register_attr_callback = MagicMock()
mock_dryer.appliance_info.data_model = "dryer"
mock_dryer.appliance_info.category = "washer_dryer"
mock_dryer.appliance_info.model_number = "12345"
mock_dryer.get_online.return_value = True
mock_dryer.get_machine_state.return_value = (
washerdryer.MachineState.RunningMainCycle
)
mock_dryer.get_door_open.return_value = False
mock_dryer.get_time_remaining.return_value = 3540
mock_dryer.get_cycle_status_filling.return_value = False
mock_dryer.get_cycle_status_rinsing.return_value = False
mock_dryer.get_cycle_status_sensing.return_value = False
mock_dryer.get_cycle_status_soaking.return_value = False
mock_dryer.get_cycle_status_spinning.return_value = False
mock_dryer.get_cycle_status_washing.return_value = False
return mock_dryer
@pytest.fixture(name="mock_sensor2_api", autouse=False)
def fixture_mock_sensor2_api():
"""Set up sensor API fixture."""
return get_sensor_mock(MOCK_SAID4, "dryer")
@pytest.fixture(name="mock_sensor_api_instances", autouse=False)
def fixture_mock_sensor_api_instances(mock_sensor1_api, mock_sensor2_api):
"""Set up sensor API fixture."""
@pytest.fixture(autouse=True)
def mock_washer_dryer_api_instances(mock_washer_api, mock_dryer_api):
"""Set up WasherDryer API fixture."""
with mock.patch(
"homeassistant.components.whirlpool.sensor.WasherDryer"
) as mock_sensor_api:
mock_sensor_api.side_effect = [
mock_sensor1_api,
mock_sensor2_api,
mock_sensor1_api,
mock_sensor2_api,
]
yield mock_sensor_api
) as mock_washer_dryer_api:
mock_washer_dryer_api.side_effect = [mock_washer_api, mock_dryer_api]
yield mock_washer_dryer_api

View File

@ -2,5 +2,3 @@
MOCK_SAID1 = "said1"
MOCK_SAID2 = "said2"
MOCK_SAID3 = "said3"
MOCK_SAID4 = "said4"

View File

@ -17,16 +17,16 @@
'ovens': dict({
}),
'washer_dryers': dict({
'WasherDryer said3': dict({
'category': 'washer_dryer',
'data_model': 'washer',
'model_number': '12345',
}),
'WasherDryer said4': dict({
'Dryer': dict({
'category': 'washer_dryer',
'data_model': 'dryer',
'model_number': '12345',
}),
'Washer': dict({
'category': 'washer_dryer',
'data_model': 'washer',
'model_number': '12345',
}),
}),
}),
'config_entry': dict({

View File

@ -0,0 +1,374 @@
# serializer version: 1
# name: test_all_entities[sensor.dryer_end_time-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.dryer_end_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': 'mdi:progress-clock',
'original_name': 'End time',
'platform': 'whirlpool',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'end_time',
'unique_id': 'said_dryer-timeremaining',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.dryer_end_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Dryer End time',
'icon': 'mdi:progress-clock',
}),
'context': <ANY>,
'entity_id': 'sensor.dryer_end_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2025-05-04T12:59:00+00:00',
})
# ---
# name: test_all_entities[sensor.dryer_state-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'standby',
'setting',
'delay_countdown',
'delay_paused',
'smart_delay',
'smart_grid_pause',
'pause',
'running_maincycle',
'running_postcycle',
'exception',
'complete',
'power_failure',
'service_diagnostic_mode',
'factory_diagnostic_mode',
'life_test',
'customer_focus_mode',
'demo_mode',
'hard_stop_or_error',
'system_initialize',
'cycle_filling',
'cycle_rinsing',
'cycle_sensing',
'cycle_soaking',
'cycle_spinning',
'cycle_washing',
'door_open',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.dryer_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'State',
'platform': 'whirlpool',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'dryer_state',
'unique_id': 'said_dryer-state',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.dryer_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Dryer State',
'options': list([
'standby',
'setting',
'delay_countdown',
'delay_paused',
'smart_delay',
'smart_grid_pause',
'pause',
'running_maincycle',
'running_postcycle',
'exception',
'complete',
'power_failure',
'service_diagnostic_mode',
'factory_diagnostic_mode',
'life_test',
'customer_focus_mode',
'demo_mode',
'hard_stop_or_error',
'system_initialize',
'cycle_filling',
'cycle_rinsing',
'cycle_sensing',
'cycle_soaking',
'cycle_spinning',
'cycle_washing',
'door_open',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.dryer_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'running_maincycle',
})
# ---
# name: test_all_entities[sensor.washer_detergent_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'unknown',
'empty',
'25',
'50',
'100',
'active',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.washer_detergent_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Detergent level',
'platform': 'whirlpool',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'whirlpool_tank',
'unique_id': 'said_washer-DispenseLevel',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.washer_detergent_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Washer Detergent level',
'options': list([
'unknown',
'empty',
'25',
'50',
'100',
'active',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.washer_detergent_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '50',
})
# ---
# name: test_all_entities[sensor.washer_end_time-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.washer_end_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': 'mdi:progress-clock',
'original_name': 'End time',
'platform': 'whirlpool',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'end_time',
'unique_id': 'said_washer-timeremaining',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.washer_end_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Washer End time',
'icon': 'mdi:progress-clock',
}),
'context': <ANY>,
'entity_id': 'sensor.washer_end_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2025-05-04T12:59:00+00:00',
})
# ---
# name: test_all_entities[sensor.washer_state-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'standby',
'setting',
'delay_countdown',
'delay_paused',
'smart_delay',
'smart_grid_pause',
'pause',
'running_maincycle',
'running_postcycle',
'exception',
'complete',
'power_failure',
'service_diagnostic_mode',
'factory_diagnostic_mode',
'life_test',
'customer_focus_mode',
'demo_mode',
'hard_stop_or_error',
'system_initialize',
'cycle_filling',
'cycle_rinsing',
'cycle_sensing',
'cycle_soaking',
'cycle_spinning',
'cycle_washing',
'door_open',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.washer_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'State',
'platform': 'whirlpool',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'washer_state',
'unique_id': 'said_washer-state',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.washer_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Washer State',
'options': list([
'standby',
'setting',
'delay_countdown',
'delay_paused',
'smart_delay',
'smart_grid_pause',
'pause',
'running_maincycle',
'running_postcycle',
'exception',
'complete',
'power_failure',
'service_diagnostic_mode',
'factory_diagnostic_mode',
'life_test',
'customer_focus_mode',
'demo_mode',
'hard_stop_or_error',
'system_initialize',
'cycle_filling',
'cycle_rinsing',
'cycle_sensing',
'cycle_soaking',
'cycle_spinning',
'cycle_washing',
'door_open',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.washer_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'running_maincycle',
})
# ---

View File

@ -1,346 +1,298 @@
"""Test the Whirlpool Sensor domain."""
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock
import pytest
from syrupy import SnapshotAssertion
from whirlpool.washerdryer import MachineState
from homeassistant.components.whirlpool.sensor import SCAN_INTERVAL
from homeassistant.core import CoreState, HomeAssistant, State
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.util.dt import as_timestamp, utc_from_timestamp, utcnow
from . import init_integration
from .const import MOCK_SAID3, MOCK_SAID4
from . import init_integration, snapshot_whirlpool_entities
from tests.common import async_fire_time_changed, mock_restore_cache_with_extra_data
WASHER_ENTITY_ID_BASE = "sensor.washer"
DRYER_ENTITY_ID_BASE = "sensor.dryer"
async def update_sensor_state(
hass: HomeAssistant, entity_id: str, mock_sensor_api_instance: MagicMock
async def trigger_attr_callback(
hass: HomeAssistant, mock_api_instance: MagicMock
) -> State:
"""Simulate an update trigger from the API."""
for call in mock_sensor_api_instance.register_attr_callback.call_args_list:
for call in mock_api_instance.register_attr_callback.call_args_list:
update_ha_state_cb = call[0][0]
update_ha_state_cb()
await hass.async_block_till_done()
return hass.states.get(entity_id)
async def test_dryer_sensor_values(
hass: HomeAssistant, mock_sensor2_api: MagicMock, entity_registry: er.EntityRegistry
) -> None:
"""Test the sensor value callbacks."""
hass.set_state(CoreState.not_running)
thetimestamp: datetime = datetime(2022, 11, 29, 00, 00, 00, 00, UTC)
mock_restore_cache_with_extra_data(
hass,
(
(
State(f"sensor.washerdryer_{MOCK_SAID3}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
(
State(f"sensor.washerdryer_{MOCK_SAID4}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
),
)
await init_integration(hass)
entity_id = f"sensor.washerdryer_{MOCK_SAID4}_none"
mock_instance = mock_sensor2_api
entry = entity_registry.async_get(entity_id)
assert entry
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "standby"
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
state_id = f"sensor.washerdryer_{MOCK_SAID3}_end_time"
state = hass.states.get(state_id)
assert state.state == thetimestamp.isoformat()
mock_instance.get_machine_state.return_value = MachineState.RunningMainCycle
mock_instance.get_cycle_status_filling.return_value = False
mock_instance.attr_value_to_bool.side_effect = [
False,
False,
False,
False,
False,
False,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "running_maincycle"
mock_instance.get_machine_state.return_value = MachineState.Complete
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "complete"
async def test_washer_sensor_values(
hass: HomeAssistant, mock_sensor1_api: MagicMock, entity_registry: er.EntityRegistry
) -> None:
"""Test the sensor value callbacks."""
hass.set_state(CoreState.not_running)
thetimestamp: datetime = datetime(2022, 11, 29, 00, 00, 00, 00, UTC)
mock_restore_cache_with_extra_data(
hass,
(
(
State(f"sensor.washerdryer_{MOCK_SAID3}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
(
State(f"sensor.washerdryer_{MOCK_SAID4}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
),
)
await init_integration(hass)
async_fire_time_changed(
hass,
utcnow() + SCAN_INTERVAL,
)
await hass.async_block_till_done()
entity_id = f"sensor.washerdryer_{MOCK_SAID3}_none"
mock_instance = mock_sensor1_api
entry = entity_registry.async_get(entity_id)
assert entry
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "standby"
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
state_id = f"sensor.washerdryer_{MOCK_SAID3}_end_time"
state = hass.states.get(state_id)
assert state.state == thetimestamp.isoformat()
state_id = f"sensor.washerdryer_{MOCK_SAID3}_detergent_level"
entry = entity_registry.async_get(state_id)
assert entry
assert entry.disabled
assert entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION
update_entry = entity_registry.async_update_entity(
entry.entity_id, disabled_by=None
)
await hass.async_block_till_done()
assert update_entry != entry
assert update_entry.disabled is False
state = hass.states.get(state_id)
assert state is None
await hass.config_entries.async_reload(entry.config_entry_id)
state = hass.states.get(state_id)
assert state is not None
assert state.state == "50"
# Test the washer cycle states
mock_instance.get_machine_state.return_value = MachineState.RunningMainCycle
mock_instance.get_cycle_status_filling.return_value = True
mock_instance.attr_value_to_bool.side_effect = [
True,
False,
False,
False,
False,
False,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "cycle_filling"
mock_instance.get_cycle_status_filling.return_value = False
mock_instance.get_cycle_status_rinsing.return_value = True
mock_instance.attr_value_to_bool.side_effect = [
False,
True,
False,
False,
False,
False,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "cycle_rinsing"
mock_instance.get_cycle_status_rinsing.return_value = False
mock_instance.get_cycle_status_sensing.return_value = True
mock_instance.attr_value_to_bool.side_effect = [
False,
False,
True,
False,
False,
False,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "cycle_sensing"
mock_instance.get_cycle_status_sensing.return_value = False
mock_instance.get_cycle_status_soaking.return_value = True
mock_instance.attr_value_to_bool.side_effect = [
False,
False,
False,
True,
False,
False,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "cycle_soaking"
mock_instance.get_cycle_status_soaking.return_value = False
mock_instance.get_cycle_status_spinning.return_value = True
mock_instance.attr_value_to_bool.side_effect = [
False,
False,
False,
False,
True,
False,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "cycle_spinning"
mock_instance.get_cycle_status_spinning.return_value = False
mock_instance.get_cycle_status_washing.return_value = True
mock_instance.attr_value_to_bool.side_effect = [
False,
False,
False,
False,
False,
True,
]
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "cycle_washing"
mock_instance.get_machine_state.return_value = MachineState.Complete
mock_instance.attr_value_to_bool.side_effect = None
mock_instance.get_door_open.return_value = True
state = await update_sensor_state(hass, entity_id, mock_instance)
assert state is not None
assert state.state == "door_open"
async def test_restore_state(hass: HomeAssistant) -> None:
"""Test sensor restore state."""
# Home assistant is not running yet
hass.set_state(CoreState.not_running)
thetimestamp: datetime = datetime(2022, 11, 29, 00, 00, 00, 00, UTC)
mock_restore_cache_with_extra_data(
hass,
(
(
State(f"sensor.washerdryer_{MOCK_SAID3}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
(
State(f"sensor.washerdryer_{MOCK_SAID4}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
),
)
# create and add entry
await init_integration(hass)
# restore from cache
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID3}_end_time")
assert state.state == thetimestamp.isoformat()
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID4}_end_time")
assert state.state == thetimestamp.isoformat()
async def test_no_restore_state(
hass: HomeAssistant, mock_sensor1_api: MagicMock
# Freeze time for WasherDryerTimeSensor
@pytest.mark.freeze_time("2025-05-04 12:00:00")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test sensor restore state with no restore."""
# create and add entry
entity_id = f"sensor.washerdryer_{MOCK_SAID3}_end_time"
"""Test all entities."""
await init_integration(hass)
# restore from cache
state = hass.states.get(entity_id)
assert state.state == "unknown"
mock_sensor1_api.get_machine_state.return_value = MachineState.RunningMainCycle
state = await update_sensor_state(hass, entity_id, mock_sensor1_api)
assert state.state != "unknown"
snapshot_whirlpool_entities(hass, entity_registry, snapshot, Platform.SENSOR)
@pytest.mark.parametrize(
("entity_id", "mock_fixture"),
[
("sensor.washer_end_time", "mock_washer_api"),
("sensor.dryer_end_time", "mock_dryer_api"),
],
)
@pytest.mark.freeze_time("2022-11-30 00:00:00")
async def test_callback(hass: HomeAssistant, mock_sensor1_api: MagicMock) -> None:
"""Test callback timestamp callback function."""
hass.set_state(CoreState.not_running)
thetimestamp: datetime = datetime(2022, 11, 29, 00, 00, 00, 00, UTC)
async def test_washer_dryer_time_sensor(
hass: HomeAssistant,
entity_id: str,
mock_fixture: str,
request: pytest.FixtureRequest,
) -> None:
"""Test Washer/Dryer end time sensors."""
now = utcnow()
restored_datetime: datetime = datetime(2022, 11, 29, 00, 00, 00, 00, UTC)
mock_restore_cache_with_extra_data(
hass,
(
[
(
State(f"sensor.washerdryer_{MOCK_SAID3}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
(
State(f"sensor.washerdryer_{MOCK_SAID4}_end_time", "1"),
{"native_value": thetimestamp, "native_unit_of_measurement": None},
),
),
State(entity_id, "1"),
{"native_value": restored_datetime, "native_unit_of_measurement": None},
)
],
)
# create and add entry
mock_instance = request.getfixturevalue(mock_fixture)
mock_instance.get_machine_state.return_value = MachineState.Pause
await init_integration(hass)
# restore from cache
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID3}_end_time")
assert state.state == thetimestamp.isoformat()
callback = mock_sensor1_api.register_attr_callback.call_args_list[1][0][0]
callback()
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID3}_end_time")
assert state.state == thetimestamp.isoformat()
mock_sensor1_api.get_machine_state.return_value = MachineState.RunningMainCycle
mock_sensor1_api.get_time_remaining.return_value = 60
callback()
# Test restored state.
state = hass.states.get(entity_id)
assert state.state == restored_datetime.isoformat()
# Test new timestamp when machine starts a cycle.
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID3}_end_time")
time = state.state
assert state.state != thetimestamp.isoformat()
# Test no time change because the machine is not running.
await trigger_attr_callback(hass, mock_instance)
# Test no timestamp change for < 60 seconds time change.
mock_sensor1_api.get_time_remaining.return_value = 65
callback()
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID3}_end_time")
assert state.state == time
state = hass.states.get(entity_id)
assert state.state == restored_datetime.isoformat()
# Test new time when machine starts a cycle.
mock_instance.get_machine_state.return_value = MachineState.RunningMainCycle
mock_instance.get_time_remaining.return_value = 60
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
expected_time = (now + timedelta(seconds=60)).isoformat()
assert state.state == expected_time
# Test no state change for < 60 seconds elapsed time.
mock_instance.get_time_remaining.return_value = 65
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert state.state == expected_time
# Test timestamp change for > 60 seconds.
mock_sensor1_api.get_time_remaining.return_value = 125
callback()
state = hass.states.get(f"sensor.washerdryer_{MOCK_SAID3}_end_time")
newtime = utc_from_timestamp(as_timestamp(time) + 65)
assert state.state == newtime.isoformat()
mock_instance.get_time_remaining.return_value = 125
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert (
state.state == utc_from_timestamp(as_timestamp(expected_time) + 65).isoformat()
)
# Test that periodic updates call the API to fetch data
mock_instance.fetch_data.reset_mock()
async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
mock_instance.fetch_data.assert_called_once()
@pytest.mark.parametrize(
("entity_id", "mock_fixture"),
[
("sensor.washer_end_time", "mock_washer_api"),
("sensor.dryer_end_time", "mock_dryer_api"),
],
)
@pytest.mark.freeze_time("2022-11-30 00:00:00")
async def test_washer_dryer_time_sensor_no_restore(
hass: HomeAssistant,
entity_id: str,
mock_fixture: str,
request: pytest.FixtureRequest,
) -> None:
"""Test Washer/Dryer end time sensors without state restore."""
now = utcnow()
mock_instance = request.getfixturevalue(mock_fixture)
mock_instance.get_machine_state.return_value = MachineState.Pause
await init_integration(hass)
state = hass.states.get(entity_id)
assert state.state == STATE_UNKNOWN
# Test no change because the machine is paused.
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert state.state == STATE_UNKNOWN
# Test new time when machine starts a cycle.
mock_instance.get_machine_state.return_value = MachineState.RunningMainCycle
mock_instance.get_time_remaining.return_value = 60
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
expected_time = (now + timedelta(seconds=60)).isoformat()
assert state.state == expected_time
@pytest.mark.parametrize(
("entity_id", "mock_fixture"),
[
("sensor.washer_state", "mock_washer_api"),
("sensor.dryer_state", "mock_dryer_api"),
],
)
@pytest.mark.parametrize(
("machine_state", "expected_state"),
[
(MachineState.Standby, "standby"),
(MachineState.Setting, "setting"),
(MachineState.DelayCountdownMode, "delay_countdown"),
(MachineState.DelayPause, "delay_paused"),
(MachineState.SmartDelay, "smart_delay"),
(MachineState.SmartGridPause, "smart_grid_pause"),
(MachineState.Pause, "pause"),
(MachineState.RunningMainCycle, "running_maincycle"),
(MachineState.RunningPostCycle, "running_postcycle"),
(MachineState.Exceptions, "exception"),
(MachineState.Complete, "complete"),
(MachineState.PowerFailure, "power_failure"),
(MachineState.ServiceDiagnostic, "service_diagnostic_mode"),
(MachineState.FactoryDiagnostic, "factory_diagnostic_mode"),
(MachineState.LifeTest, "life_test"),
(MachineState.CustomerFocusMode, "customer_focus_mode"),
(MachineState.DemoMode, "demo_mode"),
(MachineState.HardStopOrError, "hard_stop_or_error"),
(MachineState.SystemInit, "system_initialize"),
],
)
async def test_washer_dryer_machine_states(
hass: HomeAssistant,
entity_id: str,
mock_fixture: str,
machine_state: MachineState,
expected_state: str,
request: pytest.FixtureRequest,
) -> None:
"""Test Washer/Dryer machine states."""
mock_instance = request.getfixturevalue(mock_fixture)
await init_integration(hass)
mock_instance.get_machine_state.return_value = machine_state
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert state is not None
assert state.state == expected_state
@pytest.mark.parametrize(
("entity_id", "mock_fixture"),
[
("sensor.washer_state", "mock_washer_api"),
("sensor.dryer_state", "mock_dryer_api"),
],
)
@pytest.mark.parametrize(
(
"filling",
"rinsing",
"sensing",
"soaking",
"spinning",
"washing",
"expected_state",
),
[
(True, False, False, False, False, False, "cycle_filling"),
(False, True, False, False, False, False, "cycle_rinsing"),
(False, False, True, False, False, False, "cycle_sensing"),
(False, False, False, True, False, False, "cycle_soaking"),
(False, False, False, False, True, False, "cycle_spinning"),
(False, False, False, False, False, True, "cycle_washing"),
],
)
async def test_washer_dryer_running_states(
hass: HomeAssistant,
entity_id: str,
mock_fixture: str,
filling: bool,
rinsing: bool,
sensing: bool,
soaking: bool,
spinning: bool,
washing: bool,
expected_state: str,
request: pytest.FixtureRequest,
) -> None:
"""Test Washer/Dryer machine states for RunningMainCycle."""
mock_instance = request.getfixturevalue(mock_fixture)
await init_integration(hass)
mock_instance.get_machine_state.return_value = MachineState.RunningMainCycle
mock_instance.get_cycle_status_filling.return_value = filling
mock_instance.get_cycle_status_rinsing.return_value = rinsing
mock_instance.get_cycle_status_sensing.return_value = sensing
mock_instance.get_cycle_status_soaking.return_value = soaking
mock_instance.get_cycle_status_spinning.return_value = spinning
mock_instance.get_cycle_status_washing.return_value = washing
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert state is not None
assert state.state == expected_state
@pytest.mark.parametrize(
("entity_id", "mock_fixture"),
[
("sensor.washer_state", "mock_washer_api"),
("sensor.dryer_state", "mock_dryer_api"),
],
)
async def test_washer_dryer_door_open_state(
hass: HomeAssistant,
entity_id: str,
mock_fixture: str,
request: pytest.FixtureRequest,
) -> None:
"""Test Washer/Dryer machine state when door is open."""
mock_instance = request.getfixturevalue(mock_fixture)
await init_integration(hass)
state = hass.states.get(entity_id)
assert state.state == "running_maincycle"
mock_instance.get_door_open.return_value = True
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert state.state == "door_open"
mock_instance.get_door_open.return_value = False
await trigger_attr_callback(hass, mock_instance)
state = hass.states.get(entity_id)
assert state.state == "running_maincycle"