"""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, DOOR_SENSOR_ENTITY 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_open_seeded_from_existing_state_at_startup(hass, mock_node): """Sensor state already published before integration loads — no change event fires.""" hass.states.async_set(DOOR_SENSOR_ENTITY, "off") 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() assert c.data[LOCK_ENTITY_ID].door_open is False c.async_teardown() async def test_door_open_seeded_open_at_startup(hass, mock_node): hass.states.async_set(DOOR_SENSOR_ENTITY, "on") 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() assert c.data[LOCK_ENTITY_ID].door_open is True c.async_teardown() async def test_door_unavailable_at_startup_not_seeded(hass, mock_node): hass.states.async_set(DOOR_SENSOR_ENTITY, "unavailable") 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() assert c.data[LOCK_ENTITY_ID].door_open is None c.async_teardown() 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()