zlock_slots/tests/test_services.py

"""Tests for zlock_slots service handlers."""
import pytest
import voluptuous as vol
from unittest.mock import patch, AsyncMock
from pytest_homeassistant_custom_component.common import MockConfigEntry

from homeassistant.exceptions import ServiceValidationError

from custom_components.zlock_slots.coordinator import ZLockCoordinator
from custom_components.zlock_slots.services import register_services, unregister_services
from custom_components.zlock_slots.const import (
  DOMAIN, CONF_LOCK_ENTITY_IDS, CONF_GROUPS, CONF_GROUP_NAME,
)
from tests.conftest import LOCK_ENTITY_ID, FakeUserData, FakeNode

_ZWAVE_HELPER = "homeassistant.components.zwave_js.helpers.async_get_node_from_entity_id"
LOCK_2 = "lock.back_door"


@pytest.fixture
async def service_env(hass, config_entry, mock_node):
  """Coordinator wired into hass.data with services registered and mock zwave_js services ready."""
  config_entry.add_to_hass(hass)
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    coord = ZLockCoordinator(hass, config_entry)
    await coord.async_setup()
    hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coord

    set_lock_usercode = AsyncMock()
    clear_lock_usercode = AsyncMock()
    hass.services.async_register("zwave_js", "set_lock_usercode", set_lock_usercode)
    hass.services.async_register("zwave_js", "clear_lock_usercode", clear_lock_usercode)

    register_services(hass)
    yield coord, set_lock_usercode, clear_lock_usercode

  unregister_services(hass)
  coord.async_teardown()


@pytest.fixture
async def group_service_env(hass):
  """Two-lock group, each backed by its own fake node, services registered."""
  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_service_entry",
  )
  entry.add_to_hass(hass)

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

  def _node_for(hass_, entity_id):
    return nodes[entity_id]

  with patch(_ZWAVE_HELPER, side_effect=_node_for):
    coord = ZLockCoordinator(hass, entry)
    await coord.async_setup()
    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coord

    set_lock_usercode = AsyncMock()
    clear_lock_usercode = AsyncMock()
    hass.services.async_register("zwave_js", "set_lock_usercode", set_lock_usercode)
    hass.services.async_register("zwave_js", "clear_lock_usercode", clear_lock_usercode)

    register_services(hass)
    with patch(_ZWAVE_HELPER, side_effect=_node_for):
      yield coord, nodes, set_lock_usercode, clear_lock_usercode

  unregister_services(hass)
  coord.async_teardown()


# ---------------------------------------------------------------------------
# set_code
# ---------------------------------------------------------------------------

async def test_set_code_calls_zwave_service(hass, service_env, mock_node):
  coord, set_lock_usercode, _ = service_env
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"lock_entity_id": LOCK_ENTITY_ID, "slot": 5, "code": "1234"},
      blocking=True,
    )

  set_lock_usercode.assert_awaited_once()
  service_data = set_lock_usercode.call_args[0][0].data
  assert service_data["code_slot"] == 5
  assert service_data["usercode"] == "1234"
  assert service_data["entity_id"] == LOCK_ENTITY_ID


async def test_set_code_auto_assigns_lowest_free_slot(hass, service_env, mock_node):
  """Slots 1 and 2 occupied → auto-assign picks 3."""
  coord, set_lock_usercode, _ = service_env
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"lock_entity_id": LOCK_ENTITY_ID, "code": "5678"},
      blocking=True,
    )

  service_data = set_lock_usercode.call_args[0][0].data
  assert service_data["code_slot"] == 3


async def test_set_code_auto_assign_skips_occupied(hass, service_env, mock_node):
  """If slot 3 is also occupied, auto-assign picks 4."""
  node_with_3 = FakeNode(users=[
    FakeUserData(1), FakeUserData(2), FakeUserData(3),
  ])
  coord, set_lock_usercode, _ = service_env
  # Refresh to get node_with_3 slots into coordinator state.
  with patch(_ZWAVE_HELPER, return_value=node_with_3):
    await coord.async_refresh_lock(LOCK_ENTITY_ID)
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"lock_entity_id": LOCK_ENTITY_ID, "code": "9999"},
      blocking=True,
    )

  service_data = set_lock_usercode.call_args[0][0].data
  assert service_data["code_slot"] == 4


async def test_set_code_with_name_saves_name(hass, service_env, mock_node):
  coord, _, _ = service_env
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"lock_entity_id": LOCK_ENTITY_ID, "slot": 5, "code": "4321", "name": "Charlie"},
      blocking=True,
    )

  assert coord.get_name(LOCK_ENTITY_ID, 5) == "Charlie"


async def test_set_code_unknown_lock_raises(hass, service_env):
  with pytest.raises(ServiceValidationError):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"lock_entity_id": "lock.unknown", "slot": 1, "code": "1234"},
      blocking=True,
    )


# ---------------------------------------------------------------------------
# clear_code
# ---------------------------------------------------------------------------

async def test_clear_code_calls_zwave_service(hass, service_env, mock_node):
  _, _, clear_lock_usercode = service_env
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    await hass.services.async_call(
      DOMAIN, "clear_code",
      {"lock_entity_id": LOCK_ENTITY_ID, "slot": 2},
      blocking=True,
    )

  clear_lock_usercode.assert_awaited_once()
  service_data = clear_lock_usercode.call_args[0][0].data
  assert service_data["code_slot"] == 2
  assert service_data["entity_id"] == LOCK_ENTITY_ID


# ---------------------------------------------------------------------------
# set_name
# ---------------------------------------------------------------------------

async def test_set_name_updates_coordinator(hass, service_env):
  coord, _, _ = service_env
  await hass.services.async_call(
    DOMAIN, "set_name",
    {"lock_entity_id": LOCK_ENTITY_ID, "slot": 1, "name": "Diana"},
    blocking=True,
  )

  assert coord.get_name(LOCK_ENTITY_ID, 1) == "Diana"


# ---------------------------------------------------------------------------
# set_slot_enabled
# ---------------------------------------------------------------------------

async def test_set_slot_enabled_calls_node(hass, service_env, mock_node):
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    await hass.services.async_call(
      DOMAIN, "set_slot_enabled",
      {"lock_entity_id": LOCK_ENTITY_ID, "slot": 1, "enabled": False},
      blocking=True,
    )

  mock_node.access_control.set_user.assert_awaited_once()
  slot_arg, opts_arg = mock_node.access_control.set_user.call_args[0]
  assert slot_arg == 1
  assert opts_arg.active is False


async def test_set_slot_enabled_true(hass, service_env, mock_node):
  with patch(_ZWAVE_HELPER, return_value=mock_node):
    await hass.services.async_call(
      DOMAIN, "set_slot_enabled",
      {"lock_entity_id": LOCK_ENTITY_ID, "slot": 2, "enabled": True},
      blocking=True,
    )

  _, opts_arg = mock_node.access_control.set_user.call_args[0]
  assert opts_arg.active is True


# ---------------------------------------------------------------------------
# Group targeting: fan-out and target validation
# ---------------------------------------------------------------------------

async def test_set_code_group_fanout(hass, group_service_env):
  _, _, set_lock_usercode, _ = group_service_env
  await hass.services.async_call(
    DOMAIN, "set_code",
    {"group_name": "All Locks", "slot": 5, "code": "1234"},
    blocking=True,
  )

  assert set_lock_usercode.await_count == 2
  entity_ids = {c[0][0].data["entity_id"] for c in set_lock_usercode.call_args_list}
  assert entity_ids == {LOCK_ENTITY_ID, LOCK_2}


async def test_set_code_group_auto_assign_uses_union_of_members(hass, group_service_env):
  """LOCK_ENTITY_ID has slot 1 occupied, LOCK_2 has slot 2 — union picks 3 for both."""
  _, _, set_lock_usercode, _ = group_service_env
  await hass.services.async_call(
    DOMAIN, "set_code",
    {"group_name": "All Locks", "code": "5678"},
    blocking=True,
  )

  slots_used = {c[0][0].data["code_slot"] for c in set_lock_usercode.call_args_list}
  assert slots_used == {3}


async def test_clear_code_group_fanout(hass, group_service_env):
  _, _, _, clear_lock_usercode = group_service_env
  await hass.services.async_call(
    DOMAIN, "clear_code",
    {"group_name": "All Locks", "slot": 1},
    blocking=True,
  )

  assert clear_lock_usercode.await_count == 2
  entity_ids = {c[0][0].data["entity_id"] for c in clear_lock_usercode.call_args_list}
  assert entity_ids == {LOCK_ENTITY_ID, LOCK_2}


async def test_set_name_group_fanout(hass, group_service_env):
  coord, _, _, _ = group_service_env
  await hass.services.async_call(
    DOMAIN, "set_name",
    {"group_name": "All Locks", "slot": 1, "name": "Everyone"},
    blocking=True,
  )

  assert coord.get_name(LOCK_ENTITY_ID, 1) == "Everyone"
  assert coord.get_name(LOCK_2, 1) == "Everyone"


async def test_set_slot_enabled_group_fanout(hass, group_service_env):
  coord, nodes, _, _ = group_service_env
  await hass.services.async_call(
    DOMAIN, "set_slot_enabled",
    {"group_name": "All Locks", "slot": 1, "enabled": False},
    blocking=True,
  )

  nodes[LOCK_ENTITY_ID].access_control.set_user.assert_awaited_once()
  nodes[LOCK_2].access_control.set_user.assert_awaited_once()


async def test_group_fanout_continues_after_one_member_fails(hass, group_service_env):
  """One lock erroring must not block the operation from reaching the others."""
  coord, nodes, _, _ = group_service_env
  nodes[LOCK_ENTITY_ID].access_control.set_user.side_effect = RuntimeError("boom")

  await hass.services.async_call(
    DOMAIN, "set_slot_enabled",
    {"group_name": "All Locks", "slot": 1, "enabled": False},
    blocking=True,
  )

  nodes[LOCK_ENTITY_ID].access_control.set_user.assert_awaited_once()
  nodes[LOCK_2].access_control.set_user.assert_awaited_once()


async def test_unknown_group_raises(hass, group_service_env):
  with pytest.raises(ServiceValidationError):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"group_name": "Nonexistent", "slot": 1, "code": "1234"},
      blocking=True,
    )


async def test_both_lock_and_group_target_rejected(hass, group_service_env):
  with pytest.raises(vol.Invalid):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"lock_entity_id": LOCK_ENTITY_ID, "group_name": "All Locks", "slot": 1, "code": "1234"},
      blocking=True,
    )


async def test_neither_lock_nor_group_target_rejected(hass, group_service_env):
  with pytest.raises(vol.Invalid):
    await hass.services.async_call(
      DOMAIN, "set_code",
      {"slot": 1, "code": "1234"},
      blocking=True,
    )