|
|
@@ -0,0 +1,300 @@
|
|
|
+"""Tests for door sensor integration: state tracking, combined state, group aggregation."""
|
|
|
+import pytest
|
|
|
+from unittest.mock import patch
|
|
|
+
|
|
|
+from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
+
|
|
|
+from custom_components.zlock_slots.coordinator import ZLockCoordinator, compute_combined_state
|
|
|
+from custom_components.zlock_slots.sensor import (
|
|
|
+ ZLockDoorSensor,
|
|
|
+ ZLockCombinedStateSensor,
|
|
|
+ ZLockGroupCombinedStateSensor,
|
|
|
+)
|
|
|
+from custom_components.zlock_slots.const import (
|
|
|
+ DOMAIN, CONF_LOCK_ENTITY_IDS, CONF_GROUPS, CONF_GROUP_NAME,
|
|
|
+ CONF_DOOR_SENSOR_MAP, LockAction, DoorLockState,
|
|
|
+)
|
|
|
+from tests.conftest import (
|
|
|
+ LOCK_ENTITY_ID, ALARM_TYPE_ENTITY, ALARM_LEVEL_ENTITY, DOOR_SENSOR_ENTITY,
|
|
|
+ FakeNode,
|
|
|
+)
|
|
|
+
|
|
|
+LOCK_2 = "lock.back_door"
|
|
|
+DOOR_SENSOR_2 = "binary_sensor.back_door_contact"
|
|
|
+
|
|
|
+_ZWAVE_HELPER = "homeassistant.components.zwave_js.helpers.async_get_node_from_entity_id"
|
|
|
+
|
|
|
+
|
|
|
+def _door_config_entry(extra_lock_ids=None, door_sensor_map=None):
|
|
|
+ lock_ids = [LOCK_ENTITY_ID] + (extra_lock_ids or [])
|
|
|
+ return MockConfigEntry(
|
|
|
+ domain=DOMAIN,
|
|
|
+ data={CONF_LOCK_ENTITY_IDS: lock_ids},
|
|
|
+ options={CONF_DOOR_SENSOR_MAP: door_sensor_map or {LOCK_ENTITY_ID: DOOR_SENSOR_ENTITY}},
|
|
|
+ entry_id="test_door_entry",
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+async def door_coord(hass, mock_node):
|
|
|
+ entry = _door_config_entry()
|
|
|
+ entry.add_to_hass(hass)
|
|
|
+ with patch(_ZWAVE_HELPER, return_value=mock_node):
|
|
|
+ c = ZLockCoordinator(hass, entry)
|
|
|
+ await c.async_setup()
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ yield c
|
|
|
+ c.async_teardown()
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# compute_combined_state
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+@pytest.mark.parametrize("action,door_open,expected", [
|
|
|
+ (LockAction.LOCKED, False, DoorLockState.LOCKED),
|
|
|
+ (LockAction.UNLOCKED, True, DoorLockState.OPEN),
|
|
|
+ (LockAction.LOCKED, True, DoorLockState.JAMMED),
|
|
|
+ (LockAction.UNLOCKED, False, DoorLockState.CLOSED),
|
|
|
+ (None, False, None),
|
|
|
+ (LockAction.LOCKED, None, None),
|
|
|
+ (None, None, None),
|
|
|
+])
|
|
|
+def test_compute_combined_state(action, door_open, expected):
|
|
|
+ assert compute_combined_state(action, door_open) == expected
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Coordinator door sensor event handling
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+async def test_door_sensor_off_sets_closed(hass, door_coord):
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "off")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ assert door_coord.data[LOCK_ENTITY_ID].door_open is False
|
|
|
+
|
|
|
+
|
|
|
+async def test_door_sensor_on_sets_open(hass, door_coord):
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "on")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ assert door_coord.data[LOCK_ENTITY_ID].door_open is True
|
|
|
+
|
|
|
+
|
|
|
+async def test_door_sensor_unavailable_ignored(hass, door_coord):
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "on")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "unavailable")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ assert door_coord.data[LOCK_ENTITY_ID].door_open is True # unchanged
|
|
|
+
|
|
|
+
|
|
|
+async def test_door_open_initially_none(hass, door_coord):
|
|
|
+ assert door_coord.data[LOCK_ENTITY_ID].door_open is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_door_sensor_not_tracked_without_config(hass, mock_node):
|
|
|
+ entry = MockConfigEntry(
|
|
|
+ domain=DOMAIN,
|
|
|
+ data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]},
|
|
|
+ options={},
|
|
|
+ entry_id="test_no_door",
|
|
|
+ )
|
|
|
+ entry.add_to_hass(hass)
|
|
|
+ with patch(_ZWAVE_HELPER, return_value=mock_node):
|
|
|
+ c = ZLockCoordinator(hass, entry)
|
|
|
+ await c.async_setup()
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "on")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ assert c.data[LOCK_ENTITY_ID].door_open is None
|
|
|
+ c.async_teardown()
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# ZLockDoorSensor
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+async def test_door_sensor_entity_none_initially(door_coord, hass):
|
|
|
+ entry = door_coord._entry
|
|
|
+ sensor = ZLockDoorSensor(door_coord, entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_door_sensor_entity_closed(hass, door_coord):
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "off")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockDoorSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value == "Closed"
|
|
|
+
|
|
|
+
|
|
|
+async def test_door_sensor_entity_open(hass, door_coord):
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "on")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockDoorSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value == "Open"
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# ZLockCombinedStateSensor
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+async def test_combined_none_when_no_data(door_coord):
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_combined_none_when_only_door_known(hass, door_coord):
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "off")
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_combined_none_when_only_action_known(hass, door_coord):
|
|
|
+ hass.states.async_set(ALARM_TYPE_ENTITY, "24") # radio lock
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_combined_locked(hass, door_coord):
|
|
|
+ hass.states.async_set(ALARM_TYPE_ENTITY, "24") # locked
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "off") # closed
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value == "Locked"
|
|
|
+
|
|
|
+
|
|
|
+async def test_combined_open(hass, door_coord):
|
|
|
+ hass.states.async_set(ALARM_TYPE_ENTITY, "25") # unlocked
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "on") # open
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value == "Open"
|
|
|
+
|
|
|
+
|
|
|
+async def test_combined_jammed(hass, door_coord):
|
|
|
+ hass.states.async_set(ALARM_TYPE_ENTITY, "24") # locked
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "on") # open
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value == "Jammed"
|
|
|
+
|
|
|
+
|
|
|
+async def test_combined_closed(hass, door_coord):
|
|
|
+ hass.states.async_set(ALARM_TYPE_ENTITY, "25") # unlocked
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "off") # closed
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ sensor = ZLockCombinedStateSensor(door_coord, door_coord._entry, LOCK_ENTITY_ID)
|
|
|
+ assert sensor.native_value == "Closed"
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# ZLockGroupCombinedStateSensor — aggregation rules
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+async def two_lock_coord(hass, mock_node):
|
|
|
+ """Coordinator with two locks, both with door sensors."""
|
|
|
+ entry = MockConfigEntry(
|
|
|
+ domain=DOMAIN,
|
|
|
+ data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]},
|
|
|
+ options={
|
|
|
+ CONF_DOOR_SENSOR_MAP: {
|
|
|
+ LOCK_ENTITY_ID: DOOR_SENSOR_ENTITY,
|
|
|
+ LOCK_2: DOOR_SENSOR_2,
|
|
|
+ },
|
|
|
+ CONF_GROUPS: [{
|
|
|
+ CONF_GROUP_NAME: "All Locks",
|
|
|
+ CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2],
|
|
|
+ }],
|
|
|
+ },
|
|
|
+ entry_id="test_two_lock",
|
|
|
+ )
|
|
|
+ entry.add_to_hass(hass)
|
|
|
+ with patch(_ZWAVE_HELPER, return_value=mock_node):
|
|
|
+ c = ZLockCoordinator(hass, entry)
|
|
|
+ await c.async_setup()
|
|
|
+ await hass.async_block_till_done()
|
|
|
+ yield c
|
|
|
+ c.async_teardown()
|
|
|
+
|
|
|
+
|
|
|
+def _group_sensor(coord) -> ZLockGroupCombinedStateSensor:
|
|
|
+ return ZLockGroupCombinedStateSensor(
|
|
|
+ coord, coord._entry, "All Locks", [LOCK_ENTITY_ID, LOCK_2]
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def _set_lock_state(hass, alarm_type_entity, door_entity, alarm_type, door_state):
|
|
|
+ hass.states.async_set(alarm_type_entity, alarm_type)
|
|
|
+ hass.states.async_set(door_entity, door_state)
|
|
|
+ await hass.async_block_till_done()
|
|
|
+
|
|
|
+
|
|
|
+ALARM_TYPE_2 = "sensor.back_door_alarm_type"
|
|
|
+ALARM_LEVEL_2 = "sensor.back_door_alarm_level"
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_none_when_no_data(two_lock_coord):
|
|
|
+ sensor = _group_sensor(two_lock_coord)
|
|
|
+ assert sensor.native_value is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_all_locked(hass, two_lock_coord):
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_ENTITY, DOOR_SENSOR_ENTITY, "24", "off")
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_2, DOOR_SENSOR_2, "24", "off")
|
|
|
+ assert _group_sensor(two_lock_coord).native_value == "Locked"
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_any_open_wins_over_locked(hass, two_lock_coord):
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_ENTITY, DOOR_SENSOR_ENTITY, "24", "off") # Locked
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_2, DOOR_SENSOR_2, "25", "on") # Open
|
|
|
+ assert _group_sensor(two_lock_coord).native_value == "Open"
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_jammed_beats_open(hass, two_lock_coord):
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_ENTITY, DOOR_SENSOR_ENTITY, "24", "on") # Jammed
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_2, DOOR_SENSOR_2, "25", "on") # Open
|
|
|
+ assert _group_sensor(two_lock_coord).native_value == "Jammed"
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_jammed_beats_locked(hass, two_lock_coord):
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_ENTITY, DOOR_SENSOR_ENTITY, "24", "on") # Jammed
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_2, DOOR_SENSOR_2, "24", "off") # Locked
|
|
|
+ assert _group_sensor(two_lock_coord).native_value == "Jammed"
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_closed_when_mix_of_locked_and_closed(hass, two_lock_coord):
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_ENTITY, DOOR_SENSOR_ENTITY, "24", "off") # Locked
|
|
|
+ await _set_lock_state(hass, ALARM_TYPE_2, DOOR_SENSOR_2, "25", "off") # Closed
|
|
|
+ assert _group_sensor(two_lock_coord).native_value == "Closed"
|
|
|
+
|
|
|
+
|
|
|
+async def test_group_combined_excludes_lock_without_door_sensor(hass, mock_node):
|
|
|
+ """A lock with no door sensor must not affect group combined state."""
|
|
|
+ entry = MockConfigEntry(
|
|
|
+ domain=DOMAIN,
|
|
|
+ data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]},
|
|
|
+ options={
|
|
|
+ CONF_DOOR_SENSOR_MAP: {LOCK_ENTITY_ID: DOOR_SENSOR_ENTITY}, # only lock 1
|
|
|
+ CONF_GROUPS: [{
|
|
|
+ CONF_GROUP_NAME: "All Locks",
|
|
|
+ CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2],
|
|
|
+ }],
|
|
|
+ },
|
|
|
+ entry_id="test_partial_door",
|
|
|
+ )
|
|
|
+ entry.add_to_hass(hass)
|
|
|
+ with patch(_ZWAVE_HELPER, return_value=mock_node):
|
|
|
+ c = ZLockCoordinator(hass, entry)
|
|
|
+ await c.async_setup()
|
|
|
+
|
|
|
+ # lock 2 has no door sensor; lock 1 is locked+closed
|
|
|
+ hass.states.async_set(ALARM_TYPE_ENTITY, "24")
|
|
|
+ hass.states.async_set(DOOR_SENSOR_ENTITY, "off")
|
|
|
+ hass.states.async_set(ALARM_TYPE_2, "22") # lock 2 unlocked — ignored since no door sensor
|
|
|
+ await hass.async_block_till_done()
|
|
|
+
|
|
|
+ sensor = ZLockGroupCombinedStateSensor(c, entry, "All Locks", [LOCK_ENTITY_ID, LOCK_2])
|
|
|
+ assert sensor.native_value == "Locked"
|
|
|
+ c.async_teardown()
|