zlock_slots/tests/test_sensor.py

"""Tests for sensor entities: native_value, extra_state_attributes, device attachment."""
import pytest
from unittest.mock import patch

from homeassistant.helpers import device_registry as dr, entity_registry as er
from pytest_homeassistant_custom_component.common import MockConfigEntry

from custom_components.zlock_slots.coordinator import ZLockCoordinator
from custom_components.zlock_slots.sensor import (
  ZLockLastActionSensor,
  ZLockLastUserSensor,
  ZLockSlotsSensor,
  ZLockGroupLastActionSensor,
  ZLockGroupLastUserSensor,
  ZLockGroupSlotsSensor,
)
from custom_components.zlock_slots.const import (
  DOMAIN, CONF_GROUPS, CONF_GROUP_NAME, CONF_LOCK_ENTITY_IDS,
)
from tests.conftest import LOCK_ENTITY_ID, ALARM_TYPE_ENTITY, ALARM_LEVEL_ENTITY, FakeNode, FakeUserData

LOCK_2 = "lock.back_door"

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


# ---------------------------------------------------------------------------
# ZLockLastActionSensor
# ---------------------------------------------------------------------------

async def test_last_action_none_initially(coord, config_entry):
  sensor = ZLockLastActionSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value is None


async def test_last_action_after_event(hass, coord, config_entry):
  hass.states.async_set(ALARM_LEVEL_ENTITY, "1")
  hass.states.async_set(ALARM_TYPE_ENTITY, "18")
  await hass.async_block_till_done()

  sensor = ZLockLastActionSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == "Locked"


# ---------------------------------------------------------------------------
# ZLockLastUserSensor
# ---------------------------------------------------------------------------

async def test_last_user_none_initially(coord, config_entry):
  sensor = ZLockLastUserSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value is None


async def test_last_user_slot_number_when_unnamed(hass, coord, config_entry):
  hass.states.async_set(ALARM_LEVEL_ENTITY, "2")
  hass.states.async_set(ALARM_TYPE_ENTITY, "19")
  await hass.async_block_till_done()

  sensor = ZLockLastUserSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == "2"


async def test_last_user_name_when_named(hass, coord, config_entry):
  await coord.async_set_name(LOCK_ENTITY_ID, 1, "Eve")
  hass.states.async_set(ALARM_LEVEL_ENTITY, "1")
  hass.states.async_set(ALARM_TYPE_ENTITY, "18")
  await hass.async_block_till_done()

  sensor = ZLockLastUserSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == "Eve"


async def test_last_user_manual(hass, coord, config_entry):
  hass.states.async_set(ALARM_TYPE_ENTITY, "21")
  await hass.async_block_till_done()

  sensor = ZLockLastUserSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == "<Manual>"


async def test_last_user_radio(hass, coord, config_entry):
  hass.states.async_set(ALARM_TYPE_ENTITY, "25")
  await hass.async_block_till_done()

  sensor = ZLockLastUserSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == "<Radio>"


# ---------------------------------------------------------------------------
# ZLockSlotsSensor
# ---------------------------------------------------------------------------

async def test_slots_sensor_count(coord, config_entry):
  sensor = ZLockSlotsSensor(coord, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == 2


async def test_slots_sensor_zero_when_empty(hass, config_entry):
  from tests.conftest import FakeNode
  config_entry.add_to_hass(hass)
  with patch(_ZWAVE_HELPER, return_value=FakeNode(users=[])):
    c = ZLockCoordinator(hass, config_entry)
    await c.async_setup()

  sensor = ZLockSlotsSensor(c, config_entry, LOCK_ENTITY_ID)
  assert sensor.native_value == 0
  assert sensor.extra_state_attributes == {"slots": []}
  c.async_teardown()


async def test_slots_sensor_attributes_structure(coord, config_entry):
  sensor = ZLockSlotsSensor(coord, config_entry, LOCK_ENTITY_ID)
  attrs = sensor.extra_state_attributes
  assert "slots" in attrs
  slots = attrs["slots"]
  assert len(slots) == 2
  assert slots[0] == {"slot_num": 1, "status": "enabled", "name": None}
  assert slots[1] == {"slot_num": 2, "status": "disabled", "name": None}


async def test_slots_sensor_attributes_sorted(hass, config_entry):
  from tests.conftest import FakeNode, FakeUserData
  node = FakeNode(users=[FakeUserData(5), FakeUserData(1), FakeUserData(3)])
  config_entry.add_to_hass(hass)
  with patch(_ZWAVE_HELPER, return_value=node):
    c = ZLockCoordinator(hass, config_entry)
    await c.async_setup()

  sensor = ZLockSlotsSensor(c, config_entry, LOCK_ENTITY_ID)
  slot_nums = [s["slot_num"] for s in sensor.extra_state_attributes["slots"]]
  assert slot_nums == [1, 3, 5]
  c.async_teardown()


async def test_slots_sensor_attributes_include_names(hass, config_entry, mock_node):
  config_entry.add_to_hass(hass)
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    c = ZLockCoordinator(hass, config_entry)
    c._names[f"{LOCK_ENTITY_ID}:1"] = "Frank"
    await c.async_setup()

  sensor = ZLockSlotsSensor(c, config_entry, LOCK_ENTITY_ID)
  slot1 = sensor.extra_state_attributes["slots"][0]
  assert slot1["name"] == "Frank"
  c.async_teardown()


# ---------------------------------------------------------------------------
# Group sensors
# ---------------------------------------------------------------------------

@pytest.fixture
def group_config_entry():
  from pytest_homeassistant_custom_component.common import MockConfigEntry
  from custom_components.zlock_slots.const import DOMAIN
  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_group_entry",
  )


async def test_group_last_action_initially_none(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()

  sensor = ZLockGroupLastActionSensor(c, group_config_entry, "All Locks")
  assert sensor.native_value is None
  c.async_teardown()


async def test_group_last_action_after_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()
    hass.states.async_set(ALARM_TYPE_ENTITY, "24")
    await hass.async_block_till_done()

  sensor = ZLockGroupLastActionSensor(c, group_config_entry, "All Locks")
  assert sensor.native_value == "Locked"
  c.async_teardown()


async def test_group_last_user_after_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()
    hass.states.async_set(ALARM_TYPE_ENTITY, "22")
    await hass.async_block_till_done()

  sensor = ZLockGroupLastUserSensor(c, group_config_entry, "All Locks")
  assert sensor.native_value == "<Manual>"
  c.async_teardown()


async def test_group_slots_sensor_naive_union(hass):
  """Slot numbers from every member merge; a slot present on only one member still appears."""
  entry = MockConfigEntry(
    domain=DOMAIN,
    data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]},
    options={
      CONF_GROUPS: [{
        CONF_GROUP_NAME: "All Locks",
        CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2],
      }]
    },
    entry_id="test_group_union_entry",
  )
  entry.add_to_hass(hass)

  nodes = {
    LOCK_ENTITY_ID: FakeNode(users=[FakeUserData(1, active=True)]),
    LOCK_2: FakeNode(users=[FakeUserData(2, active=False)]),
  }

  with patch(_ZWAVE_HELPER, side_effect=lambda hass_, eid: nodes[eid]):
    c = ZLockCoordinator(hass, entry)
    await c.async_setup()

  sensor = ZLockGroupSlotsSensor(c, entry, "All Locks", [LOCK_ENTITY_ID, LOCK_2])
  assert sensor.native_value == 2
  slots = {s["slot_num"]: s["status"] for s in sensor.extra_state_attributes["slots"]}
  assert slots == {1: "enabled", 2: "disabled"}
  c.async_teardown()


# ---------------------------------------------------------------------------
# Device attachment
# ---------------------------------------------------------------------------

def _register_lock_device(hass) -> dr.DeviceEntry:
  """Create a fake zwave_js device with a registered lock.front_door entity."""
  zwave_entry = MockConfigEntry(domain="zwave_js", entry_id="fake_zwave_entry")
  zwave_entry.add_to_hass(hass)
  device = dr.async_get(hass).async_get_or_create(
    config_entry_id=zwave_entry.entry_id,
    identifiers={("zwave_js", "3245-89")},
    name="Front Door",
  )
  er.async_get(hass).async_get_or_create(
    "lock", "zwave_js", "front_door_lock_unique",
    device_id=device.id,
    suggested_object_id="front_door",
  )
  return device


async def _setup_integration(hass, entry, mock_node):
  entry.add_to_hass(hass)
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    assert await hass.config_entries.async_setup(entry.entry_id)
    await hass.async_block_till_done()


async def test_lock_sensors_attach_to_lock_device(hass, config_entry, mock_node):
  device = _register_lock_device(hass)
  await _setup_integration(hass, config_entry, mock_node)

  ent_reg = er.async_get(hass)
  for suffix in ("last_action", "last_user", "slots"):
    entity_id = ent_reg.async_get_entity_id(
      "sensor", DOMAIN, f"{config_entry.entry_id}_{LOCK_ENTITY_ID}_{suffix}"
    )
    assert entity_id is not None
    assert ent_reg.async_get(entity_id).device_id == device.id


async def test_lock_sensors_without_zwave_device(hass, config_entry, mock_node):
  """Lock not in the entity registry — sensors still created, just device-less."""
  await _setup_integration(hass, config_entry, mock_node)

  ent_reg = er.async_get(hass)
  entity_id = ent_reg.async_get_entity_id(
    "sensor", DOMAIN, f"{config_entry.entry_id}_{LOCK_ENTITY_ID}_last_action"
  )
  assert entity_id is not None
  assert ent_reg.async_get(entity_id).device_id is None


async def test_group_sensors_get_service_device(hass, group_config_entry, mock_node):
  await _setup_integration(hass, group_config_entry, mock_node)

  device = dr.async_get(hass).async_get_device(identifiers={(DOMAIN, "group_All Locks")})
  assert device is not None
  assert device.entry_type is dr.DeviceEntryType.SERVICE
  assert device.name == "All Locks"

  ent_reg = er.async_get(hass)
  entity_id = ent_reg.async_get_entity_id(
    "sensor", DOMAIN, f"{group_config_entry.entry_id}_group_All Locks_last_action"
  )
  assert ent_reg.async_get(entity_id).device_id == device.id


async def test_stale_group_device_and_entities_removed_on_reload(
  hass, group_config_entry, mock_node
):
  await _setup_integration(hass, group_config_entry, mock_node)
  dev_reg = dr.async_get(hass)
  assert dev_reg.async_get_device(identifiers={(DOMAIN, "group_All Locks")})

  with patch(_ZWAVE_HELPER, return_value=mock_node):
    # Update listener reloads the entry; cleanup runs during sensor setup.
    hass.config_entries.async_update_entry(group_config_entry, options={CONF_GROUPS: []})
    await hass.async_block_till_done()

  assert dev_reg.async_get_device(identifiers={(DOMAIN, "group_All Locks")}) is None
  ent_reg = er.async_get(hass)
  assert ent_reg.async_get_entity_id(
    "sensor", DOMAIN, f"{group_config_entry.entry_id}_group_All Locks_last_action"
  ) is None