|
|
@@ -0,0 +1,281 @@
|
|
|
+"""Tests for ZLockCoordinator: slot fetch, alarm event handling, name/event persistence."""
|
|
|
+import pytest
|
|
|
+from unittest.mock import patch, AsyncMock
|
|
|
+
|
|
|
+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,
|
|
|
+)
|
|
|
+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 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 == "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 == "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 == "locked"
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_user == "<Manual>"
|
|
|
+
|
|
|
+
|
|
|
+async def test_manual_unlock(hass, coord):
|
|
|
+ await _fire_alarm(hass, "22")
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_action == "unlocked"
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_user == "<Manual>"
|
|
|
+
|
|
|
+
|
|
|
+async def test_radio_lock(hass, coord):
|
|
|
+ await _fire_alarm(hass, "24")
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_action == "locked"
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_user == "<Radio>"
|
|
|
+
|
|
|
+
|
|
|
+async def test_radio_unlock(hass, coord):
|
|
|
+ await _fire_alarm(hass, "25")
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_action == "unlocked"
|
|
|
+ assert coord.data[LOCK_ENTITY_ID].last_user == "<Radio>"
|
|
|
+
|
|
|
+
|
|
|
+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
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# 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()
|
|
|
+
|
|
|
+ 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()
|
|
|
+ c1.async_teardown()
|
|
|
+
|
|
|
+ c2 = ZLockCoordinator(hass, config_entry)
|
|
|
+ await c2.async_setup()
|
|
|
+
|
|
|
+ assert c2.data[LOCK_ENTITY_ID].last_action == "unlocked"
|
|
|
+ assert c2.data[LOCK_ENTITY_ID].last_user == "<Manual>"
|
|
|
+ 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")
|
|
|
+
|
|
|
+ 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()
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# 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 == "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")
|
|
|
+ c1.async_teardown()
|
|
|
+
|
|
|
+ c2 = ZLockCoordinator(hass, group_config_entry)
|
|
|
+ await c2.async_setup()
|
|
|
+
|
|
|
+ assert c2.group_data["All Locks"].last_action == "locked"
|
|
|
+ assert c2.group_data["All Locks"].last_user == "<Radio>"
|
|
|
+ c2.async_teardown()
|