zlock_slots/tests/test_door_sensor.py

"""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()