"""Tests for ZLockCoordinator: slot fetch, alarm event handling, name/event persistence.""" import pytest from unittest.mock import patch from pytest_homeassistant_custom_component.common import flush_store from custom_components.zlock_slots.coordinator import ZLockCoordinator from custom_components.zlock_slots.const import ( DOMAIN, CONF_LOCK_ENTITY_IDS, CONF_GROUPS, CONF_GROUP_NAME, LockAction, ) from tests.conftest import ( LOCK_ENTITY_ID, ALARM_TYPE_ENTITY, ALARM_LEVEL_ENTITY, FakeUserData, FakeNode, ) _ZWAVE_HELPER = "homeassistant.components.zwave_js.helpers.async_get_node_from_entity_id" @pytest.fixture async def coord(hass, config_entry, mock_node): config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c = ZLockCoordinator(hass, config_entry) await c.async_setup() await hass.async_block_till_done() yield c c.async_teardown() # --------------------------------------------------------------------------- # Slot fetch # --------------------------------------------------------------------------- async def test_slot_fetch_populates_data(coord): slots = coord.data[LOCK_ENTITY_ID].slots assert set(slots) == {1, 2} async def test_slot_enabled_status(coord): assert coord.data[LOCK_ENTITY_ID].slots[1].status == "enabled" assert coord.data[LOCK_ENTITY_ID].slots[2].status == "disabled" async def test_slot_0_excluded(hass, config_entry): node = FakeNode(users=[FakeUserData(0), FakeUserData(1)]) config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=node): c = ZLockCoordinator(hass, config_entry) await c.async_setup() assert 0 not in c.data[LOCK_ENTITY_ID].slots assert 1 in c.data[LOCK_ENTITY_ID].slots c.async_teardown() async def test_slot_name_applied_during_fetch(hass, config_entry, mock_node): config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c = ZLockCoordinator(hass, config_entry) # Pre-seed a name before setup fetches slots. c._names[f"{LOCK_ENTITY_ID}:1"] = "Alice" await c.async_setup() assert c.data[LOCK_ENTITY_ID].slots[1].name == "Alice" assert c.data[LOCK_ENTITY_ID].slots[2].name is None c.async_teardown() async def test_fetch_failure_returns_empty(hass, config_entry): config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, side_effect=Exception("zwave unavailable")): c = ZLockCoordinator(hass, config_entry) await c.async_setup() assert c.data[LOCK_ENTITY_ID].slots == {} c.async_teardown() # --------------------------------------------------------------------------- # Name persistence # --------------------------------------------------------------------------- async def test_set_name_stores_and_retrieves(coord): await coord.async_set_name(LOCK_ENTITY_ID, 1, "Alice") assert coord.get_name(LOCK_ENTITY_ID, 1) == "Alice" async def test_set_name_empty_clears_entry(coord): await coord.async_set_name(LOCK_ENTITY_ID, 1, "Alice") await coord.async_set_name(LOCK_ENTITY_ID, 1, "") assert coord.get_name(LOCK_ENTITY_ID, 1) is None async def test_set_name_updates_cached_slot(coord): await coord.async_set_name(LOCK_ENTITY_ID, 1, "Bob") assert coord.data[LOCK_ENTITY_ID].slots[1].name == "Bob" async def test_set_name_persists_to_store(hass, config_entry, mock_node): """Names survive a coordinator restart via the Store.""" config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c1 = ZLockCoordinator(hass, config_entry) await c1.async_setup() await c1.async_set_name(LOCK_ENTITY_ID, 1, "Persisted") c1.async_teardown() c2 = ZLockCoordinator(hass, config_entry) await c2.async_setup() assert c2.get_name(LOCK_ENTITY_ID, 1) == "Persisted" c2.async_teardown() # --------------------------------------------------------------------------- # Alarm event handling # --------------------------------------------------------------------------- async def _fire_alarm(hass, alarm_type: str, alarm_level: str = "0"): hass.states.async_set(ALARM_LEVEL_ENTITY, alarm_level) hass.states.async_set(ALARM_TYPE_ENTITY, alarm_type) await hass.async_block_till_done() async def _flush_delayed_save(coord): """Alarm-driven Store writes are debounced; force pending writes to flush.""" await flush_store(coord._store) async def test_keypad_lock_sets_action_and_user(hass, coord): hass.states.async_set(ALARM_LEVEL_ENTITY, "1") hass.states.async_set(ALARM_TYPE_ENTITY, "18") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.LOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "1" async def test_keypad_lock_uses_slot_name(hass, coord): await coord.async_set_name(LOCK_ENTITY_ID, 1, "Alice") hass.states.async_set(ALARM_LEVEL_ENTITY, "1") hass.states.async_set(ALARM_TYPE_ENTITY, "18") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_user == "Alice" async def test_keypad_unlock(hass, coord): await _fire_alarm(hass, "19", alarm_level="2") assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "2" async def test_manual_lock(hass, coord): await _fire_alarm(hass, "21") assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.LOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "" async def test_manual_unlock(hass, coord): await _fire_alarm(hass, "22") assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "" async def test_radio_lock(hass, coord): await _fire_alarm(hass, "24") assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.LOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "" async def test_radio_unlock(hass, coord): await _fire_alarm(hass, "25") assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "" async def test_unknown_alarm_type_ignored(hass, coord): await _fire_alarm(hass, "99") assert coord.data[LOCK_ENTITY_ID].last_action is None assert coord.data[LOCK_ENTITY_ID].last_user is None async def test_non_integer_alarm_state_ignored(hass, coord): hass.states.async_set(ALARM_TYPE_ENTITY, "unavailable") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_action is None # --------------------------------------------------------------------------- # alarm_type / alarm_level arrival ordering # --------------------------------------------------------------------------- async def test_alarm_type_before_level_resolves_when_level_arrives(hass, coord): """Reproduces the reported bug: type fires first with stale level in state machine.""" # Fire alarm_type with NO prior alarm_level update — nothing should emit yet. hass.states.async_set(ALARM_TYPE_ENTITY, "19") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_action is None # Now alarm_level arrives — should resolve the pending action. hass.states.async_set(ALARM_LEVEL_ENTITY, "87") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "87" async def test_alarm_type_before_level_uses_name(hass, coord): await coord.async_set_name(LOCK_ENTITY_ID, 87, "Grace") hass.states.async_set(ALARM_TYPE_ENTITY, "18") await hass.async_block_till_done() hass.states.async_set(ALARM_LEVEL_ENTITY, "87") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_user == "Grace" async def test_stale_alarm_level_not_used(hass, coord): """An alarm_level older than the window must not be consumed by a later alarm_type.""" with patch("custom_components.zlock_slots.coordinator.time") as mock_time: mock_time.monotonic.return_value = 0.0 hass.states.async_set(ALARM_LEVEL_ENTITY, "3") # level=3 at t=0 await hass.async_block_till_done() mock_time.monotonic.return_value = 10.0 # 10 s later — outside window hass.states.async_set(ALARM_TYPE_ENTITY, "19") await hass.async_block_till_done() # Stale level rejected; event goes pending — nothing emitted yet. assert coord.data[LOCK_ENTITY_ID].last_action is None # Fresh level arrives and resolves the pending action. hass.states.async_set(ALARM_LEVEL_ENTITY, "87") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED assert coord.data[LOCK_ENTITY_ID].last_user == "87" async def test_new_alarm_type_clears_pending_slot_alarm(hass, coord): """A second alarm_type before level arrives should replace the first pending.""" hass.states.async_set(ALARM_TYPE_ENTITY, "18") # lock pending await hass.async_block_till_done() hass.states.async_set(ALARM_TYPE_ENTITY, "19") # unlock replaces it await hass.async_block_till_done() hass.states.async_set(ALARM_LEVEL_ENTITY, "5") await hass.async_block_till_done() assert coord.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED # --------------------------------------------------------------------------- # last_events persistence and restore # --------------------------------------------------------------------------- async def test_last_events_persisted_after_alarm(hass, config_entry, mock_node): config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c = ZLockCoordinator(hass, config_entry) await c.async_setup() hass.states.async_set(ALARM_LEVEL_ENTITY, "1") hass.states.async_set(ALARM_TYPE_ENTITY, "18") await hass.async_block_till_done() await _flush_delayed_save(c) stored = await c._store.async_load() assert stored["last_events"]["locks"][LOCK_ENTITY_ID]["last_action"] == "Locked" assert stored["last_events"]["locks"][LOCK_ENTITY_ID]["last_user"] == "1" c.async_teardown() async def test_last_events_restored_on_startup(hass, config_entry, mock_node): """last_action/last_user are non-None immediately after a cold start.""" config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c1 = ZLockCoordinator(hass, config_entry) await c1.async_setup() hass.states.async_set(ALARM_LEVEL_ENTITY, "2") hass.states.async_set(ALARM_TYPE_ENTITY, "22") await hass.async_block_till_done() await _flush_delayed_save(c1) c1.async_teardown() c2 = ZLockCoordinator(hass, config_entry) await c2.async_setup() assert c2.data[LOCK_ENTITY_ID].last_action == LockAction.UNLOCKED assert c2.data[LOCK_ENTITY_ID].last_user == "" c2.async_teardown() async def test_names_still_present_after_alarm_save(hass, config_entry, mock_node): """_async_save merges names + last_events; names must not be clobbered.""" config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c = ZLockCoordinator(hass, config_entry) await c.async_setup() await c.async_set_name(LOCK_ENTITY_ID, 1, "Alice") await _fire_alarm(hass, "18", "1") await _flush_delayed_save(c) stored = await c._store.async_load() assert stored["names"][f"{LOCK_ENTITY_ID}:1"] == "Alice" assert stored["last_events"]["locks"][LOCK_ENTITY_ID]["last_action"] == "Locked" c.async_teardown() async def test_alarm_saves_are_debounced(hass, coord): """Back-to-back alarms coalesce into a single delayed Store write.""" calls = [] real_data_to_save = coord._data_to_save def counting_data_to_save(): calls.append(1) return real_data_to_save() coord._data_to_save = counting_data_to_save await _fire_alarm(hass, "21") await _fire_alarm(hass, "22") assert not calls # nothing written yet await _flush_delayed_save(coord) assert len(calls) == 1 # single coalesced write stored = await coord._store.async_load() assert stored["last_events"]["locks"][LOCK_ENTITY_ID]["last_action"] == "Unlocked" # --------------------------------------------------------------------------- # Group state # --------------------------------------------------------------------------- @pytest.fixture def group_config_entry(): from pytest_homeassistant_custom_component.common import MockConfigEntry return MockConfigEntry( domain=DOMAIN, data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}, options={ CONF_GROUPS: [{ CONF_GROUP_NAME: "All Locks", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID], }] }, entry_id="test_entry_group", ) async def test_group_state_updated_on_alarm(hass, group_config_entry, mock_node): group_config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c = ZLockCoordinator(hass, group_config_entry) await c.async_setup() await _fire_alarm(hass, "19", "1") assert c.group_data["All Locks"].last_action == LockAction.UNLOCKED assert c.group_data["All Locks"].last_user == "1" c.async_teardown() async def test_group_last_events_restored(hass, group_config_entry, mock_node): group_config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): c1 = ZLockCoordinator(hass, group_config_entry) await c1.async_setup() await _fire_alarm(hass, "24") await _flush_delayed_save(c1) c1.async_teardown() c2 = ZLockCoordinator(hass, group_config_entry) await c2.async_setup() assert c2.group_data["All Locks"].last_action == LockAction.LOCKED assert c2.group_data["All Locks"].last_user == "" c2.async_teardown()