zlock_slots/tests/test_coordinator.py

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